From bb1b7fd81f5c2373ec2b083027a92737a2af1d7a Mon Sep 17 00:00:00 2001
From: Oleg Zhurakousky <ozhurakousky@pivotal.io>
Date: Wed, 21 Oct 2020 11:46:18 +0200
Subject: [PATCH] GH-2021 Add bridge to SI created error channel if not
 PublisheSubscribe

When @ServiceActivator is used to configure error handler to a binding specific error channel, if such channel is not created manually as pubsub, SI will create it as DirctChannel. This fix ensures that in such case a new PubSub channel is created and bridged to an SI created channel to ensure there can always be multiple subscribers to error channel

Resolves #2021
---
 .../binder/AbstractMessageChannelBinder.java  | 19 ++++++++++++-------
 1 file changed, 12 insertions(+), 7 deletions(-)

diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
index 5dcd9b518..52eee73f3 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
@@ -47,6 +47,7 @@ import org.springframework.context.support.GenericApplicationContext;
 import org.springframework.expression.Expression;
 import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.integration.channel.AbstractSubscribableChannel;
+import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.PublishSubscribeChannel;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.core.MessageProducer;
@@ -585,20 +586,24 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 			ProducerDestination destination) {
 
 		String errorChannelName = errorsBaseName(destination);
-		SubscribableChannel errorChannel;
+		SubscribableChannel errorChannel = new PublishSubscribeChannel();
+
 		if (getApplicationContext().containsBean(errorChannelName)) {
 			Object errorChannelObject = getApplicationContext().getBean(errorChannelName);
 			if (!(errorChannelObject instanceof SubscribableChannel)) {
 				throw new IllegalStateException("Error channel '" + errorChannelName
 						+ "' must be a SubscribableChannel");
 			}
-			errorChannel = (SubscribableChannel) errorChannelObject;
-		}
-		else {
-			errorChannel = new PublishSubscribeChannel();
-			((GenericApplicationContext) getApplicationContext()).registerBean(
-					errorChannelName, SubscribableChannel.class, () -> errorChannel);
+			if (errorChannelObject instanceof DirectChannel) {
+				errorChannelName = "bridged." + errorChannelName;
+				BridgeHandler bridge = new BridgeHandler();
+				bridge.setOutputChannel((MessageChannel) errorChannelObject);
+				errorChannel.subscribe(bridge);
+			}
 		}
+
+		((GenericApplicationContext) getApplicationContext()).registerBean(
+				errorChannelName, SubscribableChannel.class, () -> errorChannel);
 		MessageChannel defaultErrorChannel = null;
 		if (getApplicationContext()
 				.containsBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)) {