Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-9215: Honor back-pressure in FluxMessageChannel

Browse Source 
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`**
This commit is contained in:
Artem Bilan
2024-06-06 11:38:45 -04:00
parent 91de12c786
commit e9561b41f1
1 changed files with 3 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java
View File

@@ -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<Message<?>>, ReactiveStreamsSubscribableChannel {
private final Scheduler scheduler = Schedulers.boundedElastic();
private final Sinks.Many<Message<?>> sink = Sinks.many().multicast().onBackpressureBuffer(1, false);
private final Sinks.Many<Boolean> subscribedSignal = Sinks.many().replay().limit(1);
@@ -107,7 +103,8 @@ public class FluxMessageChannel extends AbstractMessageChannel
public void subscribe(Subscriber<? super Message<?>> subscriber) {
this.sink.asFlux()
.doFinally((s) -> this.subscribedSignal.tryEmitNext(this.sink.currentSubscriberCount() > 0))
.share()
.publish(1)
.refCount()
.subscribe(subscriber);
Mono<Boolean> subscribersBarrier =
@@ -148,7 +145,7 @@ public class FluxMessageChannel extends AbstractMessageChannel
Flux<Object> 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();
}