Merge pull request #758 from garyrussell/INT-2951

* garyrussell-INT-2951:
  INT-2951 Add Context Id to Dispatcher has no Subs.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 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.
@@ -39,7 +39,6 @@ import org.springframework.integration.dispatcher.AbstractDispatcher;
 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
@@ -104,7 +103,7 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple
 				? ((RabbitTemplate) this.getAmqpTemplate()).getMessageConverter()
 				: new SimpleMessageConverter();
 		MessageListener listener = new DispatchingMessageListener(converter,
-				this.dispatcher, this.channelName, this.isPubSub);
+				this.dispatcher, this, this.isPubSub);
 		this.container.setMessageListener(listener);
 		if (!this.container.isActive()) {
 			this.container.afterPropertiesSet();
@@ -124,17 +123,17 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple
 
 		private final MessageConverter converter;
 
-		private final String channelName;
+		private final AbstractSubscribableAmqpChannel channel;
 
 		private final boolean isPubSub;
 
 		private DispatchingMessageListener(MessageConverter converter,
-				MessageDispatcher dispatcher, String channelName, boolean isPubSub) {
+				MessageDispatcher dispatcher, AbstractSubscribableAmqpChannel channel, 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.channel = channel;
 			this.isPubSub = isPubSub;
 		}
 
@@ -153,9 +152,8 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple
 				}
 			}
 			catch (MessageDispatchingException e) {
-				String channelName = StringUtils.hasText(this.channelName) ? this.channelName : "unknown";
-				String exceptionMessage = e.getMessage() + " for amqp-channel "
-						+ channelName + ".";
+				String exceptionMessage = e.getMessage() + " for amqp-channel '"
+						+ this.channel.getFullChannelName() + "'.";
 				if (this.isPubSub) {
 					// log only for backwards compatibility with pub/sub
 					if (logger.isWarnEnabled()) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 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.
@@ -71,6 +71,7 @@ public class DispatcherHasNoSubscribersTests {
 
 		PointToPointSubscribableAmqpChannel amqpChannel = new PointToPointSubscribableAmqpChannel("noSubscribersChannel",
 				container, amqpTemplate);
+		amqpChannel.setBeanName("noSubscribersChannel");
 		amqpChannel.afterPropertiesSet();
 
 		MessageListener listener = (MessageListener) container.getMessageListener();
@@ -79,7 +80,7 @@ public class DispatcherHasNoSubscribersTests {
 			fail("Exception expected");
 		}
 		catch (MessageDeliveryException e) {
-			assertEquals("Dispatcher has no subscribers for amqp-channel noSubscribersChannel.", e.getMessage());
+			assertEquals("Dispatcher has no subscribers for amqp-channel 'noSubscribersChannel'.", e.getMessage());
 		}
 	}
 
@@ -98,12 +99,13 @@ public class DispatcherHasNoSubscribersTests {
 		AmqpTemplate amqpTemplate = mock(AmqpTemplate.class);
 		final Queue queue = new Queue("noSubscribersQueue");
 		PublishSubscribeAmqpChannel amqpChannel = new PublishSubscribeAmqpChannel("noSubscribersChannel",
-				container, amqpTemplate) { 
+				container, amqpTemplate) {
 					@Override
 					protected Queue initializeQueue(AmqpAdmin admin,
 							String channelName) {
 						return queue;
 					}};
+		amqpChannel.setBeanName("noSubscribersChannel");
 		amqpChannel.afterPropertiesSet();
 
 		List<String> logList = insertMockLoggerInListener(amqpChannel);
@@ -142,7 +144,7 @@ public class DispatcherHasNoSubscribersTests {
 			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);
+				assertEquals("Dispatcher has no subscribers for amqp-channel 'noSubscribersChannel'.", message);
 				break;
 			}
 		}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2013 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.
@@ -40,9 +40,10 @@ import org.springframework.util.StringUtils;
  * properties such as the channel name. Also provides the common functionality
  * for sending and receiving {@link Message Messages} including the invocation
  * of any {@link ChannelInterceptor ChannelInterceptors}.
- * 
+ *
  * @author Mark Fisher
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  */
 public abstract class AbstractMessageChannel extends IntegrationObjectSupport implements MessageChannel, TrackableComponent {
 
@@ -54,7 +55,10 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport im
 
 	private final ChannelInterceptorList interceptors = new ChannelInterceptorList();
 
+	private volatile String fullChannelName;
 
+
+	@Override
 	public String getComponentType() {
 		return "channel";
 	}
@@ -66,7 +70,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport im
 	/**
 	 * Specify the Message payload datatype(s) supported by this channel. If a
 	 * payload type does not match directly, but the 'conversionService' is
-	 * available, then type conversion will be attempted in the order of the 
+	 * available, then type conversion will be attempted in the order of the
 	 * elements provided in this array.
 	 * <p>
 	 * If this property is not set explicitly, any Message payload type will be
@@ -103,6 +107,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport im
 	 * Finally, if that bean is not available, it will fallback to the
 	 * "conversionService" bean, if available.
 	 */
+	@Override
 	public void setConversionService(ConversionService conversionService) {
 		super.setConversionService(conversionService);
 	}
@@ -114,13 +119,29 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport im
 		return this.interceptors;
 	}
 
+	/**
+	 * Returns the fully qualified channel name including the application context
+	 * id, if available.
+	 * @return The name.
+	 */
+	public String getFullChannelName() {
+		if (this.fullChannelName == null) {
+			String contextId = this.getApplicationContextId();
+			String componentName = this.getComponentName();
+			componentName = (StringUtils.hasText(contextId) ? contextId + "." : "") +
+					(StringUtils.hasText(componentName) ? componentName : "unknown.channel.name");
+			this.fullChannelName = componentName;
+		}
+		return this.fullChannelName;
+	}
+
 	/**
 	 * Send a message on this channel. If the channel is at capacity, this
 	 * method will block until either space becomes available or the sending
 	 * thread is interrupted.
-	 * 
+	 *
 	 * @param message the Message to send
-	 * 
+	 *
 	 * @return <code>true</code> if the message is sent successfully or
 	 * <code>false</code> if the sending thread is interrupted.
 	 */
@@ -134,10 +155,10 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport im
 	 * is interrupted. If the specified timeout is 0, the method will return
 	 * immediately. If less than zero, it will block indefinitely (see
 	 * {@link #send(Message)}).
-	 * 
+	 *
 	 * @param message the Message to send
 	 * @param timeout the timeout in milliseconds
-	 * 
+	 *
 	 * @return <code>true</code> if the message is sent successfully,
 	 * <code>false</code> if the message cannot be sent within the allotted
 	 * time or the sending thread is interrupted.
@@ -185,8 +206,8 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport im
 			}
 		}
 		throw new MessageDeliveryException(message, "Channel '" + this.getComponentName() +
-				"' expected one of the following datataypes [" + 
-				StringUtils.arrayToCommaDelimitedString(this.datatypes) + 
+				"' expected one of the following datataypes [" +
+				StringUtils.arrayToCommaDelimitedString(this.datatypes) +
 				"], but received [" + message.getPayload().getClass() + "]");
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 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.
@@ -27,7 +27,6 @@ import org.springframework.integration.core.SubscribableChannel;
 import org.springframework.integration.dispatcher.AbstractDispatcher;
 import org.springframework.integration.dispatcher.MessageDispatcher;
 import org.springframework.util.Assert;
-import org.springframework.util.StringUtils;
 
 /**
  * Base implementation of {@link MessageChannel} that invokes the subscribed
@@ -37,7 +36,8 @@ import org.springframework.util.StringUtils;
  * @author Oleg Zhurakousky
  * @author Gary Russell
  */
-public abstract class AbstractSubscribableChannel extends AbstractMessageChannel implements SubscribableChannel {
+public abstract class AbstractSubscribableChannel extends AbstractMessageChannel
+		implements SubscribableChannel {
 
 	private final AtomicInteger handlerCounter = new AtomicInteger();
 
@@ -66,7 +66,7 @@ public abstract class AbstractSubscribableChannel extends AbstractMessageChannel
 				counter = handlerCounter.addAndGet(delta);
 			}
 			if (logger.isInfoEnabled()) {
-				logger.info("Channel '" + this.getComponentName() + "' has " + counter + " subscriber(s).");
+				logger.info("Channel '" + this.getFullChannelName() + "' has " + counter + " subscriber(s).");
 			}
 		}
 	}
