Use EmitterProcessor in the FluxMessageChannel

The `EmitterProcessor` has a good logic to block upstream producer
when its downstream subscriber cannot keep up with overproducing.

* Rework `FluxMessageChannel` logic to rely on the `EmitterProcessor`
instead of `Flux.create()`
* Cancel `FluxMessageChannel` internal subscriptions in the `destroy()`
* Fix `ReactiveStreamsTests.testFluxTransform()` for the splitter's
delimiter
* Ensure in the `FluxMessageChannelTests.testFluxMessageChannel`
that we can have several concurrent subscribers to the
`FluxMessageChannel`

Author: artembilan

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

@@ -27,7 +27,9 @@
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 
+import reactor.core.Disposable;
 import reactor.core.publisher.ConnectableFlux;
+import reactor.core.publisher.EmitterProcessor;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.FluxSink;
 
@@ -47,15 +49,15 @@ public class FluxMessageChannel extends AbstractMessageChannel
 
 	private final Map<Publisher<? extends Message<?>>, ConnectableFlux<?>> publishers = new ConcurrentHashMap<>();
 
-	private final Flux<Message<?>> flux;
+	private final Map<ConnectableFlux<?>, Disposable> disposables = new ConcurrentHashMap<>();
+
+	private final EmitterProcessor<Message<?>> flux;
 
 	private FluxSink<Message<?>> sink;
 
 	public FluxMessageChannel() {
-		this.flux =
-				Flux.<Message<?>>create(emitter -> this.sink = emitter, FluxSink.OverflowStrategy.IGNORE)
-						.publish()
-						.autoConnect();
+		this.flux = EmitterProcessor.create(1, false);
+		this.sink = this.flux.sink();
 	}
 
 	@Override
@@ -70,11 +72,12 @@ protected boolean doSend(Message<?> message, long timeout) {
 	public void subscribe(Subscriber<? super Message<?>> subscriber) {
 		this.subscribers.add(subscriber);
 
-		this.flux.doOnCancel(() -> this.subscribers.remove(subscriber))
+		this.flux.doFinally((signal) -> this.subscribers.remove(subscriber))
 				.retry()
 				.subscribe(subscriber);
 
-		this.publishers.values().forEach(ConnectableFlux::connect);
+		this.publishers.values()
+				.forEach((connectableFlux) -> this.disposables.put(connectableFlux, connectableFlux.connect()));
 	}
 
 	@Override
@@ -84,14 +87,22 @@ public void subscribeTo(Publisher<? extends Message<?>> publisher) {
 						.handle((message, sink) -> sink.next(send(message)))
 						.onErrorContinue((throwable, event) ->
 								logger.warn("Error during processing event: " + event, throwable))
-						.doOnComplete(() -> this.publishers.remove(publisher))
+						.doFinally((signal) -> this.disposables.remove(this.publishers.remove(publisher)))
 						.publish();
 
 		this.publishers.put(publisher, connectableFlux);
 
 		if (!this.subscribers.isEmpty()) {
-			connectableFlux.connect();
+			connectableFlux.connect((disposable) -> this.disposables.put(connectableFlux, disposable));
 		}
 	}
 
+	@Override
+	public void destroy() {
+		super.destroy();
+
+		this.subscribers.forEach(Subscriber::onComplete);
+		this.disposables.values().forEach(Disposable::dispose);
+	}
+
 }

spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/FluxMessageChannelTests.java

@@ -23,14 +23,15 @@
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.annotation.BridgeFrom;
 import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.channel.FluxMessageChannel;
 import org.springframework.integration.channel.MessageChannelReactiveUtils;
@@ -47,24 +48,25 @@
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
+import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
 
 /**
  * @author Artem Bilan
  *
  * @since 5.0
  */
