GH-1425 Added support for type-aware channels

Resolves #1425

spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java

@@ -31,6 +31,7 @@ import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.cloud.stream.binder.BinderHeaders;
 import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
+import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel;
 import org.springframework.cloud.stream.messaging.Sink;
 import org.springframework.cloud.stream.messaging.Source;
 import org.springframework.cloud.stream.test.binder.MessageCollector;
@@ -68,6 +69,14 @@ public class MessageChannelConfigurerTests {
 	@Autowired
 	private MessageCollector messageCollector;
 
+	@Test
+	public void testChannelTypes() throws Exception {
+		DirectWithAttributesChannel inputChannel = (DirectWithAttributesChannel) testSink.input();
+		DirectWithAttributesChannel outputChannel = (DirectWithAttributesChannel) testSource.output();
+		assertThat(inputChannel.getAttribute("type")).isEqualTo(Sink.INPUT);
+		assertThat(outputChannel.getAttribute("type")).isEqualTo(Source.OUTPUT);
+	}
+
 	@Test
 	public void testMessageConverterConfigurer() throws Exception {
 		final CountDownLatch latch = new CountDownLatch(1);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-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.
@@ -16,7 +16,9 @@
 
 package org.springframework.cloud.stream.binding;
 
-import org.springframework.integration.channel.DirectChannel;
+import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel;
+import org.springframework.cloud.stream.messaging.Sink;
+import org.springframework.cloud.stream.messaging.Source;
 import org.springframework.messaging.SubscribableChannel;
 
 /**
@@ -26,6 +28,7 @@ import org.springframework.messaging.SubscribableChannel;
  * @author Marius Bogoevici
  * @author David Syer
  * @author Ilayaperumal Gopinathan
+ * @author Oleg Zhurakousky
  */
 public class SubscribableChannelBindingTargetFactory extends AbstractBindingTargetFactory<SubscribableChannel> {
 
@@ -38,14 +41,16 @@ public class SubscribableChannelBindingTargetFactory extends AbstractBindingTarg
 
 	@Override
 	public SubscribableChannel createInput(String name) {
-		SubscribableChannel subscribableChannel = new DirectChannel();
+		DirectWithAttributesChannel subscribableChannel = new DirectWithAttributesChannel();
+		subscribableChannel.setAttribute("type", Sink.INPUT);
 		this.messageChannelConfigurer.configureInputChannel(subscribableChannel, name);
 		return subscribableChannel;
 	}
 
 	@Override
 	public SubscribableChannel createOutput(String name) {
-		SubscribableChannel subscribableChannel = new DirectChannel();
+		DirectWithAttributesChannel subscribableChannel = new DirectWithAttributesChannel();
+		subscribableChannel.setAttribute("type", Source.OUTPUT);
 		this.messageChannelConfigurer.configureOutputChannel(subscribableChannel, name);
 		return subscribableChannel;
 	}

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

@@ -0,0 +1,40 @@
+/*
+ * 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.messaging;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.integration.channel.DirectChannel;
+
+/**
+ *
+ * @author Oleg Zhurakousky
+ * @since 2.1
+ */
+public class DirectWithAttributesChannel extends DirectChannel {
+
+	private final Map<String, Object> attributes = new HashMap<>();
+
+	public void setAttribute(String key, Object value) {
+		this.attributes.put(key, value);
+	}
+
+	public Object getAttribute(String key) {
+		return this.attributes.get(key);
+	}
+}

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java

@@ -38,6 +38,7 @@ import org.springframework.cloud.stream.binding.DynamicDestinationsBindable;
 import org.springframework.cloud.stream.binding.SubscribableChannelBindingTargetFactory;
 import org.springframework.cloud.stream.config.BindingProperties;
 import org.springframework.cloud.stream.config.BindingServiceProperties;
+import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -176,8 +177,8 @@ public class BinderAwareChannelResolverTests {
 				matches("foo"), any(DirectChannel.class), any(ProducerProperties.class))).thenReturn(fooBinding);
 		when(binder2.bindProducer(
 				matches("bar"), any(DirectChannel.class), any(ProducerProperties.class))).thenReturn(barBinding);
-		when(mockBinderFactory.getBinder(null, DirectChannel.class)).thenReturn(binder);
-		when(mockBinderFactory.getBinder("someTransport", DirectChannel.class)).thenReturn(binder2);
+		when(mockBinderFactory.getBinder(null, DirectWithAttributesChannel.class)).thenReturn(binder);
+		when(mockBinderFactory.getBinder("someTransport", DirectWithAttributesChannel.class)).thenReturn(binder2);
 		BindingService bindingService = new BindingService(bindingServiceProperties,
 				mockBinderFactory);
 		BinderAwareChannelResolver resolver = new BinderAwareChannelResolver(bindingService, this.bindingTargetFactory,