@@ -77,10 +77,8 @@ public abstract class AbstractSubscribableChannel extends AbstractMessageChannel
 			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);
+			String description = e.getMessage() + " for channel '" + this.getFullChannelName() + "'.";
+			throw new MessageDeliveryException(message, description, e);
 		}
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java

@@ -18,11 +18,14 @@ package org.springframework.integration.context;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.beans.factory.BeanInitializationException;
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.beans.factory.InitializingBean;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
 import org.springframework.core.convert.ConversionService;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.util.Assert;
@@ -41,8 +44,10 @@ import org.springframework.util.StringUtils;
  * @author Oleg Zhurakousky
  * @author Josh Long
  * @author Stefan Ferstl
+ * @author Gary Russell
  */
-public abstract class IntegrationObjectSupport implements BeanNameAware, NamedComponent, BeanFactoryAware, InitializingBean {
+public abstract class IntegrationObjectSupport implements BeanNameAware, NamedComponent,
+		ApplicationContextAware, BeanFactoryAware, InitializingBean {
 
 	/**
 	 * Logger that is available to subclasses
@@ -59,6 +64,7 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
 
 	private volatile ConversionService conversionService;
 
+	private volatile ApplicationContext applicationContext;
 
 	public final void setBeanName(String beanName) {
 		this.beanName = beanName;
@@ -89,10 +95,15 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
 	}
 
 	public final void setBeanFactory(BeanFactory beanFactory) {
-		Assert.notNull(beanFactory, "beanFactory must not be null");
+		Assert.notNull(beanFactory, "'beanFactory' must not be null");
 		this.beanFactory = beanFactory;
 	}
 
+	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+		Assert.notNull(applicationContext, "'applicationContext' must not be null");
+		this.applicationContext = applicationContext;
+	}
+
 	public final void afterPropertiesSet() {
 		try {
 			this.onInit();
@@ -147,6 +158,15 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
 		this.conversionService = conversionService;
 	}
 
+	/**
+	 * Returns the {@link ApplicationContext#getId()} if the
+	 * {@link ApplicationContext} is available.
+	 * @return The id, or null if there is no application context.
+	 */
+	public String getApplicationContextId() {
+		return this.applicationContext == null ? null : this.applicationContext.getId();
+	}
+
 	@Override
 	public String toString() {
 		return (this.beanName != null) ? this.beanName : super.toString();

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 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,11 +15,14 @@
  */
 package org.springframework.integration.channel;
 
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
 
+import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.support.AbstractApplicationContext;
 import org.springframework.integration.MessageChannel;
 import org.springframework.integration.MessagingException;
 import org.springframework.integration.message.GenericMessage;
@@ -41,14 +44,21 @@ public class DispatcherHasNoSubscribersTests {
 	@Autowired
 	MessageChannel subscribedChannel;
 
+	@Autowired
+	AbstractApplicationContext applicationContext;
+
+	@Before
+	public void setup() {
+		applicationContext.setId("foo");
+	}
+
 	@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());
+			assertEquals("Dispatcher has no subscribers for channel 'foo.noSubscribersChannel'.", e.getMessage());
 		}
 	}
 
@@ -58,8 +68,19 @@ public class DispatcherHasNoSubscribersTests {
 			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());
+			assertEquals("Dispatcher has no subscribers for channel 'foo.noSubscribersChannel'.", e.getMessage());
+		}
+	}
+
+	@Test
+	public void withNoContext() {
+		DirectChannel channel = new DirectChannel();
+		channel.setBeanName("bar");
+		try {
+			channel.send(new GenericMessage<String>("Hello, world!"));
+			fail("Exception expected");
+		} catch (MessagingException e) {
+			assertEquals("Dispatcher has no subscribers for channel 'bar'.", e.getMessage());
 		}
 	}
 

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

