GH-1590 Added support for implicit function bindings

- Added support for implicit function binding where you no longer required to use EnableBinding making your application simply a boot app with stream/function bits coming being auto configured.
- Added support for binding function without it being specified via ‘spring.cloud.strea.function.definition’ property if such function is unique (single per catalog)

Resolves #1590

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2018 the original author or authors.
+ * Copyright 2015-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.
@@ -79,7 +79,7 @@ import org.springframework.validation.Validator;
 @Configuration
 @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
 @EnableConfigurationProperties({ BindingServiceProperties.class })
-@Import({ ContentTypeConfiguration.class })
+@Import({ ContentTypeConfiguration.class, ImplicitFunctionBindingConfiguration.class })
 @Deprecated
 public class BinderFactoryConfiguration {
 

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingBeansRegistrar.java

@@ -30,6 +30,7 @@ import org.springframework.util.ClassUtils;
  * @author Marius Bogoevici
  * @author Dave Syer
  * @author Artem Bilan
+ * @author Oleg Zhurakousky
  */
 public class BindingBeansRegistrar implements ImportBeanDefinitionRegistrar {
 
@@ -56,5 +57,4 @@ public class BindingBeansRegistrar implements ImportBeanDefinitionRegistrar {
 				EnableBinding.class, ClassUtils.resolveClassName(className, null));
 		return enableBinding.value();
 	}
-
 }

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ImplicitFunctionBindingConfiguration.java

@@ -0,0 +1,101 @@
+/*
+ * Copyright 2019-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.
+ * 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.config;
+
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.beans.factory.support.BeanDefinitionRegistry;
+import org.springframework.cloud.function.context.FunctionCatalog;
+import org.springframework.cloud.function.context.FunctionRegistry;
+import org.springframework.cloud.function.context.catalog.FunctionInspector;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils;
+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.core.env.Environment;
+import org.springframework.core.env.StandardEnvironment;
+import org.springframework.lang.Nullable;
+import org.springframework.util.ObjectUtils;
+import org.springframework.util.StringUtils;
+
+/**
+ *
+ * @author Oleg Zhurakousky
+ * @since 2.2
+ *
+ */
+class ImplicitFunctionBindingConfiguration {
+
+	@Bean
+	public BeanFactoryPostProcessor implicitFunctionBinder(Environment environment,
+			@Nullable FunctionRegistry functionCatalog, @Nullable FunctionInspector inspector) {
+		return new BeanFactoryPostProcessor() {
+			@Override
+			public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
+				if (functionCatalog != null && ObjectUtils.isEmpty(beanFactory.getBeanNamesForAnnotation(EnableBinding.class))) {
+					BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory;
+					String name = determineFunctionName(functionCatalog, environment);
+					if (StringUtils.hasText(name)) {
+						Object definedFunction = functionCatalog.lookup(name);
+						Class<?> inputType = inspector.getInputType(definedFunction);
+						Class<?> outputType = inspector.getOutputType(definedFunction);
+						if (Void.class.isAssignableFrom(outputType)) {
+							bind(Sink.class, registry);
+						}
+						else if (Void.class.isAssignableFrom(inputType)) {
+							bind(Source.class, registry);
+						}
+						else {
+							bind(Processor.class, registry);
+						}
+					}
+				}
+			}
+		};
+	}
+
+	private void bind(Class<?> type, BeanDefinitionRegistry registry) {
+		if (!registry.containsBeanDefinition(type.getName())) {
+			BindingBeanDefinitionRegistryUtils.registerBindingTargetBeanDefinitions(type, type.getName(), registry);
+			BindingBeanDefinitionRegistryUtils.registerBindingTargetsQualifiedBeanDefinitions(type, type, registry);
+		}
+	}
+
+	private String determineFunctionName(FunctionCatalog catalog, Environment environment) {
+		String name = environment.getProperty("spring.cloud.stream.function.definition");
+		if (!StringUtils.hasText(name)) {
+			name = catalog.size() == 1
+					? catalog.getNames(Function.class).stream().findFirst()
+							.orElseGet(() -> catalog.getNames(Consumer.class).stream().findFirst()
+									.orElseGet(() -> catalog.getNames(Supplier.class).stream().findFirst().orElse(null)))
+							: null;
+			if (StringUtils.hasText(name)) {
+				((StandardEnvironment) environment).getSystemProperties()
+						.putIfAbsent("spring.cloud.stream.function.definition", name);
+			}
+		}
+
+		return name;
+	}
+}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java

@@ -20,12 +20,13 @@ import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
-import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.AutoConfigureBefore;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-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.BinderFactoryConfiguration;
+import org.springframework.cloud.stream.config.BindingServiceConfiguration;
 import org.springframework.cloud.stream.config.BindingServiceProperties;
 import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
 import org.springframework.cloud.stream.messaging.Processor;
