diff --git a/spring-cloud-stream/pom.xml b/spring-cloud-stream/pom.xml
index 88bb1a633..efb74c64b 100644
--- a/spring-cloud-stream/pom.xml
+++ b/spring-cloud-stream/pom.xml
@@ -55,6 +55,11 @@
spring-boot-configuration-processor
true
+
+ org.springframework.integration
+ spring-integration-test
+ test
+
org.springframework.boot
spring-boot-starter-test
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 1b06a0409..82a8e60e7 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
@@ -1,5 +1,5 @@
/*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-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.
@@ -16,8 +16,11 @@
package org.springframework.cloud.stream.binder;
+
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.beans.factory.support.DefaultSingletonBeanRegistry;
import org.springframework.cloud.stream.provisioning.ConsumerDestination;
import org.springframework.cloud.stream.provisioning.ProducerDestination;
import org.springframework.cloud.stream.provisioning.ProvisioningException;
@@ -26,10 +29,14 @@ import org.springframework.context.Lifecycle;
import org.springframework.expression.ExpressionParser;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.channel.FixedSubscriberChannel;
+import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.integration.handler.BridgeHandler;
+import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer;
+import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
@@ -240,8 +247,10 @@ public abstract class AbstractMessageChannelBinder message) throws MessagingException {
+ if (this.errorChannel.subscribers() > (this.defaultErrorChannelPresent ? 2 : 1)) {
+ // user has subscribed; default is 2, this and the bridge to the
+ // errorChannel
+ return;
+ }
+ if (message.getPayload() instanceof MessagingException) {
+ throw (MessagingException) message.getPayload();
+ }
+ else {
+ throw new MessagingException((Message>) null, (Throwable) message.getPayload());
+ }
+ }
+
+}
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/LastSubscriberAwareChannel.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/LastSubscriberAwareChannel.java
new file mode 100644
index 000000000..b0d818f2c
--- /dev/null
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/LastSubscriberAwareChannel.java
@@ -0,0 +1,37 @@
+/*
+ * 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.binder;
+
+import org.springframework.messaging.SubscribableChannel;
+
+/**
+ * A channel that can ensure a {@code LastSubscriberMessageHandler} is always
+ * the last subscriber.
+ *
+ * @author Gary Russell
+ * @since 1.3
+ *
+ */
+interface LastSubscriberAwareChannel extends SubscribableChannel {
+
+ /**
+ * Return the current subscribers.
+ * @return the subscribers.
+ */
+ int subscribers();
+
+}
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/LastSubscriberMessageHandler.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/LastSubscriberMessageHandler.java
new file mode 100644
index 000000000..0579a8a2e
--- /dev/null
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/LastSubscriberMessageHandler.java
@@ -0,0 +1,30 @@
+/*
+ * 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.binder;
+
+import org.springframework.messaging.MessageHandler;
+
+/**
+ * A marker interface designating a subscriber that must be the last.
+ *
+ * @author Gary Russell
+ * @since 1.3
+ *
+ */
+public interface LastSubscriberMessageHandler extends MessageHandler {
+
+}
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 f68ce61b1..a71f28da2 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
@@ -16,24 +16,41 @@
package org.springframework.cloud.stream.binder;
+import java.util.Iterator;
+import java.util.Set;
+
import org.junit.Test;
+import org.mockito.Matchers;
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.cloud.stream.provisioning.ProvisioningProvider;
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;
import org.springframework.messaging.MessageChannel;
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.anyString;
+import static org.mockito.Mockito.mock;
/**
* @author Marius Bogoevici
+ * @author Gary Russell
* @since 1.2.2
*/
public class AbstractMessageChannelBinderTests {
@@ -41,7 +58,10 @@ public class AbstractMessageChannelBinderTests {
@Test
public void testEndpointLifecycle() throws Exception {
StubMessageChannelBinder binder = new StubMessageChannelBinder();
- binder.setApplicationContext(new GenericApplicationContext());
+ GenericApplicationContext context = new GenericApplicationContext();
+ context.refresh();
+ context.getBeanFactory().registerSingleton("errorChannel", new PublishSubscribeChannel());
+ binder.setApplicationContext(context);
Binding consumerBinding = binder.bindConsumer("foo", "fooGroup", new DirectChannel(),
new ConsumerProperties());
@@ -51,7 +71,25 @@ public class AbstractMessageChannelBinderTests {
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();
+ assertThat(errorChannel).isNotNull();
+ @SuppressWarnings("unchecked")
+ 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()).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();
+ assertThat(context.containsBean("foo.fooGroup.errors.bridge")).isTrue();
consumerBinding.unbind();
+ assertThat(context.containsBean("foo.fooGroup.errors")).isFalse();
+ assertThat(context.containsBean("foo.fooGroup.errors.recoverer")).isFalse();
+ 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);
@@ -69,12 +107,65 @@ public class AbstractMessageChannelBinderTests {
Mockito.verifyNoMoreInteractions(messageHandler);
}
+ @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);
+
+ Binding consumerBinding = binder.bindConsumer("foo", "fooGroup", new DirectChannel(),
+ new ConsumerProperties());
+ ErrorInfrastructure errorInfra = binder.errorInfrastructure;
+ SubscribableChannel errorChannel = errorInfra.getErrorChannel();
+ assertThat(errorChannel).isNotNull();
+ @SuppressWarnings("unchecked")
+ 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(context.containsBean("foo.fooGroup.errors")).isTrue();
+ assertThat(context.containsBean("foo.fooGroup.errors.recoverer")).isTrue();
+ assertThat(context.containsBean("foo.fooGroup.errors.handler")).isTrue();
+ assertThat(context.containsBean("foo.fooGroup.errors.bridge")).isTrue();
+ consumerBinding.unbind();
+ assertThat(context.containsBean("foo.fooGroup.errors")).isFalse();
+ assertThat(context.containsBean("foo.fooGroup.errors.recoverer")).isFalse();
+ assertThat(context.containsBean("foo.fooGroup.errors.handler")).isFalse();
+ assertThat(context.containsBean("foo.fooGroup.errors.bridge")).isFalse();
+ }
+
private static class StubMessageChannelBinder extends
- AbstractMessageChannelBinder> {
+ AbstractMessageChannelBinder> {
+
+ private final boolean hasRecoverer;
+
+ private ErrorInfrastructure errorInfrastructure;
+
+ StubMessageChannelBinder() {
+ this(false);
+ }
@SuppressWarnings("unchecked")
- private StubMessageChannelBinder() {
+ StubMessageChannelBinder(boolean hasRecoverer) {
super(true, 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.getArgumentAt(0, String.class));
+ }
+
+ }).given(this.provisioningProvider).provisionConsumerDestination(anyString(), anyString(),
+ Matchers.any(ConsumerProperties.class));
}
@Override
@@ -88,10 +179,39 @@ public class AbstractMessageChannelBinderTests {
@Override
protected MessageProducer createConsumerEndpoint(ConsumerDestination destination, String group,
ConsumerProperties properties) throws Exception {
- return Mockito.mock(MessageProducer.class,
+ 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;
+ }
+
}
}