Merge branch '3.1.x'

org.springframework.jms/src/main/java/org/springframework/jms/connection/CachingConnectionFactory.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.
@@ -34,6 +34,8 @@ import javax.jms.MessageConsumer;
 import javax.jms.MessageProducer;
 import javax.jms.QueueSession;
 import javax.jms.Session;
+import javax.jms.TemporaryQueue;
+import javax.jms.TemporaryTopic;
 import javax.jms.Topic;
 import javax.jms.TopicSession;
 
@@ -323,16 +325,18 @@ public class CachingConnectionFactory extends SingleConnectionFactory {
 					// let raw JMS invocation throw an exception if Destination (i.e. args[0]) is null
 					if ((methodName.equals("createConsumer") || methodName.equals("createReceiver") ||
 							methodName.equals("createSubscriber"))) {
-						if (args[0] != null) {
-							return getCachedConsumer((Destination) args[0],
+						Destination dest = (Destination) args[0];
+						if (dest != null && !(dest instanceof TemporaryQueue || dest instanceof TemporaryTopic)) {
+							return getCachedConsumer(dest,
 									(args.length > 1 ? (String) args[1] : null),
 									(args.length > 2 && (Boolean) args[2]),
 									null);
 						}
 					}
 					else if (methodName.equals("createDurableSubscriber")) {
-						if (args[0] != null) {
-							return getCachedConsumer((Destination) args[0],
+						Destination dest = (Destination) args[0];
+						if (dest != null) {
+							return getCachedConsumer(dest,
 									(args.length > 2 ? (String) args[2] : null),
 									(args.length > 3 && (Boolean) args[3]),
 									(String) args[1]);

org.springframework.jms/src/main/java/org/springframework/jms/listener/AbstractJmsListeningContainer.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.
@@ -151,6 +151,7 @@ public abstract class AbstractJmsListeningContainer extends JmsDestinationAccess
 	/**
 	 * Delegates to {@link #validateConfiguration()} and {@link #initialize()}.
 	 */
+	@Override
 	public void afterPropertiesSet() {
 		super.afterPropertiesSet();
 		validateConfiguration();

org.springframework.jms/src/main/java/org/springframework/jms/listener/AbstractMessageListenerContainer.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.
@@ -410,6 +410,7 @@ public abstract class AbstractMessageListenerContainer extends AbstractJmsListen
 		return this.acceptMessagesWhileStopping;
 	}
 
+	@Override
 	protected void validateConfiguration() {
 		if (this.destination == null) {
 			throw new IllegalArgumentException("Property 'destination' or 'destinationName' is required");

org.springframework.jms/src/main/java/org/springframework/jms/listener/AbstractPollingMessageListenerContainer.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.
@@ -75,8 +75,7 @@ import org.springframework.transaction.support.TransactionSynchronizationUtils;
  * @see #receiveAndExecute
  * @see #setTransactionManager
  */
-public abstract class AbstractPollingMessageListenerContainer extends AbstractMessageListenerContainer
-		implements BeanNameAware {
+public abstract class AbstractPollingMessageListenerContainer extends AbstractMessageListenerContainer {
 
 	/**
 	 * The default receive timeout: 1000 ms = 1 second.
@@ -100,6 +99,7 @@ public abstract class AbstractPollingMessageListenerContainer extends AbstractMe
 	private volatile Boolean commitAfterNoMessageReceived;
 
 
+	@Override
 	public void setSessionTransacted(boolean sessionTransacted) {
 		super.setSessionTransacted(sessionTransacted);
 		this.sessionTransactedCalled = true;
@@ -188,6 +188,7 @@ public abstract class AbstractPollingMessageListenerContainer extends AbstractMe
 	}
 
 
+	@Override
 	public void initialize() {
 		// Set sessionTransacted=true in case of a non-JTA transaction manager.
 		if (!this.sessionTransactedCalled &&
@@ -374,6 +375,7 @@ public abstract class AbstractPollingMessageListenerContainer extends AbstractMe
 	 * container's "sessionTransacted" flag being set to "true".
 	 * @see org.springframework.jms.connection.JmsResourceHolder
 	 */
+	@Override
 	protected boolean isSessionLocallyTransacted(Session session) {
 		if (!super.isSessionLocallyTransacted(session)) {
 			return false;

org.springframework.jms/src/main/java/org/springframework/jms/listener/DefaultMessageListenerContainer.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.
@@ -470,6 +470,7 @@ public class DefaultMessageListenerContainer extends AbstractPollingMessageListe
 		}
 	}
 
+	@Override
 	protected void validateConfiguration() {
 		super.validateConfiguration();
 		synchronized (this.lifecycleMonitor) {
@@ -484,6 +485,7 @@ public class DefaultMessageListenerContainer extends AbstractPollingMessageListe
 	// Implementation of AbstractMessageListenerContainer's template methods
 	//-------------------------------------------------------------------------
 
+	@Override
 	public void initialize() {
 		// Adapt default cache level.
 		if (this.cacheLevel == CACHE_AUTO) {
@@ -516,6 +518,7 @@ public class DefaultMessageListenerContainer extends AbstractPollingMessageListe
 	 * @see #scheduleNewInvoker
 	 * @see #setTaskExecutor
 	 */
+	@Override
 	protected void doInitialize() throws JMSException {
 		synchronized (this.lifecycleMonitor) {
 			for (int i = 0; i < this.concurrentConsumers; i++) {
@@ -527,6 +530,7 @@ public class DefaultMessageListenerContainer extends AbstractPollingMessageListe
 	/**
 	 * Destroy the registered JMS Sessions and associated MessageConsumers.
 	 */
+	@Override
 	protected void doShutdown() throws JMSException {
 		logger.debug("Waiting for shutdown of message listener invokers");
 		try {
@@ -549,6 +553,7 @@ public class DefaultMessageListenerContainer extends AbstractPollingMessageListe
 	/**
 	 * Overridden to reset the stop callback, if any.
 	 */
+	@Override
 	public void start() throws JmsException {
 		synchronized (this.lifecycleMonitor) {
 			this.stopCallback = null;
@@ -658,6 +663,7 @@ public class DefaultMessageListenerContainer extends AbstractPollingMessageListe
 	 * @see #setCacheLevel
 	 * @see #CACHE_CONNECTION
 	 */
+	@Override
 	protected final boolean sharedConnectionEnabled() {
 		return (getCacheLevel() >= CACHE_CONNECTION);
 	}
@@ -666,6 +672,7 @@ public class DefaultMessageListenerContainer extends AbstractPollingMessageListe
 	 * Re-executes the given task via this listener container's TaskExecutor.
 	 * @see #setTaskExecutor
 	 */
+	@Override
 	protected void doRescheduleTask(Object task) {
 		this.taskExecutor.execute((Runnable) task);
 	}
@@ -674,6 +681,7 @@ public class DefaultMessageListenerContainer extends AbstractPollingMessageListe
 	 * Tries scheduling a new invoker, since we know messages are coming in...
 	 * @see #scheduleNewInvokerIfAppropriate()
 	 */
+	@Override
 	protected void messageReceived(Object invoker, Session session) {
 		((AsyncMessageListenerInvoker) invoker).setIdle(false);
 		scheduleNewInvokerIfAppropriate();
@@ -682,6 +690,7 @@ public class DefaultMessageListenerContainer extends AbstractPollingMessageListe
 	/**
 	 * Marks the affected invoker as idle.
 	 */
+	@Override
 	protected void noMessageReceived(Object invoker, Session session) {
 		((AsyncMessageListenerInvoker) invoker).setIdle(true);
 	}
@@ -745,6 +754,7 @@ public class DefaultMessageListenerContainer extends AbstractPollingMessageListe
 	 * asynchronous invokers to establish the shared Connection on first access.
 	 * @see #refreshConnectionUntilSuccessful()
 	 */
+	@Override
 	protected void establishSharedConnection() {
 		try {
 			super.establishSharedConnection();
@@ -760,6 +770,7 @@ public class DefaultMessageListenerContainer extends AbstractPollingMessageListe
 	 * <code>Connection.start()</code>, relying on listeners to perform
 	 * appropriate recovery.
 	 */
+	@Override
 	protected void startSharedConnection() {
 		try {
 			super.startSharedConnection();
@@ -774,6 +785,7 @@ public class DefaultMessageListenerContainer extends AbstractPollingMessageListe
 	 * <code>Connection.stop()</code>, relying on listeners to perform
 	 * appropriate recovery after a restart.
 	 */
+	@Override
 	protected void stopSharedConnection() {
 		try {
 			super.stopSharedConnection();

org.springframework.jms/src/main/java/org/springframework/jms/listener/SimpleMessageListenerContainer.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.
@@ -62,6 +62,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 
 	private boolean pubSubNoLocal = false;
 
+	private boolean connectLazily = false;
+
 	private int concurrentConsumers = 1;
 
 	private Executor taskExecutor;
@@ -89,6 +91,20 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 		return this.pubSubNoLocal;
 	}
 
+	/**
+	 * Specify whether to connect lazily, i.e. whether to establish the JMS Connection
+	 * and the corresponding Sessions and MessageConsumers as late as possible -
+	 * in the start phase of this container.
+	 * <p>Default is "false": connecting early, i.e. during the bean initialization phase.
+	 * Set this flag to "true" in order to switch to lazy connecting if your target broker
+	 * is likely to not have started up yet and you prefer to not even try a connection.
+	 * @see #start()
+	 * @see #initialize()
+	 */
+	public void setConnectLazily(boolean connectLazily) {
+		this.connectLazily = connectLazily;
+	}
+
 	/**
 	 * Specify concurrency limits via a "lower-upper" String, e.g. "5-10", or a simple
 	 * upper limit String, e.g. "10".
@@ -159,6 +175,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 		this.taskExecutor = taskExecutor;
 	}
 
+	@Override
 	protected void validateConfiguration() {
 		super.validateConfiguration();
 		if (isSubscriptionDurable() && this.concurrentConsumers != 1) {
@@ -174,6 +191,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 	/**
 	 * Always use a shared JMS Connection.
 	 */
+	@Override
 	protected final boolean sharedConnectionEnabled() {
 		return true;
 	}
@@ -183,15 +201,25 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 	 * in the form of a JMS Session plus associated MessageConsumer.
 	 * @see #createListenerConsumer
 	 */
+	@Override
 	protected void doInitialize() throws JMSException {
-		establishSharedConnection();
-		initializeConsumers();
+		if (!this.connectLazily) {
+			try {
+				establishSharedConnection();
+			}
+			catch (JMSException ex) {
+				logger.debug("Could not connect on initialization - registering message consumers lazily", ex);
+				return;
+			}
+			initializeConsumers();
+		}
 	}
 
 	/**
 	 * Re-initializes this container's JMS message consumers,
 	 * if not initialized already.
 	 */
+	@Override
 	protected void doStart() throws JMSException {
 		super.doStart();
 		initializeConsumers();
@@ -200,6 +228,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 	/**
 	 * Registers this listener container as JMS ExceptionListener on the shared connection.
 	 */
+	@Override
 	protected void prepareSharedConnection(Connection connection) throws JMSException {
 		super.prepareSharedConnection(connection);
 		connection.setExceptionListener(this);
@@ -320,6 +349,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 	/**
 	 * Destroy the registered JMS Sessions and associated MessageConsumers.
 	 */
+	@Override
 	protected void doShutdown() throws JMSException {
 		logger.debug("Closing JMS MessageConsumers");
 		for (MessageConsumer consumer : this.consumers) {