@@ -33,8 +34,10 @@ 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.context.annotation.Import;
 import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.SubscribableChannel;
 
@@ -45,20 +48,11 @@ import org.springframework.messaging.SubscribableChannel;
  * @since 2.1
  */
 @Configuration
-@ConditionalOnProperty("spring.cloud.stream.function.definition")
 @EnableConfigurationProperties(StreamFunctionProperties.class)
+@Import(BinderFactoryConfiguration.class)
+@AutoConfigureBefore(BindingServiceConfiguration.class)
 public class FunctionConfiguration {
 
-
-	@Autowired(required = false)
-	private Source source;
-
-	@Autowired(required = false)
-	private Processor processor;
-
-	@Autowired(required = false)
-	private Sink sink;
-
 	@Bean
 	public IntegrationFlowFunctionSupport functionSupport(
 			FunctionCatalog functionCatalog, FunctionInspector functionInspector,
@@ -69,11 +63,6 @@ public class FunctionConfiguration {
 				messageConverterFactory, functionProperties, bindingServiceProperties);
 	}
 
-//	@Bean
-//	public FunctionCatalogWrapper functionCatalogWrapper(FunctionCatalog catalog) {
-//		return new FunctionCatalogWrapper(catalog);
-//	}
-
 	/**
 	 * This configuration creates an instance of the {@link IntegrationFlow} from standard
 	 * Spring Cloud Stream bindings such as {@link Source}, {@link Processor} and
@@ -88,42 +77,46 @@ public class FunctionConfiguration {
 	 * the IntegrationFlow that may have been already defined by the existing (extended)
 	 * app.
 	 * @param functionSupport support for registering beans
+	 * @param source source binding
+	 * @param processor processor binding
+	 * @param sink sink binding
 	 * @return integration flow for Stream
 	 */
 	@ConditionalOnMissingBean
 	@Bean
 	public IntegrationFlow integrationFlowCreator(
-			IntegrationFlowFunctionSupport functionSupport) {
+			IntegrationFlowFunctionSupport functionSupport,
+			@Nullable Source source, @Nullable Processor processor, @Nullable Sink sink) {
 		if (functionSupport.containsFunction(Consumer.class)
-				&& consumerBindingPresent()) {
+				&& consumerBindingPresent(processor, sink)) {
 			return functionSupport
-					.integrationFlowForFunction(getInputChannel(), getOutputChannel())
+					.integrationFlowForFunction(getInputChannel(processor, sink), getOutputChannel(processor, source))
 					.get();
 		}
 		else if (functionSupport.containsFunction(Function.class)
-				&& consumerBindingPresent()) {
+				&& consumerBindingPresent(processor, sink)) {
 			return functionSupport
-					.integrationFlowForFunction(getInputChannel(), getOutputChannel())
+					.integrationFlowForFunction(getInputChannel(processor, sink), getOutputChannel(processor, source))
 					.get();
 		}
 		else if (functionSupport.containsFunction(Supplier.class)) {
 			return functionSupport.integrationFlowFromNamedSupplier()
-					.channel(getOutputChannel()).get();
+					.channel(getOutputChannel(processor, source)).get();
 		}
 		return null;
 	}
 
-	private boolean consumerBindingPresent() {
-		return this.processor != null || this.sink != null;
+	private boolean consumerBindingPresent(Processor processor, Sink sink) {
+		return processor != null || sink != null;
 	}
 
-	private SubscribableChannel getInputChannel() {
-		return this.processor != null ? this.processor.input() : this.sink.input();
+	private SubscribableChannel getInputChannel(Processor processor, Sink sink) {
+		return processor != null ? processor.input() : sink.input();
 	}
 
-	private MessageChannel getOutputChannel() {
-		return this.processor != null ? this.processor.output()
-				: (this.source != null ? this.source.output() : new NullChannel());
+	private MessageChannel getOutputChannel(Processor processor, Source source) {
+		return processor != null ? processor.output()
+				: (source != null ? source.output() : new NullChannel());
 	}
 
 }

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java

@@ -26,7 +26,6 @@ import java.util.function.Supplier;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.junit.Test;
 
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.WebApplicationType;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.builder.SpringApplicationBuilder;
@@ -208,21 +207,18 @@ public class GreenfieldFunctionEnableBindingTests {
 	@EnableBinding(Source.class)
 	public static class HttpInboundEndpoint {
 
-		@Autowired
-		private Source source;
-
 		@Bean
 		public Function<String, String> upperCase() {
 			return String::toUpperCase;
 		}
 
 		@Bean
-		public HttpRequestHandlingEndpointSupport doFoo() {
+		public HttpRequestHandlingEndpointSupport doFoo(Source source) {
 			HttpRequestHandlerEndpointSpec httpRequestHandler = Http
 					.inboundChannelAdapter("/*")
 					.requestMapping(requestMapping -> requestMapping
 							.methods(HttpMethod.POST).consumes("*/*"))
-					.requestChannel(this.source.output());
+					.requestChannel(source.output());
 			return httpRequestHandler.get();
 		}
 

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java

@@ -0,0 +1,169 @@
+/*
+ * Copyright 2019-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.
+ * 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.Consumer;
+import java.util.function.Function;
+
+import org.junit.Test;
+
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.binder.test.InputDestination;
+import org.springframework.cloud.stream.binder.test.OutputDestination;
+import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
+import org.springframework.cloud.stream.messaging.Processor;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ *
+ * @author Oleg Zhurakousky
+ *
+ */
+public class ImplicitFunctionBindingTests {
+
+	@Test
+	public void testBindingWithNoEnableBindingConfiguration() {
+
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(
+						NoEnableBindingConfiguration.class))
+								.web(WebApplicationType.NONE)
+								.run("--spring.jmx.enabled=false",
+										"--spring.cloud.stream.function.definition=func")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context
+					.getBean(OutputDestination.class);
+
+			Message<byte[]> inputMessage = MessageBuilder
+					.withPayload("Hello".getBytes()).build();
+			inputDestination.send(inputMessage);
+
+			Message<byte[]> outputMessage = outputDestination.receive();
+			assertThat(outputMessage.getPayload()).isEqualTo("Hello".getBytes());
+
+		}
+	}
+
+	@Test
+	public void testBindingWithEnableBindingConfiguration() {
+
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(
+						EnableBindingConfiguration.class))
+								.web(WebApplicationType.NONE)
+								.run("--spring.jmx.enabled=false",
+										"--spring.cloud.stream.function.definition=func")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context
+					.getBean(OutputDestination.class);
+
+			Message<byte[]> inputMessage = MessageBuilder
+					.withPayload("Hello".getBytes()).build();
+			inputDestination.send(inputMessage);
+
+			Message<byte[]> outputMessage = outputDestination.receive();
+			assertThat(outputMessage.getPayload()).isEqualTo("Hello".getBytes());
+
+		}
+	}
+
+	@Test
+	public void testBindingWithNoEnableBindingAndNoDefinitionPropertyConfiguration() {
+
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(
+						SingleFunctionConfiguration.class))
+								.web(WebApplicationType.NONE)
+								.run("--spring.jmx.enabled=false")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context
+					.getBean(OutputDestination.class);
+
+			Message<byte[]> inputMessage = MessageBuilder
+					.withPayload("Hello".getBytes()).build();
+			inputDestination.send(inputMessage);
+
+			Message<byte[]> outputMessage = outputDestination.receive();
+			assertThat(outputMessage.getPayload()).isEqualTo("Hello".getBytes());
+
+		}
+	}
+
+	@EnableAutoConfiguration
+	public static class NoEnableBindingConfiguration  {
+
+		@Bean
+		public Function<String, String> func() {
+			return x -> {
+				System.out.println("Function");
+				return x;
+			};
+		}
+
+		@Bean
+		public Consumer<String> cons() {
+			return x -> {
+				System.out.println("Consumer");
+			};
+		}
+	}
+
+	@EnableAutoConfiguration
+	@EnableBinding(Processor.class)
+	public static class EnableBindingConfiguration  {
+
+		@Bean
+		public Function<String, String> func() {
+			return x -> {
+				System.out.println("Function");
+				return x;
+			};
+		}
+
+		@Bean
+		public Consumer<String> cons() {
+			return x -> {
+				System.out.println("Consumer");
+			};
+		}
+	}
+
+	@EnableAutoConfiguration
+	public static class SingleFunctionConfiguration {
+
+		@Bean
+		public Function<String, String> func() {
+			return x -> {
+				System.out.println("Function");
+				return x;
+			};
+		}
+
+	}
+
+}

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java

@@ -144,13 +144,10 @@ public class ProcessorToFunctionsSupportTests {
 	@Import(BaseProcessorConfiguration.class)
 	public static class ConsumerConfiguration {
 
-		@Autowired
-		OutputDestination out;
-
 		@Bean
-		public Consumer<String> log() {
+		public Consumer<String> log(OutputDestination out) {
 			return x -> {
-				DirectFieldAccessor dfa = new DirectFieldAccessor(this.out);
+				DirectFieldAccessor dfa = new DirectFieldAccessor(out);
 				MessageChannel channel = (MessageChannel) dfa.getPropertyValue("channel");
 				channel.send(new GenericMessage<byte[]>(x.getBytes()));
 			};