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

GH-9999: Fix FluxMessageChannel for restarted application

Browse Source 
Fixes: #9999
Issue link: https://github.com/spring-projects/spring-integration/issues/9999

After the application context stop, the `FluxMessageChannel` continues to try to emit messages.
Just because subscriptions to the provided publishers are not cancelled.
More over, we would like to come back to the production from those publishers when we start
application back.

* Fix `FluxMessageChannel` implementing a `Lifecycle` contract.
Gather provided publisher in a local cache to come back to them when we call `start()`,
essentially, initiating a new subscription to those provided publishers

**Auto-cherry-pick to `6.3.x`**

# Conflicts:
#	spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java
This commit is contained in:
Artem Bilan
2025-04-30 17:04:31 -04:00
parent c045873d7f
commit 4a8caef467
2 changed files with 99 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -17,6 +17,8 @@
package org.springframework.integration.channel;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.LockSupport;
@@ -30,6 +32,7 @@ import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.util.context.ContextView;
import org.springframework.context.Lifecycle;
import org.springframework.core.log.LogMessage;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.StaticMessageHeaderAccessor;
@@ -42,6 +45,9 @@ import org.springframework.util.Assert;
/**
* The {@link AbstractMessageChannel} implementation for the
* Reactive Streams {@link Publisher} based on the Project Reactor {@link Flux}.
* <p>
* This class implements {@link Lifecycle} to control subscriptions to publishers
* attached via {@link #subscribeTo(Publisher)}, when this channel is restarted.
*
* @author Artem Bilan
* @author Gary Russell
@@ -50,11 +56,13 @@ import org.springframework.util.Assert;
* @since 5.0
*/
public class FluxMessageChannel extends AbstractMessageChannel
implements Publisher<Message<?>>, ReactiveStreamsSubscribableChannel {
implements Publisher<Message<?>>, ReactiveStreamsSubscribableChannel, Lifecycle {
private final Sinks.Many<Message<?>> sink = Sinks.many().multicast().onBackpressureBuffer(1, false);
private final Disposable.Composite upstreamSubscriptions = Disposables.composite();
private final List<Publisher<? extends Message<?>>> sourcePublishers = new ArrayList<>();
private volatile Disposable.Composite upstreamSubscriptions = Disposables.composite();
private volatile boolean active = true;
@@ -111,6 +119,56 @@ public class FluxMessageChannel extends AbstractMessageChannel
.subscribe(subscriber);
}
@Override
public void start() {
this.active = true;
this.upstreamSubscriptions = Disposables.composite();
this.sourcePublishers.forEach(this::doSubscribeTo);
}
@Override
public void stop() {
this.active = false;
this.upstreamSubscriptions.dispose();
}
@Override
public boolean isRunning() {
return this.active;
}
private void disposeUpstreamSubscription(AtomicReference<Disposable> disposableReference) {
Disposable disposable = disposableReference.get();
if (disposable != null) {
this.upstreamSubscriptions.remove(disposable);
disposable.dispose();
}
}
@Override
public void subscribeTo(Publisher<? extends Message<?>> publisher) {
this.sourcePublishers.add(publisher);
doSubscribeTo(publisher);
}
private void doSubscribeTo(Publisher<? extends Message<?>> publisher) {
Flux<Object> upstreamPublisher =
Flux.from(publisher)
.doOnComplete(() -> this.sourcePublishers.remove(publisher))
.delaySubscription(
Mono.fromCallable(this.sink::currentSubscriberCount)
.filter((value) -> value > 0)
.repeatWhenEmpty((repeat) ->
this.active ? repeat.delayElements(Duration.ofMillis(100)) : repeat))
.flatMap((message) ->
Mono.just(message)
.handle((messageToHandle, syncSink) -> sendReactiveMessage(messageToHandle))
.contextWrite(StaticMessageHeaderAccessor.getReactorContext(message)))
.contextCapture();
addPublisherToSubscribe(upstreamPublisher);
}
private void addPublisherToSubscribe(Flux<?> publisher) {
AtomicReference<Disposable> disposableReference = new AtomicReference<>();
@@ -126,32 +184,6 @@ public class FluxMessageChannel extends AbstractMessageChannel
}
}
private void disposeUpstreamSubscription(AtomicReference<Disposable> disposableReference) {
Disposable disposable = disposableReference.get();
if (disposable != null) {
this.upstreamSubscriptions.remove(disposable);
disposable.dispose();
}
}
@Override
public void subscribeTo(Publisher<? extends Message<?>> publisher) {
Flux<Object> upstreamPublisher =
Flux.from(publisher)
.delaySubscription(
Mono.fromCallable(this.sink::currentSubscriberCount)
.filter((value) -> value > 0)
.repeatWhenEmpty((repeat) ->
this.active ? repeat.delayElements(Duration.ofMillis(100)) : repeat))
.flatMap((message) ->
Mono.just(message)
.handle((messageToHandle, syncSink) -> sendReactiveMessage(messageToHandle))
.contextWrite(StaticMessageHeaderAccessor.getReactorContext(message)))
.contextCapture();
addPublisherToSubscribe(upstreamPublisher);
}
private void sendReactiveMessage(Message<?> message) {
Message<?> messageToSend = message;
// We have just restored Reactor context, so no need in a header anymore.
@@ -169,7 +201,7 @@ public class FluxMessageChannel extends AbstractMessageChannel
}
}
catch (Exception ex) {
logger.warn(ex, LogMessage.format("Error during processing event: %s", messageToSend));
logger.error(ex, LogMessage.format("Error during processing event: %s", messageToSend));
}
}
@@ -177,6 +209,7 @@ public class FluxMessageChannel extends AbstractMessageChannel
public void destroy() {
this.active = false;
this.upstreamSubscriptions.dispose();
this.sourcePublishers.clear();
this.sink.emitComplete(Sinks.EmitFailureHandler.busyLooping(Duration.ofSeconds(1)));
super.destroy();
}

37
spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java
View File

@@ -38,6 +38,7 @@ import reactor.test.StepVerifier;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.Lifecycle;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -266,6 +267,35 @@ public class ReactiveStreamsTests {
.verify(Duration.ofSeconds(10));
}
@Autowired
QueueChannel fromPublisherResult;
@Autowired
ConfigurableApplicationContext applicationContext;
@Test
@DirtiesContext
// Use disruptive this.applicationContext.start()
void verifyFluxMessageChannelRestart() {
for (long i = 0; i < 3L; i++) {
assertThat(this.fromPublisherResult.receive(10_000)).extracting(Message::getPayload).isEqualTo(i);
}
this.applicationContext.stop();
this.fromPublisherResult.purge(null);
this.applicationContext.start();
// The applicationContext restart causes all the endpoint to be started,
// while we really don't have a subscription to this producer
this.testMessageProducer.stop();
for (long i = 0; i < 3L; i++) {
assertThat(this.fromPublisherResult.receive(10_000)).extracting(Message::getPayload).isEqualTo(i);
}
}
@Configuration
@EnableIntegration
public static class ContextConfiguration {
@@ -325,6 +355,13 @@ public class ReactiveStreamsTests {
.toReactivePublisher(true);
}
@Bean
IntegrationFlow fromPublisher() {
return IntegrationFlow.from(Flux.interval(Duration.ofMillis(100)).map(GenericMessage::new))
.channel(c -> c.queue("fromPublisherResult"))
.get();
}
}
private static class TestMessageProducerSpec