From 794c75f5364b51d7ec89335b08bfaca0f6d4d139 Mon Sep 17 00:00:00 2001
From: Oleg Zhurakousky <ozhurakousky@pivotal.io>
Date: Fri, 18 Jan 2019 15:05:00 +0100
Subject: [PATCH] GH-1581 Fixed multi-IntegrationFlow conflicts

Removed dependency on IntegrationFlow from AbstractMessageChannelBinder as it was unnecessary
Fixed and re-enabled ignored tests

Resolves #1581
---
 .../binder/AbstractMessageChannelBinder.java  | 26 ++++++++++++-------
 .../GreenfieldFunctionEnableBindingTests.java |  4 ++-
 .../ProcessorToFunctionsSupportTests.java     |  9 +++----
 3 files changed, 24 insertions(+), 15 deletions(-)

diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
index e047d9dce..7f0bce482 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 the original author or authors.
+ * Copyright 2016-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -51,7 +51,6 @@ import org.springframework.integration.channel.PublishSubscribeChannel;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.core.MessageSource;
-import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlowBuilder;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.handler.AbstractMessageHandler;
@@ -120,8 +119,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 	@Autowired(required = false)
 	private StreamFunctionProperties streamFunctionProperties;
 
-	@Autowired(required = false)
-	private IntegrationFlow integrationFlow;
+	private boolean producerBindingExist;
 
 	public AbstractMessageChannelBinder(String[] headersToEmbed, PP provisioningProvider) {
 		this(headersToEmbed, provisioningProvider, null);
@@ -197,7 +195,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		}
 		this.postProcessOutputChannel(outputChannel, producerProperties);
 
-		if (this.streamFunctionProperties != null && StringUtils.hasText(this.streamFunctionProperties.getDefinition()) && isInheritedIntegrationFlow()) {
+		if (shouldWireDunctionToChannel(true)) {
 			outputChannel = this.postProcessOutboundChannelForFunction(outputChannel, producerProperties);
 		}
 
@@ -232,6 +230,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		};
 
 		doPublishEvent(new BindingCreatedEvent(binding));
+		this.producerBindingExist = true;
 		return binding;
 	}
 
@@ -346,7 +345,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		try {
 			ConsumerDestination destination = this.provisioningProvider.provisionConsumerDestination(name, group, properties);
 			// the function support for the inbound channel is only for Sink
-			if (this.streamFunctionProperties != null && StringUtils.hasText(this.streamFunctionProperties.getDefinition()) && isInheritedIntegrationFlow()) {
+			if (shouldWireDunctionToChannel(false)) {
 				inputChannel = this.postProcessInboundChannelForFunction(inputChannel, (ConsumerProperties) properties);
 			}
 			if (HeaderMode.embeddedHeaders.equals(properties.getHeaderMode())) {
@@ -818,9 +817,18 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 	 * bootstrap (e.g. brand new function based app). For that please see
 	 * FunctionConfiguration.integrationFlowCreator
 	 */
-
-	private boolean isInheritedIntegrationFlow() {
-		return !this.getApplicationContext().containsBean("integrationFlowCreator") || integrationFlow == null;
+	private boolean shouldWireDunctionToChannel(boolean producer) {
+		if (!producer && this.producerBindingExist) {
+			return false;
+		}
+		else {
+			return this.streamFunctionProperties != null
+					&& StringUtils.hasText(this.streamFunctionProperties.getDefinition())
+					&& (
+							!this.getApplicationContext().containsBean("integrationFlowCreator")
+							|| this.getApplicationContext().getBean("integrationFlowCreator").equals(null)
+						);
+		}
 	}
 
 	private SubscribableChannel postProcessOutboundChannelForFunction(MessageChannel outputChannel, ProducerProperties producerProperties) {
diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java
index 58fa36cd7..d04ca43da 100644
--- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java
+++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java
@@ -116,7 +116,9 @@ public class GreenfieldFunctionEnableBindingTests {
 			restTemplate.postForLocation("http://localhost:8080", "hello");
 
 			OutputDestination target = context.getBean(OutputDestination.class);
-			assertThat(target.receive(10000).getPayload()).isEqualTo("HELLO".getBytes(StandardCharsets.UTF_8));
+			String result = new String(target.receive(10000).getPayload());
+			System.out.println(result);
+			assertThat(result).isEqualTo("HELLO");
 		}
 	}
 
diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java
index 071544e8b..b88c0fbdc 100644
--- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java
+++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,7 +21,6 @@ import java.util.function.Consumer;
 import java.util.function.Function;
 
 import org.junit.After;
-import org.junit.Ignore;
 import org.junit.Test;
 
 import org.springframework.beans.DirectFieldAccessor;
@@ -70,7 +69,6 @@ public class ProcessorToFunctionsSupportTests {
 	}
 
 	@Test
-	@Ignore
 	public void testSingleFunction() {
 		context = new SpringApplicationBuilder(
 			TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class)).web(
@@ -84,7 +82,6 @@ public class ProcessorToFunctionsSupportTests {
 	}
 
 	@Test
-	@Ignore
 	public void testComposedFunction() {
 	context = new SpringApplicationBuilder(
 			TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class)).web(
@@ -95,7 +92,9 @@ public class ProcessorToFunctionsSupportTests {
 		InputDestination source = context.getBean(InputDestination.class);
 		OutputDestination target = context.getBean(OutputDestination.class);
 		source.send(new GenericMessage<byte[]>("hello".getBytes(StandardCharsets.UTF_8)));
-		assertThat(target.receive(1000).getPayload()).isEqualTo("HELLO:HELLO".getBytes(StandardCharsets.UTF_8));
+		String result = new String(target.receive(1000).getPayload());
+		System.out.println(result);
+		assertThat(result).isEqualTo("HELLO:HELLO");
 	}
 
 	@Test