Use EmitterProcessor in the FluxMessageChannel (#3104)

* 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`

* * Use `flux.onComplete()` instead of iteration over subscribers
* Change `subscribers` list into just `AtomicInteger` count marker
* fix `DefaultSplitterTests` according a new logic in the `FluxMessageChannel`

* GH-3107: Add errorOnTimeout for TcpInboundGateway

Fixes #3107

The `MessagingGatewaySupport` has an `errorOnTimeout` option to throw
a `MessageTimeoutException` when downstream reply doesn't come back in
time for configured reply timeout

* Expose an `errorOnTimeout` option as a `TcpInboundGateway` ctor
property
* Add new factory methods into a `Tcp` factory for Java DSL
* Ensure a property works as expected in the `IpIntegrationTests`
* Document a new option

* * Use `delaySubscription()` for subscribing publishers in the `FluxMessageChannel`
to wait until this one subscribed.
* Use an `EmitterProcessor` to catch subscriptions and pass them as a
signal to delayed upstream publishers
* Fix  `FluxMessageChannelTests.testFluxMessageChannelCleanUp` to
verify an actual property instead of removed.
* Fix `RSocketOutboundGatewayIntegrationTests` for the proper subscription
into a `FluxMessageChannel` before actual interaction with an RSocket
gateway.
This should help us also to avoid some race conditions in the future

* Revert "GH-3107: Add errorOnTimeout for TcpInboundGateway"

This reverts commit fa6119d.

* * Refactor `FluxMessageChannel` to use `ReplayProcessor` for `subscribedSignal`.
This one is used `delaySubscription` for the upstream publishers
* Use a `AtomicBoolean` for subscription state since `doOnSubscribe()`
is called before `EmitterProcessor` adds subscribers for its `downstreams`
* Use `publishOn(Schedulers.boundedElastic())` for upstream publishers
to avoid blocking over there when our `EmitterProcessor` doesn't have
enough demand
* Refactor reactive tests to have a subscription into the `FluxMessageChannel`
earlier than emission happens for it

* * Use `Flux.subscribe(Consumer)` instead of `doOnNext(Consumer).subscribe()`

* * Emit `subscribedSignal` value after `.subscribe(subscriber)`
instead of `doOnSubscribe`
* Check for `this.processor.hasDownstreams()` before emitting such an event

* * Use `this.processor.hasDownstreams()` as a value to emit for `subscribedSignal`.
This way we are less vulnerable race conditions when subscribers are changed
actively

Author: artembilan

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

@@ -16,82 +16,79 @@
 
 package org.springframework.integration.channel;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 
-import reactor.core.publisher.ConnectableFlux;
+import reactor.core.publisher.EmitterProcessor;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.FluxSink;
+import reactor.core.publisher.ReplayProcessor;
+import reactor.core.scheduler.Schedulers;
 
 /**
  * The {@link AbstractMessageChannel} implementation for the
  * Reactive Streams {@link Publisher} based on the Project Reactor {@link Flux}.
  *
  * @author Artem Bilan
  * @author Gary Russell
+ * @author Sergei Egorov
  *
  * @since 5.0
  */
 public class FluxMessageChannel extends AbstractMessageChannel
 		implements Publisher<Message<?>>, ReactiveStreamsSubscribableChannel {
 
-	private final List<Subscriber<? super Message<?>>> subscribers = new ArrayList<>();
-
-	private final Map<Publisher<? extends Message<?>>, ConnectableFlux<?>> publishers = new ConcurrentHashMap<>();
+	private final EmitterProcessor<Message<?>> processor;
 
-	private final Flux<Message<?>> flux;
+	private final FluxSink<Message<?>> sink;
 
-	private FluxSink<Message<?>> sink;
+	private final ReplayProcessor<Boolean> subscribedSignal = ReplayProcessor.create(1);
 
 	public FluxMessageChannel() {
-		this.flux =
-				Flux.<Message<?>>create(emitter -> this.sink = emitter, FluxSink.OverflowStrategy.IGNORE)
-						.publish()
-						.autoConnect();
+		this.processor = EmitterProcessor.create(1, false);
+		this.sink = this.processor.sink(FluxSink.OverflowStrategy.BUFFER);
 	}
 
 	@Override
 	protected boolean doSend(Message<?> message, long timeout) {
-		Assert.state(this.subscribers.size() > 0,
+		Assert.state(this.processor.hasDownstreams(),
 				() -> "The [" + this + "] doesn't have subscribers to accept messages");
 		this.sink.next(message);
 		return true;
 	}
 
 	@Override
 	public void subscribe(Subscriber<? super Message<?>> subscriber) {
-		this.subscribers.add(subscriber);
-
-		this.flux.doOnCancel(() -> this.subscribers.remove(subscriber))
-				.retry()
+		this.processor
+				.doFinally((s) -> this.subscribedSignal.onNext(this.processor.hasDownstreams()))
 				.subscribe(subscriber);
-
-		this.publishers.values().forEach(ConnectableFlux::connect);
+		this.subscribedSignal.onNext(this.processor.hasDownstreams());
 	}
 
 	@Override
 	public void subscribeTo(Publisher<? extends Message<?>> publisher) {
-		ConnectableFlux<?> connectableFlux =
-				Flux.from(publisher)
-						.handle((message, sink) -> sink.next(send(message)))
-						.onErrorContinue((throwable, event) ->
-								logger.warn("Error during processing event: " + event, throwable))
-						.doOnComplete(() -> this.publishers.remove(publisher))
-						.publish();
-
-		this.publishers.put(publisher, connectableFlux);
+		Flux.from(publisher)
+				.delaySubscription(this.subscribedSignal.filter(Boolean::booleanValue).next())
+				.publishOn(Schedulers.boundedElastic())
+				.doOnNext((message) -> {
+					try {
+						send(message);
+					}
+					catch (Exception e) {
+						logger.warn("Error during processing event: " + message, e);
+					}
+				})
+				.subscribe();
+	}
 
-		if (!this.subscribers.isEmpty()) {
-			connectableFlux.connect();
-		}
+	@Override
+	public void destroy() {
+		this.subscribedSignal.onNext(false);
+		this.processor.onComplete();
+		super.destroy();
 	}
 
 }

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

@@ -20,17 +20,17 @@
 
 import java.util.ArrayList;
 import java.util.List;
-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 +47,26 @@
 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.EmitterProcessor;
 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);
@@ -130,9 +139,9 @@ public void testFluxMessageChannelCleanUp() throws InterruptedException {
 
 		assertThat(finishLatch.await(10, TimeUnit.SECONDS)).isTrue();
 
-		assertThat(TestUtils.getPropertyValue(flux, "publishers", Map.class).isEmpty()).isTrue();
-
 		flowRegistration.destroy();
+
+		assertThat(TestUtils.getPropertyValue(flux, "processor", EmitterProcessor.class).isTerminated()).isTrue();
 	}
 
 	@Configuration
@@ -158,6 +167,7 @@ public String handle(int payload) {
 		}
 
 		@Bean
+		@BridgeFrom("fluxMessageChannel")
 		public MessageChannel queueChannel() {
 			return new QueueChannel();
 		}