-@RunWith(SpringRunner.class)
+@SpringJUnitConfig
 @DirtiesContext
 public class FluxMessageChannelTests {
 
 	@Autowired
 	private MessageChannel fluxMessageChannel;
 
 	@Autowired
-	private MessageChannel queueChannel;
+	private QueueChannel queueChannel;
 
 	@Autowired
 	private PollableChannel errorChannel;
@@ -73,7 +75,7 @@ public class FluxMessageChannelTests {
 	private IntegrationFlowContext integrationFlowContext;
 
 	@Test
-	public void testFluxMessageChannel() {
+	void testFluxMessageChannel() {
 		QueueChannel replyChannel = new QueueChannel();
 
 		for (int i = 0; i < 10; i++) {
@@ -90,28 +92,35 @@ public void testFluxMessageChannel() {
 		Message<?> error = this.errorChannel.receive(0);
 		assertThat(error).isNotNull();
 		assertThat(((MessagingException) error.getPayload()).getFailedMessage().getPayload()).isEqualTo(5);
+
+		List<Message<?>> messages = this.queueChannel.clear();
+		assertThat(messages).extracting((message) -> (Integer) message.getPayload())
+				.containsAll(IntStream.range(0, 10).boxed().collect(Collectors.toList()));
 	}
 
 	@Test
-	public void testMessageChannelReactiveAdaptation() throws InterruptedException {
+	void testMessageChannelReactiveAdaptation() throws InterruptedException {
 		CountDownLatch done = new CountDownLatch(2);
 		List<String> results = new ArrayList<>();
 
-		Flux.from(MessageChannelReactiveUtils.<String>toPublisher(this.queueChannel))
-				.map(Message::getPayload)
-				.map(String::toUpperCase)
-				.doOnNext(results::add)
-				.subscribe(v -> done.countDown());
+		Disposable disposable =
+				Flux.from(MessageChannelReactiveUtils.<String>toPublisher(this.queueChannel))
+						.map(Message::getPayload)
+						.map(String::toUpperCase)
+						.doOnNext(results::add)
+						.subscribe(v -> done.countDown());
 
 		this.queueChannel.send(new GenericMessage<>("foo"));
 		this.queueChannel.send(new GenericMessage<>("bar"));
 
 		assertThat(done.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(results).containsExactly("FOO", "BAR");
+
+		disposable.dispose();
 	}
 
 	@Test
-	public void testFluxMessageChannelCleanUp() throws InterruptedException {
+	void testFluxMessageChannelCleanUp() throws InterruptedException {
 		FluxMessageChannel flux = MessageChannels.flux().get();
 
 		CountDownLatch finishLatch = new CountDownLatch(1);
@@ -158,6 +167,7 @@ public String handle(int payload) {
 		}
 
 		@Bean
+		@BridgeFrom("fluxMessageChannel")
 		public MessageChannel queueChannel() {
 			return new QueueChannel();
 		}

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

@@ -31,8 +31,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.Test;
 import org.reactivestreams.Publisher;
 
 import org.springframework.beans.factory.annotation.Autowired;
@@ -51,7 +50,7 @@
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 import reactor.core.publisher.Flux;
 
@@ -62,7 +61,7 @@
  *
  * @since 5.0
  */
-@RunWith(SpringRunner.class)
+@SpringJUnitConfig
 @DirtiesContext
 public class ReactiveStreamsTests {
 
@@ -98,7 +97,7 @@ public class ReactiveStreamsTests {
 	private Publisher<Message<String>> fixedSubscriberChannelFlow;
 
 	@Test
-	public void testReactiveFlow() throws Exception {
+	void testReactiveFlow() throws Exception {
 		List<String> results = new ArrayList<>();
 		CountDownLatch latch = new CountDownLatch(6);
 		Flux.from(this.publisher)
@@ -115,7 +114,7 @@ public void testReactiveFlow() throws Exception {
 	}
 
 	@Test
-	public void testPollableReactiveFlow() throws Exception {
+	void testPollableReactiveFlow() throws Exception {
 		this.inputChannel.send(new GenericMessage<>("1,2,3,4,5"));
 
 		CountDownLatch latch = new CountDownLatch(6);
@@ -152,7 +151,7 @@ public void testPollableReactiveFlow() throws Exception {
 	}
 
 	@Test
-	public void testFromPublisher() {
+	void testFromPublisher() {
 		Flux<Message<?>> messageFlux = Flux.just("1,2,3,4")
 				.map(v -> v.split(","))
 				.flatMapIterable(Arrays::asList)
@@ -178,11 +177,11 @@ public void testFromPublisher() {
 	}
 
 	@Test
-	public void testFluxTransform() {
+	void testFluxTransform() {
 		QueueChannel resultChannel = new QueueChannel();
 
 		IntegrationFlow integrationFlow = f -> f
-				.split()
+				.split((splitter) -> splitter.delimiters(","))
 				.<String, String>fluxTransform(flux -> flux
 						.map(Message::getPayload)
 						.map(String::toUpperCase))
@@ -212,7 +211,7 @@ public void testFluxTransform() {
 	}
 
 	@Test
-	public void singleChannelFlowTest() throws InterruptedException {
+	void singleChannelFlowTest() throws InterruptedException {
 		CountDownLatch latch = new CountDownLatch(1);
 		Flux.from(this.singleChannelFlow)
 				.map(m -> m.getPayload().toUpperCase())
@@ -224,7 +223,7 @@ public void singleChannelFlowTest() throws InterruptedException {
 	}
 
 	@Test
-	public void fixedSubscriberChannelFlowTest() throws InterruptedException {
+	void fixedSubscriberChannelFlowTest() throws InterruptedException {
 		CountDownLatch latch = new CountDownLatch(1);
 		Flux.from(this.fixedSubscriberChannelFlow)
 				.map(m -> m.getPayload().toUpperCase())