diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/AbstractBindingLifecycle.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/AbstractBindingLifecycle.java new file mode 100644 index 000000000..684c831bf --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/AbstractBindingLifecycle.java @@ -0,0 +1,81 @@ +/* + * Copyright 2017 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binding; + +import java.util.Map; + +import org.springframework.context.SmartLifecycle; + +/** + * Base implementation of lifecycle operations for {@link BindingService} + * aware {@link Bindable}s. + * + * @see InputBindingLifecycle + * @see OutputBindingLifecycle + * + * @author Oleg Zhurakousky + */ +abstract class AbstractBindingLifecycle implements SmartLifecycle { + final BindingService bindingService; + + private final Map bindables; + + private volatile boolean running; + + AbstractBindingLifecycle(BindingService bindingService, Map bindables) { + this.bindingService = bindingService; + this.bindables = bindables; + } + + @Override + public void start() { + if (!running) { + bindables.values().stream().forEach(this::doStartWithBindable); + this.running = true; + } + } + + @Override + public void stop() { + if (running) { + bindables.values().stream().forEach(this::doStopWithBindable); + this.running = false; + } + } + + @Override + public boolean isRunning() { + return running; + } + + @Override + public boolean isAutoStartup() { + return true; + } + + @Override + public void stop(Runnable callback) { + stop(); + if (callback != null) { + callback.run(); + } + } + + abstract void doStartWithBindable(Bindable bindable); + + abstract void doStopWithBindable(Bindable bindable); +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/InputBindingLifecycle.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/InputBindingLifecycle.java index de3821eb3..164397bab 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/InputBindingLifecycle.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/InputBindingLifecycle.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-2017 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. @@ -18,89 +18,17 @@ package org.springframework.cloud.stream.binding; import java.util.Map; -import org.springframework.beans.BeansException; -import org.springframework.context.ApplicationContext; -import org.springframework.context.ApplicationContextAware; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.context.SmartLifecycle; - /** * Coordinates binding/unbinding of input binding targets in accordance to the lifecycle * of the host context. * @author Marius Bogoevici * @author Ilayaperumal Gopinathan + * @author Oleg Zhurakousky */ -public class InputBindingLifecycle implements SmartLifecycle, ApplicationContextAware { +public class InputBindingLifecycle extends AbstractBindingLifecycle { - private volatile boolean running; - - private ConfigurableApplicationContext applicationContext; - - @Override - public void setApplicationContext(ApplicationContext applicationContext) - throws BeansException { - this.applicationContext = (ConfigurableApplicationContext) applicationContext; - } - - @Override - public void start() { - if (!running) { - // retrieve the BindingService lazily, avoiding early initialization - try { - BindingService bindingService = this.applicationContext - .getBean(BindingService.class); - Map bindables = this.applicationContext - .getBeansOfType(Bindable.class); - for (Bindable bindable : bindables.values()) { - bindable.bindInputs(bindingService); - } - } - catch (BeansException e) { - throw new IllegalStateException( - "Cannot perform binding, no proper implementation found", e); - } - this.running = true; - } - } - - @Override - public void stop() { - if (running) { - try { - // retrieve the BindingService lazily, avoiding early - // initialization - BindingService bindingService = this.applicationContext - .getBean(BindingService.class); - Map bindables = this.applicationContext - .getBeansOfType(Bindable.class); - for (Bindable bindable : bindables.values()) { - bindable.unbindInputs(bindingService); - } - } - catch (BeansException e) { - throw new IllegalStateException( - "Cannot perform unbinding, no proper implementation found", e); - } - this.running = false; - } - } - - @Override - public boolean isRunning() { - return running; - } - - @Override - public boolean isAutoStartup() { - return true; - } - - @Override - public void stop(Runnable callback) { - stop(); - if (callback != null) { - callback.run(); - } + public InputBindingLifecycle(BindingService bindingService, Map bindables) { + super(bindingService, bindables); } /** @@ -111,4 +39,14 @@ public class InputBindingLifecycle implements SmartLifecycle, ApplicationContext public int getPhase() { return Integer.MAX_VALUE - 1000; } + + @Override + void doStartWithBindable(Bindable bindable) { + bindable.bindInputs(bindingService); + } + + @Override + void doStopWithBindable(Bindable bindable) { + bindable.unbindInputs(bindingService); + } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/OutputBindingLifecycle.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/OutputBindingLifecycle.java index 4f897ba97..d4a2c6037 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/OutputBindingLifecycle.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/OutputBindingLifecycle.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2017 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. @@ -18,91 +18,18 @@ package org.springframework.cloud.stream.binding; import java.util.Map; -import org.springframework.beans.BeansException; -import org.springframework.context.ApplicationContext; -import org.springframework.context.ApplicationContextAware; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.context.SmartLifecycle; - /** * Coordinates binding/unbinding of output binding targets in accordance to the lifecycle * of the host context. * * @author Marius Bogoevici * @author Ilayaperumal Gopinathan + * @author Oleg Zhurakousky */ -public class OutputBindingLifecycle implements SmartLifecycle, ApplicationContextAware { +public class OutputBindingLifecycle extends AbstractBindingLifecycle { - private volatile boolean running; - - private ConfigurableApplicationContext applicationContext; - - @Override - public void setApplicationContext(ApplicationContext applicationContext) - throws BeansException { - this.applicationContext = (ConfigurableApplicationContext) applicationContext; - } - - @Override - public void start() { - if (!running) { - - // retrieve the BindingService lazily, avoiding early initialization - try { - BindingService bindingService = this.applicationContext - .getBean(BindingService.class); - Map bindables = this.applicationContext - .getBeansOfType(Bindable.class); - for (Bindable bindable : bindables.values()) { - bindable.bindOutputs(bindingService); - } - } - catch (BeansException e) { - throw new IllegalStateException( - "Cannot perform binding, no proper implementation found", e); - } - this.running = true; - } - } - - @Override - public void stop() { - if (running) { - try { - // retrieve the BindingService lazily, avoiding early - // initialization - BindingService bindingService = this.applicationContext - .getBean(BindingService.class); - Map bindables = this.applicationContext - .getBeansOfType(Bindable.class); - for (Bindable bindable : bindables.values()) { - bindable.unbindOutputs(bindingService); - } - } - catch (BeansException e) { - throw new IllegalStateException( - "Cannot perform unbinding, no proper implementation found", e); - } - this.running = false; - } - } - - @Override - public boolean isRunning() { - return running; - } - - @Override - public boolean isAutoStartup() { - return true; - } - - @Override - public void stop(Runnable callback) { - stop(); - if (callback != null) { - callback.run(); - } + public OutputBindingLifecycle(BindingService bindingService, Map bindables) { + super(bindingService, bindables); } /** @@ -113,4 +40,14 @@ public class OutputBindingLifecycle implements SmartLifecycle, ApplicationContex public int getPhase() { return Integer.MIN_VALUE + 1000; } + + @Override + void doStartWithBindable(Bindable bindable) { + bindable.bindOutputs(bindingService); + } + + @Override + void doStopWithBindable(Bindable bindable) { + bindable.unbindOutputs(bindingService); + } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java index 173764f3c..721179ff8 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java @@ -32,6 +32,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.stream.binder.BinderFactory; import org.springframework.cloud.stream.binding.AbstractBindingTargetFactory; +import org.springframework.cloud.stream.binding.Bindable; import org.springframework.cloud.stream.binding.BinderAwareChannelResolver; import org.springframework.cloud.stream.binding.BinderAwareRouterBeanPostProcessor; import org.springframework.cloud.stream.binding.BindingService; @@ -136,14 +137,14 @@ public class BindingServiceConfiguration { @Bean @DependsOn("bindingService") - public OutputBindingLifecycle outputBindingLifecycle() { - return new OutputBindingLifecycle(); + public OutputBindingLifecycle outputBindingLifecycle(BindingService bindingService, Map bindables) { + return new OutputBindingLifecycle(bindingService, bindables); } @Bean @DependsOn("bindingService") - public InputBindingLifecycle inputBindingLifecycle() { - return new InputBindingLifecycle(); + public InputBindingLifecycle inputBindingLifecycle(BindingService bindingService, Map bindables) { + return new InputBindingLifecycle(bindingService, bindables); } @Bean diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinderTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinderTests.java index 5a6114ae6..60234d286 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinderTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinderTests.java @@ -19,22 +19,18 @@ package org.springframework.cloud.stream.binder; import java.util.Iterator; import java.util.Set; +import org.junit.Before; import org.junit.Test; -import org.mockito.Mockito; -import org.mockito.invocation.InvocationOnMock; -import org.mockito.stubbing.Answer; import org.springframework.beans.DirectFieldAccessor; -import org.springframework.beans.factory.DisposableBean; -import org.springframework.beans.factory.InitializingBean; -import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.ErrorInfrastructure; -import org.springframework.cloud.stream.provisioning.ConsumerDestination; -import org.springframework.cloud.stream.provisioning.ProducerDestination; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.stream.binder.integration.SpringIntegrationBinderConfiguration; import org.springframework.cloud.stream.provisioning.ProvisioningProvider; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.Lifecycle; -import org.springframework.context.support.GenericApplicationContext; import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.handler.BridgeHandler; import org.springframework.integration.test.util.TestUtils; @@ -43,36 +39,43 @@ import org.springframework.messaging.MessageHandler; import org.springframework.messaging.SubscribableChannel; import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.BDDMockito.willAnswer; -import static org.mockito.Matchers.any; -import static org.mockito.Matchers.anyString; -import static org.mockito.Mockito.mock; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; /** * @author Marius Bogoevici * @author Gary Russell + * @author Oleg Zhurakousky + * * @since 1.2.2 */ public class AbstractMessageChannelBinderTests { + private ApplicationContext context; + + @Before + public void prepare() { + this.context = new SpringApplicationBuilder(SpringIntegrationBinderConfiguration.getCompleteConfiguration()) + .web(WebApplicationType.NONE) + .run(); + } + @Test public void testEndpointLifecycle() throws Exception { - StubMessageChannelBinder binder = new StubMessageChannelBinder(); - GenericApplicationContext context = new GenericApplicationContext(); - context.refresh(); - context.getBeanFactory().registerSingleton("errorChannel", new PublishSubscribeChannel()); - binder.setApplicationContext(context); + AbstractMessageChannelBinder> binder = + context.getBean(AbstractMessageChannelBinder.class); - Binding consumerBinding = binder.bindConsumer("foo", "fooGroup", new DirectChannel(), - new ConsumerProperties()); + ConsumerProperties consumerProperties = new ConsumerProperties(); + consumerProperties.setMaxAttempts(1); // to force error infrastructure creation + + Binding consumerBinding = binder.bindConsumer("foo", "fooGroup", new DirectChannel(), consumerProperties); DirectFieldAccessor consumerBindingAccessor = new DirectFieldAccessor(consumerBinding); - Object messageProducer = consumerBindingAccessor.getPropertyValue("lifecycle"); - Mockito.verify((Lifecycle) messageProducer).start(); - Mockito.verify((InitializingBean) messageProducer).afterPropertiesSet(); - Mockito.verify((MessageProducer) messageProducer).setOutputChannel(Mockito.any(MessageChannel.class)); - Mockito.verifyNoMoreInteractions(messageProducer); - ErrorInfrastructure errorInfra = binder.errorInfrastructure; - SubscribableChannel errorChannel = errorInfra.getErrorChannel(); + MessageProducer messageProducer = (MessageProducer) consumerBindingAccessor.getPropertyValue("lifecycle"); + assertTrue(((Lifecycle)messageProducer).isRunning()); + assertNotNull(messageProducer.getOutputChannel()); + + SubscribableChannel errorChannel = (SubscribableChannel) consumerBindingAccessor.getPropertyValue("lifecycle.errorChannel"); assertThat(errorChannel).isNotNull(); Set handlers = TestUtils.getPropertyValue(errorChannel, "dispatcher.handlers", Set.class); assertThat(handlers.size()).isEqualTo(2); @@ -89,46 +92,37 @@ public class AbstractMessageChannelBinderTests { assertThat(context.containsBean("foo.fooGroup.errors.handler")).isFalse(); assertThat(context.containsBean("foo.fooGroup.errors.bridge")).isFalse(); - Mockito.verify((Lifecycle) messageProducer).stop(); - Mockito.verify((DisposableBean) messageProducer).destroy(); - Mockito.verifyNoMoreInteractions(messageProducer); + assertFalse(((Lifecycle) messageProducer).isRunning()); ProducerProperties producerProps = new ProducerProperties(); producerProps.setErrorChannelEnabled(true); Binding producerBinding = binder.bindProducer("bar", new DirectChannel(), producerProps); - DirectFieldAccessor producerBindingAccessor = new DirectFieldAccessor(producerBinding); - Object messageHandler = producerBindingAccessor.getPropertyValue("lifecycle"); - Mockito.verify((Lifecycle) messageHandler).start(); - Mockito.verify((InitializingBean) messageHandler).afterPropertiesSet(); - Mockito.verifyNoMoreInteractions(messageHandler); assertThat(context.containsBean("bar.errors")).isTrue(); assertThat(context.containsBean("bar.errors.bridge")).isTrue(); producerBinding.unbind(); - Mockito.verify((Lifecycle) messageHandler).stop(); - Mockito.verify((DisposableBean) messageHandler).destroy(); - Mockito.verifyNoMoreInteractions(messageHandler); assertThat(context.containsBean("bar.errors")).isFalse(); assertThat(context.containsBean("bar.errors.bridge")).isFalse(); } @Test public void testEndpointBinderHasRecoverer() throws Exception { - StubMessageChannelBinder binder = new StubMessageChannelBinder(true); - GenericApplicationContext context = new GenericApplicationContext(); - context.refresh(); - context.getBeanFactory().registerSingleton("errorChannel", new PublishSubscribeChannel()); - binder.setApplicationContext(context); + ConfigurableApplicationContext context = + new SpringApplicationBuilder(SpringIntegrationBinderConfiguration.getCompleteConfiguration()).web(WebApplicationType.NONE).run(); - Binding consumerBinding = binder.bindConsumer("foo", "fooGroup", new DirectChannel(), - new ConsumerProperties()); - ErrorInfrastructure errorInfra = binder.errorInfrastructure; - SubscribableChannel errorChannel = errorInfra.getErrorChannel(); + AbstractMessageChannelBinder> binder = + context.getBean(AbstractMessageChannelBinder.class); + + Binding consumerBinding = binder.bindConsumer("foo", "fooGroup", new DirectChannel(), new ConsumerProperties()); + DirectFieldAccessor consumerBindingAccessor = new DirectFieldAccessor(consumerBinding); + SubscribableChannel errorChannel = (SubscribableChannel) consumerBindingAccessor.getPropertyValue("lifecycle.errorChannel"); + assertThat(errorChannel).isNull(); + errorChannel = (SubscribableChannel) consumerBindingAccessor.getPropertyValue("lifecycle.recoveryCallback.channel"); assertThat(errorChannel).isNotNull(); Set handlers = TestUtils.getPropertyValue(errorChannel, "dispatcher.handlers", Set.class); assertThat(handlers.size()).isEqualTo(2); Iterator iterator = handlers.iterator(); assertThat(iterator.next()).isInstanceOf(BridgeHandler.class); - assertThat(iterator.next()).isNotInstanceOf(LastSubscriberMessageHandler.class); + assertThat(iterator.next()).isInstanceOf(LastSubscriberMessageHandler.class); assertThat(context.containsBean("foo.fooGroup.errors")).isTrue(); assertThat(context.containsBean("foo.fooGroup.errors.recoverer")).isTrue(); assertThat(context.containsBean("foo.fooGroup.errors.handler")).isTrue(); @@ -139,111 +133,4 @@ public class AbstractMessageChannelBinderTests { assertThat(context.containsBean("foo.fooGroup.errors.handler")).isFalse(); assertThat(context.containsBean("foo.fooGroup.errors.bridge")).isFalse(); } - - private static class StubMessageChannelBinder extends - AbstractMessageChannelBinder> { - - private final boolean hasRecoverer; - - private ErrorInfrastructure errorInfrastructure; - - StubMessageChannelBinder() { - this(false); - } - - StubMessageChannelBinder(boolean hasRecoverer) { - super(null, Mockito.mock(ProvisioningProvider.class)); - mockProvisioner(); - this.hasRecoverer = hasRecoverer; - } - - private void mockProvisioner() { - willAnswer(new Answer() { - - @Override - public SimpleConsumerDestination answer(final InvocationOnMock invocation) throws Throwable { - return new SimpleConsumerDestination(invocation.getArgument(0)); - } - - }).given(this.provisioningProvider).provisionConsumerDestination(anyString(), anyString(), - any(ConsumerProperties.class)); - willAnswer(new Answer() { - - @Override - public SimpleProducerDestination answer(final InvocationOnMock invocation) throws Throwable { - return new SimpleProducerDestination(invocation.getArgument(0)); - } - - }).given(this.provisioningProvider).provisionProducerDestination(anyString(), - any(ProducerProperties.class)); - } - - @Override - protected MessageHandler createProducerMessageHandler(ProducerDestination destination, - ProducerProperties producerProperties, MessageChannel errorChannel) throws Exception { - MessageHandler mock = Mockito.mock(MessageHandler.class, Mockito.withSettings() - .extraInterfaces(Lifecycle.class, InitializingBean.class, DisposableBean.class)); - return mock; - } - - @Override - protected MessageProducer createConsumerEndpoint(ConsumerDestination destination, String group, - ConsumerProperties properties) throws Exception { - this.errorInfrastructure = registerErrorInfrastructure(destination, group, properties); - MessageProducer adapter = Mockito.mock(MessageProducer.class, - Mockito.withSettings().extraInterfaces(Lifecycle.class, InitializingBean.class, - DisposableBean.class)); - return adapter; - } - - @Override - protected MessageHandler getErrorMessageHandler(ConsumerDestination destination, String group, - ConsumerProperties consumerProperties) { - if (this.hasRecoverer) { - return mock(MessageHandler.class); - } - else { - return null; - } - } - - } - - private static class SimpleConsumerDestination implements ConsumerDestination { - - private final String name; - - SimpleConsumerDestination(String name) { - this.name = name; - } - - @Override - public String getName() { - return this.name; - } - - } - - private static class SimpleProducerDestination implements ProducerDestination { - - private final String name; - - - SimpleProducerDestination(String name) { - this.name = name; - } - - @Override - public String getName() { - return this.name; - } - - @Override - public String getNameForPartition(int partition) { - return getName() + partition; - } - - } - } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithGlobalPropertiesOnlyTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithGlobalPropertiesOnlyTest.java index 84ffb56cc..fa26a1e84 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithGlobalPropertiesOnlyTest.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithGlobalPropertiesOnlyTest.java @@ -24,11 +24,10 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.binder.integration.SpringIntegrationBinderConfiguration; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.messaging.Source; -import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration; -import org.springframework.context.annotation.Import; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** @@ -36,7 +35,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; * @author Ilayaperumal Gopinathan */ @RunWith(SpringJUnit4ClassRunner.class) -@SpringBootTest(classes = SourceBindingWithGlobalPropertiesOnlyTest.TestSource.class, properties = { +@SpringBootTest(classes = {SpringIntegrationBinderConfiguration.class, SourceBindingWithGlobalPropertiesOnlyTest.TestSource.class}, properties = { "spring.cloud.stream.default.contentType=application/json", "spring.cloud.stream.default.producer.partitionKeyExpression=key" }) public class SourceBindingWithGlobalPropertiesOnlyTest { @@ -54,7 +53,6 @@ public class SourceBindingWithGlobalPropertiesOnlyTest { @EnableBinding(Source.class) @EnableAutoConfiguration - @Import(MockBinderRegistryConfiguration.class) public static class TestSource { } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationChannelBinder.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationChannelBinder.java index aa1d8921b..3f5d565b4 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationChannelBinder.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationChannelBinder.java @@ -127,6 +127,7 @@ class SpringIntegrationChannelBinder extends AbstractMessageChannelBinder 1) { adapter.setRetryTemplate(buildRetryTemplate(properties)); + adapter.setRecoveryCallback(errorInfrastructure.getRecoverer()); } else { adapter.setErrorMessageStrategy(errorMessageStrategy);