GH-9999: Fix FluxMessageChannel for restarted application

Fixes: #9999
Issue link: #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

Author: artembilan

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

@@ -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.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 @@
 /**
  * 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 @@
  * @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,19 +119,22 @@ public void subscribe(Subscriber<? super Message<?>> subscriber) {
 				.subscribe(subscriber);
 	}
 
-	private void addPublisherToSubscribe(Flux<?> publisher) {
-		AtomicReference<Disposable> disposableReference = new AtomicReference<>();
+	@Override
+	public void start() {
+		this.active = true;
+		this.upstreamSubscriptions = Disposables.composite();
+		this.sourcePublishers.forEach(this::doSubscribeTo);
+	}
 
-		Disposable disposable =
-				publisher
-						.doOnTerminate(() -> disposeUpstreamSubscription(disposableReference))
-						.subscribe();
+	@Override
+	public void stop() {
+		this.active = false;
+		this.upstreamSubscriptions.dispose();
+	}
 
-		if (!disposable.isDisposed()) {
-			if (this.upstreamSubscriptions.add(disposable)) {
-				disposableReference.set(disposable);
-			}
-		}
+	@Override
+	public boolean isRunning() {
+		return this.active;
 	}
 
 	private void disposeUpstreamSubscription(AtomicReference<Disposable> disposableReference) {
@@ -136,8 +147,14 @@ private void disposeUpstreamSubscription(AtomicReference<Disposable> disposableR
 
 	@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)
@@ -152,6 +169,21 @@ public void subscribeTo(Publisher<? extends Message<?>> publisher) {
 		addPublisherToSubscribe(upstreamPublisher);
 	}
 
+	private void addPublisherToSubscribe(Flux<?> publisher) {
+		AtomicReference<Disposable> disposableReference = new AtomicReference<>();
+
+		Disposable disposable =
+				publisher
+						.doOnTerminate(() -> disposeUpstreamSubscription(disposableReference))
+						.subscribe();
+
+		if (!disposable.isDisposed()) {
+			if (this.upstreamSubscriptions.add(disposable)) {
+				disposableReference.set(disposable);
+			}
+		}
+	}
+
 	private void sendReactiveMessage(Message<?> message) {
 		Message<?> messageToSend = message;
 		// We have just restored Reactor context, so no need in a header anymore.
@@ -169,14 +201,15 @@ private void sendReactiveMessage(Message<?> message) {
 			}
 		}
 		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));
 		}
 	}
 
 	@Override
 	public void destroy() {
 		this.active = false;
 		this.upstreamSubscriptions.dispose();
+		this.sourcePublishers.clear();
 		this.sink.emitComplete(Sinks.EmitFailureHandler.busyLooping(Duration.ofSeconds(1)));
 		super.destroy();
 	}

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

@@ -38,6 +38,7 @@
 
 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 @@ void messageProducerIsNotStartedAutomatically() {
 				.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 Publisher<Message<String>> messageProducerFlow() {
 					.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