GH-2066 Ensure global error channel distributes messages in case of exception

Resolves #2066

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() {

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<String>("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 {
+
+	}
+
 }