diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/ErrorChannelTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/ErrorChannelTests.java new file mode 100644 index 000000000..ab493db8d --- /dev/null +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/ErrorChannelTests.java @@ -0,0 +1,107 @@ +/* + * Copyright 2016 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; + +import static org.junit.Assert.assertTrue; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.SpringApplicationConfiguration; +import org.springframework.cloud.stream.annotation.Bindings; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.binder.redis.config.RedisMessageChannelBinderConfiguration; +import org.springframework.cloud.stream.messaging.Sink; +import org.springframework.cloud.stream.messaging.Source; +import org.springframework.cloud.stream.test.junit.redis.RedisTestSupport; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Import; +import org.springframework.context.annotation.PropertySource; +import org.springframework.integration.annotation.InboundChannelAdapter; +import org.springframework.integration.annotation.Poller; +import org.springframework.integration.core.MessageSource; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.support.ErrorMessage; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.util.Assert; + +/** + * @author Ilayaperumal Gopinathan + */ +@RunWith(SpringJUnit4ClassRunner.class) +@SpringApplicationConfiguration({ErrorChannelTests.TestSource.class, ErrorChannelTests.TestErrorSink.class}) +public class ErrorChannelTests { + + @Rule + public RedisTestSupport redisTestSupport = new RedisTestSupport(); + + @Autowired + @Bindings(TestErrorSink.class) + private Sink testErrorSink; + + @Test + public void testErrorChannelBinding() throws Exception { + final CountDownLatch latch = new CountDownLatch(1); + MessageHandler errorMessageHandler = new MessageHandler() { + @Override + public void handleMessage(Message message) throws MessagingException { + Assert.isTrue(message instanceof ErrorMessage, "Message should be an instance of ErrorMessage"); + Assert.isTrue(message.getPayload() instanceof MessagingException, "Message payload should be an instance" + + "of MessagingException"); + Assert.isTrue(message.getPayload().toString() + .equals("org.springframework.messaging.MessagingException: test")); + latch.countDown(); + } + }; + testErrorSink.input().subscribe(errorMessageHandler); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + } + + @EnableBinding(Source.class) + @EnableAutoConfiguration + @Import(RedisMessageChannelBinderConfiguration.class) + @PropertySource("classpath:/org/springframework/cloud/stream/config/errorchannel/source-channel.properties") + public static class TestSource { + + @Bean + @InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "5000", maxMessagesPerPoll = "1")) + public MessageSource timerMessageSource() { + return new MessageSource() { + @Override + public Message receive() { + throw new MessagingException("test"); + } + }; + } + } + + @EnableBinding(Sink.class) + @EnableAutoConfiguration + @Import(RedisMessageChannelBinderConfiguration.class) + @PropertySource("classpath:/org/springframework/cloud/stream/config/errorchannel/errorsink-channel.properties") + public static class TestErrorSink { + + } +} + 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 31182146f..80954ecc7 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 @@ -117,7 +117,7 @@ public class MessageChannelConfigurerTests { @EnableBinding(Source.class) @EnableAutoConfiguration @Import(RedisMessageChannelBinderConfiguration.class) - @PropertySource("classpath:/org/springframework/cloud/stream/config/source-channel-configurers.properties") + @PropertySource("classpath:/org/springframework/cloud/stream/config/channel/source-channel-configurers.properties") public static class TestSource { } @@ -125,7 +125,7 @@ public class MessageChannelConfigurerTests { @EnableBinding(Sink.class) @EnableAutoConfiguration @Import(RedisMessageChannelBinderConfiguration.class) - @PropertySource("classpath:/org/springframework/cloud/stream/config/sink-channel-configurers.properties") + @PropertySource("classpath:/org/springframework/cloud/stream/config/channel/sink-channel-configurers.properties") public static class TestSink { } diff --git a/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/sink-channel-configurers.properties b/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/sink-channel-configurers.properties similarity index 100% rename from spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/sink-channel-configurers.properties rename to spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/sink-channel-configurers.properties diff --git a/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/source-channel-configurers.properties b/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/source-channel-configurers.properties similarity index 100% rename from spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/source-channel-configurers.properties rename to spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/source-channel-configurers.properties diff --git a/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/errorchannel/errorsink-channel.properties b/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/errorchannel/errorsink-channel.properties new file mode 100644 index 000000000..aa6464b4e --- /dev/null +++ b/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/errorchannel/errorsink-channel.properties @@ -0,0 +1 @@ +spring.cloud.stream.bindings.input.destination=errorchannel-test diff --git a/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/errorchannel/source-channel.properties b/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/errorchannel/source-channel.properties new file mode 100644 index 000000000..d91c5dd27 --- /dev/null +++ b/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/errorchannel/source-channel.properties @@ -0,0 +1,2 @@ +spring.cloud.stream.bindings.output.destination=source-output +spring.cloud.stream.bindings.error.destination=errorchannel-test diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java index 4923b732e..41a91c4b1 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java @@ -31,7 +31,7 @@ import org.springframework.context.annotation.Import; import org.springframework.integration.config.EnableIntegration; /** - * Enables the binding of inputs and outputs to a broker, according to the list + * Enables the binding of {@link Input} and {@link Output} - annotated components to a broker, according to the list * of interfaces passed as value to the annotation. * * @author Dave Syer diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableAdapter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableAdapter.java new file mode 100644 index 000000000..88a00de84 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableAdapter.java @@ -0,0 +1,54 @@ +/* + * Copyright 2016 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.binding; + +import java.util.Collections; +import java.util.Set; + +/** + * Default adapter implementation for {@Bindable}. + * + * @author Ilayaperumal Gopinathan + */ +public class BindableAdapter implements Bindable { + + + @Override + public void bindInputs(ChannelBindingService adapter) { + } + + @Override + public void bindOutputs(ChannelBindingService adapter) { + } + + @Override + public void unbindInputs(ChannelBindingService adapter) { + } + + @Override + public void unbindOutputs(ChannelBindingService adapter) { + } + + @Override + public Set getInputs() { + return Collections.unmodifiableSet(Collections.EMPTY_SET); + } + + @Override + public Set getOutputs() { + return Collections.unmodifiableSet(Collections.EMPTY_SET); + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableErrorChannel.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableErrorChannel.java new file mode 100644 index 000000000..b62b648d4 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableErrorChannel.java @@ -0,0 +1,55 @@ +/* + * Copyright 2016 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.binding; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.Set; + +import org.springframework.integration.channel.PublishSubscribeChannel; + +/** + * A {@link Bindable} component that represents an error channel. + * + * @author Ilayaperumal Gopinathan + */ +public class BindableErrorChannel extends BindableAdapter { + + private final String name; + + private final PublishSubscribeChannel errorChannel; + + public BindableErrorChannel(String name, PublishSubscribeChannel errorChannel) { + this.name = name; + this.errorChannel = errorChannel; + } + + @Override + public void bindOutputs(ChannelBindingService adapter) { + adapter.bindProducer(errorChannel, name); + } + + @Override + public void unbindOutputs(ChannelBindingService adapter) { + adapter.unbindProducers(name); + } + + @Override + public Set getOutputs() { + return Collections.unmodifiableSet(new HashSet(Arrays.asList(name))); + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/OutputBindingLifecycle.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/OutputBindingLifecycle.java index a8c878dfc..e95e7ac82 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/OutputBindingLifecycle.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/OutputBindingLifecycle.java @@ -17,7 +17,6 @@ package org.springframework.cloud.stream.binding; import java.util.Map; -import java.util.concurrent.atomic.AtomicBoolean; import org.springframework.beans.BeansException; import org.springframework.context.ApplicationContext; diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java index 8817bf6ba..bc0eaa6f5 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java @@ -24,13 +24,16 @@ import java.util.Map; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactoryUtils; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.config.BeanPostProcessor; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.ConfigurationPropertiesBinding; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.stream.binder.BinderFactory; import org.springframework.cloud.stream.binding.BindableChannelFactory; +import org.springframework.cloud.stream.binding.BindableErrorChannel; import org.springframework.cloud.stream.binding.BinderAwareChannelResolver; import org.springframework.cloud.stream.binding.BinderAwareRouterBeanPostProcessor; import org.springframework.cloud.stream.binding.ChannelBindingService; @@ -47,6 +50,7 @@ import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.DependsOn; import org.springframework.core.convert.converter.Converter; import org.springframework.expression.PropertyAccessor; +import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.config.IntegrationEvaluationContextFactoryBean; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.json.JsonPropertyAccessor; @@ -69,6 +73,8 @@ import org.springframework.tuple.spel.TuplePropertyAccessor; @EnableConfigurationProperties(ChannelBindingServiceProperties.class) public class ChannelBindingServiceConfiguration { + private static final String ERROR_CHANNEL_NAME = "error"; + @Autowired MessageBuilderFactory messageBuilderFactory; @@ -201,4 +207,11 @@ public class ChannelBindingServiceConfiguration { } } + @Bean + @ConditionalOnProperty("spring.cloud.stream.bindings." + ERROR_CHANNEL_NAME + ".destination") + public BindableErrorChannel bindableErrorChannel( + @Qualifier(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME) PublishSubscribeChannel errorChannel) { + return new BindableErrorChannel(ERROR_CHANNEL_NAME, errorChannel); + } + } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithBindingTargets.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithBindingTargets.java index f0fdf048d..357a41c78 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithBindingTargets.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithBindingTargets.java @@ -27,6 +27,7 @@ import org.junit.runner.RunWith; import org.mockito.Mockito; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.test.SpringApplicationConfiguration; import org.springframework.cloud.stream.annotation.Bindings; @@ -35,10 +36,13 @@ import org.springframework.cloud.stream.messaging.Source; import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration; import org.springframework.context.annotation.Import; import org.springframework.context.annotation.PropertySource; +import org.springframework.integration.channel.PublishSubscribeChannel; +import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Marius Bogoevici + * @author Ilayaperumal Gopinathan */ @RunWith(SpringJUnit4ClassRunner.class) @SpringApplicationConfiguration(SourceBindingTestsWithBindingTargets.TestSource.class) @@ -51,10 +55,16 @@ public class SourceBindingTestsWithBindingTargets { @Autowired @Bindings(TestSource.class) private Source testSource; + @Autowired + @Qualifier(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME) + private PublishSubscribeChannel errorChannel; + @SuppressWarnings("unchecked") @Test public void testSourceOutputChannelBound() { verify(binder).bindProducer(eq("testtock"), eq(testSource.output()), Mockito.any()); + //Check error channel binding + verify(binder).bindProducer(eq("error-test"), eq(errorChannel), Mockito.any()); verifyNoMoreInteractions(binder); } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java index 895549f67..b45ca8ef0 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java @@ -27,7 +27,6 @@ import org.junit.Assert; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.ArgumentCaptor; -import org.mockito.ArgumentMatcher; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; @@ -77,14 +76,4 @@ public class PartitionedProducerTest { } - class PropertiesArgumentMatcher extends ArgumentMatcher { - @Override - public boolean matches(Object argument) { - if (!(argument instanceof Properties)) { - return false; - } - return true; - } - } - } diff --git a/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/sink-binding-test.properties b/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/sink-binding-test.properties index 4c11271fe..9d96ac395 100644 --- a/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/sink-binding-test.properties +++ b/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/sink-binding-test.properties @@ -1 +1 @@ -spring.cloud.stream.bindings.input=testtock +spring.cloud.stream.bindings.input.destination=testtock diff --git a/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/source-binding-test.properties b/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/source-binding-test.properties index bb70f4f1c..2238eca4e 100644 --- a/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/source-binding-test.properties +++ b/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/source-binding-test.properties @@ -1 +1,2 @@ -spring.cloud.stream.bindings.output=testtock +spring.cloud.stream.bindings.output.destination=testtock +spring.cloud.stream.bindings.error.destination=error-test