@@ -37,7 +37,6 @@ 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
@@ -89,7 +88,7 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
 		this.configureDispatcher(isPubSub);
 		MessageListener listener = new DispatchingMessageListener(
 				this.getJmsTemplate(), this.dispatcher,
-				this.getComponentName(), isPubSub);
+				this, isPubSub);
 		this.container.setMessageListener(listener);
 		if (!this.container.isActive()) {
 			this.container.afterPropertiesSet();
@@ -118,16 +117,16 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
 
 		private final MessageDispatcher dispatcher;
 
-		private final String channelName;
+		private final SubscribableJmsChannel channel;
 
 		private final boolean isPubSub;
 
 
 		private DispatchingMessageListener(JmsTemplate jmsTemplate,
-				MessageDispatcher dispatcher, String channelName, boolean isPubSub) {
+				MessageDispatcher dispatcher, SubscribableJmsChannel channel, boolean isPubSub) {
 			this.jmsTemplate = jmsTemplate;
 			this.dispatcher = dispatcher;
-			this.channelName = channelName;
+			this.channel = channel;
 			this.isPubSub = isPubSub;
 		}
 
@@ -146,9 +145,8 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
 				}
 			}
 			catch (MessageDispatchingException e) {
-				String channelName = StringUtils.hasText(this.channelName) ? this.channelName : "unknown";
-				String exceptionMessage = e.getMessage() + " for jms-channel "
-						+ channelName + ".";
+				String exceptionMessage = e.getMessage() + " for jms-channel '"
+						+ this.channel.getFullChannelName() + "'.";
 				if (this.isPubSub) {
 					// log only for backwards compatibility with pub/sub
 					if (logger.isWarnEnabled()) {

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

@@ -292,7 +292,7 @@ public class SubscribableJmsChannelTests {
 			fail("Exception expected");
 		}
 		catch (MessageDeliveryException e) {
-			assertEquals("Dispatcher has no subscribers for jms-channel noSubscribersChannel.", e.getMessage());
+			assertEquals("Dispatcher has no subscribers for jms-channel 'noSubscribersChannel'.", e.getMessage());
 		}
 	}
 
@@ -346,7 +346,7 @@ public class SubscribableJmsChannelTests {
 			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);
+				assertEquals("Dispatcher has no subscribers for jms-channel 'noSubscribersChannel'.", message);
 				break;
 			}
 		}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 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.
