Use TopicProcessor for PublisherIntegrationFlow

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

Even if `EmitterProcessor` looks good for us not requesting until real `subscriber()`, it doesn't keep backlog for late subscribers like it happens some times in our subscriber in the separate Thread.

The solution should be considered as tentative because the real one must be based on the fact that `AMPH` has to perform `subscription.request(n)` as a fact of the subscription from downstream.
In other words mid-flow components should work as passive subscribers and be based on downstream demand

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.EmitterProcessor;
+import reactor.core.publisher.TopicProcessor;
 import reactor.util.function.Tuple2;
 
 /**
@@ -2728,7 +2728,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 		}
 		else {
 			if (channelForPublisher != null) {
-				Processor<?, ?> processor = EmitterProcessor.create(false);
+				Processor<?, ?> processor = TopicProcessor.share(false);
 				publisher = (Publisher<Message<T>>) processor;
 				Subscriber<Message<?>> subscriber = (Subscriber<Message<?>>) processor;
 				addComponent(new ReactiveConsumer(channelForPublisher, subscriber));

spring-integration-core/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java

@@ -55,6 +55,7 @@ import org.springframework.messaging.support.GenericMessage;
 public class ChannelAdapterParserTests {
 
 	private AbstractApplicationContext applicationContext;
+
 	private AbstractApplicationContext applicationContextInner;
 
 
@@ -69,6 +70,7 @@ public class ChannelAdapterParserTests {
 	@After
 	public void tearDown() {
 		this.applicationContext.close();
+		this.applicationContextInner.close();
 	}
 
 
@@ -307,17 +309,21 @@ public class ChannelAdapterParserTests {
 		assertEquals("test", message.getPayload());
 
 		MessageSource<?> testMessageSource = this.applicationContext.getBean("testMessageSource", MessageSource.class);
-		SourcePollingChannelAdapter adapterWithMessageSourceRef = this.applicationContext.getBean("adapterWithMessageSourceRef", SourcePollingChannelAdapter.class);
+		SourcePollingChannelAdapter adapterWithMessageSourceRef =
+				this.applicationContext.getBean("adapterWithMessageSourceRef", SourcePollingChannelAdapter.class);
 		MessageSource<?> source = TestUtils.getPropertyValue(adapterWithMessageSourceRef, "source", MessageSource.class);
 		assertSame(testMessageSource, source);
 	}
 
 	public static class SampleBean {
+
 		private final String message = "hello";
 
 		String getMessage() {
 			return message;
 		}
+
 	}
+
 }
 

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

@@ -27,11 +27,9 @@ import java.util.Arrays;
 import java.util.Date;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.log4j.Level;
@@ -79,7 +77,7 @@ public class ReactiveStreamsTests {
 	private Publisher<Message<Integer>> pollablePublisher;
 
 	@Autowired
-	@Qualifier("reactiveSteamsMessageSource")
+	@Qualifier("reactiveStreamsMessageSource")
 	private Lifecycle messageSource;
 
 	@Autowired
@@ -87,7 +85,7 @@ public class ReactiveStreamsTests {
 	private MessageChannel inputChannel;
 
 	@Test
-	public void testReactiveFlow() throws InterruptedException {
+	public void testReactiveFlow() throws Exception {
 		List<String> results = new ArrayList<>();
 		CountDownLatch latch = new CountDownLatch(6);
 		Flux.from(this.publisher)
@@ -100,10 +98,11 @@ public class ReactiveStreamsTests {
 		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		String[] strings = results.toArray(new String[results.size()]);
 		assertArrayEquals(new String[] { "A", "B", "C", "D", "E", "F" }, strings);
+		this.messageSource.stop();
 	}
 
 	@Test
-	public void testPollableReactiveFlow() throws InterruptedException, TimeoutException, ExecutionException {
+	public void testPollableReactiveFlow() throws Exception {
 		this.inputChannel.send(new GenericMessage<>("1,2,3,4,5"));
 
 		CountDownLatch warmUpLatch = new CountDownLatch(3);
@@ -154,7 +153,7 @@ public class ReactiveStreamsTests {
 					.from(() -> new GenericMessage<>("a,b,c,d,e,f"),
 							e -> e.poller(p -> p.trigger(ctx -> this.invoked.getAndSet(true) ? null : new Date()))
 									.autoStartup(false)
-									.id("reactiveSteamsMessageSource"))
+									.id("reactiveStreamsMessageSource"))
 					.split(String.class, p -> p.split(","))
 					.toReactivePublisher();
 		}