From 28990fe995575bdccb31eee527114ab147ae8962 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 23 Oct 2018 12:07:58 +0200 Subject: [PATCH] GH-1425 Added support for type-aware channels Resolves #1425 --- .../config/MessageChannelConfigurerTests.java | 9 +++++ ...bscribableChannelBindingTargetFactory.java | 13 ++++-- .../DirectWithAttributesChannel.java | 40 +++++++++++++++++++ .../BinderAwareChannelResolverTests.java | 5 ++- 4 files changed, 61 insertions(+), 6 deletions(-) create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/messaging/DirectWithAttributesChannel.java diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java index 050122da3..a7a12f930 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java +++ b/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); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SubscribableChannelBindingTargetFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SubscribableChannelBindingTargetFactory.java index e01aec339..8e31a90e9 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SubscribableChannelBindingTargetFactory.java +++ b/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 { @@ -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; } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/messaging/DirectWithAttributesChannel.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/messaging/DirectWithAttributesChannel.java new file mode 100644 index 000000000..1a759c810 --- /dev/null +++ b/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 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); + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java index 70fd80923..8ee46b59e 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java +++ b/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,