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

GH-2997 Fix support for producer's error-handler-definition

Browse Source 
Resolves #2997
This commit is contained in:
Oleg Zhurakousky
2024-09-24 17:18:48 +02:00
parent add6b0178e
commit 8fc89bc4c0
3 changed files with 16 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestContainer.java
View File

@@ -25,7 +25,7 @@ import org.testcontainers.containers.RabbitMQContainer;
*
* @author Chris Bono
*/
public class RabbitTestContainer {
public final class RabbitTestContainer {
private static final RabbitMQContainer RABBITMQ;
static {
@@ -42,6 +42,9 @@ public class RabbitTestContainer {
RABBITMQ.start();
}
private RabbitTestContainer() {
}
/**
* Should be called early by test that wants to ensure a shared {@link RabbitMQContainer} is up and running.
*/

6
core/spring-cloud-stream-test-binder/src/test/java/org/springframework/cloud/stream/binder/test/TestChannelBinderTests.java
View File

@@ -43,7 +43,7 @@ class TestChannelBinderTests {
"--spring.cloud.stream.bindings.function-in-0.destination=input")) {
TestChannelBinder binder = context.getBean(TestChannelBinder.class);
Method registerErrorInfrastructure = ReflectionUtils
.findMethod(TestChannelBinder.class, "registerErrorInfrastructure", ProducerDestination.class, String.class);
.findMethod(TestChannelBinder.class, "registerErrorInfrastructure", ProducerDestination.class, String.class, boolean.class);
registerErrorInfrastructure.setAccessible(true);
ProducerDestination destination = new ProducerDestination() {
@Override
@@ -56,7 +56,7 @@ class TestChannelBinderTests {
return "sample";
}
};
registerErrorInfrastructure.invoke(binder, destination, "function-in-0");
registerErrorInfrastructure.invoke(binder, destination, "function-in-0", false);
destination = new ProducerDestination() {
@Override
public String getNameForPartition(int partition) {
@@ -68,7 +68,7 @@ class TestChannelBinderTests {
return "sample";
}
};
registerErrorInfrastructure.invoke(binder, destination, "function-in-0");
registerErrorInfrastructure.invoke(binder, destination, "function-in-0", false);
}
}

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

@@ -305,8 +305,13 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
bp = bsp.getBindingProperties(bindingName);
}
SubscribableChannel errorChannel = (bp != null && StringUtils.hasText(bp.getErrorHandlerDefinition())) || producerProperties.isErrorChannelEnabled()
? registerErrorInfrastructure(producerDestination, producerProperties.getBindingName()) : null;
boolean errorHandlerDefined = bp != null && StringUtils.hasText(bp.getErrorHandlerDefinition());
SubscribableChannel errorChannel = errorHandlerDefined || producerProperties.isErrorChannelEnabled()
? registerErrorInfrastructure(producerDestination, producerProperties.getBindingName(), errorHandlerDefined)
: null;
String errorChannelName = errorsBaseName(producerDestination, producerProperties.getBindingName());
this.subscribeFunctionErrorHandler(errorChannelName, producerProperties.getBindingName());
producerMessageHandler = createProducerMessageHandler(producerDestination,
producerProperties, outputChannel, errorChannel);
@@ -727,7 +732,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
* @return the channel.
*/
private SubscribableChannel registerErrorInfrastructure(
ProducerDestination destination, String bindingName) {
ProducerDestination destination, String bindingName, boolean errorHandlerDefinitionAvailable) {
String errorChannelName = errorsBaseName(destination, bindingName);
SubscribableChannel errorChannel = new PublishSubscribeChannel();
@@ -751,7 +756,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
}
MessageChannel defaultErrorChannel = null;
if (getApplicationContext()
if (!errorHandlerDefinitionAvailable && getApplicationContext()
.containsBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)) {
defaultErrorChannel = getApplicationContext().getBean(
IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME,