diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/aggregate/AggregateApplicationTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/aggregate/AggregateApplicationTests.java new file mode 100644 index 000000000..f36d91679 --- /dev/null +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/aggregate/AggregateApplicationTests.java @@ -0,0 +1,54 @@ +/* + * Copyright 2017 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.aggregate; + +import java.util.concurrent.TimeUnit; + +import org.junit.Assert; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.cloud.stream.aggregate.AggregateApplicationBuilder; +import org.springframework.cloud.stream.binder.BinderFactory; +import org.springframework.cloud.stream.config.aggregate.processor.TestProcessor; +import org.springframework.cloud.stream.config.aggregate.source.TestSource; +import org.springframework.cloud.stream.test.binder.TestSupportBinder; +import org.springframework.cloud.stream.test.binder.TestSupportBinderAutoConfiguration; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +import static org.hamcrest.Matchers.notNullValue; + +/** + * @author Ilayaperumal Gopinathan + */ +@RunWith(SpringJUnit4ClassRunner.class) +public class AggregateApplicationTests { + + @Test + @SuppressWarnings("unchecked") + public void testAggregateApplication() throws Exception { + ConfigurableApplicationContext context = new AggregateApplicationBuilder(TestSupportBinderAutoConfiguration.class).from(TestSource.class).to(TestProcessor.class).run(); + TestSupportBinder testSupportBinder = (TestSupportBinder) context.getBean(BinderFactory.class).getBinder(null, MessageChannel.class); + MessageChannel processorOutput = testSupportBinder.getChannelForName("output"); + Message received = (Message) (testSupportBinder.messageCollector().forChannel(processorOutput).poll(5, TimeUnit.SECONDS)); + Assert.assertThat(received, notNullValue()); + Assert.assertTrue(received.getPayload().endsWith("processed")); + } +} diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/aggregate/processor/TestProcessor.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/aggregate/processor/TestProcessor.java new file mode 100644 index 000000000..e7b76b367 --- /dev/null +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/aggregate/processor/TestProcessor.java @@ -0,0 +1,39 @@ +/* + * Copyright 2017 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.aggregate.processor; + +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.messaging.Processor; +import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.handler.annotation.SendTo; + +/** + * @author Ilayaperumal Gopinathan + */ +@EnableBinding(Processor.class) +@EnableAutoConfiguration +@Configuration +public class TestProcessor { + + @StreamListener(Processor.INPUT) + @SendTo(Processor.OUTPUT) + public String process(String message) { + return message + " processed"; + } +} diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/aggregate/source/TestSource.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/aggregate/source/TestSource.java new file mode 100644 index 000000000..e6421af15 --- /dev/null +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/aggregate/source/TestSource.java @@ -0,0 +1,50 @@ +/* + * Copyright 2017 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.aggregate.source; + +import java.text.SimpleDateFormat; +import java.util.Date; + +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.messaging.Source; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.annotation.InboundChannelAdapter; +import org.springframework.integration.core.MessageSource; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.GenericMessage; + +/** + * @author Ilayaperumal Gopinathan + */ +@EnableBinding(Source.class) +@EnableAutoConfiguration +@Configuration +public class TestSource { + + @Bean + @InboundChannelAdapter(Source.OUTPUT) + public MessageSource timerMessageSource() { + return new MessageSource() { + @Override + public Message receive() { + return new GenericMessage<>(new SimpleDateFormat("DDMMMYYYY").format(new Date())); + } + }; + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java index 58cf37119..baca9be31 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java @@ -102,21 +102,18 @@ public class BindingServiceConfiguration { } @Bean - @ConditionalOnMissingBean(MessageConverterConfigurer.class) public MessageConverterConfigurer messageConverterConfigurer(BindingServiceProperties bindingServiceProperties, CompositeMessageConverterFactory compositeMessageConverterFactory) { return new MessageConverterConfigurer(bindingServiceProperties, compositeMessageConverterFactory); } @Bean - @ConditionalOnMissingBean(SubscribableChannelBindingTargetFactory.class) public SubscribableChannelBindingTargetFactory channelFactory( CompositeMessageChannelConfigurer compositeMessageChannelConfigurer) { return new SubscribableChannelBindingTargetFactory(compositeMessageChannelConfigurer); } @Bean - @ConditionalOnMissingBean(CompositeMessageChannelConfigurer.class) public CompositeMessageChannelConfigurer compositeMessageChannelConfigurer( MessageConverterConfigurer messageConverterConfigurer) { List configurerList = new ArrayList<>(); @@ -126,27 +123,23 @@ public class BindingServiceConfiguration { @Bean @DependsOn("bindingService") - @ConditionalOnMissingBean(OutputBindingLifecycle.class) public OutputBindingLifecycle outputBindingLifecycle() { return new OutputBindingLifecycle(); } @Bean @DependsOn("bindingService") - @ConditionalOnMissingBean(InputBindingLifecycle.class) public InputBindingLifecycle inputBindingLifecycle() { return new InputBindingLifecycle(); } @Bean @DependsOn("bindingService") - @ConditionalOnMissingBean(ContextStartAfterRefreshListener.class) public ContextStartAfterRefreshListener contextStartAfterRefreshListener() { return new ContextStartAfterRefreshListener(); } @Bean - @ConditionalOnMissingBean(BinderAwareChannelResolver.class) public BinderAwareChannelResolver binderAwareChannelResolver(BindingService bindingService, AbstractBindingTargetFactory bindingTargetFactory, DynamicDestinationsBindable dynamicDestinationsBindable) { @@ -155,20 +148,17 @@ public class BindingServiceConfiguration { @Bean @ConditionalOnProperty("spring.cloud.stream.bindings." + ERROR_CHANNEL_NAME + ".destination") - @ConditionalOnMissingBean(SingleBindingTargetBindable.class) public SingleBindingTargetBindable errorChannelBindable( @Qualifier(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME) PublishSubscribeChannel errorChannel) { return new SingleBindingTargetBindable(ERROR_CHANNEL_NAME, errorChannel); } @Bean - @ConditionalOnMissingBean(DynamicDestinationsBindable.class) public DynamicDestinationsBindable dynamicDestinationsBindable() { return new DynamicDestinationsBindable(); } @Bean - @ConditionalOnMissingBean(CompositeMessageConverterFactory.class) public CompositeMessageConverterFactory compositeMessageConverterFactory() { List messageConverters = new ArrayList<>(); if (!CollectionUtils.isEmpty(this.customMessageConverters)) { @@ -178,7 +168,6 @@ public class BindingServiceConfiguration { } @Bean - @ConditionalOnMissingBean(MessageHandlerMethodFactory.class) public static MessageHandlerMethodFactory messageHandlerMethodFactory( CompositeMessageConverterFactory compositeMessageConverterFactory) { DefaultMessageHandlerMethodFactory messageHandlerMethodFactory = new DefaultMessageHandlerMethodFactory();