Use EmitterProcessor for Channels adaptation
#3100
Conversation
Related spring-cloud/spring-cloud-stream#1835 To honor a back-pressure after `MessageChannel` adaptation it is better to use an `EmitterProcessor.create(1)` instead of `Flux.create()`. This way whenever an emitter buffer is full, we block upstream producer and don't allow it to produce more messages **Cherry-pick to 5.1.x**
| EmitterProcessor<Message<T>> publisher = EmitterProcessor.create(1); | ||
| @SuppressWarnings("unchecked") | ||
| MessageHandler messageHandler = (message) -> publisher.onNext((Message<T>) message); | ||
| return publisher | ||
| .doOnSubscribe((sub) -> inputChannel.subscribe(messageHandler)) | ||
| .doOnCancel(() -> inputChannel.unsubscribe(messageHandler)); |
There was a problem hiding this comment.
Consider using Flux.defer, so that you return a cold Publisher that can be subscribed multiple times
| @@ -60,37 +61,18 @@ else if (messageChannel instanceof PollableChannel) { | |||
| } | |||
|
|
|||
| private static <T> Publisher<Message<T>> adaptSubscribableChannelToPublisher(SubscribableChannel inputChannel) { | |||
| return new SubscribableChannelPublisherAdapter<>(inputChannel); | |||
| EmitterProcessor<Message<T>> publisher = EmitterProcessor.create(1); | |||
There was a problem hiding this comment.
hint: for performance reasons, you may want to allocate a bigger buffer (e.g. 32, or 256, or even use the default one from EmitterProcessor.create().
There was a problem hiding this comment.
Yeah... We had such a discussion with @olegz and our position is to not lose messages in case of crash.
When messages sit upstream blocked, there is more guarantee that they survive restart.
When we buffer them on the emitter level, it looks there is a possibility that we lose them.
No? What am I missing?
There was a problem hiding this comment.
prefetch will keep 1 message in memory anyways (mmm... I think)
Let me test it...
There was a problem hiding this comment.
Ok, so, the thing is:
A processor is okay, but the downstream most probably will prefetch (e.g. flatMap or concatMap), so it will request more than the user is ready to process.
Even if they set prefetch to 1 (the lowest possible value), it still means that there will be 1 element dangling in memory and it will be lost on crash
An acknowledgment mechanism may help here
There was a problem hiding this comment.
That is true and we also discussed it with Artem and we'll need to deal with it at some point, but in the science of sensible defaults it is my strong belief there are only two lowest and highest ;), so we chose lowest
There was a problem hiding this comment.
I think we will take that risk 😄
At least we have only one message dangling around.
We have many ways to ack messages.
Different story though...
Thank so far.
I have failing test in the ReactiveStreamsConsumerTests. It looks like I can't subscribe a second time, even if I use Flux.defer().
Dare to Zoom? 😉
* Fix `ReactiveStreamsConsumerTests` to use a new `Subscription` after each `stop()/start()` on the `ReactiveStreamsConsumer`
|
OK. Looks like Travis is green. Thanks |
|
@garyrussell , are you going to back-port this into |
|
Sorry; I forgot; please do so if you think there will be many. |
* Use `EmitterProcessor` for Channels adaptation Related spring-cloud/spring-cloud-stream#1835 To honor a back-pressure after `MessageChannel` adaptation it is better to use an `EmitterProcessor.create(1)` instead of `Flux.create()`. This way whenever an emitter buffer is full, we block upstream producer and don't allow it to produce more messages **Cherry-pick to 5.1.x** * * Wrap every new subscription into a `Flux.defer()` * Fix `ReactiveStreamsConsumerTests` to use a new `Subscription` after each `stop()/start()` on the `ReactiveStreamsConsumer` * * Remove unused imports # Conflicts: # spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java # spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTests.java # spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java * Fixing conflicts in tests
|
Cherry-picked to |
Related spring-cloud/spring-cloud-stream#1835
To honor a back-pressure after
MessageChanneladaptation it is betterto use an
EmitterProcessor.create(1)instead ofFlux.create().This way whenever an emitter buffer is full, we block upstream producer
and don't allow it to produce more messages
Cherry-pick to 5.1.x