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

Revise subscribed state in the FluxMessageChannel

Browse Source 
The `Sinks.Many<Boolean> subscribedSignal` is a bit of overhead in the logic.

* Use `Mono.fromCallable(this.sink::currentSubscriberCount)` instead for `delaySubscription()`
when we perform `subscribeTo(Publisher)`

**Auto-cherry-pick to `6.3.x` & `6.2.x`**
This commit is contained in:
Artem Bilan
2024-06-25 14:47:12 -04:00
parent f2399a4e13
commit bdcb856a90
1 changed files with 5 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -54,8 +54,6 @@ public class FluxMessageChannel extends AbstractMessageChannel
private final Sinks.Many<Message<?>> sink = Sinks.many().multicast().onBackpressureBuffer(1, false); private final Sinks.Many<Message<?>> sink = Sinks.many().multicast().onBackpressureBuffer(1, false);
private final Sinks.Many<Boolean> subscribedSignal = Sinks.many().replay().limit(1);
private final Disposable.Composite upstreamSubscriptions = Disposables.composite(); private final Disposable.Composite upstreamSubscriptions = Disposables.composite();
private volatile boolean active = true; private volatile boolean active = true;
@@ -102,19 +100,9 @@ public class FluxMessageChannel extends AbstractMessageChannel
@Override @Override
public void subscribe(Subscriber<? super Message<?>> subscriber) { public void subscribe(Subscriber<? super Message<?>> subscriber) {
this.sink.asFlux() this.sink.asFlux()
.doFinally((s) -> this.subscribedSignal.tryEmitNext(this.sink.currentSubscriberCount() > 0))
.publish(1) .publish(1)
.refCount() .refCount()
.subscribe(subscriber); .subscribe(subscriber);
Mono<Boolean> subscribersBarrier =
Mono.fromCallable(() -> this.sink.currentSubscriberCount() > 0)
.filter(Boolean::booleanValue)
.doOnNext(this.subscribedSignal::tryEmitNext)
.repeatWhenEmpty((repeat) ->
this.active ? repeat.delayElements(Duration.ofMillis(100)) : repeat); // NOSONAR
addPublisherToSubscribe(Flux.from(subscribersBarrier));
} }
private void addPublisherToSubscribe(Flux<?> publisher) { private void addPublisherToSubscribe(Flux<?> publisher) {
@@ -144,8 +132,11 @@ public class FluxMessageChannel extends AbstractMessageChannel
public void subscribeTo(Publisher<? extends Message<?>> publisher) { public void subscribeTo(Publisher<? extends Message<?>> publisher) {
Flux<Object> upstreamPublisher = Flux<Object> upstreamPublisher =
Flux.from(publisher) Flux.from(publisher)
.delaySubscription(this.subscribedSignal.asFlux().filter(Boolean::booleanValue).next()) .delaySubscription(
// .publishOn(this.scheduler) Mono.fromCallable(this.sink::currentSubscriberCount)
.filter((value) -> value > 0)
.repeatWhenEmpty((repeat) ->
this.active ? repeat.delayElements(Duration.ofMillis(100)) : repeat))
.flatMap((message) -> .flatMap((message) ->
Mono.just(message) Mono.just(message)
.handle((messageToHandle, syncSink) -> sendReactiveMessage(messageToHandle)) .handle((messageToHandle, syncSink) -> sendReactiveMessage(messageToHandle))
@@ -180,7 +171,6 @@ public class FluxMessageChannel extends AbstractMessageChannel
public void destroy() { public void destroy() {
this.active = false; this.active = false;
this.upstreamSubscriptions.dispose(); this.upstreamSubscriptions.dispose();
this.subscribedSignal.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST);
this.sink.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST); this.sink.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST);
super.destroy(); super.destroy();
} }