Use EmitterProcessor in the FluxMessageChannel #3104

artembilan · 2019-11-06T20:30:07Z

The EmitterProcessor has a good logic to block upstream producer
when its downstream subscriber cannot keep up with overproducing.

Rework FluxMessageChannel logic to rely on the EmitterProcessor
instead of Flux.create()
Cancel FluxMessageChannel internal subscriptions in the destroy()
Fix ReactiveStreamsTests.testFluxTransform() for the splitter's
delimiter
Ensure in the FluxMessageChannelTests.testFluxMessageChannel
that we can have several concurrent subscribers to the
FluxMessageChannel

artembilan · 2019-11-08T15:58:56Z

@bsideup ,

Would you mind to take a look into this fix, please?

Thanks

bsideup

(here was a comment but I missed an important detail so I am updating it :D)

...tegration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java

artembilan · 2019-11-13T15:06:42Z

I think I found a solution with the delaySubscription(Publisher<U> subscriptionDelay).

Stay tuned! Will come back to you today 😄

artembilan · 2019-11-13T20:09:51Z

So, here you are a fix based on the delaySubscription() and some additional internal EmitterProcessor to handle external subscriptions to this FluxMessageChannel.

artembilan · 2019-11-15T17:06:04Z

...tegration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java

+						Mono.fromSupplier(this.subscribed::get)
+								.filter((subscribers) -> subscribers > 0)
+								.switchIfEmpty(this.subscriptionDelay.next()))
+				.subscribe();


I think the proper way would be to have a subscribeTo() to return Mono<Void> instead to let that caller to subscribe or compose further its reactive flow.
This way we would defer an emission from this source publisher as late as possible.

But I'd like to revise that stuff in a separate PR.

Thanks

artembilan · 2019-11-19T17:19:32Z

@bsideup , gentle ping. 😄

artembilan · 2019-11-19T20:30:54Z

OK! I have minimized the change for this PR as much as possible - 2 files. 😄

Looking forward for your feedback!

artembilan · 2019-11-22T18:12:19Z

OK. I pushed what we have discussed with some workaround for subscription race condition.

Thank you for your feedback, @bsideup, looking forward for more!

artembilan · 2019-12-02T19:07:11Z

@bsideup ,

so, I have rebased to master and decreased a change as much as possible. Again 😄

Would be great to have some feedback before release this Wednesday.

Thanks

Fixes spring-projects#3107 The `MessagingGatewaySupport` has an `errorOnTimeout` option to throw a `MessageTimeoutException` when downstream reply doesn't come back in time for configured reply timeout * Expose an `errorOnTimeout` option as a `TcpInboundGateway` ctor property * Add new factory methods into a `Tcp` factory for Java DSL * Ensure a property works as expected in the `IpIntegrationTests` * Document a new option

The `EmitterProcessor` has a good logic to block upstream producer when its downstream subscriber cannot keep up with overproducing. * Rework `FluxMessageChannel` logic to rely on the `EmitterProcessor` instead of `Flux.create()` * Cancel `FluxMessageChannel` internal subscriptions in the `destroy()` * Fix `ReactiveStreamsTests.testFluxTransform()` for the splitter's delimiter * Ensure in the `FluxMessageChannelTests.testFluxMessageChannel` that we can have several concurrent subscribers to the `FluxMessageChannel`

* Change `subscribers` list into just `AtomicInteger` count marker * fix `DefaultSplitterTests` according a new logic in the `FluxMessageChannel`

…ssageChannel` to wait until this one subscribed. * Use an `EmitterProcessor` to catch subscriptions and pass them as a signal to delayed upstream publishers * Fix `FluxMessageChannelTests.testFluxMessageChannelCleanUp` to verify an actual property instead of removed. * Fix `RSocketOutboundGatewayIntegrationTests` for the proper subscription into a `FluxMessageChannel` before actual interaction with an RSocket gateway. This should help us also to avoid some race conditions in the future

…way" This reverts commit fa6119d.

…bedSignal`. This one is used `delaySubscription` for the upstream publishers * Use a `AtomicBoolean` for subscription state since `doOnSubscribe()` is called before `EmitterProcessor` adds subscribers for its `downstreams` * Use `publishOn(Schedulers.boundedElastic())` for upstream publishers to avoid blocking over there when our `EmitterProcessor` doesn't have enough demand * Refactor reactive tests to have a subscription into the `FluxMessageChannel` earlier than emission happens for it

