From 6ac0c7316a668eddedda3da3d762f4955290808c Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@pivotal.io>
Date: Fri, 20 Jan 2017 10:40:23 -0500
Subject: [PATCH] Add `IntegrationFlowDefinition.implicitChannel` state to be
 sure that we don't add `nullChannel` if we wire-tap the real channel.
 Otherwise we end up with the case to break the flow which expect message from
 the explicit channel. Add JavDocs to `.wireTap()` to explain behaviour in the
 end of flow

---
 .../dsl/IntegrationFlowDefinition.java            | 15 +++++++++++----
 .../dsl/flowservices/FlowServiceTests.java        |  3 ++-
 2 files changed, 13 insertions(+), 5 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 78bf9e672a..c8a2e5239a 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
@@ -127,6 +127,8 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 
 	private StandardIntegrationFlow integrationFlow;
 
+	private boolean implicitChannel;
+
 	IntegrationFlowDefinition() {
 	}
 
@@ -396,7 +398,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 * Populate the {@code Wire Tap} EI Pattern specific
 	 * {@link org.springframework.messaging.support.ChannelInterceptor} implementation
 	 * to the current {@link #currentMessageChannel}.
-	 * It is useful when an implicit {@link MessageChannel} is used between endpoints:
+	 * <p> It is useful when an implicit {@link MessageChannel} is used between endpoints:
 	 * <pre class="code">
 	 * {@code
 	 *  .transform("payload")
@@ -407,11 +409,14 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 * This method can be used after any {@link #channel} for explicit {@link MessageChannel},
 	 * but with the caution do not impact existing {@link org.springframework.messaging.support.ChannelInterceptor}s.
 	 * @param wireTapSpec the {@link WireTapSpec} to use.
+	 * <p> When this EIP-method is used in the end of flow, it appends {@code nullChannel} to terminate flow properly,
+	 * Otherwise {@code Dispatcher has no subscribers} exception is thrown for implicit {@link DirectChannel}.
 	 * @return the current {@link IntegrationFlowDefinition}.
 	 */
 	public B wireTap(WireTapSpec wireTapSpec) {
 		WireTap interceptor = wireTapSpec.get();
 		if (this.currentMessageChannel == null || !(this.currentMessageChannel instanceof ChannelInterceptorAware)) {
+			this.implicitChannel = true;
 			channel(new DirectChannel());
 		}
 		addComponent(wireTapSpec);
@@ -2889,9 +2894,11 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 				}
 			}
 
-			Optional<Object> lastComponent = this.integrationComponents.stream().reduce((first, second) -> second);
-			if (lastComponent.get() instanceof WireTapSpec) {
-//				channel(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME);
+			if (this.implicitChannel) {
+				Optional<Object> lastComponent = this.integrationComponents.stream().reduce((first, second) -> second);
+				if (lastComponent.get() instanceof WireTapSpec) {
+					channel(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME);
+				}
 			}
 
 			this.integrationFlow = new StandardIntegrationFlow(this.integrationComponents);
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java
index 3e5fb96776..7c46c27c2f 100644
--- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java
@@ -167,7 +167,8 @@ public class FlowServiceTests {
 					.enrichHeaders(Collections.singletonMap("foo", "FOO"))
 					.filter(this)
 					.handle(this)
-					.channel(MessageChannels.queue("myFlowAdapterOutput"));
+					.channel(MessageChannels.queue("myFlowAdapterOutput"))
+					.log();
 		}
 
 		public String messageSource() {