From 6c379d71997769684caf9a9a39cb809c1db02e16 Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@pivotal.io>
Date: Mon, 17 Sep 2018 10:50:29 -0400
Subject: [PATCH] Fix a logic in the IntFlowDef.toReactivePublisher (#2553)

* Fix a logic in the IntFlowDef.toReactivePublisher

* We may consider to start a `Publisher<Message<?>>` just from one
channel.
So, and an implicit `bridge()` to meet and internal `IntegrationFlow`
 logic
* The `MessageChannelReference` and `FixedSubscriberChannelPrototype`
can't be converted to the reactive `Publisher`.
So, an implicit `bridge()` in between them and target `FluxMessageChannel`

NOTE: This maybe considered for back-port, but the workaround is
simple: just add extra `bridge()` after the mentioned channels

* * Allow `log()` before `toReactivePublisher()` and the same time fix
the problem with not resetted `implicitChannel` flag in the
`IntegrationFlowDefinition`
---
 .../dsl/IntegrationFlowDefinition.java        | 11 ++--
 .../reactivestreams/ReactiveStreamsTests.java | 54 +++++++++++++++++++
 2 files changed, 62 insertions(+), 3 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 db4d9c6731..0f67cf6580 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
@@ -210,6 +210,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 */
 	public B channel(MessageChannel messageChannel) {
 		Assert.notNull(messageChannel, "'messageChannel' must not be null");
+		this.implicitChannel = false;
 		if (this.currentMessageChannel != null) {
 			bridge();
 		}
@@ -431,9 +432,9 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 */
 	public B wireTap(WireTapSpec wireTapSpec) {
 		WireTap interceptor = wireTapSpec.get();
-		if (this.currentMessageChannel == null || !(this.currentMessageChannel instanceof ChannelInterceptorAware)) {
-			this.implicitChannel = true;
+		if (!(this.currentMessageChannel instanceof ChannelInterceptorAware)) {
 			channel(new DirectChannel());
+			this.implicitChannel = true;
 		}
 		addComponent(wireTapSpec);
 		((ChannelInterceptorAware) this.currentMessageChannel).addInterceptor(interceptor);
@@ -2941,7 +2942,9 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 			publisher = (Publisher<Message<T>>) channelForPublisher;
 		}
 		else {
-			if (channelForPublisher != null) {
+			if (channelForPublisher != null && this.integrationComponents.size() > 1
+					&& !(channelForPublisher instanceof MessageChannelReference) &&
+					!(channelForPublisher instanceof FixedSubscriberChannelPrototype)) {
 				publisher = MessageChannelReactiveUtils.toPublisher(channelForPublisher);
 			}
 			else {
@@ -2951,6 +2954,8 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 			}
 		}
 
+		this.implicitChannel = false;
+
 		get();
 
 		return new PublisherIntegrationFlow<>(this.integrationComponents, publisher);
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 129c548f41..5077253b00 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
@@ -89,6 +89,18 @@ public class ReactiveStreamsTests {
 	@Autowired
 	private IntegrationFlowContext integrationFlowContext;
 
+	@Autowired
+	private MessageChannel singleChannel;
+
+	@Autowired
+	private Publisher<Message<String>> singleChannelFlow;
+
+	@Autowired
+	private MessageChannel fixedSubscriberChannel;
+
+	@Autowired
+	private Publisher<Message<String>> fixedSubscriberChannelFlow;
+
 	@Test
 	public void testReactiveFlow() throws Exception {
 		List<String> results = new ArrayList<>();
@@ -207,6 +219,30 @@ public class ReactiveStreamsTests {
 		integrationFlowRegistration.destroy();
 	}
 
+	@Test
+	public void singleChannelFlowTest() throws InterruptedException {
+		CountDownLatch latch = new CountDownLatch(1);
+		Flux.from(this.singleChannelFlow)
+				.map(m -> m.getPayload().toUpperCase())
+				.subscribe(p -> {
+					latch.countDown();
+				});
+		this.singleChannel.send(new GenericMessage<>("foo"));
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
+	}
+
+	@Test
+	public void fixedSubscriberChannelFlowTest() throws InterruptedException {
+		CountDownLatch latch = new CountDownLatch(1);
+		Flux.from(this.fixedSubscriberChannelFlow)
+				.map(m -> m.getPayload().toUpperCase())
+				.subscribe(p -> {
+					latch.countDown();
+				});
+		this.fixedSubscriberChannel.send(new GenericMessage<>("bar"));
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
+	}
+
 	@Configuration
 	@EnableIntegration
 	public static class ContextConfiguration {
@@ -221,6 +257,7 @@ public class ReactiveStreamsTests {
 									.autoStartup(false)
 									.id("reactiveStreamsMessageSource"))
 					.split(String.class, p -> p.split(","))
+					.log()
 					.toReactivePublisher();
 		}
 
@@ -231,6 +268,23 @@ public class ReactiveStreamsTests {
 					.split(s -> s.delimiters(","))
 					.<String, Integer>transform(Integer::parseInt)
 					.channel(MessageChannels.queue())
+					.log()
+					.toReactivePublisher();
+		}
+
+		@Bean
+		public Publisher<Message<String>> singleChannelFlow() {
+			return IntegrationFlows
+					.from(MessageChannels.direct("singleChannel"))
+					.log()
+					.toReactivePublisher();
+		}
+
+		@Bean
+		public Publisher<Message<String>> fixedSubscriberChannelFlow() {
+			return IntegrationFlows
+					.from("fixedSubscriberChannel", true)
+					.log()
 					.toReactivePublisher();
 		}