diff --git a/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc b/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc index ba4b68714..0f67be74b 100644 --- a/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc +++ b/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc @@ -737,18 +737,6 @@ public void error(Message message) { This may be a convenient option if error handling logic is the same regardless of which handler produced the error. -Also, error messages sent to the `errorChannel` can be published to the specific destination at the broker by configuring a binding named `error` for the outbound target. -This option provides a mechanism to automatically send error messages to another application bound to that destination or for later retrieval (for example, audit). -For example, to publish error messages to a broker destination named `myErrors`, set the following property: - -[source,text] ----- -spring.cloud.stream.bindings.error.destination=myErrors. ----- - -NOTE: The ability to bridge global error channel to a broker destination essentially provides a mechanism which connects -the _application-level_ error handling with the _system-level_ error handling. - ===== System Error Handling System-level error handling implies that the errors are communicated back to the messaging system and, given that not every messaging system 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 deleted file mode 100644 index e0f1f785e..000000000 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/ErrorChannelTests.java +++ /dev/null @@ -1,82 +0,0 @@ -/* - * 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 java.util.concurrent.TimeUnit; - -import org.junit.Test; -import org.junit.runner.RunWith; - -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.context.SpringBootTest; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.binder.BinderFactory; -import org.springframework.cloud.stream.messaging.Source; -import org.springframework.cloud.stream.test.binder.TestSupportBinder; -import org.springframework.context.annotation.Bean; -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.MessageChannel; -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) -@SpringBootTest(classes = ErrorChannelTests.TestSource.class) -public class ErrorChannelTests { - - @Autowired - @Qualifier(BindingServiceConfiguration.ERROR_BRIDGE_CHANNEL) - private MessageChannel errorBridgeChannel; - - @Autowired - private BinderFactory binderFactory; - - @Test - public void testErrorChannelBinding() throws Exception { - Message message = ((TestSupportBinder) binderFactory.getBinder(null, MessageChannel.class)) - .messageCollector().forChannel(errorBridgeChannel).poll(10, TimeUnit.SECONDS); - 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"), "Text did not match"); - } - - @EnableBinding(Source.class) - @EnableAutoConfiguration - @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 () -> { - throw new MessagingException("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 deleted file mode 100644 index d91c5dd27..000000000 --- a/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/errorchannel/source-channel.properties +++ /dev/null @@ -1,2 +0,0 @@ -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/binding/SingleBindingTargetBindable.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SingleBindingTargetBindable.java index e4c7a7736..e71e17f5d 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SingleBindingTargetBindable.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SingleBindingTargetBindable.java @@ -31,7 +31,10 @@ import org.springframework.cloud.stream.binder.Binding; * * @author Ilayaperumal Gopinathan * @author Marius Bogoevici + * + * @deprecated This class is no longer used by the framework and maybe removed in a future release. */ +@Deprecated public class SingleBindingTargetBindable implements Bindable { private final String name; 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 2c741ae29..99e19da3a 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 @@ -24,7 +24,6 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.stream.binder.BinderFactory; import org.springframework.cloud.stream.binding.AbstractBindingTargetFactory; @@ -40,7 +39,6 @@ import org.springframework.cloud.stream.binding.MessageChannelStreamListenerResu import org.springframework.cloud.stream.binding.MessageConverterConfigurer; import org.springframework.cloud.stream.binding.MessageSourceBindingTargetFactory; import org.springframework.cloud.stream.binding.OutputBindingLifecycle; -import org.springframework.cloud.stream.binding.SingleBindingTargetBindable; import org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor; import org.springframework.cloud.stream.binding.SubscribableChannelBindingTargetFactory; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; @@ -52,17 +50,13 @@ import org.springframework.context.annotation.DependsOn; import org.springframework.context.annotation.Import; import org.springframework.context.annotation.Role; import org.springframework.context.event.ContextRefreshedEvent; -import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.config.GlobalChannelInterceptorProcessor; import org.springframework.integration.config.HandlerMethodArgumentResolversHolder; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; -import org.springframework.integration.handler.BridgeHandler; import org.springframework.integration.router.AbstractMappingMessageRouter; import org.springframework.lang.Nullable; import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.core.DestinationResolver; import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory; import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory; @@ -90,11 +84,6 @@ public class BindingServiceConfiguration { public static final String STREAM_LISTENER_ANNOTATION_BEAN_POST_PROCESSOR_NAME = "streamListenerAnnotationBeanPostProcessor"; - public static final String ERROR_BRIDGE_CHANNEL = "errorBridgeChannel"; - - private static final String ERROR_KEY_NAME = "error"; - - @Bean public MessageChannelStreamListenerResultAdapter messageChannelStreamListenerResultAdapter() { return new MessageChannelStreamListenerResultAdapter(); @@ -180,32 +169,11 @@ public class BindingServiceConfiguration { callback, globalChannelInterceptorProcessor); } - @Bean - @ConditionalOnProperty("spring.cloud.stream.bindings." + ERROR_KEY_NAME + ".destination") - public MessageChannel errorBridgeChannel( - @Qualifier(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME) PublishSubscribeChannel errorChannel) { - SubscribableChannel errorBridgeChannel = new DirectChannel(); - BridgeHandler handler = new BridgeHandler(); - handler.setOutputChannel(errorBridgeChannel); - errorChannel.subscribe(handler); - return errorBridgeChannel; - } - - @Bean - @ConditionalOnProperty("spring.cloud.stream.bindings." + ERROR_KEY_NAME + ".destination") - public SingleBindingTargetBindable errorBridgeChannelBindable( - @Qualifier(ERROR_BRIDGE_CHANNEL) MessageChannel errorBridgeChannel, - CompositeMessageChannelConfigurer compositeMessageChannelConfigurer) { - compositeMessageChannelConfigurer.configureOutputChannel(errorBridgeChannel, ERROR_KEY_NAME); - return new SingleBindingTargetBindable<>(ERROR_KEY_NAME, errorBridgeChannel); - } - @Bean public DynamicDestinationsBindable dynamicDestinationsBindable() { return new DynamicDestinationsBindable(); } - @SuppressWarnings("deprecation") @Bean @ConditionalOnMissingBean diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ErrorBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ErrorBindingTests.java index 2440a12bf..d99075413 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ErrorBindingTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ErrorBindingTests.java @@ -16,29 +16,21 @@ package org.springframework.cloud.stream.binder; -import java.util.concurrent.atomic.AtomicBoolean; - import org.junit.Test; import org.mockito.Mockito; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.config.BindingServiceConfiguration; import org.springframework.cloud.stream.messaging.Processor; import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Import; -import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.SubscribableChannel; -import org.springframework.messaging.support.GenericMessage; -import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.isNull; -import static org.mockito.ArgumentMatchers.same; /** * @author Marius Bogoevici @@ -61,95 +53,10 @@ public class ErrorBindingTests { applicationContext.close(); } - @SuppressWarnings({"rawtypes", "unchecked"}) - @Test - public void testErrorChannelBoundIfConfigured() { - ConfigurableApplicationContext applicationContext = SpringApplication.run(TestProcessor.class, - "--spring.cloud.stream.bindings.error.destination=foo", "--server.port=0"); - BinderFactory binderFactory = applicationContext.getBean(BinderFactory.class, MessageChannel.class); - - Binder binder = binderFactory.getBinder(null, MessageChannel.class); - - MessageChannel errorChannel = applicationContext.getBean(BindingServiceConfiguration.ERROR_BRIDGE_CHANNEL, - MessageChannel.class); - - Mockito.verify(binder).bindConsumer(eq("input"), isNull(), any(MessageChannel.class), - any(ConsumerProperties.class)); - Mockito.verify(binder).bindProducer(eq("output"), any(MessageChannel.class), any(ProducerProperties.class)); - Mockito.verify(binder).bindProducer(eq("foo"), same(errorChannel), any(ProducerProperties.class)); - Mockito.verifyNoMoreInteractions(binder); - applicationContext.close(); - } - - @Test - public void testErrorChannelIsBoundWithCorrectContentTypeConverter() { - final AtomicBoolean received = new AtomicBoolean(false); - ConfigurableApplicationContext applicationContext = SpringApplication.run(TestProcessor.class, - "--spring.cloud.stream.bindings.error.destination=foo", - "--spring.cloud.stream.bindings.error.content-type=application/json", - "--server.port=0"); - - MessageChannel errorChannel = applicationContext.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, - MessageChannel.class); - MessageChannel errorBridgeChannel = applicationContext.getBean(BindingServiceConfiguration.ERROR_BRIDGE_CHANNEL, - MessageChannel.class); - - ((SubscribableChannel) errorBridgeChannel).subscribe(message -> { - assertThat(new String((byte[]) message.getPayload())).isEqualTo("{\"foo\":\"bar\"}"); - received.set(true); - }); - - Foo foo = new Foo(); - foo.setFoo("bar"); - - errorChannel.send(new GenericMessage<>(foo)); - assertThat(received.get()).isTrue(); - applicationContext.close(); - } - - @Test - public void testErrorChannelForExceptionWhenContentTypeIsSet() { - final AtomicBoolean received = new AtomicBoolean(false); - ConfigurableApplicationContext applicationContext = SpringApplication.run(TestProcessor.class, - "--spring.cloud.stream.bindings.error.destination=foo", - "--spring.cloud.stream.bindings.error.content-type=application/json", - "--server.port=0"); - - MessageChannel errorChannel = applicationContext.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, - MessageChannel.class); - MessageChannel errorBridgeChannel = applicationContext.getBean(BindingServiceConfiguration.ERROR_BRIDGE_CHANNEL, - MessageChannel.class); - - ((SubscribableChannel) errorBridgeChannel).subscribe(message -> { - String payload = new String((byte[]) message.getPayload()); - assertThat(payload.contains("cause")).isTrue(); - assertThat(payload.contains("stackTrace")).isTrue(); - assertThat(payload.contains("throwing exception")).isTrue(); - received.set(true); - }); - - errorChannel.send(new GenericMessage<>(new Exception("throwing exception"))); - assertThat(received.get()).isTrue(); - applicationContext.close(); - } - @EnableBinding(Processor.class) @EnableAutoConfiguration @Import(MockBinderRegistryConfiguration.class) public static class TestProcessor { } - - private class Foo { - String foo; - - @SuppressWarnings("unused") // used json ser/deser - public String getFoo() { - return foo; - } - - public void setFoo(String foo) { - this.foo = foo; - } - } }