From fd989cd712313ad62a8403b52a1a67d7ee9122d3 Mon Sep 17 00:00:00 2001
From: Oleg Zhurakousky <ozhurakousky@pivotal.io>
Date: Tue, 11 Dec 2018 19:45:44 +0100
Subject: [PATCH] GH-1553 Fixed inconsistencies in function bootstrap

Resolves #1553
---
 .../binder/AbstractMessageChannelBinder.java  | 42 ++++++++++++----
 .../function/FunctionConfiguration.java       | 50 +++++++++++++++----
 .../IntegrationFlowFunctionSupport.java       | 23 ++++++++-
 .../SourceToFunctionsSupportTests.java        |  8 ---
 4 files changed, 95 insertions(+), 28 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 fca8bbd82..e047d9dce 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
@@ -51,6 +51,7 @@ 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;
@@ -119,6 +120,9 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 	@Autowired(required = false)
 	private StreamFunctionProperties streamFunctionProperties;
 
+	@Autowired(required = false)
+	private IntegrationFlow integrationFlow;
+
 	public AbstractMessageChannelBinder(String[] headersToEmbed, PP provisioningProvider) {
 		this(headersToEmbed, provisioningProvider, null);
 	}
@@ -193,7 +197,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		}
 		this.postProcessOutputChannel(outputChannel, producerProperties);
 
-		if (this.streamFunctionProperties != null && StringUtils.hasText(this.streamFunctionProperties.getDefinition())) {
+		if (this.streamFunctionProperties != null && StringUtils.hasText(this.streamFunctionProperties.getDefinition()) && isInheritedIntegrationFlow()) {
 			outputChannel = this.postProcessOutboundChannelForFunction(outputChannel, producerProperties);
 		}
 
@@ -342,7 +346,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())) {
+			if (this.streamFunctionProperties != null && StringUtils.hasText(this.streamFunctionProperties.getDefinition()) && isInheritedIntegrationFlow()) {
 				inputChannel = this.postProcessInboundChannelForFunction(inputChannel, (ConsumerProperties) properties);
 			}
 			if (HeaderMode.embeddedHeaders.equals(properties.getHeaderMode())) {
@@ -806,6 +810,19 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		}
 	}
 
