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 1b4170425..344d99d22 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 @@ -23,10 +23,12 @@ import java.util.Iterator; import java.util.List; import java.util.Map; +import org.springframework.beans.BeansException; import org.springframework.beans.factory.ObjectProvider; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.config.BeanDefinition; +import org.springframework.beans.factory.config.BeanPostProcessor; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.condition.SearchStrategy; @@ -59,6 +61,7 @@ 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.PublishSubscribeChannel; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.router.AbstractMappingMessageRouter; import org.springframework.lang.Nullable; @@ -157,6 +160,19 @@ public class BindingServiceConfiguration { return binderConfigurations; } + @Bean + public BeanPostProcessor globalErrorChannelCustomizer() { + return new BeanPostProcessor() { + @Override + public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { + if ("errorChannel".equals(beanName)) { + ((PublishSubscribeChannel) bean).setIgnoreFailures(true); + } + return bean; + } + }; + } + @Bean(name = STREAM_LISTENER_ANNOTATION_BEAN_POST_PROCESSOR_NAME) @ConditionalOnMissingBean(search = SearchStrategy.CURRENT) public static StreamListenerAnnotationBeanPostProcessor streamListenerAnnotationBeanPostProcessor() { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BindingServiceConfigurationTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BindingServiceConfigurationTests.java index fe287f6fe..79a4fcb02 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BindingServiceConfigurationTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BindingServiceConfigurationTests.java @@ -17,6 +17,7 @@ package org.springframework.cloud.stream.config; import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; @@ -24,10 +25,13 @@ import org.springframework.boot.WebApplicationType; import org.springframework.boot.builder.SpringApplicationBuilder; import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; import org.springframework.context.ApplicationContext; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.messaging.support.GenericMessage; import static org.assertj.core.api.Assertions.assertThat; @@ -37,6 +41,32 @@ import static org.assertj.core.api.Assertions.assertThat; */ public class BindingServiceConfigurationTests { + @Test + public void testErroChannelDistributesMessagesInCaseOfException() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration + .getCompleteConfiguration(EmptyConfiguration.class)) + .web(WebApplicationType.NONE).run()) { + AtomicInteger counter = new AtomicInteger(); + PublishSubscribeChannel channel = context.getBean("errorChannel", PublishSubscribeChannel.class); + //channel.setIgnoreFailures(true); + channel.subscribe(m -> { + counter.incrementAndGet(); + throw new RuntimeException("one"); + }); + channel.subscribe(m -> { + counter.incrementAndGet(); + throw new RuntimeException("two"); + }); + channel.subscribe(m -> { + counter.incrementAndGet(); + throw new RuntimeException("three"); + }); + channel.send(new GenericMessage("foo")); + assertThat(counter.get()).isEqualTo(3); + } + } + @Test public void valdateImportedConfiguartionHandlerPostProcessing() { ApplicationContext context = new SpringApplicationBuilder( @@ -70,4 +100,9 @@ public class BindingServiceConfigurationTests { } + @Configuration + public static class EmptyConfiguration { + + } + }