GH-2132 Fix channel creation/registration logic

SubscribableChannelBindingTargetFactory now capable of dealing with channels that may have been created outside of the normal process. In other words if channel exists as bean it will only cofigure it

Resolves #2132

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SubscribableChannelBindingTargetFactory.java

@@ -16,10 +16,9 @@
 
 package org.springframework.cloud.stream.binding;
 
+import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel;
-import org.springframework.cloud.stream.messaging.Sink;
-import org.springframework.cloud.stream.messaging.Source;
 import org.springframework.context.support.GenericApplicationContext;
 import org.springframework.messaging.SubscribableChannel;
 
@@ -48,25 +47,59 @@ public class SubscribableChannelBindingTargetFactory
 
 	@Override
 	public SubscribableChannel createInput(String name) {
-		DirectWithAttributesChannel subscribableChannel = new DirectWithAttributesChannel();
-		subscribableChannel.setComponentName(name);
-		subscribableChannel.setAttribute("type", Sink.INPUT);
-		this.messageChannelConfigurer.configureInputChannel(subscribableChannel, name);
-		if (context != null && !context.containsBean(name)) {
-			context.registerBean(name, DirectWithAttributesChannel.class, () -> subscribableChannel);
+		SubscribableChannel subscribableChannel = null;
+		if (context != null && context.containsBean(name)) {
+			try {
+				subscribableChannel = context.getBean(name, SubscribableChannel.class);
+			}
+			catch (BeanCreationException e) {
+				// ignore
+				/*
+				 * Since we still support annotation-based programming model, this exception happens
+				 * because of proxies related to @Input @Output
+				 */
+			}
 		}
+		if (subscribableChannel == null) {
+			DirectWithAttributesChannel channel = new DirectWithAttributesChannel();
+			channel.setComponentName(name);
+			if (context != null && !context.containsBean(name)) {
+				context.registerBean(name, DirectWithAttributesChannel.class, () -> channel);
+			}
+			subscribableChannel = channel;
+		}
+		if (subscribableChannel instanceof DirectWithAttributesChannel) {
+			((DirectWithAttributesChannel) subscribableChannel).setAttribute("type", "input");
+			this.messageChannelConfigurer.configureInputChannel(subscribableChannel, name);
+		}
+
 		return subscribableChannel;
 	}
 
 	@Override
 	public SubscribableChannel createOutput(String name) {
-		DirectWithAttributesChannel subscribableChannel = new DirectWithAttributesChannel();
-		subscribableChannel.setComponentName(name);
-		subscribableChannel.setAttribute("type", Source.OUTPUT);
-		this.messageChannelConfigurer.configureOutputChannel(subscribableChannel, name);
-		if (context != null && !context.containsBean(name)) {
-			context.registerBean(name, DirectWithAttributesChannel.class, () -> subscribableChannel);
+		SubscribableChannel subscribableChannel = null;
+		if (context != null && context.containsBean(name)) {
+			try {
+				subscribableChannel = context.getBean(name, SubscribableChannel.class);
+			}
+			catch (BeanCreationException e) {
+				// ignore
+			}
 		}
+		if (subscribableChannel == null) {
+			DirectWithAttributesChannel channel = new DirectWithAttributesChannel();
+			channel.setComponentName(name);
+			if (context != null && !context.containsBean(name)) {
+				context.registerBean(name, DirectWithAttributesChannel.class, () -> channel);
+			}
+			subscribableChannel = channel;
+		}
+		if (subscribableChannel instanceof DirectWithAttributesChannel) {
+			((DirectWithAttributesChannel) subscribableChannel).setAttribute("type", "output");
+			this.messageChannelConfigurer.configureOutputChannel(subscribableChannel, name);
+		}
+
 		return subscribableChannel;
 	}
 

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

@@ -50,6 +50,7 @@ import org.springframework.cloud.stream.binder.test.OutputDestination;
 import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
 import org.springframework.cloud.stream.binding.BindingsLifecycleController;
 import org.springframework.cloud.stream.binding.BindingsLifecycleController.State;
+import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel;
 import org.springframework.cloud.stream.messaging.Sink;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
@@ -61,6 +62,7 @@ import org.springframework.integration.support.MessageBuilder;
 import org.springframework.kafka.support.KafkaNull;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.SubscribableChannel;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.scheduling.support.PeriodicTrigger;
 
@@ -79,6 +81,23 @@ public class ImplicitFunctionBindingTests {
 		System.clearProperty("spring.cloud.function.definition");
 	}
 
+
+	@Test
+	public void testExplicitChannelConfiguration() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(ExplicitChannelConfiguration.class))
+						.web(WebApplicationType.NONE)
+						.run("--spring.jmx.enabled=false", "--spring.cloud.function.definition=echo")) {
+
+			InputDestination input = context.getBean(InputDestination.class);
+			input.send(new GenericMessage<byte[]>("hello".getBytes()), "echo-in-0");
+
+			OutputDestination output = context.getBean(OutputDestination.class);
+			Message<byte[]> result = output.receive(1000, "echo-out-0");
+			assertThat(result.getPayload()).isEqualTo("hello".getBytes());
+		}
+	}
+
 	@SuppressWarnings({"rawtypes" })
 	@Test
 	public void testDisableAutodetect() {
@@ -1457,6 +1476,20 @@ public class ImplicitFunctionBindingTests {
 		}
 	}
 
+	@EnableAutoConfiguration
+	public static class ExplicitChannelConfiguration {
+
+		@Bean("echo-in-0")
+		public SubscribableChannel myChannel() {
+			return new DirectWithAttributesChannel();
+		}
+
+		@Bean
+		public Function<String, String> echo() {
+			return x -> x;
+		}
+	}
+
 	public static class Person {
 		private String name;
 		private int id;