From 61e77435ab5ef0daa53f9058789f9228797fbdbd Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@pivotal.io>
Date: Mon, 12 Dec 2016 19:54:32 -0500
Subject: [PATCH] Revert to `EmitterProcessor`

https://build.spring.io/browse/INT-MASTER-468/

Looks like demand in the `FluxTake` isn't honored:

```
public void request(long n) {
if (wip != 0) {
			s.request(n);
		}
		else if (WIP.compareAndSet(this, 0, 1)) {
				if (n >= this.n) {
					s.request(Long.MAX_VALUE);
				}
				else {
					s.request(n);
				}
		}
}
````
It isn't clear why we request from upstream `Long.MAX_VALUE`, if we have only strong `n` limit

So, revert to the `secondSubscriberLatch` and don't send the next message until the subscription from the second `Flux`
---
 .../dsl/IntegrationFlowDefinition.java        |  4 +-
 .../reactivestreams/ReactiveStreamsTests.java | 38 ++++++++++---------
 2 files changed, 23 insertions(+), 19 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 0f6d9aa7ab..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
@@ -96,7 +96,7 @@ import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
 
-import reactor.core.publisher.TopicProcessor;
+import reactor.core.publisher.EmitterProcessor;
 import reactor.util.function.Tuple2;
 
 /**
@@ -2728,7 +2728,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 		}
 		else {
 			if (channelForPublisher != null) {
-				Processor<?, ?> processor = TopicProcessor.share(false);
+				Processor<?, ?> processor = EmitterProcessor.create(false);
 				publisher = (Publisher<Message<T>>) processor;
 				Subscriber<Message<?>> subscriber = (Subscriber<Message<?>>) processor;
 				addComponent(new ReactiveConsumer(channelForPublisher, subscriber));
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 76a3cf443a..02187b1899 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
@@ -105,31 +105,35 @@ public class ReactiveStreamsTests {
 	public void testPollableReactiveFlow() throws Exception {
 		this.inputChannel.send(new GenericMessage<>("1,2,3,4,5"));
 
-		CountDownLatch warmUpLatch = new CountDownLatch(3);
 		CountDownLatch latch = new CountDownLatch(6);
 
 		Flux.from(this.pollablePublisher)
+				.take(6)
 				.filter(m -> m.getHeaders().containsKey(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER))
 				.doOnNext(p -> latch.countDown())
-				.doOnNext(p -> warmUpLatch.countDown())
-				.take(6)
 				.subscribe();
 
-		Future<List<Integer>> future =
-				Executors.newSingleThreadExecutor().submit(() ->
-						Flux.just("11,12,13")
-								.map(v -> v.split(","))
-								.flatMapIterable(Arrays::asList)
-								.map(Integer::parseInt)
-								.<Message<Integer>>map(GenericMessage<Integer>::new)
-								.concatWith(this.pollablePublisher)
-								.map(Message::getPayload)
-								.take(7)
-								.log("org.springframework.integration.flux")
-								.collectList()
-								.block(Duration.ofSeconds(10)));
+		CountDownLatch secondSubscriberLatch = new CountDownLatch(1);
 
-		assertTrue(warmUpLatch.await(10, TimeUnit.SECONDS));
+		Future<List<Integer>> future =
+				Executors.newSingleThreadExecutor().submit(() -> {
+					Thread.sleep(100);
+					return Flux.just("11,12,13")
+							.map(v -> v.split(","))
+							.flatMapIterable(Arrays::asList)
+							.map(Integer::parseInt)
+							.<Message<Integer>>map(GenericMessage<Integer>::new)
+							.concatWith(this.pollablePublisher)
+							.take(7)
+							.map(Message::getPayload)
+							.log("org.springframework.integration.flux")
+							.collectList()
+							.doOnSubscribe(s -> secondSubscriberLatch.countDown())
+							.block(Duration.ofSeconds(10));
+				}
+				);
+
+		assertTrue(secondSubscriberLatch.await(10, TimeUnit.SECONDS));
 
 		this.inputChannel.send(new GenericMessage<>("6,7,8,9,10"));