Pausable channels #2667
Pausable channels #2667
Conversation
ozangunalp
force-pushed
the
pausable_channels
branch
from
899f83f
to
e8758a7
Compare
| import io.smallrye.mutiny.subscription.MultiSubscriber; | ||
| import io.smallrye.reactive.messaging.PausableChannel; | ||
|
|
||
| public class PausableMulti<T> extends MultiOperator<T, T> implements PausableChannel { |
There was a problem hiding this comment.
Would like to have @jponge review on this one.
There was a problem hiding this comment.
Something that could be done is passing that operator through the Reactive Streams (Flow) TCK. I don't think there's anything incorrect in that implementation, but you know... 🤣
...g-provider/src/main/java/io/smallrye/reactive/messaging/providers/helpers/PausableMulti.java
Show resolved
Hide resolved
| } | ||
|
|
||
| @Override | ||
| public void request(long numberOfItems) { |
There was a problem hiding this comment.
What happens when Long.MAX value is requested? (unbounded subscription)
My understanding is that the pausing mechanism delays upstream requests when paused (aka a kind of passive pausing).
Also this should be documented as per reactive streams expectations.
There was a problem hiding this comment.
It is much like demand capping. pause/resume won't work if the subscription is unbounded.
How would you define active pausing? To buffer items until resumed?
There was a problem hiding this comment.
Yes, but TBH in this case what you have totally makes sense. Just make sure you document the case of unbounded requests, or even "large" requests 😉 (e.g., Long.MAX_VALUE - 100)
There was a problem hiding this comment.
Also, we could've contributed this operator to Mutiny itself, but I couldn't find a Mutiny-like API to give the control of pause/resume. Would you've any ideas on that ?
There was a problem hiding this comment.
We'd need an extended subscription to support these additional operations I guess.
| import io.smallrye.mutiny.subscription.MultiSubscriber; | ||
| import io.smallrye.reactive.messaging.PausableChannel; | ||
|
|
||
| public class PausableMulti<T> extends MultiOperator<T, T> implements PausableChannel { |
There was a problem hiding this comment.
Something that could be done is passing that operator through the Reactive Streams (Flow) TCK. I don't think there's anything incorrect in that implementation, but you know... 🤣
ozangunalp
force-pushed
the
pausable_channels
branch
from
e8758a7
to
02293ec
Compare
|
Pushed a small change to pass the flow publisher tck. |
ozangunalp
force-pushed
the
pausable_channels
branch
from
02293ec
to
3be8cab
Compare
| ``` | ||
|
|
||
| !!!warning | ||
| Pausable channels only work with back-pressure aware subscribers, with bounded downstream requests. |
Adds a new decorator for pausing and accumulating downstream requests to connector channels.
A channel needs to set the
pausableflag to true. The pause/resume is controlled by thePausableChannelthat is accessible fromChannelRegistryby channel name.Closes #1649