+	/*
+	 * FUNCTION-TO-EXISTING-APP section
+	 *
+	 * To support composing functions into the existing apps.
+	 * These methods do/should not participate in any way with general function
+	 * 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 SubscribableChannel postProcessOutboundChannelForFunction(MessageChannel outputChannel, ProducerProperties producerProperties) {
 		if (this.integrationFlowFunctionSupport != null) {
 			Publisher<?> publisher = MessageChannelReactiveUtils.toPublisher(outputChannel);
@@ -846,22 +863,25 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 	}
 
 	// we're doing it reflectively so we don't expose this as a property to the user
-	private void propagateConsumerPropertiesToFunction(ConsumerProperties consumerProperties) {
+	private void propagateProducerPropertiesToFunction(ProducerProperties producerProperties) {
 		try {
-			Method setConsumerProperties = ReflectionUtils.findMethod(StreamFunctionProperties.class, "setConsumerProperties", ConsumerProperties.class);
-			setConsumerProperties.setAccessible(true);
-			setConsumerProperties.invoke(this.streamFunctionProperties, consumerProperties);
+			Method setProducerProperties = ReflectionUtils.findMethod(StreamFunctionProperties.class,
+					"setProducerProperties", ProducerProperties.class);
+			setProducerProperties.setAccessible(true);
+			setProducerProperties.invoke(this.streamFunctionProperties, producerProperties);
 		}
 		catch (Exception e) {
 			throw new IllegalStateException(e);
 		}
 	}
+
 	// we're doing it reflectively so we don't expose this as a property to the user
-	private void propagateProducerPropertiesToFunction(ProducerProperties producerProperties) {
+	private void propagateConsumerPropertiesToFunction(ConsumerProperties consumerProperties) {
 		try {
-			Method setProducerProperties = ReflectionUtils.findMethod(StreamFunctionProperties.class, "setProducerProperties", ProducerProperties.class);
-			setProducerProperties.setAccessible(true);
-			setProducerProperties.invoke(this.streamFunctionProperties, producerProperties);
+			Method setConsumerProperties = ReflectionUtils.findMethod(StreamFunctionProperties.class,
+					"setConsumerProperties", ConsumerProperties.class);
+			setConsumerProperties.setAccessible(true);
+			setConsumerProperties.invoke(this.streamFunctionProperties, consumerProperties);
 		}
 		catch (Exception e) {
 			throw new IllegalStateException(e);
@@ -876,6 +896,8 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		}
 	}
 
+	// END FUNCTION-TO-EXISTING-APP section
+
 	private final class SendingHandler extends AbstractMessageHandler implements Lifecycle {
 
 		private final boolean embedHeaders;
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
index 24812a526..279935346 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
@@ -16,6 +16,7 @@
 
 package org.springframework.cloud.stream.function;
 
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
@@ -25,12 +26,17 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.function.context.FunctionCatalog;
 import org.springframework.cloud.function.context.catalog.FunctionInspector;
+import org.springframework.cloud.stream.config.BindingServiceProperties;
 import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
 import org.springframework.cloud.stream.messaging.Processor;
+import org.springframework.cloud.stream.messaging.Sink;
 import org.springframework.cloud.stream.messaging.Source;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.SubscribableChannel;
 
 /**
  * @author Oleg Zhurakousky
@@ -49,12 +55,15 @@ public class FunctionConfiguration {
 	@Autowired(required = false)
 	private Processor processor;
 
+	@Autowired(required = false)
+	private Sink sink;
+
 	@Bean
 	public IntegrationFlowFunctionSupport functionSupport(FunctionCatalogWrapper functionCatalog,
 		FunctionInspector functionInspector, CompositeMessageConverterFactory messageConverterFactory,
-		StreamFunctionProperties functionProperties) {
+		StreamFunctionProperties functionProperties, BindingServiceProperties bindingServiceProperties) {
 		return new IntegrationFlowFunctionSupport(functionCatalog, functionInspector, messageConverterFactory,
-			functionProperties);
+			functionProperties, bindingServiceProperties);
 	}
 
 	@Bean
@@ -63,19 +72,42 @@ public class FunctionConfiguration {
 	}
 
 	/**
-	 * This configuration creates an instance of {@link IntegrationFlow} appropriate for binding declared using EnableBinding.
+	 * This configuration creates an instance of the {@link IntegrationFlow} from standard
+	 * Spring Cloud Stream bindings such as {@link Source}, {@link Processor} and {@link Sink}
+	 * ONLY if there are no existing instances of the {@link IntegrationFlow} already available
+	 * in the context. This means that it only plays a role in green-field Spring Cloud Stream apps.
+	 *
+	 * For logic to compose functions into the existing apps please see "FUNCTION-TO-EXISTING-APP"
+	 * section of AbstractMessageChannelBinder.
+	 *
+	 * The @ConditionalOnMissingBean ensures it does not collide with the the instance of the IntegrationFlow
+	 * that may have been already defined by the existing (extended) app.
 	 */
 	@ConditionalOnMissingBean
 	@Bean
 	public IntegrationFlow integrationFlowCreator(IntegrationFlowFunctionSupport functionSupport) {
-		if (this.processor != null) {
-			return functionSupport.containsFunction(Function.class) ?
-					functionSupport.integrationFlowForFunction(this.processor.input(), this.processor.output()).get() : null;
+		if (functionSupport.containsFunction(Consumer.class) && consumerBindingPresent()) {
+			return functionSupport.integrationFlowForFunction(getInputChannel(), getOutputChannel()).get();
 		}
-		else if (this.source != null && this.processor == null) {
-			return functionSupport.containsFunction(Supplier.class) ?
-				functionSupport.integrationFlowFromNamedSupplier().channel(this.source.output()).get() : null;
+		else if (functionSupport.containsFunction(Function.class) && consumerBindingPresent()) {
+			return functionSupport.integrationFlowForFunction(getInputChannel(), getOutputChannel()).get();
+		}
+		else if (functionSupport.containsFunction(Supplier.class)) {
+			return functionSupport.integrationFlowFromNamedSupplier().channel(getOutputChannel()).get();
 		}
 		return null;
 	}
+
+	private boolean consumerBindingPresent() {
+		return this.processor != null || this.sink != null;
+	}
+
+	private SubscribableChannel getInputChannel() {
+		return this.processor != null ? this.processor.input() : this.sink.input();
+	}
+
+	private MessageChannel getOutputChannel() {
+		return this.processor != null ? this.processor.output()
+				: (this.source != null ? this.source.output() : new NullChannel());
+	}
 }
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java
index f987be8dc..29705f4f0 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java
@@ -29,7 +29,9 @@ import org.springframework.cloud.function.context.FunctionCatalog;
 import org.springframework.cloud.function.context.FunctionType;
 import org.springframework.cloud.function.context.catalog.FunctionInspector;
 import org.springframework.cloud.function.core.FluxSupplier;
+import org.springframework.cloud.stream.config.BindingServiceProperties;
 import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
+import org.springframework.integration.context.IntegrationObjectSupport;
 import org.springframework.integration.dsl.IntegrationFlowBuilder;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.messaging.Message;
@@ -56,6 +58,8 @@ public class IntegrationFlowFunctionSupport {
 
 	private final StreamFunctionProperties functionProperties;
 
+	private final BindingServiceProperties bindingServiceProperties;
+
 	@Autowired
 	private MessageChannel errorChannel;
 
@@ -66,7 +70,8 @@ public class IntegrationFlowFunctionSupport {
 	 * @param functionProperties
 	 */
 	IntegrationFlowFunctionSupport(FunctionCatalogWrapper functionCatalog, FunctionInspector functionInspector,
-			CompositeMessageConverterFactory messageConverterFactory, StreamFunctionProperties functionProperties) {
+			CompositeMessageConverterFactory messageConverterFactory, StreamFunctionProperties functionProperties,
+			BindingServiceProperties bindingServiceProperties) {
 
 		Assert.notNull(functionCatalog, "'functionCatalog' must not be null");
 		Assert.notNull(functionInspector, "'functionInspector' must not be null");
@@ -76,6 +81,7 @@ public class IntegrationFlowFunctionSupport {
 		this.functionInspector = functionInspector;
 		this.messageConverterFactory = messageConverterFactory;
 		this.functionProperties = functionProperties;
+		this.bindingServiceProperties = bindingServiceProperties;
 	}
 
 	/**
@@ -148,6 +154,21 @@ public class IntegrationFlowFunctionSupport {
 
 	public <O> IntegrationFlowBuilder integrationFlowForFunction(SubscribableChannel inputChannel,
 			MessageChannel outputChannel) {
+
+		if (inputChannel instanceof IntegrationObjectSupport) {
+			String inputBindingName = ((IntegrationObjectSupport)inputChannel).getComponentName();
+			if (StringUtils.hasText(inputBindingName)) {
+				this.functionProperties.setConsumerProperties(this.bindingServiceProperties.getConsumerProperties(inputBindingName));
+			}
+		}
+
+		if (outputChannel instanceof IntegrationObjectSupport) {
+			String outputBindingName = ((IntegrationObjectSupport)outputChannel).getComponentName();
+			if (StringUtils.hasText(outputBindingName)) {
+				this.functionProperties.setProducerProperties(this.bindingServiceProperties.getProducerProperties(outputBindingName));
+			}
+		}
+
 		IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(inputChannel).bridge();
 
 		if (!this.andThenFunction(flowBuilder, outputChannel, this.functionProperties)) {
diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java
index 5b47bf548..08ab6c9ee 100644
--- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java
+++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java
@@ -26,7 +26,6 @@ import org.junit.Test;
 import org.junit.rules.ExpectedException;
 import reactor.core.publisher.Flux;
 
-import org.springframework.beans.BeanInstantiationException;
 import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.WebApplicationType;
@@ -49,10 +48,6 @@ import org.springframework.util.Assert;
 import org.springframework.util.MimeTypeUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.hamcrest.CoreMatchers.allOf;
-import static org.hamcrest.CoreMatchers.endsWith;
-import static org.hamcrest.CoreMatchers.isA;
-import static org.hamcrest.Matchers.hasProperty;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 
@@ -169,9 +164,6 @@ public class SourceToFunctionsSupportTests {
 	public void testFunctionDoesNotExist() {
 
 		expectedException.expect(BeanCreationException.class);
-		expectedException.expectCause(
-			allOf(isA(BeanInstantiationException.class), hasProperty("cause", isA(IllegalArgumentException.class)),
-				hasProperty("message", endsWith("'doesNotExist' cannot be located."))));
 
 		new SpringApplicationBuilder(
 			TestChannelBinderConfiguration.getCompleteConfiguration(SupplierConfiguration.class)).web(