Merge pull request #351 from garyrussell/INT-2431

* INT-2431:
  INT-2431 Improve 'Dispatcher Has No Subscribers'

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2012 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.
@@ -31,15 +31,19 @@ import org.springframework.amqp.support.converter.SimpleMessageConverter;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.integration.Message;
+import org.springframework.integration.MessageDeliveryException;
+import org.springframework.integration.MessageDispatchingException;
 import org.springframework.integration.MessagingException;
 import org.springframework.integration.core.MessageHandler;
 import org.springframework.integration.core.SubscribableChannel;
 import org.springframework.integration.dispatcher.MessageDispatcher;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
 
 /**
  * @author Mark Fisher
+ * @author Gary Russell
  * @since 2.1
  */
 abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel implements SubscribableChannel, SmartLifecycle, DisposableBean {
@@ -50,16 +54,23 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple
 
 	private volatile MessageDispatcher dispatcher;
 
+	private final boolean isPubSub;
 
 	public AbstractSubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container, AmqpTemplate amqpTemplate) {
+		this(channelName, container, amqpTemplate, false);
+	}
+
+	public AbstractSubscribableAmqpChannel(String channelName,
+			SimpleMessageListenerContainer container,
+			AmqpTemplate amqpTemplate, boolean isPubSub) {
 		super(amqpTemplate);
 		Assert.notNull(container, "container must not be null");
 		Assert.hasText(channelName, "channel name must not be empty");
 		this.channelName = channelName;
 		this.container = container;
+		this.isPubSub = isPubSub;
 	}
 
-
 	public boolean subscribe(MessageHandler handler) {
 		return this.dispatcher.addHandler(handler);
 	}
@@ -78,7 +89,8 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple
 		MessageConverter converter = (this.getAmqpTemplate() instanceof RabbitTemplate)
 				? ((RabbitTemplate) this.getAmqpTemplate()).getMessageConverter()
 				: new SimpleMessageConverter();
-		MessageListener listener = new DispatchingMessageListener(converter, this.dispatcher);
+		MessageListener listener = new DispatchingMessageListener(converter,
+				this.dispatcher, this.channelName, this.isPubSub);
 		this.container.setMessageListener(listener);
 		if (!this.container.isActive()) {
 			this.container.afterPropertiesSet();
@@ -98,20 +110,27 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple
 
 		private final MessageConverter converter;
 
+		private final String channelName;
 
-		private DispatchingMessageListener(MessageConverter converter, MessageDispatcher dispatcher) {
+		private final boolean isPubSub;
+
+		private DispatchingMessageListener(MessageConverter converter,
+				MessageDispatcher dispatcher, String channelName, boolean isPubSub) {
 			Assert.notNull(converter, "MessageConverter must not be null");
 			Assert.notNull(dispatcher, "MessageDispatcher must not be null");
 			this.converter = converter;
 			this.dispatcher = dispatcher;
+			this.channelName = channelName;
+			this.isPubSub = isPubSub;
 		}
 
 
 		public void onMessage(org.springframework.amqp.core.Message message) {
+			Message<?> messageToSend = null;
 			try {
 				Object converted = this.converter.fromMessage(message);
 				if (converted != null) {
-					Message<?> messageToSend = (converted instanceof Message<?>) ? (Message<?>) converted
+					messageToSend = (converted instanceof Message<?>) ? (Message<?>) converted
 							: MessageBuilder.withPayload(converted).build();
 					this.dispatcher.dispatch(messageToSend);
 				}
@@ -119,6 +138,21 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple
 					logger.warn("MessageConverter returned null, no Message to dispatch");
 				}
 			}
+			catch (MessageDispatchingException e) {
+				String channelName = StringUtils.hasText(this.channelName) ? this.channelName : "unknown";
+				String exceptionMessage = e.getMessage() + " for amqp-channel "
+						+ channelName + ".";
+				if (this.isPubSub) {
+					// log only for backwards compatibility with pub/sub
+					if (logger.isWarnEnabled()) {
+						logger.warn(exceptionMessage, e);
+					}
+				}
+				else {
+					throw new MessageDeliveryException(
+							messageToSend, exceptionMessage, e);
+				}
+			}
 			catch (Exception e) {
 				throw new MessagingException("Failure occured in AMQP listener while attempting to convert and dispatch Message.", e);
 			}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2012 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.
@@ -28,6 +28,7 @@ import org.springframework.integration.dispatcher.MessageDispatcher;
 
 /**
  * @author Mark Fisher
+ * @author Gary Russell
  * @since 2.1
  */
 public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel {
@@ -36,7 +37,7 @@ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel
 
 
 	public PublishSubscribeAmqpChannel(String channelName, SimpleMessageListenerContainer container, AmqpTemplate amqpTemplate) {
-		super(channelName, container, amqpTemplate);
+		super(channelName, container, amqpTemplate, true);
 	}
 
 
@@ -65,7 +66,7 @@ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel
 
 	@Override
 	protected MessageDispatcher createDispatcher() {
-		return new BroadcastingDispatcher();
+		return new BroadcastingDispatcher(true);
 	}
 
 	@Override

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java

@@ -0,0 +1,152 @@
+/*
+ * Copyright 2002-2012 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.integration.amqp.channel;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyBoolean;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.junit.Test;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+import org.springframework.amqp.core.AmqpAdmin;
+import org.springframework.amqp.core.AmqpTemplate;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageListener;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.connection.Connection;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.integration.MessageDeliveryException;
+import org.springframework.integration.test.util.TestUtils;
+
+import com.rabbitmq.client.Channel;
+
+
+/**
+ * @author Gary Russell
+ * @since 2.1
+ *
+ */
+public class DispatcherHasNoSubscribersTests {
+
+	@Test
+	public void testPtP() {
+		final Channel channel = mock(Channel.class);
+		Connection connection = mock(Connection.class);
+		doAnswer(new Answer<Channel>() {
+			public Channel answer(InvocationOnMock invocation) throws Throwable {
+				return channel;
+			}}).when(connection).createChannel(anyBoolean());
+		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
+		when(connectionFactory.createConnection()).thenReturn(connection);
+		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
+		container.setConnectionFactory(connectionFactory);
+		AmqpTemplate amqpTemplate = mock(AmqpTemplate.class);
+
+		PointToPointSubscribableAmqpChannel amqpChannel = new PointToPointSubscribableAmqpChannel("noSubscribersChannel",
+				container, amqpTemplate);
+		amqpChannel.afterPropertiesSet();
+
+		MessageListener listener = (MessageListener) container.getMessageListener();
+		try {
+			listener.onMessage(new Message("Hello world!".getBytes(), null));
+			fail("Exception expected");
+		}
+		catch (MessageDeliveryException e) {
+			assertEquals("Dispatcher has no subscribers for amqp-channel noSubscribersChannel.", e.getMessage());
+		}
+	}
+
+	@Test
+	public void testPubSub() {
+		final Channel channel = mock(Channel.class);
+		Connection connection = mock(Connection.class);
+		doAnswer(new Answer<Channel>() {
+			public Channel answer(InvocationOnMock invocation) throws Throwable {
+				return channel;
+			}}).when(connection).createChannel(anyBoolean());
+		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
+		when(connectionFactory.createConnection()).thenReturn(connection);
+		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
+		container.setConnectionFactory(connectionFactory);
+		AmqpTemplate amqpTemplate = mock(AmqpTemplate.class);
+		final Queue queue = new Queue("noSubscribersQueue");
+		PublishSubscribeAmqpChannel amqpChannel = new PublishSubscribeAmqpChannel("noSubscribersChannel",
+				container, amqpTemplate) { 
+					@Override
+					protected Queue initializeQueue(AmqpAdmin admin,
+							String channelName) {
+						return queue;
+					}};
+		amqpChannel.afterPropertiesSet();
+
+		List<String> logList = insertMockLoggerInListener(amqpChannel);
+		MessageListener listener = (MessageListener) container.getMessageListener();
+		listener.onMessage(new Message("Hello world!".getBytes(), null));
+		verifyLogReceived(logList);
+	}
+
+	private List<String> insertMockLoggerInListener(
+			PublishSubscribeAmqpChannel channel) {
+		SimpleMessageListenerContainer container = TestUtils.getPropertyValue(
+				channel, "container", SimpleMessageListenerContainer.class);
+		Log logger = mock(Log.class);
+		final ArrayList<String> logList = new ArrayList<String>();
+		doAnswer(new Answer<Object>() {
+			public Object answer(InvocationOnMock invocation)
+					throws Throwable {
+				String message = (String) invocation.getArguments()[0];
+				if (message.startsWith("Dispatcher has no subscribers")) {
+					logList.add(message);
+				}
+				return null;
+			}}).when(logger).warn(anyString(), any(Exception.class));
+		when(logger.isWarnEnabled()).thenReturn(true);
+		Object listener = container.getMessageListener();
+		DirectFieldAccessor dfa = new DirectFieldAccessor(listener);
+		dfa.setPropertyValue("logger", logger);
+		return logList;
+	}
+
+	private void verifyLogReceived(final List<String> logList) {
+		assertTrue("Failed to get expected exception", logList.size() > 0);
+		boolean expectedExceptionFound = false;
+		while (logList.size() > 0) {
+			String message = logList.remove(0);
+			assertNotNull("Failed to get expected exception", message);
+			if (message.startsWith("Dispatcher has no subscribers")) {
+				expectedExceptionFound = true;
+				assertEquals("Dispatcher has no subscribers for amqp-channel noSubscribersChannel.", message);
+				break;
+			}
+		}
+		assertTrue("Failed to get expected exception", expectedExceptionFound);
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/MessageDispatchingException.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2002-2012 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.integration;
+
+import org.springframework.integration.dispatcher.MessageDispatcher;
+
+/**
+ * Exception that indicates an internal error occurred within
+ * a {@link MessageDispatcher} preventing message delivery.
+ *
+ * @author Gary Russell
+ * @since 2.1
+ *
+ */
+@SuppressWarnings("serial")
+public class MessageDispatchingException extends MessageDeliveryException {
+
+	public MessageDispatchingException(String description) {
+		super(description);
+	}
+
+	public MessageDispatchingException(Message<?> undeliveredMessage,
+			String description, Throwable cause) {
+		super(undeliveredMessage, description, cause);
+	}
+
+	public MessageDispatchingException(Message<?> undeliveredMessage,
+			String description) {
+		super(undeliveredMessage, description);
+	}
+
+	public MessageDispatchingException(Message<?> undeliveredMessage) {
+		super(undeliveredMessage);
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2012 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.
@@ -20,11 +20,14 @@ import java.util.concurrent.atomic.AtomicInteger;
 
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageChannel;
+import org.springframework.integration.MessageDeliveryException;
+import org.springframework.integration.MessageDispatchingException;
 import org.springframework.integration.core.MessageHandler;
 import org.springframework.integration.core.SubscribableChannel;
 import org.springframework.integration.dispatcher.MessageDispatcher;
 import org.springframework.integration.dispatcher.UnicastingDispatcher;
 import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
 
 /**
  * Base implementation of {@link MessageChannel} that invokes the subscribed
@@ -32,6 +35,7 @@ import org.springframework.util.Assert;
  * 
  * @author Mark Fisher
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  */
 public abstract class AbstractSubscribableChannel extends AbstractMessageChannel implements SubscribableChannel {
 	
@@ -58,7 +62,15 @@ public abstract class AbstractSubscribableChannel extends AbstractMessageChannel
 
 	@Override
 	protected boolean doSend(Message<?> message, long timeout) {
-		return this.getRequiredDispatcher().dispatch(message);
+		try {
+			return this.getRequiredDispatcher().dispatch(message);
+		}
+		catch (MessageDispatchingException e) {
+			String componentName = this.getComponentName();
+			componentName = StringUtils.hasText(componentName) ? componentName : "unknown";
+			throw new MessageDeliveryException(message, e.getMessage()
+					+ " for channel " + componentName + ".", e);
+		}
 	}
 
 	private MessageDispatcher getRequiredDispatcher() {

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2012 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.
@@ -20,6 +20,7 @@ import java.util.List;
 import java.util.concurrent.Executor;
 
 import org.springframework.integration.Message;
+import org.springframework.integration.MessageDispatchingException;
 import org.springframework.integration.MessagingException;
 import org.springframework.integration.core.MessageHandler;
 import org.springframework.integration.support.MessageBuilder;
@@ -40,6 +41,8 @@ import org.springframework.integration.support.MessageBuilder;
  */
 public class BroadcastingDispatcher extends AbstractDispatcher {
 
+	private final boolean requireSubscribers;
+
 	private volatile boolean ignoreFailures;
 
 	private volatile boolean applySequence;
@@ -47,10 +50,19 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
 	private final Executor executor;
 
 	public BroadcastingDispatcher() {
-		this.executor = null;
+		this(null, false);
 	}
 
 	public BroadcastingDispatcher(Executor executor) {
+		this(executor, false);
+	}
+
+	public BroadcastingDispatcher(boolean requireSubscribers) {
+		this(null, requireSubscribers);
+	}
+
+	public BroadcastingDispatcher(Executor executor, boolean requireSubscribers) {
+		this.requireSubscribers = requireSubscribers;
 		this.executor = executor;
 	}
 
@@ -80,6 +92,9 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
 		boolean dispatched = false;
 		int sequenceNumber = 1;
 		List<MessageHandler> handlers = this.getHandlers();
+		if (this.requireSubscribers && handlers.size() == 0) {
+			throw new MessageDispatchingException(message, "Dispatcher has no subscribers");
+		}
 		int sequenceSize = handlers.size();
 		for (final MessageHandler handler : handlers) {
 			final Message<?> messageToSend = (!this.applySequence) ? message : MessageBuilder.fromMessage(message)

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java

@@ -1,4 +1,4 @@
-/* Copyright 2002-2010 the original author or authors.
+/* Copyright 2002-2012 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.
@@ -25,6 +25,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageDeliveryException;
+import org.springframework.integration.MessageDispatchingException;
 import org.springframework.integration.MessagingException;
 import org.springframework.integration.core.MessageHandler;
 
@@ -105,7 +106,7 @@ public class UnicastingDispatcher extends AbstractDispatcher {
 		boolean success = false;
 		Iterator<MessageHandler> handlerIterator = this.getHandlerIterator(message);
 		if (!handlerIterator.hasNext()) {
-			throw new MessageDeliveryException(message, "Dispatcher has no subscribers.");
+			throw new MessageDispatchingException(message, "Dispatcher has no subscribers");
 		}
 		List<RuntimeException> exceptions = new ArrayList<RuntimeException>();
 		while (success == false && handlerIterator.hasNext()) {

spring-integration-core/src/test/java/org/springframework/integration/channel/DispatcherHasNoSubscribersTests-context.xml

@@ -0,0 +1,14 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns="http://www.springframework.org/schema/integration"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.1.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
+
+	<channel id="subscribedChannel" />
+
+	<bridge input-channel="subscribedChannel" output-channel="noSubscribersChannel" />
+
+	<channel id="noSubscribersChannel" />
+
+</beans:beans>

spring-integration-core/src/test/java/org/springframework/integration/channel/DispatcherHasNoSubscribersTests.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2002-2012 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.integration.channel;
+
+import static org.junit.Assert.*;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.integration.MessageChannel;
+import org.springframework.integration.MessagingException;
+import org.springframework.integration.message.GenericMessage;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Gary Russell
+ * @since 2.1
+ *
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class DispatcherHasNoSubscribersTests {
+
+	@Autowired
+	MessageChannel noSubscribersChannel;
+
+	@Autowired
+	MessageChannel subscribedChannel;
+
+	@Test
+	public void oneChannel() {
+		try {
+			noSubscribersChannel.send(new GenericMessage<String>("Hello, world!"));
+			fail("Exception expected");
+		} catch (MessagingException e) {
+			assertEquals("Dispatcher has no subscribers for channel noSubscribersChannel.", e.getMessage());
+			assertEquals("Dispatcher has no subscribers for channel noSubscribersChannel.", e.getLocalizedMessage());
+		}
+	}
+
+	@Test
+	public void stackedChannels() {
+		try {
+			subscribedChannel.send(new GenericMessage<String>("Hello, world!"));
+			fail("Exception expected");
+		} catch (MessagingException e) {
+			assertEquals("Dispatcher has no subscribers for channel noSubscribersChannel.", e.getMessage());
+			assertEquals("Dispatcher has no subscribers for channel noSubscribersChannel.", e.getLocalizedMessage());
+		}
+	}
+
+}

spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2012 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.
@@ -24,6 +24,8 @@ import org.apache.commons.logging.LogFactory;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.integration.Message;
+import org.springframework.integration.MessageDeliveryException;
+import org.springframework.integration.MessageDispatchingException;
 import org.springframework.integration.MessagingException;
 import org.springframework.integration.core.MessageHandler;
 import org.springframework.integration.core.SubscribableChannel;
@@ -35,9 +37,11 @@ import org.springframework.integration.support.MessageBuilder;
 import org.springframework.jms.core.JmsTemplate;
 import org.springframework.jms.listener.AbstractMessageListenerContainer;
 import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
 
 /**
  * @author Mark Fisher
+ * @author Gary Russell
  * @since 2.0
  */
 public class SubscribableJmsChannel extends AbstractJmsChannel implements SubscribableChannel, SmartLifecycle, DisposableBean {
@@ -71,8 +75,11 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
 			return;
 		}
 		super.onInit();
-		this.configureDispatcher(this.container.isPubSubDomain());
-		MessageListener listener = new DispatchingMessageListener(this.getJmsTemplate(), this.dispatcher);
+		boolean isPubSub = this.container.isPubSubDomain();
+		this.configureDispatcher(isPubSub);
+		MessageListener listener = new DispatchingMessageListener(
+				this.getJmsTemplate(), this.dispatcher,
+				this.getComponentName(), isPubSub);
 		this.container.setMessageListener(listener);
 		if (!this.container.isActive()) {
 			this.container.afterPropertiesSet();
@@ -82,7 +89,7 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
 
 	private void configureDispatcher(boolean isPubSub) {
 		if (isPubSub) {
-			this.dispatcher = new BroadcastingDispatcher();
+			this.dispatcher = new BroadcastingDispatcher(true);
 		}
 		else {
 			UnicastingDispatcher unicastingDispatcher = new UnicastingDispatcher();
@@ -100,18 +107,26 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
 
 		private final MessageDispatcher dispatcher;
 
+		private final String channelName;
 
-		private DispatchingMessageListener(JmsTemplate jmsTemplate, MessageDispatcher dispatcher) {
+		private final boolean isPubSub;
+
+
+		private DispatchingMessageListener(JmsTemplate jmsTemplate,
+				MessageDispatcher dispatcher, String channelName, boolean isPubSub) {
 			this.jmsTemplate = jmsTemplate;
 			this.dispatcher = dispatcher;
+			this.channelName = channelName;
+			this.isPubSub = isPubSub;
 		}
 
 
 		public void onMessage(javax.jms.Message message) {
+			Message<?> messageToSend = null;
 			try {
 				Object converted = this.jmsTemplate.getMessageConverter().fromMessage(message);
 				if (converted != null) {
-					Message<?> messageToSend = (converted instanceof Message<?>) ? (Message<?>) converted
+					messageToSend = (converted instanceof Message<?>) ? (Message<?>) converted
 							: MessageBuilder.withPayload(converted).build();
 					this.dispatcher.dispatch(messageToSend);
 				}
@@ -119,6 +134,21 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
 					logger.warn("MessageConverter returned null, no Message to dispatch");
 				}
 			}
+			catch (MessageDispatchingException e) {
+				String channelName = StringUtils.hasText(this.channelName) ? this.channelName : "unknown";
+				String exceptionMessage = e.getMessage() + " for jms-channel "
+						+ channelName + ".";
+				if (this.isPubSub) {
+					// log only for backwards compatibility with pub/sub
+					if (logger.isWarnEnabled()) {
+						logger.warn(exceptionMessage, e);
+					}
+				}
+				else {
+					throw new MessageDeliveryException(
+							messageToSend, exceptionMessage, e);
+				}
+			}
 			catch (Exception e) {
 				throw new MessagingException("failed to handle incoming JMS Message", e);
 			}

spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2012 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.
@@ -48,6 +48,7 @@ import org.springframework.util.StringUtils;
 /**
  * @author Mark Fisher
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  * @since 2.0
  */
 public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChannel> implements SmartLifecycle, DisposableBean, BeanNameAware {
@@ -337,8 +338,8 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChanne
 		if (!CollectionUtils.isEmpty(this.interceptors)) {
 			this.channel.setInterceptors(this.interceptors);
 		}
-		this.channel.afterPropertiesSet();
 		this.channel.setBeanName(this.beanName);
+		this.channel.afterPropertiesSet();
 		return this.channel;
 	}
 

spring-integration-jms/src/test/java/org/springframework/integration/jms/SubscribableJmsChannelTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2012 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.
@@ -21,6 +21,11 @@ import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -29,25 +34,32 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import javax.jms.Destination;
+import javax.jms.MessageListener;
 
 import org.apache.activemq.ActiveMQConnectionFactory;
 import org.apache.activemq.command.ActiveMQQueue;
 import org.apache.activemq.command.ActiveMQTopic;
+import org.apache.commons.logging.Log;
 import org.junit.Before;
 import org.junit.Test;
-
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.context.support.StaticApplicationContext;
 import org.springframework.integration.Message;
+import org.springframework.integration.MessageDeliveryException;
 import org.springframework.integration.core.MessageHandler;
 import org.springframework.integration.jms.config.JmsChannelFactoryBean;
 import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.jms.listener.AbstractMessageListenerContainer;
 import org.springframework.jms.listener.DefaultMessageListenerContainer;
 
 /**
  * @author Mark Fisher
+ * @author Gary Russell
+ * @since 2.0
  */
 public class SubscribableJmsChannelTests {
 
@@ -247,6 +259,85 @@ public class SubscribableJmsChannelTests {
 		assertFalse(channel.isRunning());
 	}
 
+	@Test
+	public void dispatcherHasNoSubscribersQueue() throws Exception {
+		JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(true);
+		factoryBean.setConnectionFactory(this.connectionFactory);
+		factoryBean.setDestinationName("noSubscribersQueue");
+		factoryBean.setBeanName("noSubscribersChannel");
+		factoryBean.afterPropertiesSet();
+		SubscribableJmsChannel channel = (SubscribableJmsChannel) factoryBean.getObject();
+		channel.afterPropertiesSet();
+
+		AbstractMessageListenerContainer container = TestUtils
+				.getPropertyValue(channel, "container",
+						AbstractMessageListenerContainer.class);
+		MessageListener listener = (MessageListener) container.getMessageListener();
+		try {
+			listener.onMessage(new StubTextMessage("Hello, world!"));
+			fail("Exception expected");
+		}
+		catch (MessageDeliveryException e) {
+			assertEquals("Dispatcher has no subscribers for jms-channel noSubscribersChannel.", e.getMessage());
+		}
+	}
+
+	@Test
+	public void dispatcherHasNoSubscribersTopic() throws Exception {
+		JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(true);
+		factoryBean.setConnectionFactory(this.connectionFactory);
+		factoryBean.setDestinationName("noSubscribersTopic");
+		factoryBean.setBeanName("noSubscribersChannel");
+		factoryBean.setPubSubDomain(true);
+		factoryBean.afterPropertiesSet();
+		SubscribableJmsChannel channel = (SubscribableJmsChannel) factoryBean.getObject();
+		channel.afterPropertiesSet();
+
+		AbstractMessageListenerContainer container = TestUtils
+				.getPropertyValue(channel, "container",
+						AbstractMessageListenerContainer.class);
+		MessageListener listener = (MessageListener) container.getMessageListener();
+		List<String> logList  = insertMockLoggerInListener(channel);
+		listener.onMessage(new StubTextMessage("Hello, world!"));
+		verifyLogReceived(logList);
+	}
+
+	private List<String> insertMockLoggerInListener(
+			SubscribableJmsChannel channel) {
+		AbstractMessageListenerContainer container = TestUtils.getPropertyValue(
+				channel, "container", AbstractMessageListenerContainer.class);
+		Log logger = mock(Log.class);
+		final ArrayList<String> logList = new ArrayList<String>();
+		doAnswer(new Answer<Object>() {
+			public Object answer(InvocationOnMock invocation)
+					throws Throwable {
+				String message = (String) invocation.getArguments()[0];
+				if (message.startsWith("Dispatcher has no subscribers")) {
+					logList.add(message);
+				}
+				return null;
+			}}).when(logger).warn(anyString(), any(Exception.class));
+		when(logger.isWarnEnabled()).thenReturn(true);
+		Object listener = container.getMessageListener();
+		DirectFieldAccessor dfa = new DirectFieldAccessor(listener);
+		dfa.setPropertyValue("logger", logger);
+		return logList;
+	}
+
+	private void verifyLogReceived(final List<String> logList) {
+		assertTrue("Failed to get expected exception", logList.size() > 0);
+		boolean expectedExceptionFound = false;
+		while (logList.size() > 0) {
+			String message = logList.remove(0);
+			assertNotNull("Failed to get expected exception", message);
+			if (message.startsWith("Dispatcher has no subscribers")) {
+				expectedExceptionFound = true;
+				assertEquals("Dispatcher has no subscribers for jms-channel noSubscribersChannel.", message);
+				break;
+			}
+		}
+		assertTrue("Failed to get expected exception", expectedExceptionFound);
+	}
 
 	/**
 	 * Blocks until the listener container has subscribed; if the container does not support

spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2012 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.
@@ -30,6 +30,8 @@ import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
 import org.springframework.data.redis.serializer.RedisSerializer;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
 import org.springframework.integration.Message;
+import org.springframework.integration.MessageDeliveryException;
+import org.springframework.integration.MessageDispatchingException;
 import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.integration.channel.MessagePublishingErrorHandler;
 import org.springframework.integration.core.MessageHandler;
@@ -42,9 +44,11 @@ import org.springframework.integration.support.converter.SimpleMessageConverter;
 import org.springframework.integration.util.ErrorHandlingTaskExecutor;
 import org.springframework.util.Assert;
 import org.springframework.util.ErrorHandler;
+import org.springframework.util.StringUtils;
 
 /**
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  * @since 2.0
  */
 @SuppressWarnings("rawtypes")
@@ -55,7 +59,7 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements
 	private final RedisTemplate redisTemplate;
 	private final String topicName;
 	
-	private final MessageDispatcher dispatcher = new BroadcastingDispatcher();
+	private final MessageDispatcher dispatcher = new BroadcastingDispatcher(true);
 	
 	private volatile boolean initialized;
 	
@@ -171,7 +175,16 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements
 		@SuppressWarnings("unused")
 		public void handleMessage(String s) {
 			Message<?> siMessage = messageConverter.toMessage(s);
-			dispatcher.dispatch(siMessage);
+			try {
+				dispatcher.dispatch(siMessage);
+			}
+			catch (MessageDispatchingException e) {
+				String topicName = SubscribableRedisChannel.this.topicName;
+				topicName = StringUtils.hasText(topicName) ? topicName : "unknown";
+				throw new MessageDeliveryException(siMessage, e.getMessage()
+						+ " for redis-channel "
+						+ topicName + ".", e);
+			}
 		}
 	}
 }

spring-integration-redis/src/test/java/org/springframework/integration/redis/channel/SubscribableRedisChannelTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2012 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.
@@ -15,21 +15,33 @@
  */
 package org.springframework.integration.redis.channel;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import java.lang.reflect.InvocationTargetException;
+import java.util.Map;
+import java.util.Set;
+
 import org.junit.Test;
 import org.mockito.Mockito;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+import org.springframework.data.redis.listener.RedisMessageListenerContainer;
+import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
 import org.springframework.integration.Message;
 import org.springframework.integration.core.MessageHandler;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.redis.rules.RedisAvailable;
 import org.springframework.integration.redis.rules.RedisAvailableTests;
-
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.util.ReflectionUtils;
 /**
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  * @since 2.0
  */
 public class SubscribableRedisChannelTests extends RedisAvailableTests{
@@ -56,4 +68,34 @@ public class SubscribableRedisChannelTests extends RedisAvailableTests{
 		verify(handler, times(3)).handleMessage(Mockito.any(Message.class));
 		channel.stop();
 	}
+
+	@Test
+	@RedisAvailable
+	public void dispatcherHasNoSubscribersTest() throws Exception{
+		JedisConnectionFactory connectionFactory = new JedisConnectionFactory();
+		connectionFactory.setPort(7379);
+		connectionFactory.afterPropertiesSet();
+
+		SubscribableRedisChannel channel = new SubscribableRedisChannel(connectionFactory, "si.test.channel.no.subs");
+		channel.setBeanFactory(mock(BeanFactory.class));
+		channel.afterPropertiesSet();
+
+		RedisMessageListenerContainer container = TestUtils.getPropertyValue(
+				channel, "container", RedisMessageListenerContainer.class);
+		@SuppressWarnings("unchecked")
+		Map<?, Set<MessageListenerAdapter>> channelMapping = (Map<?, Set<MessageListenerAdapter>>) TestUtils
+				.getPropertyValue(container, "channelMapping");
+		MessageListenerAdapter listener = channelMapping.entrySet().iterator().next().getValue().iterator().next();
+		Object delegate = TestUtils.getPropertyValue(listener, "delegate");
+		try {
+			ReflectionUtils.findMethod(delegate.getClass(), "handleMessage", String.class).invoke(delegate, "Hello, world!");
+			fail("Exception expected");
+		}
+		catch (InvocationTargetException e) {
+			Throwable cause = e.getCause();
+			assertNotNull(cause);
+			assertEquals("Dispatcher has no subscribers for redis-channel si.test.channel.no.subs.", cause.getMessage());
+		}
+
+	}
 }