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

GH-1510 Fixed multiple subscribers to error channel

Browse Source 
Ensured that in the event the binding specific error channel is an instance of the DirectChannel, no more subscribers are addeed after the initial one
Added tests

Resolves #1510
Resolves #1514
This commit is contained in:
Oleg Zhurakousky
2018-10-23 20:06:39 +02:00
parent 28990fe995
commit d4b5431a67
2 changed files with 112 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -44,6 +44,7 @@ import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.channel.AbstractSubscribableChannel;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.MessageChannelReactiveUtils;
import org.springframework.integration.channel.PublishSubscribeChannel;
@@ -607,22 +608,46 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
handler = getDefaultErrorMessageHandler((LastSubscriberAwareChannel) errorChannel, defaultErrorChannel != null);
}
String errorMessageHandlerName = getErrorMessageHandlerName(destination, group, consumerProperties);
if (handler != null) {
this.registerComponentWithBeanFactory(errorMessageHandlerName, handler);
beanFactory.initializeBean(handler, errorMessageHandlerName);
errorChannel.subscribe(handler);
if (this.isSubscribable(errorChannel)) {
this.registerComponentWithBeanFactory(errorMessageHandlerName, handler);
beanFactory.initializeBean(handler, errorMessageHandlerName);
errorChannel.subscribe(handler);
}
else {
logger.warn("The provided errorChannel '" + errorChannelName + "' is an instance of DirectChannel, "
+ "so no more subscribers could be added which may affect DLQ processing. Resolution: Configure your own errorChannel as "
+ "an instance of PublishSubscribeChannel");
}
}
if (defaultErrorChannel != null) {
BridgeHandler errorBridge = new BridgeHandler();
errorBridge.setOutputChannel(defaultErrorChannel);
errorChannel.subscribe(errorBridge);
String errorBridgeHandlerName = getErrorBridgeName(destination, group, consumerProperties);
this.registerComponentWithBeanFactory(errorBridgeHandlerName, errorBridge);
beanFactory.initializeBean(errorBridge, errorBridgeHandlerName);
if (this.isSubscribable(errorChannel)) {
BridgeHandler errorBridge = new BridgeHandler();
errorBridge.setOutputChannel(defaultErrorChannel);
errorChannel.subscribe(errorBridge);
String errorBridgeHandlerName = getErrorBridgeName(destination, group, consumerProperties);
this.registerComponentWithBeanFactory(errorBridgeHandlerName, errorBridge);
beanFactory.initializeBean(errorBridge, errorBridgeHandlerName);
}
else {
logger.warn("The provided errorChannel '" + errorChannelName + "' is an instance of DirectChannel, "
+ "so no more subscribers could be added and no error messages will be sent to global error channel. Resolution: Configure your own errorChannel as "
+ "an instance of PublishSubscribeChannel");
}
}
return new ErrorInfrastructure(errorChannel, recoverer, handler);
}
private boolean isSubscribable(SubscribableChannel errorChannel) {
if (errorChannel instanceof PublishSubscribeChannel) {
return true;
}
return errorChannel instanceof AbstractSubscribableChannel
? ((AbstractSubscribableChannel)errorChannel).getSubscriberCount() == 0 : true;
}
private void destroyErrorInfrastructure(ProducerDestination destination) {
String errorChannelName = errorsBaseName(destination);
String errorBridgeHandlerName = getErrorBridgeName(destination);
@@ -785,7 +810,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
private SubscribableChannel postProcessOutboundChannelForFunction(MessageChannel outputChannel) {
if (this.integrationFlowFunctionSupport != null) {
Publisher publisher = MessageChannelReactiveUtils.toPublisher(outputChannel);
Publisher<?> publisher = MessageChannelReactiveUtils.toPublisher(outputChannel);
// If the app has an explicit Supplier bean defined, make that as the publisher
if (this.integrationFlowFunctionSupport.containsFunction(Supplier.class)) {
IntegrationFlowBuilder integrationFlowBuilder = IntegrationFlows.from(outputChannel).bridge();

79
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ErrorBindingTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -17,21 +17,35 @@
package org.springframework.cloud.stream.binder;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.binder.test.InputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.GenericMessage;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
/**
* @author Marius Bogoevici
* @author Oleg Zhurakousky
*/
public class ErrorBindingTests {
@@ -39,7 +53,7 @@ public class ErrorBindingTests {
@Test
public void testErrorChannelNotBoundByDefault() {
ConfigurableApplicationContext applicationContext = SpringApplication.run(TestProcessor.class,
"--server.port=0", "--spring.cloud.stream.default-binder=mock");
"--server.port=0", "--spring.cloud.stream.default-binder=mock", "--spring.jmx.enabled=false");
BinderFactory binderFactory = applicationContext.getBean(BinderFactory.class);
Binder binder = binderFactory.getBinder(null, MessageChannel.class);
@@ -51,9 +65,70 @@ public class ErrorBindingTests {
applicationContext.close();
}
@Test
public void testConfigurationWithDefaultErrorHandler() {
ApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
.getCompleteConfiguration(ErrorBindingTests.ErrorConfigurationDefault.class))
.web(WebApplicationType.NONE).run("--spring.cloud.stream.bindings.input.consumer.max-attempts=1", "--spring.jmx.enabled=false");
InputDestination source = context.getBean(InputDestination.class);
source.send(new GenericMessage<byte[]>("Hello".getBytes()));
source.send(new GenericMessage<byte[]>("Hello".getBytes()));
source.send(new GenericMessage<byte[]>("Hello".getBytes()));
ErrorConfigurationDefault errorConfiguration = context.getBean(ErrorConfigurationDefault.class);
assertThat(errorConfiguration.counter == 3);
}
@Test
public void testConfigurationWithCustomErrorHandler() {
ApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
.getCompleteConfiguration(ErrorBindingTests.ErrorConfigurationWithCustomErrorHandler.class))
.web(WebApplicationType.NONE).run("--spring.cloud.stream.bindings.input.consumer.max-attempts=1", "--spring.jmx.enabled=false");
InputDestination source = context.getBean(InputDestination.class);
source.send(new GenericMessage<byte[]>("Hello".getBytes()));
source.send(new GenericMessage<byte[]>("Hello".getBytes()));
source.send(new GenericMessage<byte[]>("Hello".getBytes()));
ErrorConfigurationWithCustomErrorHandler errorConfiguration = context.getBean(ErrorConfigurationWithCustomErrorHandler.class);
assertThat(errorConfiguration.counter == 6);
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestProcessor {
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ErrorConfigurationDefault {
private int counter;
@StreamListener(Sink.INPUT)
public void handle(Object value) {
counter++;
throw new RuntimeException("BOOM!");
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ErrorConfigurationWithCustomErrorHandler {
private int counter;
@StreamListener(Sink.INPUT)
public void handle(Object value) {
counter++;
throw new RuntimeException("BOOM!");
}
@ServiceActivator(inputChannel = "input.anonymous.errors")
public void error(Message<?> message) {
counter++;
}
}
}