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

GH-2218 Fix support for multiple consumers to the same destination

Browse Source 
This fix implies that a single error infristructure will be shared by such consumers

Resolves #2218
This commit is contained in:
Oleg Zhurakousky
2021-08-28 09:51:38 +02:00
parent 9a8ff1521e
commit d462a81559
1 changed files with 16 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -684,8 +684,11 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
String recovererBeanName = getErrorRecovererName(destination, group,
consumerProperties);
((GenericApplicationContext) getApplicationContext()).registerBean(
recovererBeanName, ErrorMessageSendingRecoverer.class, () -> recoverer);
if (!getApplicationContext().containsBean(recovererBeanName)) {
((GenericApplicationContext) getApplicationContext()).registerBean(
recovererBeanName, ErrorMessageSendingRecoverer.class, () -> recoverer);
}
MessageHandler handler;
if (polled) {
handler = getPolledConsumerErrorMessageHandler(destination, group,
@@ -711,11 +714,13 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
if (handler != null) {
if (this.isSubscribable(errorChannel)) {
MessageHandler errorHandler = handler;
((GenericApplicationContext) getApplicationContext()).registerBean(
errorMessageHandlerName, MessageHandler.class,
() -> errorHandler);
errorChannel.subscribe(handler);
if (!getApplicationContext().containsBean(errorMessageHandlerName)) {
MessageHandler errorHandler = handler;
((GenericApplicationContext) getApplicationContext()).registerBean(
errorMessageHandlerName, MessageHandler.class,
() -> errorHandler);
errorChannel.subscribe(handler);
}
}
else {
this.logger.warn("The provided errorChannel '" + errorChannelName
@@ -734,8 +739,10 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
String errorBridgeHandlerName = getErrorBridgeName(destination, group,
consumerProperties);
((GenericApplicationContext) getApplicationContext()).registerBean(
errorBridgeHandlerName, BridgeHandler.class, () -> errorBridge);
if (getApplicationContext().containsBean(errorBridgeHandlerName)) {
((GenericApplicationContext) getApplicationContext()).registerBean(
errorBridgeHandlerName, BridgeHandler.class, () -> errorBridge);
}
}
else {
this.logger.warn("The provided errorChannel '" + errorChannelName