From d4b5431a67ad73a457757856054d79fcb60548ef Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 23 Oct 2018 20:06:39 +0200 Subject: [PATCH] GH-1510 Fixed multiple subscribers to error channel 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 --- .../binder/AbstractMessageChannelBinder.java | 45 ++++++++--- .../stream/binder/ErrorBindingTests.java | 79 ++++++++++++++++++- 2 files changed, 112 insertions(+), 12 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 4b7744f4b..e6fad5acc 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 @@ -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 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(); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ErrorBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ErrorBindingTests.java index bd872f87e..1f46a1781 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ErrorBindingTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ErrorBindingTests.java @@ -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("Hello".getBytes())); + source.send(new GenericMessage("Hello".getBytes())); + source.send(new GenericMessage("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("Hello".getBytes())); + source.send(new GenericMessage("Hello".getBytes())); + source.send(new GenericMessage("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++; + } + } }