@@ -191,8 +191,9 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements
 				String topicName = SubscribableRedisChannel.this.topicName;
 				topicName = StringUtils.hasText(topicName) ? topicName : "unknown";
 				throw new MessageDeliveryException(siMessage, e.getMessage()
-						+ " for redis-channel "
-						+ topicName + ".", e);
+						+ " for redis-channel '"
+						+ topicName + "' (" + SubscribableRedisChannel.this.getFullChannelName()
+						+ ").", e);
 			}
 		}
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 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.
@@ -45,9 +45,9 @@ import org.springframework.util.ReflectionUtils;
  * @since 2.0
  */
 public class SubscribableRedisChannelTests extends RedisAvailableTests{
-	
 
-	@Test 
+
+	@Test
 	@RedisAvailable
 	public void pubSubChanneTest() throws Exception{
 		JedisConnectionFactory connectionFactory = new JedisConnectionFactory();
@@ -77,6 +77,7 @@ public class SubscribableRedisChannelTests extends RedisAvailableTests{
 		connectionFactory.afterPropertiesSet();
 
 		SubscribableRedisChannel channel = new SubscribableRedisChannel(connectionFactory, "si.test.channel.no.subs");
+		channel.setBeanName("dhnsChannel");
 		channel.setBeanFactory(mock(BeanFactory.class));
 		channel.afterPropertiesSet();
 
@@ -94,7 +95,7 @@ public class SubscribableRedisChannelTests extends RedisAvailableTests{
 		catch (InvocationTargetException e) {
 			Throwable cause = e.getCause();
 			assertNotNull(cause);
-			assertEquals("Dispatcher has no subscribers for redis-channel si.test.channel.no.subs.", cause.getMessage());
+			assertEquals("Dispatcher has no subscribers for redis-channel 'si.test.channel.no.subs' (dhnsChannel).", cause.getMessage());
 		}
 
 	}