GH-1876 Some cleanup on functional and annotation mix

Some plishing and cleanup as well as re-enabling few tests,. but as explained in GH issue there is really nothing to fix here since the behav ior is by design.

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

@@ -79,6 +79,7 @@ import org.springframework.util.ObjectUtils;
  * @author Oleg Zhurakousky
  * @author Soby Chacko
  */
+@SuppressWarnings("deprecation")
 @Configuration
 @EnableConfigurationProperties({ BindingServiceProperties.class,
 		SpringIntegrationProperties.class, StreamFunctionProperties.class })
@@ -88,13 +89,11 @@ import org.springframework.util.ObjectUtils;
 @ConditionalOnBean(value = BinderTypeRegistry.class, search = SearchStrategy.CURRENT)
 public class BindingServiceConfiguration {
 
-	// @checkstyle:off
 	/**
 	 * Name of the Spring Cloud Stream stream listener annotation bean post processor.
 	 */
 	public static final String STREAM_LISTENER_ANNOTATION_BEAN_POST_PROCESSOR_NAME = "streamListenerAnnotationBeanPostProcessor";
 
-	// @checkstyle:on
 
 	@Autowired(required = false)
 	private Collection<DefaultBinderFactory.Listener> binderFactoryListeners;

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

@@ -59,7 +59,6 @@ import org.springframework.cloud.function.context.config.FunctionContextUtils;
 import org.springframework.cloud.function.context.config.RoutingFunction;
 import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.cloud.stream.binder.BinderHeaders;
-import org.springframework.cloud.stream.binder.BinderTypeRegistry;
 import org.springframework.cloud.stream.binder.BindingCreatedEvent;
 import org.springframework.cloud.stream.binder.ConsumerProperties;
 import org.springframework.cloud.stream.binder.PartitionHandler;
@@ -121,8 +120,8 @@ public class FunctionConfiguration {
 
 	@Bean
 	public InitializingBean functionBindingRegistrar(Environment environment, FunctionCatalog functionCatalog,
-			StreamFunctionProperties streamFunctionProperties, BinderTypeRegistry binderTypeRegistry) {
-		return new FunctionBindingRegistrar(binderTypeRegistry, functionCatalog, streamFunctionProperties);
+			StreamFunctionProperties streamFunctionProperties) {
+		return new FunctionBindingRegistrar(functionCatalog, streamFunctionProperties);
 	}
 
 	@Bean
@@ -599,13 +598,14 @@ public class FunctionConfiguration {
 	/**
 	 * Creates and registers instances of BindableFunctionProxyFactory for each user defined function
 	 * thus triggering destination bindings between function arguments and destinations.
+	 *
+	 * In other words this class is responsible to do the same work as EnableBinding except that it derives the input/output names
+	 * from the names of the function (e.g., function-in-0).
 	 */
 	private static class FunctionBindingRegistrar implements InitializingBean, ApplicationContextAware, EnvironmentAware {
 
 		protected final Log logger = LogFactory.getLog(getClass());
 
-		private final BinderTypeRegistry binderTypeRegistry;
-
 		private final FunctionCatalog functionCatalog;
 
 		private final StreamFunctionProperties streamFunctionProperties;
@@ -618,8 +618,7 @@ public class FunctionConfiguration {
 
 		private int outputCount;
 
-		FunctionBindingRegistrar(BinderTypeRegistry binderTypeRegistry, FunctionCatalog functionCatalog, StreamFunctionProperties streamFunctionProperties) {
-			this.binderTypeRegistry = binderTypeRegistry;
+		FunctionBindingRegistrar(FunctionCatalog functionCatalog, StreamFunctionProperties streamFunctionProperties) {
 			this.functionCatalog = functionCatalog;
 			this.streamFunctionProperties = streamFunctionProperties;
 		}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/messaging/DirectWithAttributesChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 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.
@@ -20,6 +20,7 @@ import java.util.HashMap;
 import java.util.Map;
 
 import org.springframework.integration.channel.DirectChannel;
+import org.springframework.messaging.MessageHandler;
 
 /**
  * @author Oleg Zhurakousky
@@ -41,4 +42,9 @@ public class DirectWithAttributesChannel extends DirectChannel {
 	public String getBeanName() {
 		return this.getComponentName();
 	}
+
+	@Override
+	public boolean subscribe(MessageHandler handler) {
+		return this.getDispatcher().getHandlerCount() == 1 ? false : super.subscribe(handler);
+	}
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2019 the original author or authors.
+ * Copyright 2019-2020 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.
@@ -17,11 +17,11 @@
 package org.springframework.cloud.stream.function;
 
 import java.nio.charset.StandardCharsets;
+import java.util.List;
 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;
@@ -59,7 +59,6 @@ public class ProcessorToFunctionsSupportTests {
 	}
 
 	@Test
-	@Ignore
 	public void testPathThrough() {
 		this.context = new SpringApplicationBuilder(TestChannelBinderConfiguration
 				.getCompleteConfiguration(FunctionsConfiguration.class))
@@ -72,7 +71,6 @@ public class ProcessorToFunctionsSupportTests {
 	}
 
 	@Test
-	@Ignore
 	public void testSingleFunction() {
 		this.context = new SpringApplicationBuilder(TestChannelBinderConfiguration
 				.getCompleteConfiguration(FunctionsConfiguration.class))
@@ -85,10 +83,13 @@ public class ProcessorToFunctionsSupportTests {
 		source.send(new GenericMessage<byte[]>("hello".getBytes(StandardCharsets.UTF_8)));
 		assertThat(target.receive(1000).getPayload())
 				.isEqualTo("HELLO".getBytes(StandardCharsets.UTF_8));
+		//to ensure there is no possibility of load balancing to the EnableBinding
+		source.send(new GenericMessage<byte[]>("hello".getBytes(StandardCharsets.UTF_8)));
+		assertThat(target.receive(1000).getPayload())
+				.isEqualTo("HELLO".getBytes(StandardCharsets.UTF_8));
 	}
 
 	@Test
-	@Ignore
 	public void testComposedFunction() {
 		this.context = new SpringApplicationBuilder(TestChannelBinderConfiguration
 				.getCompleteConfiguration(FunctionsConfiguration.class))
@@ -101,12 +102,10 @@ public class ProcessorToFunctionsSupportTests {
 		OutputDestination target = this.context.getBean(OutputDestination.class);
 		source.send(new GenericMessage<byte[]>("hello".getBytes(StandardCharsets.UTF_8)));
 		String result = new String(target.receive(1000).getPayload());
-		System.out.println(result);
 		assertThat(result).isEqualTo("HELLO:HELLO");
 	}
 
 	@Test
-	@Ignore
 	public void testConsumer() {
 		this.context = new SpringApplicationBuilder(TestChannelBinderConfiguration
 				.getCompleteConfiguration(ConsumerConfiguration.class))
@@ -135,7 +134,10 @@ public class ProcessorToFunctionsSupportTests {
 
 		@Bean
 		public Function<String, String> toUpperCase() {
-			return String::toUpperCase;
+			return v -> {
+				System.out.println();
+				return v.toUpperCase();
+			};
 		}
 
 		@Bean
@@ -149,11 +151,12 @@ public class ProcessorToFunctionsSupportTests {
 	@Import(BaseProcessorConfiguration.class)
 	public static class ConsumerConfiguration {
 
+		@SuppressWarnings("unchecked")
 		@Bean
 		public Consumer<String> log(OutputDestination out) {
 			return x -> {
 				DirectFieldAccessor dfa = new DirectFieldAccessor(out);
-				MessageChannel channel = (MessageChannel) dfa.getPropertyValue("channel");
+				MessageChannel channel = ((List<MessageChannel>) dfa.getPropertyValue("channels")).get(0);
 				channel.send(new GenericMessage<byte[]>(x.getBytes()));
 			};
 		}