From e4a82107d86ae065a86015df598be02e28a9d4e0 Mon Sep 17 00:00:00 2001
From: Oleg Zhurakousky <ozhurakousky@pivotal.io>
Date: Sun, 9 Sep 2018 14:32:00 +0200
Subject: [PATCH] GH-1465 Polishing function support for channel binder

Removed BinderFunctionSupport in favor of private method (at least for now)
Removed 'context.getBean()` from IntegrationFlowFunctionSupport in favor of propper DI

Resolves #1465
---
 .../binder/AbstractMessageChannelBinder.java  | 34 +++++++--------
 .../function/BinderFunctionSupport.java       | 42 -------------------
 .../IntegrationFlowFunctionSupport.java       | 20 ++-------
 3 files changed, 20 insertions(+), 76 deletions(-)
 delete mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BinderFunctionSupport.java

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 431886634..9508c9375 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
@@ -18,17 +18,17 @@ package org.springframework.cloud.stream.binder;
 
 import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.function.Function;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.commons.logging.Log;
 
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
-import org.springframework.beans.factory.NoSuchBeanDefinitionException;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.beans.factory.support.DefaultSingletonBeanRegistry;
 import org.springframework.cloud.stream.config.ListenerContainerCustomizer;
-import org.springframework.cloud.stream.function.BinderFunctionSupport;
 import org.springframework.cloud.stream.function.IntegrationFlowFunctionSupport;
 import org.springframework.cloud.stream.provisioning.ConsumerDestination;
 import org.springframework.cloud.stream.provisioning.ProducerDestination;
@@ -39,6 +39,8 @@ import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.channel.AbstractMessageChannel;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.MessageChannelReactiveUtils;
 import org.springframework.integration.channel.PublishSubscribeChannel;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.core.MessageProducer;
@@ -103,6 +105,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 
 	private ApplicationEventPublisher applicationEventPublisher;
 
+	@Autowired(required = false)
 	private IntegrationFlowFunctionSupport integrationFlowFunctionSupport;
 
 	public AbstractMessageChannelBinder(String[] headersToEmbed, PP provisioningProvider) {
@@ -178,10 +181,9 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 			((Lifecycle) producerMessageHandler).start();
 		}
 
-		postProcessOutputChannel(outputChannel, producerProperties);
+		this.postProcessOutputChannel(outputChannel, producerProperties);
 
-		outputChannel = BinderFunctionSupport.andThenFunctionDefinition(this.integrationFlowFunctionSupport,
-			outputChannel);
+		outputChannel = this.postProcessChannelForFunction(outputChannel);
 
 		((SubscribableChannel) outputChannel).subscribe(
 				new SendingHandler(producerMessageHandler, HeaderMode.embeddedHeaders
@@ -740,18 +742,6 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		return destination.getName() + ".errors";
 	}
 
-
-	@Override
-	protected void onInit() throws Exception {
-		super.onInit();
-		try {
-			this.integrationFlowFunctionSupport = getApplicationContext().getBean(IntegrationFlowFunctionSupport.class);
-		}
-		catch (NoSuchBeanDefinitionException e) {
-			//ignore
-		}
-	}
-
 	private Map<String, Object> doGetExtendedInfo(Object destination, Object properties) {
 		Map<String, Object> extendedInfo = new LinkedHashMap<>();
 		extendedInfo.put("bindingDestination", destination.toString());
@@ -777,6 +767,16 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		}
 	}
 
+	private SubscribableChannel postProcessChannelForFunction(MessageChannel outputChannel) {
+		if (integrationFlowFunctionSupport != null && integrationFlowFunctionSupport.containsFunction(Function.class)) {
+			DirectChannel actualOutputChannel = new DirectChannel();
+			integrationFlowFunctionSupport.andThenFunction(MessageChannelReactiveUtils.toPublisher(outputChannel),
+				actualOutputChannel);
+			return actualOutputChannel;
+		}
+		return (SubscribableChannel) outputChannel;
+	}
+
 	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/BinderFunctionSupport.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BinderFunctionSupport.java
deleted file mode 100644
index fae10d95b..000000000
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BinderFunctionSupport.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * Copyright 2018 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.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.stream.function;
-
-import java.util.function.Function;
-
-import org.springframework.integration.channel.DirectChannel;
-import org.springframework.integration.channel.MessageChannelReactiveUtils;
-import org.springframework.messaging.MessageChannel;
-import org.springframework.messaging.SubscribableChannel;
-
-/**
- * @author David Turanski
- * @since 2.1
- **/
-public abstract class BinderFunctionSupport {
-
-	public static SubscribableChannel andThenFunctionDefinition(IntegrationFlowFunctionSupport
-		integrationFlowFunctionSupport, MessageChannel outputChannel) {
-		if (integrationFlowFunctionSupport != null && integrationFlowFunctionSupport.containsFunction(Function.class)) {
-			DirectChannel actualOutputChannel = new DirectChannel();
-			integrationFlowFunctionSupport.andThenFunction(MessageChannelReactiveUtils.toPublisher(outputChannel),
-				actualOutputChannel);
-			return actualOutputChannel;
-		}
-		return (SubscribableChannel) outputChannel;
-	}
-}
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 70ddee1a4..f63e4dc14 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
@@ -24,16 +24,12 @@ import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
-import org.springframework.beans.BeansException;
-import org.springframework.beans.factory.InitializingBean;
+import org.springframework.beans.factory.annotation.Autowired;
 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.converter.CompositeMessageConverterFactory;
-import org.springframework.context.ApplicationContext;
-import org.springframework.context.ApplicationContextAware;
-import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.dsl.IntegrationFlowBuilder;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.messaging.Message;
@@ -50,7 +46,7 @@ import org.springframework.util.StringUtils;
  *
  * @since 2.1
  */
-public class IntegrationFlowFunctionSupport implements ApplicationContextAware, InitializingBean {
+public class IntegrationFlowFunctionSupport {
 
 	private final FunctionCatalogWrapper functionCatalog;
 
@@ -60,8 +56,8 @@ public class IntegrationFlowFunctionSupport implements ApplicationContextAware,
 
 	private final StreamFunctionProperties functionProperties;
 
+	@Autowired
 	private MessageChannel errorChannel;
-	private ApplicationContext applicationContext;
 
 	/**
 	 * @param functionCatalog
@@ -201,14 +197,4 @@ public class IntegrationFlowFunctionSupport implements ApplicationContextAware,
 		Flux<?> inputPublisher = Flux.from(publisher);
 		subscribeToOutput(outputProcessor, functionInvoker.apply((Flux<Message<I>>) inputPublisher)).subscribe();
 	}
-
-	@Override
-	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
-		this.applicationContext = applicationContext;
-	}
-
-	@Override
-	public void afterPropertiesSet() throws Exception {
-		this.errorChannel = (MessageChannel) applicationContext.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME);
-	}
 }