SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2021 Add bridge to SI created error channel if not PublisheSubscribe

Browse Source 
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
This commit is contained in:
Oleg Zhurakousky
2020-10-21 11:46:18 +02:00
parent d23cb9b1fb
commit bb1b7fd81f
1 changed files with 12 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
View File

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