…ribe()`

instead of `doOnSubscribe` * Check for `this.processor.hasDownstreams()` before emitting such an event

artembilan · 2019-12-03T16:20:57Z

Pushed the fix as we discussed in private chat.

Thanks

bsideup · 2019-12-03T16:35:10Z

...tegration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java

-
-		this.publishers.values().forEach(ConnectableFlux::connect);
+		if (this.processor.hasDownstreams()) {
+			this.subscribedSignal.onNext(true);


probably this.subscribedSignal.onNext(this.processor.hasDownstreams()) would be better

He-he! True. Will fix soon.

…ribedSignal`. This way we are less vulnerable race conditions when subscribers are changed actively

artembilan · 2019-12-03T17:05:49Z

@garyrussell ,

We're done here with @bsideup . So, let us know if you are OK and we merge (or you).

Thanks

artembilan added this to the 5.2.2 milestone Nov 6, 2019

bsideup suggested changes Nov 8, 2019

View reviewed changes

bsideup reviewed Nov 8, 2019

View reviewed changes

...tegration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java Outdated Show resolved Hide resolved

bsideup reviewed Nov 8, 2019

View reviewed changes

...tegration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java Outdated Show resolved Hide resolved

bsideup reviewed Nov 8, 2019

View reviewed changes

...tegration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java Outdated Show resolved Hide resolved

artembilan force-pushed the EmitterProcessor_for_FluxMessageChannel branch from fb52b63 to e066b8c Compare November 8, 2019 18:19

artembilan force-pushed the EmitterProcessor_for_FluxMessageChannel branch from 86bee7f to cefe941 Compare November 13, 2019 20:07

artembilan commented Nov 15, 2019

View reviewed changes

artembilan force-pushed the EmitterProcessor_for_FluxMessageChannel branch 2 times, most recently from 1f1c315 to f267182 Compare November 19, 2019 17:16

artembilan force-pushed the EmitterProcessor_for_FluxMessageChannel branch from 13bd6a0 to 15d2628 Compare November 19, 2019 20:29

artembilan force-pushed the EmitterProcessor_for_FluxMessageChannel branch from 15d2628 to 5d0fa19 Compare November 22, 2019 18:11

artembilan force-pushed the EmitterProcessor_for_FluxMessageChannel branch from 5d0fa19 to b60bc55 Compare December 2, 2019 19:04

artembilan added 8 commits December 3, 2019 11:07

* Use flux.onComplete() instead of iteration over subscribers

fbd2c71

* Change `subscribers` list into just `AtomicInteger` count marker * fix `DefaultSplitterTests` according a new logic in the `FluxMessageChannel`

Revert "spring-projectsGH-3107: Add errorOnTimeout for TcpInboundGate…

66480b6

…way" This reverts commit fa6119d.

* Use Flux.subscribe(Consumer) instead of `doOnNext(Consumer).subsc…

368ad5c

…ribe()`

* Emit subscribedSignal value after .subscribe(subscriber)

9a35815

instead of `doOnSubscribe` * Check for `this.processor.hasDownstreams()` before emitting such an event

artembilan force-pushed the EmitterProcessor_for_FluxMessageChannel branch from f75e027 to 9a35815 Compare December 3, 2019 16:20

bsideup approved these changes Dec 3, 2019

View reviewed changes

bsideup reviewed Dec 3, 2019

View reviewed changes

* Use this.processor.hasDownstreams() as a value to emit for `subsc…

8d88759

…ribedSignal`. This way we are less vulnerable race conditions when subscribers are changed actively

garyrussell merged commit 94e0816 into spring-projects:master Dec 3, 2019

artembilan deleted the EmitterProcessor_for_FluxMessageChannel branch May 15, 2023 20:50

Use EmitterProcessor in the FluxMessageChannel #3104

Use EmitterProcessor in the FluxMessageChannel #3104

Uh oh!

Conversation

artembilan commented Nov 6, 2019

Uh oh!

artembilan commented Nov 8, 2019

Uh oh!

bsideup left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

artembilan commented Nov 13, 2019

Uh oh!

artembilan commented Nov 13, 2019

Uh oh!

artembilan Nov 15, 2019

Choose a reason for hiding this comment

Uh oh!

artembilan commented Nov 19, 2019

Uh oh!

artembilan commented Nov 19, 2019

Uh oh!

artembilan commented Nov 22, 2019

Uh oh!

artembilan commented Dec 2, 2019

Uh oh!

artembilan commented Dec 3, 2019

Uh oh!

bsideup Dec 3, 2019

Choose a reason for hiding this comment

Uh oh!

artembilan Dec 3, 2019

Choose a reason for hiding this comment

Uh oh!

artembilan commented Dec 3, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

bsideup left a comment •

edited

Loading