From e9561b41f170e045f1dddc9d7caa5944fbf484e5 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 6 Jun 2024 11:38:45 -0400 Subject: [PATCH] GH-9215: Honor back-pressure in `FluxMessageChannel` Fixes: #9215 * Instead of `share()` use `publish(1).refCount()` to prefetch only item from upstream. * Also remove `publishOn(this.scheduler)` for upstream publishers in favor of opt-in on the consumer side. **Auto-cherry-pick to `6.3.x` & `6.2.x`** --- .../integration/channel/FluxMessageChannel.java | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java index 01de10f5b2..d071ebc9d5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java @@ -28,8 +28,6 @@ import reactor.core.Disposables; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.Sinks; -import reactor.core.scheduler.Scheduler; -import reactor.core.scheduler.Schedulers; import reactor.util.context.ContextView; import org.springframework.core.log.LogMessage; @@ -54,8 +52,6 @@ import org.springframework.util.Assert; public class FluxMessageChannel extends AbstractMessageChannel implements Publisher>, ReactiveStreamsSubscribableChannel { - private final Scheduler scheduler = Schedulers.boundedElastic(); - private final Sinks.Many> sink = Sinks.many().multicast().onBackpressureBuffer(1, false); private final Sinks.Many subscribedSignal = Sinks.many().replay().limit(1); @@ -107,7 +103,8 @@ public class FluxMessageChannel extends AbstractMessageChannel public void subscribe(Subscriber> subscriber) { this.sink.asFlux() .doFinally((s) -> this.subscribedSignal.tryEmitNext(this.sink.currentSubscriberCount() > 0)) - .share() + .publish(1) + .refCount() .subscribe(subscriber); Mono subscribersBarrier = @@ -148,7 +145,7 @@ public class FluxMessageChannel extends AbstractMessageChannel Flux upstreamPublisher = Flux.from(publisher) .delaySubscription(this.subscribedSignal.asFlux().filter(Boolean::booleanValue).next()) - .publishOn(this.scheduler) +// .publishOn(this.scheduler) .flatMap((message) -> Mono.just(message) .handle((messageToHandle, syncSink) -> sendReactiveMessage(messageToHandle)) @@ -185,7 +182,6 @@ public class FluxMessageChannel extends AbstractMessageChannel this.upstreamSubscriptions.dispose(); this.subscribedSignal.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST); this.sink.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST); - this.scheduler.dispose(); super.destroy(); }