From 7be7968e05dc6f099010150435f485582d85781e Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@pivotal.io>
Date: Tue, 29 Nov 2016 11:58:59 -0500
Subject: [PATCH] Use `EmitterProcessor` do not drop polled messages

The `BridgeHandler` is Reactive `Subscriber` with `MAX_VALUE` demand just `onSubscribe()`.
That causes a drain of the upstream `QueueChannel` to the `ReactiveChannel` with `DirectProcessor`.
 The last one just drops messages if there is no subscribers

`EmitterProcessor` doesn't request upstream until real subscriber arrives to it.
Therefore change `BridgeHandler` logic in the `IntegrationFlowDefinition.toReactivePublisher()` to the `EmitterProcessor` to allow late `Subscriber`s and don't lose messages from the `QueueChannel` if there is no downstream `Subscriber`s
---
 .../dsl/IntegrationFlowDefinition.java        | 20 +++++++++----------
 .../reactivestreams/ReactiveStreamsTests.java | 15 ++++++--------
 2 files changed, 15 insertions(+), 20 deletions(-)

diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java
index 1d49312317..7a4c9600c6 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java
@@ -25,8 +25,9 @@ import java.util.concurrent.Executor;
 import java.util.function.Consumer;
 import java.util.function.Function;
 
+import org.reactivestreams.Processor;
 import org.reactivestreams.Publisher;
-
+import org.reactivestreams.Subscriber;
 import org.springframework.aop.framework.Advised;
 import org.springframework.aop.support.AopUtils;
 import org.springframework.beans.factory.BeanCreationException;
@@ -95,6 +96,7 @@ import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
 
+import reactor.core.publisher.EmitterProcessor;
 import reactor.util.function.Tuple2;
 
 /**
@@ -2725,21 +2727,17 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 			publisher = (Publisher<Message<T>>) channelForPublisher;
 		}
 		else {
-			MessageChannel reactiveChannel = new ReactiveChannel();
-			publisher = (Publisher<Message<T>>) reactiveChannel;
-
 			if (channelForPublisher != null) {
-				BridgeHandler bridge = new BridgeHandler();
-				bridge.setOutputChannel(reactiveChannel);
-
-				addComponent(bridge)
-						.addComponent(new ReactiveConsumer(channelForPublisher, bridge))
-						.addComponent(reactiveChannel);
+				Processor<?, ?> processor = EmitterProcessor.create(false);
+				publisher = (Publisher<Message<T>>) processor;
+				Subscriber<Message<?>> subscriber = (Subscriber<Message<?>>) processor;
+				addComponent(new ReactiveConsumer(channelForPublisher, subscriber));
 			}
 			else {
+				MessageChannel reactiveChannel = new ReactiveChannel();
+				publisher = (Publisher<Message<T>>) reactiveChannel;
 				channel(reactiveChannel);
 			}
-
 		}
 
 		get();
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java
index 21a32be707..8f256c92c5 100644
--- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java
@@ -39,7 +39,6 @@ import org.junit.Rule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.reactivestreams.Publisher;
-
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.Lifecycle;
@@ -107,16 +106,16 @@ public class ReactiveStreamsTests {
 	public void testPollableReactiveFlow() throws InterruptedException, TimeoutException, ExecutionException {
 		this.inputChannel.send(new GenericMessage<>("1,2,3,4,5"));
 
+		CountDownLatch warmUpLatch = new CountDownLatch(3);
 		CountDownLatch latch = new CountDownLatch(6);
 
 		Flux.from(this.pollablePublisher)
 				.filter(m -> m.getHeaders().containsKey(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER))
 				.doOnNext(p -> latch.countDown())
+				.doOnNext(p -> warmUpLatch.countDown())
 				.take(6)
 				.subscribe();
 
-		final CountDownLatch asyncSetupLatch = new CountDownLatch(1);
-
 		Future<List<Integer>> future =
 				Executors.newSingleThreadExecutor().submit(() ->
 						Flux.just("11,12,13")
@@ -127,23 +126,21 @@ public class ReactiveStreamsTests {
 								.concatWith(this.pollablePublisher)
 								.map(Message::getPayload)
 								.take(7)
+								.log("org.springframework.integration.flux")
 								.collectList()
-								.block(getDuration(asyncSetupLatch)));
+								.block(Duration.ofSeconds(10)));
 
-		assertTrue(asyncSetupLatch.await(10, TimeUnit.SECONDS));
+		assertTrue(warmUpLatch.await(10, TimeUnit.SECONDS));
 
 		this.inputChannel.send(new GenericMessage<>("6,7,8,9,10"));
 
 		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		List<Integer> integers = future.get(20, TimeUnit.SECONDS);
+
 		assertNotNull(integers);
 		assertEquals(7, integers.size());
 	}
 
-	private Duration getDuration(CountDownLatch asyncSetupLatch) {
-		asyncSetupLatch.countDown();
-		return Duration.ofSeconds(10);
-	}
 
 	@Configuration
 	@EnableIntegration