gt: 						messageReceived = (invokeListener() || messageReceived);
context: /*
   * 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.
   * 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.jms.listener;
  
  import java.util.HashSet;
  import java.util.Set;
  import java.util.concurrent.Executor;
  import javax.jms.Connection;
  import javax.jms.JMSException;
  import javax.jms.MessageConsumer;
  import javax.jms.Session;
  
  import org.springframework.core.Constants;
  import org.springframework.core.task.SimpleAsyncTaskExecutor;
  import org.springframework.core.task.TaskExecutor;
  import org.springframework.jms.JmsException;
  import org.springframework.jms.support.JmsUtils;
  import org.springframework.jms.support.destination.CachingDestinationResolver;
  import org.springframework.jms.support.destination.DestinationResolver;
  import org.springframework.scheduling.SchedulingAwareRunnable;
  import org.springframework.scheduling.SchedulingTaskExecutor;
  import org.springframework.util.Assert;
  import org.springframework.util.ClassUtils;
  
  /**
   * Message listener container variant that uses plain JMS client APIs, specifically
   * a loop of {@code MessageConsumer.receive()} calls that also allow for
   * transactional reception of messages (registering them with XA transactions).
   * Designed to work in a native JMS environment as well as in a J2EE environment,
   * with only minimal differences in configuration.
   *
   * <p>This is a simple but nevertheless powerful form of message listener container.
   * On startup, it obtains a fixed number of JMS Sessions to invoke the listener,
   * and optionally allows for dynamic adaptation at runtime (up to a maximum number).
   * Like {@link SimpleMessageListenerContainer}, its main advantage is its low level
   * of runtime complexity, in particular the minimal requirements on the JMS provider:
   * not even the JMS {@code ServerSessionPool} facility is required. Beyond that, it is
   * fully self-recovering in case the broker is temporarily unavailable, and allows
   * for stops/restarts as well as runtime changes to its configuration.
   *
   * <p>Actual {@code MessageListener} execution happens in asynchronous work units which are
   * created through Spring's {@link org.springframework.core.task.TaskExecutor TaskExecutor}
   * abstraction. By default, the specified number of invoker tasks will be created
   * on startup, according to the {@link #setConcurrentConsumers "concurrentConsumers"}
   * setting. Specify an alternative {@code TaskExecutor} to integrate with an existing
   * thread pool facility (such as a J2EE server's), for example using a
   * {@link org.springframework.scheduling.commonj.WorkManagerTaskExecutor CommonJ WorkManager}.
   * With a native JMS setup, each of those listener threads is going to use a
   * cached JMS {@code Session} and {@code MessageConsumer} (only refreshed in case
   * of failure), using the JMS provider's resources as efficiently as possible.
   *
   * <p>Message reception and listener execution can automatically be wrapped
   * in transactions by passing a Spring
   * {@link org.springframework.transaction.PlatformTransactionManager} into the
   * {@link #setTransactionManager "transactionManager"} property. This will usually
   * be a {@link org.springframework.transaction.jta.JtaTransactionManager} in a
   * J2EE environment, in combination with a JTA-aware JMS {@code ConnectionFactory}
   * obtained from JNDI (check your J2EE server's documentation). Note that this
   * listener container will automatically reobtain all JMS handles for each transaction
   * in case an external transaction manager is specified, for compatibility with
   * all J2EE servers (in particular JBoss). This non-caching behavior can be
   * overridden through the {@link #setCacheLevel "cacheLevel"} /
   * {@link #setCacheLevelName "cacheLevelName"} property, enforcing caching of
   * the {@code Connection} (or also {@code Session} and {@code MessageConsumer})
   * even if an external transaction manager is involved.
   *
   * <p>Dynamic scaling of the number of concurrent invokers can be activated
   * by specifying a {@link #setMaxConcurrentConsumers "maxConcurrentConsumers"}
   * value that is higher than the {@link #setConcurrentConsumers "concurrentConsumers"}
   * value. Since the latter's default is 1, you can also simply specify a
   * "maxConcurrentConsumers" of e.g. 5, which will lead to dynamic scaling up to
   * 5 concurrent consumers in case of increasing message load, as well as dynamic
   * shrinking back to the standard number of consumers once the load decreases.
   * Consider adapting the {@link #setIdleTaskExecutionLimit "idleTaskExecutionLimit"}
   * setting to control the lifespan of each new task, to avoid frequent scaling up
   * and down, in particular if the {@code ConnectionFactory} does not pool JMS
   * {@code Sessions} and/or the {@code TaskExecutor} does not pool threads (check
   * your configuration!). Note that dynamic scaling only really makes sense for a
   * queue in the first place; for a topic, you will typically stick with the default
   * number of 1 consumer, otherwise you'd receive the same message multiple times on
   * the same node.
   *
   * <p><b>Note: Don't use Spring's {@link org.springframework.jms.connection.CachingConnectionFactory}
   * in combination with dynamic scaling.</b> Ideally, don't use it with a message
   * listener container at all, since it is generally preferable to let the
   * listener container itself handle appropriate caching within its lifecycle.
   * Also, stopping and restarting a listener container will only work with an
   * independent, locally cached Connection - not with an externally cached one.
   *
   * <p><b>It is strongly recommended to either set {@link #setSessionTransacted
   * "sessionTransacted"} to "true" or specify an external {@link #setTransactionManager
   * "transactionManager"}.</b> See the {@link AbstractMessageListenerContainer}
   * javadoc for details on acknowledge modes and native transaction options,
   * as well as the {@link AbstractPollingMessageListenerContainer} javadoc
   * for details on configuring an external transaction manager.
   *
   * @author Juergen Hoeller
   * @since 2.0
   * @see #setTransactionManager
   * @see #setCacheLevel
   * @see javax.jms.MessageConsumer#receive(long)
   * @see SimpleMessageListenerContainer
   * @see org.springframework.jms.listener.endpoint.JmsMessageEndpointManager
   */
  public class DefaultMessageListenerContainer extends AbstractPollingMessageListenerContainer {
  
  	/**
  	 * Default thread name prefix: "DefaultMessageListenerContainer-".
  	 */
  	public static final String DEFAULT_THREAD_NAME_PREFIX =
  			ClassUtils.getShortName(DefaultMessageListenerContainer.class) + "-";
  
  	/**
  	 * The default recovery interval: 5000 ms = 5 seconds.
  	 */
  	public static final long DEFAULT_RECOVERY_INTERVAL = 5000;
  
  
  	/**
  	 * Constant that indicates to cache no JMS resources at all.
  	 * @see #setCacheLevel
  	 */
  	public static final int CACHE_NONE = 0;
  
  	/**
  	 * Constant that indicates to cache a shared JMS {@code Connection} for each
  	 * listener thread.
  	 * @see #setCacheLevel
  	 */
  	public static final int CACHE_CONNECTION = 1;
  
  	/**
  	 * Constant that indicates to cache a shared JMS {@code Connection} and a JMS
  	 * {@code Session} for each listener thread.
  	 * @see #setCacheLevel
  	 */
  	public static final int CACHE_SESSION = 2;
  
  	/**
  	 * Constant that indicates to cache a shared JMS {@code Connection}, a JMS
  	 * {@code Session}, and a JMS MessageConsumer for each listener thread.
  	 * @see #setCacheLevel
  	 */
  	public static final int CACHE_CONSUMER = 3;
  
  	/**
  	 * Constant that indicates automatic choice of an appropriate caching level
  	 * (depending on the transaction management strategy).
  	 * @see #setCacheLevel
  	 */
  	public static final int CACHE_AUTO = 4;
  
  
  	private static final Constants constants = new Constants(DefaultMessageListenerContainer.class);
  
  
  	private Executor taskExecutor;
  
  	private long recoveryInterval = DEFAULT_RECOVERY_INTERVAL;
  
  	private int cacheLevel = CACHE_AUTO;
  
  	private int concurrentConsumers = 1;
  
  	private int maxConcurrentConsumers = 1;
  
  	private int maxMessagesPerTask = Integer.MIN_VALUE;
  
  	private int idleConsumerLimit = 1;
  
  	private int idleTaskExecutionLimit = 1;
  
  	private final Set<AsyncMessageListenerInvoker> scheduledInvokers = new HashSet<AsyncMessageListenerInvoker>();
  
  	private int activeInvokerCount = 0;
  
  	private int registeredWithDestination = 0;
  
  	private volatile boolean recovering = false;
  
  	private Runnable stopCallback;
  
  	private Object currentRecoveryMarker = new Object();
  
  	private final Object recoveryMonitor = new Object();
  
  
  	/**
  	 * Set the Spring {@code TaskExecutor} to use for running the listener threads.
  	 * <p>Default is a {@link org.springframework.core.task.SimpleAsyncTaskExecutor},
  	 * starting up a number of new threads, according to the specified number
  	 * of concurrent consumers.
  	 * <p>Specify an alternative {@code TaskExecutor} for integration with an existing
  	 * thread pool. Note that this really only adds value if the threads are
  	 * managed in a specific fashion, for example within a J2EE environment.
  	 * A plain thread pool does not add much value, as this listener container
  	 * will occupy a number of threads for its entire lifetime.
  	 * @see #setConcurrentConsumers
  	 * @see org.springframework.core.task.SimpleAsyncTaskExecutor
  	 * @see org.springframework.scheduling.commonj.WorkManagerTaskExecutor
  	 */
  	public void setTaskExecutor(Executor taskExecutor) {
  		this.taskExecutor = taskExecutor;
  	}
  
  	/**
  	 * Specify the interval between recovery attempts, in <b>milliseconds</b>.
  	 * The default is 5000 ms, that is, 5 seconds.
  	 * @see #handleListenerSetupFailure
  	 */
  	public void setRecoveryInterval(long recoveryInterval) {
  		this.recoveryInterval = recoveryInterval;
  	}
  
  	/**
  	 * Specify the level of caching that this listener container is allowed to apply,
  	 * in the form of the name of the corresponding constant: e.g. "CACHE_CONNECTION".
  	 * @see #setCacheLevel
  	 */
  	public void setCacheLevelName(String constantName) throws IllegalArgumentException {
  		if (constantName == null || !constantName.startsWith("CACHE_")) {
  			throw new IllegalArgumentException("Only cache constants allowed");
  		}
  		setCacheLevel(constants.asNumber(constantName).intValue());
  	}
  
  	/**
  	 * Specify the level of caching that this listener container is allowed to apply.
  	 * <p>Default is {@link #CACHE_NONE} if an external transaction manager has been specified
  	 * (to reobtain all resources freshly within the scope of the external transaction),
  	 * and {@link #CACHE_CONSUMER} otherwise (operating with local JMS resources).
  	 * <p>Some J2EE servers only register their JMS resources with an ongoing XA
  	 * transaction in case of a freshly obtained JMS {@code Connection} and {@code Session},
  	 * which is why this listener container by default does not cache any of those.
  	 * However, if you want to optimize for a specific server, consider switching
  	 * this setting to at least {@link #CACHE_CONNECTION} or {@link #CACHE_SESSION}
  	 * even in conjunction with an external transaction manager.
  	 * @see #CACHE_NONE
  	 * @see #CACHE_CONNECTION
  	 * @see #CACHE_SESSION
  	 * @see #CACHE_CONSUMER
  	 * @see #setCacheLevelName
  	 * @see #setTransactionManager
  	 */
  	public void setCacheLevel(int cacheLevel) {
  		this.cacheLevel = cacheLevel;
  	}
  
  	/**
  	 * Return the level of caching that this listener container is allowed to apply.
  	 */
  	public int getCacheLevel() {
  		return this.cacheLevel;
  	}
  
  
  	/**
  	 * Specify concurrency limits via a "lower-upper" String, e.g. "5-10", or a simple
  	 * upper limit String, e.g. "10" (the lower limit will be 1 in this case).
  	 * <p>This listener container will always hold on to the minimum number of consumers
  	 * ({@link #setConcurrentConsumers}) and will slowly scale up to the maximum number
  	 * of consumers {@link #setMaxConcurrentConsumers} in case of increasing load.
  	 */
  	public void setConcurrency(String concurrency) {
  		try {
  			int separatorIndex = concurrency.indexOf('-');
  			if (separatorIndex != -1) {
  				setConcurrentConsumers(Integer.parseInt(concurrency.substring(0, separatorIndex)));
  				setMaxConcurrentConsumers(Integer.parseInt(concurrency.substring(separatorIndex + 1, concurrency.length())));
  			}
  			else {
  				setConcurrentConsumers(1);
  				setMaxConcurrentConsumers(Integer.parseInt(concurrency));
  			}
  		}
  		catch (NumberFormatException ex) {
  			throw new IllegalArgumentException("Invalid concurrency value [" + concurrency + "]: only " +
  					"single maximum integer (e.g. \"5\") and minimum-maximum combo (e.g. \"3-5\") supported.");
  		}
  	}
  
  	/**
  	 * Specify the number of concurrent consumers to create. Default is 1.
  	 * <p>Specifying a higher value for this setting will increase the standard
  	 * level of scheduled concurrent consumers at runtime: This is effectively
  	 * the minimum number of concurrent consumers which will be scheduled
  	 * at any given time. This is a static setting; for dynamic scaling,
  	 * consider specifying the "maxConcurrentConsumers" setting instead.
  	 * <p>Raising the number of concurrent consumers is recommendable in order
  	 * to scale the consumption of messages coming in from a queue. However,
  	 * note that any ordering guarantees are lost once multiple consumers are
  	 * registered. In general, stick with 1 consumer for low-volume queues.
  	 * <p><b>Do not raise the number of concurrent consumers for a topic,
  	 * unless vendor-specific setup measures clearly allow for it.</b>
  	 * With regular setup, this would lead to concurrent consumption
  	 * of the same message, which is hardly ever desirable.
  	 * <p><b>This setting can be modified at runtime, for example through JMX.</b>
  	 * @see #setMaxConcurrentConsumers
  	 */
  	public void setConcurrentConsumers(int concurrentConsumers) {
  		Assert.isTrue(concurrentConsumers > 0, "'concurrentConsumers' value must be at least 1 (one)");
  		synchronized (this.lifecycleMonitor) {
  			this.concurrentConsumers = concurrentConsumers;
  			if (this.maxConcurrentConsumers < concurrentConsumers) {
  				this.maxConcurrentConsumers = concurrentConsumers;
  			}
  		}
  	}
  
  	/**
  	 * Return the "concurrentConsumer" setting.
  	 * <p>This returns the currently configured "concurrentConsumers" value;
  	 * the number of currently scheduled/active consumers might differ.
  	 * @see #getScheduledConsumerCount()
  	 * @see #getActiveConsumerCount()
  	 */
  	public final int getConcurrentConsumers() {
  		synchronized (this.lifecycleMonitor) {
  			return this.concurrentConsumers;
  		}
  	}
  
  	/**
  	 * Specify the maximum number of concurrent consumers to create. Default is 1.
  	 * <p>If this setting is higher than "concurrentConsumers", the listener container
  	 * will dynamically schedule new consumers at runtime, provided that enough
  	 * incoming messages are encountered. Once the load goes down again, the number of
  	 * consumers will be reduced to the standard level ("concurrentConsumers") again.
  	 * <p>Raising the number of concurrent consumers is recommendable in order
  	 * to scale the consumption of messages coming in from a queue. However,
  	 * note that any ordering guarantees are lost once multiple consumers are
  	 * registered. In general, stick with 1 consumer for low-volume queues.
  	 * <p><b>Do not raise the number of concurrent consumers for a topic,
  	 * unless vendor-specific setup measures clearly allow for it.</b>
  	 * With regular setup, this would lead to concurrent consumption
  	 * of the same message, which is hardly ever desirable.
  	 * <p><b>This setting can be modified at runtime, for example through JMX.</b>
  	 * @see #setConcurrentConsumers
  	 */
  	public void setMaxConcurrentConsumers(int maxConcurrentConsumers) {
  		Assert.isTrue(maxConcurrentConsumers > 0, "'maxConcurrentConsumers' value must be at least 1 (one)");
  		synchronized (this.lifecycleMonitor) {
  			this.maxConcurrentConsumers =
  					(maxConcurrentConsumers > this.concurrentConsumers ? maxConcurrentConsumers : this.concurrentConsumers);
  		}
  	}
  
  	/**
  	 * Return the "maxConcurrentConsumer" setting.
  	 * <p>This returns the currently configured "maxConcurrentConsumers" value;
  	 * the number of currently scheduled/active consumers might differ.
  	 * @see #getScheduledConsumerCount()
  	 * @see #getActiveConsumerCount()
  	 */
  	public final int getMaxConcurrentConsumers() {
  		synchronized (this.lifecycleMonitor) {
  			return this.maxConcurrentConsumers;
  		}
  	}
  
  	/**
  	 * Specify the maximum number of messages to process in one task.
  	 * More concretely, this limits the number of message reception attempts
  	 * per task, which includes receive iterations that did not actually
  	 * pick up a message until they hit their timeout (see the
  	 * {@link #setReceiveTimeout "receiveTimeout"} property).
  	 * <p>Default is unlimited (-1) in case of a standard TaskExecutor,
  	 * reusing the original invoker threads until shutdown (at the
  	 * expense of limited dynamic scheduling).
  	 * <p>In case of a SchedulingTaskExecutor indicating a preference for
  	 * short-lived tasks, the default is 10 instead. Specify a number
  	 * of 10 to 100 messages to balance between rather long-lived and
  	 * rather short-lived tasks here.
  	 * <p>Long-lived tasks avoid frequent thread context switches through
  	 * sticking with the same thread all the way through, while short-lived
  	 * tasks allow thread pools to control the scheduling. Hence, thread
  	 * pools will usually prefer short-lived tasks.
  	 * <p><b>This setting can be modified at runtime, for example through JMX.</b>
  	 * @see #setTaskExecutor
  	 * @see #setReceiveTimeout
  	 * @see org.springframework.scheduling.SchedulingTaskExecutor#prefersShortLivedTasks()
  	 */
  	public void setMaxMessagesPerTask(int maxMessagesPerTask) {
  		Assert.isTrue(maxMessagesPerTask != 0, "'maxMessagesPerTask' must not be 0");
  		synchronized (this.lifecycleMonitor) {
  			this.maxMessagesPerTask = maxMessagesPerTask;
  		}
  	}
  
  	/**
  	 * Return the maximum number of messages to process in one task.
  	 */
  	public final int getMaxMessagesPerTask() {
  		synchronized (this.lifecycleMonitor) {
  			return this.maxMessagesPerTask;
  		}
  	}
  
  	/**
  	 * Specify the limit for the number of consumers that are allowed to be idle
  	 * at any given time.
  	 * <p>This limit is used by the {@link #scheduleNewInvokerIfAppropriate} method
  	 * to determine if a new invoker should be created. Increasing the limit causes
  	 * invokers to be created more aggressively. This can be useful to ramp up the
  	 * number of invokers faster.
  	 * <p>The default is 1, only scheduling a new invoker (which is likely to
  	 * be idle initially) if none of the existing invokers is currently idle.
  	 */
  	public void setIdleConsumerLimit(int idleConsumerLimit) {
  		Assert.isTrue(idleConsumerLimit > 0, "'idleConsumerLimit' must be 1 or higher");
  		synchronized (this.lifecycleMonitor) {
  			this.idleConsumerLimit = idleConsumerLimit;
  		}
  	}
  
  	/**
  	 * Return the limit for the number of idle consumers.
  	 */
  	public final int getIdleConsumerLimit() {
  		synchronized (this.lifecycleMonitor) {
  			return this.idleConsumerLimit;
  		}
  	}
  
  	/**
  	 * Specify the limit for idle executions of a consumer task, not having
  	 * received any message within its execution. If this limit is reached,
  	 * the task will shut down and leave receiving to other executing tasks.
  	 * <p>The default is 1, closing idle resources early once a task didn't
  	 * receive a message. This applies to dynamic scheduling only; see the
  	 * {@link #setMaxConcurrentConsumers "maxConcurrentConsumers"} setting.
  	 * The minimum number of consumers
  	 * (see {@link #setConcurrentConsumers "concurrentConsumers"})
  	 * will be kept around until shutdown in any case.
  	 * <p>Within each task execution, a number of message reception attempts
  	 * (according to the "maxMessagesPerTask" setting) will each wait for an incoming
  	 * message (according to the "receiveTimeout" setting). If all of those receive
  	 * attempts in a given task return without a message, the task is considered
  	 * idle with respect to received messages. Such a task may still be rescheduled;
  	 * however, once it reached the specified "idleTaskExecutionLimit", it will
  	 * shut down (in case of dynamic scaling).
  	 * <p>Raise this limit if you encounter too frequent scaling up and down.
  	 * With this limit being higher, an idle consumer will be kept around longer,
  	 * avoiding the restart of a consumer once a new load of messages comes in.
  	 * Alternatively, specify a higher "maxMessagesPerTask" and/or "receiveTimeout" value,
  	 * which will also lead to idle consumers being kept around for a longer time
  	 * (while also increasing the average execution time of each scheduled task).
  	 * <p><b>This setting can be modified at runtime, for example through JMX.</b>
  	 * @see #setMaxMessagesPerTask
  	 * @see #setReceiveTimeout
  	 */
  	public void setIdleTaskExecutionLimit(int idleTaskExecutionLimit) {
  		Assert.isTrue(idleTaskExecutionLimit > 0, "'idleTaskExecutionLimit' must be 1 or higher");
  		synchronized (this.lifecycleMonitor) {
  			this.idleTaskExecutionLimit = idleTaskExecutionLimit;
  		}
  	}
  
  	/**
  	 * Return the limit for idle executions of a consumer task.
  	 */
  	public final int getIdleTaskExecutionLimit() {
  		synchronized (this.lifecycleMonitor) {
  			return this.idleTaskExecutionLimit;
  		}
  	}
  
  
  	//-------------------------------------------------------------------------
  	// Implementation of AbstractMessageListenerContainer's template methods
  	//-------------------------------------------------------------------------
  
  	@Override
  	public void initialize() {
  		// Adapt default cache level.
  		if (this.cacheLevel == CACHE_AUTO) {
  			this.cacheLevel = (getTransactionManager() != null ? CACHE_NONE : CACHE_CONSUMER);
  		}
  
  		// Prepare taskExecutor and maxMessagesPerTask.
  		synchronized (this.lifecycleMonitor) {
  			if (this.taskExecutor == null) {
  				this.taskExecutor = createDefaultTaskExecutor();
  			}
  			else if (this.taskExecutor instanceof SchedulingTaskExecutor &&
  					((SchedulingTaskExecutor) this.taskExecutor).prefersShortLivedTasks() &&
  					this.maxMessagesPerTask == Integer.MIN_VALUE) {
  				// TaskExecutor indicated a preference for short-lived tasks. According to
  				// setMaxMessagesPerTask javadoc, we'll use 10 message per task in this case
  				// unless the user specified a custom value.
  				this.maxMessagesPerTask = 10;
  			}
  		}
  
  		// Proceed with actual listener initialization.
  		super.initialize();
  	}
  
  	/**
  	 * Creates the specified number of concurrent consumers,
  	 * in the form of a JMS Session plus associated MessageConsumer
  	 * running in a separate thread.
  	 * @see #scheduleNewInvoker
  	 * @see #setTaskExecutor
  	 */
  	@Override
  	protected void doInitialize() throws JMSException {
  		synchronized (this.lifecycleMonitor) {
  			for (int i = 0; i < this.concurrentConsumers; i++) {
  				scheduleNewInvoker();
  			}
  		}
  	}
  
  	/**
  	 * Destroy the registered JMS Sessions and associated MessageConsumers.
  	 */
  	@Override
  	protected void doShutdown() throws JMSException {
  		logger.debug("Waiting for shutdown of message listener invokers");
  		try {
  			synchronized (this.lifecycleMonitor) {
  				// Waiting for AsyncMessageListenerInvokers to deactivate themselves...
  				while (this.activeInvokerCount > 0) {
  					if (logger.isDebugEnabled()) {
  						logger.debug("Still waiting for shutdown of " + this.activeInvokerCount +
  								" message listener invokers");
  					}
  					this.lifecycleMonitor.wait();
  				}
  				// Clear remaining scheduled invokers, possibly left over as paused tasks...
  				for (AsyncMessageListenerInvoker scheduledInvoker : this.scheduledInvokers) {
  					scheduledInvoker.clearResources();
  				}
  				this.scheduledInvokers.clear();
  			}
  		}
  		catch (InterruptedException ex) {
  			// Re-interrupt current thread, to allow other threads to react.
  			Thread.currentThread().interrupt();
  		}
  	}
  
  	/**
  	 * Overridden to reset the stop callback, if any.
  	 */
  	@Override
  	public void start() throws JmsException {
  		synchronized (this.lifecycleMonitor) {
  			this.stopCallback = null;
  		}
  		super.start();
  	}
  
  	/**
  	 * Stop this listener container, invoking the specific callback
  	 * once all listener processing has actually stopped.
  	 * <p>Note: Further {@code stop(runnable)} calls (before processing
  	 * has actually stopped) will override the specified callback. Only the
  	 * latest specified callback will be invoked.
  	 * <p>If a subsequent {@link #start()} call restarts the listener container
  	 * before it has fully stopped, the callback will not get invoked at all.
  	 * @param callback the callback to invoke once listener processing
  	 * has fully stopped
  	 * @throws JmsException if stopping failed
  	 * @see #stop()
  	 */
  	@Override
  	public void stop(Runnable callback) throws JmsException {
  		synchronized (this.lifecycleMonitor) {
  			this.stopCallback = callback;
  		}
  		stop();
  	}
  
  	/**
  	 * Return the number of currently scheduled consumers.
  	 * <p>This number will always be between "concurrentConsumers" and
  	 * "maxConcurrentConsumers", but might be higher than "activeConsumerCount"
  	 * (in case some consumers are scheduled but not executing at the moment).
  	 * @see #getConcurrentConsumers()
  	 * @see #getMaxConcurrentConsumers()
  	 * @see #getActiveConsumerCount()
  	 */
  	public final int getScheduledConsumerCount() {
  		synchronized (this.lifecycleMonitor) {
  			return this.scheduledInvokers.size();
  		}
  	}
  
  	/**
  	 * Return the number of currently active consumers.
  	 * <p>This number will always be between "concurrentConsumers" and
  	 * "maxConcurrentConsumers", but might be lower than "scheduledConsumerCount"
  	 * (in case some consumers are scheduled but not executing at the moment).
  	 * @see #getConcurrentConsumers()
  	 * @see #getMaxConcurrentConsumers()
  	 * @see #getActiveConsumerCount()
  	 */
  	public final int getActiveConsumerCount() {
  		synchronized (this.lifecycleMonitor) {
  			return this.activeInvokerCount;
  		}
  	}
  
  	/**
  	 * Return whether at least one consumer has entered a fixed registration with the
  	 * target destination. This is particularly interesting for the pub-sub case where
  	 * it might be important to have an actual consumer registered that is guaranteed
  	 * not to miss any messages that are just about to be published.
  	 * <p>This method may be polled after a {@link #start()} call, until asynchronous
  	 * registration of consumers has happened which is when the method will start returning
  	 * {@code true} &ndash; provided that the listener container ever actually establishes
  	 * a fixed registration. It will then keep returning {@code true} until shutdown,
  	 * since the container will hold on to at least one consumer registration thereafter.
  	 * <p>Note that a listener container is not bound to having a fixed registration in
  	 * the first place. It may also keep recreating consumers for every invoker execution.
  	 * This particularly depends on the {@link #setCacheLevel cache level} setting:
  	 * only {@link #CACHE_CONSUMER} will lead to a fixed registration.
  	 */
  	public boolean isRegisteredWithDestination() {
  		synchronized (this.lifecycleMonitor) {
  			return (this.registeredWithDestination > 0);
  		}
  	}
  
  
  	/**
  	 * Create a default TaskExecutor. Called if no explicit TaskExecutor has been specified.
  	 * <p>The default implementation builds a {@link org.springframework.core.task.SimpleAsyncTaskExecutor}
  	 * with the specified bean name (or the class name, if no bean name specified) as thread name prefix.
  	 * @see org.springframework.core.task.SimpleAsyncTaskExecutor#SimpleAsyncTaskExecutor(String)
  	 */
  	protected TaskExecutor createDefaultTaskExecutor() {
  		String beanName = getBeanName();
  		String threadNamePrefix = (beanName != null ? beanName + "-" : DEFAULT_THREAD_NAME_PREFIX);
  		return new SimpleAsyncTaskExecutor(threadNamePrefix);
  	}
  
  	/**
  	 * Schedule a new invoker, increasing the total number of scheduled
  	 * invokers for this listener container.
  	 */
  	private void scheduleNewInvoker() {
  		AsyncMessageListenerInvoker invoker = new AsyncMessageListenerInvoker();
  		if (rescheduleTaskIfNecessary(invoker)) {
  			// This should always be true, since we're only calling this when active.
  			this.scheduledInvokers.add(invoker);
  		}
  	}
  
  	/**
  	 * Use a shared JMS Connection depending on the "cacheLevel" setting.
  	 * @see #setCacheLevel
  	 * @see #CACHE_CONNECTION
  	 */
  	@Override
  	protected final boolean sharedConnectionEnabled() {
  		return (getCacheLevel() >= CACHE_CONNECTION);
  	}
  
  	/**
  	 * Re-executes the given task via this listener container's TaskExecutor.
  	 * @see #setTaskExecutor
  	 */
  	@Override
  	protected void doRescheduleTask(Object task) {
  		this.taskExecutor.execute((Runnable) task);
  	}
  
  	/**
  	 * 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();
  	}
  
  	/**
  	 * Marks the affected invoker as idle.
  	 */
  	@Override
  	protected void noMessageReceived(Object invoker, Session session) {
  		((AsyncMessageListenerInvoker) invoker).setIdle(true);
  	}
  
  	/**
  	 * Schedule a new invoker, increasing the total number of scheduled
  	 * invokers for this listener container, but only if the specified
  	 * "maxConcurrentConsumers" limit has not been reached yet, and only
  	 * if the specified "idleConsumerLimit" has not been reached either.
  	 * <p>Called once a message has been received, in order to scale up while
  	 * processing the message in the invoker that originally received it.
  	 * @see #setTaskExecutor
  	 * @see #getMaxConcurrentConsumers()
  	 * @see #getIdleConsumerLimit()
  	 */
  	protected void scheduleNewInvokerIfAppropriate() {
  		if (isRunning()) {
  			resumePausedTasks();
  			synchronized (this.lifecycleMonitor) {
  				if (this.scheduledInvokers.size() < this.maxConcurrentConsumers &&
  						getIdleInvokerCount() < this.idleConsumerLimit) {
  					scheduleNewInvoker();
  					if (logger.isDebugEnabled()) {
  						logger.debug("Raised scheduled invoker count: " + this.scheduledInvokers.size());
  					}
  				}
  			}
  		}
  	}
  
  	/**
  	 * Determine whether the current invoker should be rescheduled,
  	 * given that it might not have received a message in a while.
  	 * @param idleTaskExecutionCount the number of idle executions
  	 * that this invoker task has already accumulated (in a row)
  	 */
  	private boolean shouldRescheduleInvoker(int idleTaskExecutionCount) {
  		boolean superfluous =
  				(idleTaskExecutionCount >= this.idleTaskExecutionLimit && getIdleInvokerCount() > 1);
  		return (this.scheduledInvokers.size() <=
  				(superfluous ? this.concurrentConsumers : this.maxConcurrentConsumers));
  	}
  
  	/**
  	 * Determine whether this listener container currently has more
  	 * than one idle instance among its scheduled invokers.
  	 */
  	private int getIdleInvokerCount() {
  		int count = 0;
  		for (AsyncMessageListenerInvoker invoker : this.scheduledInvokers) {
  			if (invoker.isIdle()) {
  				count++;
  			}
  		}
  		return count;
  	}
  
  
  	/**
  	 * Overridden to accept a failure in the initial setup - leaving it up to the
  	 * asynchronous invokers to establish the shared Connection on first access.
  	 * @see #refreshConnectionUntilSuccessful()
  	 */
  	@Override
  	protected void establishSharedConnection() {
  		try {
  			super.establishSharedConnection();
  		}
  		catch (Exception ex) {
  			if (ex instanceof JMSException) {
  				invokeExceptionListener((JMSException) ex);
  			}
  			logger.debug("Could not establish shared JMS Connection - " +
  					"leaving it up to asynchronous invokers to establish a Connection as soon as possible", ex);
  		}
  	}
  
  	/**
  	 * This implementations proceeds even after an exception thrown from
  	 * {@code Connection.start()}, relying on listeners to perform
  	 * appropriate recovery.
  	 */
  	@Override
  	protected void startSharedConnection() {
  		try {
  			super.startSharedConnection();
  		}
  		catch (Exception ex) {
  			logger.debug("Connection start failed - relying on listeners to perform recovery", ex);
  		}
  	}
  
  	/**
  	 * This implementations proceeds even after an exception thrown from
  	 * {@code Connection.stop()}, relying on listeners to perform
  	 * appropriate recovery after a restart.
  	 */
  	@Override
  	protected void stopSharedConnection() {
  		try {
  			super.stopSharedConnection();
  		}
  		catch (Exception ex) {
  			logger.debug("Connection stop failed - relying on listeners to perform recovery after restart", ex);
  		}
  	}
  
  	/**
  	 * Handle the given exception that arose during setup of a listener.
  	 * Called for every such exception in every concurrent listener.
  	 * <p>The default implementation logs the exception at warn level
  	 * if not recovered yet, and at debug level if already recovered.
  	 * Can be overridden in subclasses.
  	 * @param ex the exception to handle
  	 * @param alreadyRecovered whether a previously executing listener
  	 * already recovered from the present listener setup failure
  	 * (this usually indicates a follow-up failure than can be ignored
  	 * other than for debug log purposes)
  	 * @see #recoverAfterListenerSetupFailure()
  	 */
  	protected void handleListenerSetupFailure(Throwable ex, boolean alreadyRecovered) {
  		if (ex instanceof JMSException) {
  			invokeExceptionListener((JMSException) ex);
  		}
  		if (ex instanceof SharedConnectionNotInitializedException) {
  			if (!alreadyRecovered) {
  				logger.info("JMS message listener invoker needs to establish shared Connection");
  			}
  		}
  		else {
  			// Recovery during active operation..
  			if (alreadyRecovered) {
  				logger.debug("Setup of JMS message listener invoker failed - already recovered by other invoker", ex);
  			}
  			else {
  				StringBuilder msg = new StringBuilder();
  				msg.append("Setup of JMS message listener invoker failed for destination '");
  				msg.append(getDestinationDescription()).append("' - trying to recover. Cause: ");
  				msg.append(ex instanceof JMSException ? JmsUtils.buildExceptionMessage((JMSException) ex) : ex.getMessage());
  				if (logger.isDebugEnabled()) {
  					logger.warn(msg, ex);
  				}
  				else {
  					logger.warn(msg);
  				}
  			}
  		}
  	}
  
  	/**
  	 * Recover this listener container after a listener failed to set itself up,
  	 * for example re-establishing the underlying Connection.
  	 * <p>The default implementation delegates to DefaultMessageListenerContainer's
  	 * recovery-capable {@link #refreshConnectionUntilSuccessful()} method, which will
  	 * try to re-establish a Connection to the JMS provider both for the shared
  	 * and the non-shared Connection case.
  	 * @see #refreshConnectionUntilSuccessful()
  	 * @see #refreshDestination()
  	 */
  	protected void recoverAfterListenerSetupFailure() {
  		this.recovering = true;
  		try {
  			refreshConnectionUntilSuccessful();
  			refreshDestination();
  		}
  		finally {
  			this.recovering = false;
  		}
  	}
  
  	/**
  	 * Refresh the underlying Connection, not returning before an attempt has been
  	 * successful. Called in case of a shared Connection as well as without shared
  	 * Connection, so either needs to operate on the shared Connection or on a
  	 * temporary Connection that just gets established for validation purposes.
  	 * <p>The default implementation retries until it successfully established a
  	 * Connection, for as long as this message listener container is running.
  	 * Applies the specified recovery interval between retries.
  	 * @see #setRecoveryInterval
  	 * @see #start()
  	 * @see #stop()
  	 */
  	protected void refreshConnectionUntilSuccessful() {
  		while (isRunning()) {
  			try {
  				if (sharedConnectionEnabled()) {
  					refreshSharedConnection();
  				}
  				else {
  					Connection con = createConnection();
  					JmsUtils.closeConnection(con);
  				}
  				logger.info("Successfully refreshed JMS Connection");
  				break;
  			}
  			catch (Exception ex) {
  				if (ex instanceof JMSException) {
  					invokeExceptionListener((JMSException) ex);
  				}
  				StringBuilder msg = new StringBuilder();
  				msg.append("Could not refresh JMS Connection for destination '");
  				msg.append(getDestinationDescription()).append("' - retrying in ");
  				msg.append(this.recoveryInterval).append(" ms. Cause: ");
  				msg.append(ex instanceof JMSException ? JmsUtils.buildExceptionMessage((JMSException) ex) : ex.getMessage());
  				if (logger.isDebugEnabled()) {
  					logger.error(msg, ex);
  				}
  				else {
  					logger.error(msg);
  				}
  			}
  			sleepInbetweenRecoveryAttempts();
  		}
  	}
  
  	/**
  	 * Refresh the JMS destination that this listener container operates on.
  	 * <p>Called after listener setup failure, assuming that a cached Destination
  	 * object might have become invalid (a typical case on WebLogic JMS).
  	 * <p>The default implementation removes the destination from a
  	 * DestinationResolver's cache, in case of a CachingDestinationResolver.
  	 * @see #setDestinationName
  	 * @see org.springframework.jms.support.destination.CachingDestinationResolver
  	 */
  	protected void refreshDestination() {
  		String destName = getDestinationName();
  		if (destName != null) {
  			DestinationResolver destResolver = getDestinationResolver();
  			if (destResolver instanceof CachingDestinationResolver) {
  				((CachingDestinationResolver) destResolver).removeFromCache(destName);
  			}
  		}
  	}
  
  	/**
  	 * Sleep according to the specified recovery interval.
  	 * Called between recovery attempts.
  	 */
  	protected void sleepInbetweenRecoveryAttempts() {
  		if (this.recoveryInterval > 0) {
  			try {
  				Thread.sleep(this.recoveryInterval);
  			}
  			catch (InterruptedException interEx) {
  				// Re-interrupt current thread, to allow other threads to react.
  				Thread.currentThread().interrupt();
  			}
  		}
  	}
  
  	/**
  	 * Return whether this listener container is currently in a recovery attempt.
  	 * <p>May be used to detect recovery phases but also the end of a recovery phase,
  	 * with {@code isRecovering()} switching to {@code false} after having been found
  	 * to return {@code true} before.
  	 * @see #recoverAfterListenerSetupFailure()
  	 */
  	public final boolean isRecovering() {
  		return this.recovering;
  	}
  
  
  	//-------------------------------------------------------------------------
  	// Inner classes used as internal adapters
  	//-------------------------------------------------------------------------
  
  	/**
  	 * Runnable that performs looped {@code MessageConsumer.receive()} calls.
  	 */
  	private class AsyncMessageListenerInvoker implements SchedulingAwareRunnable {
  
  		private Session session;
  
  		private MessageConsumer consumer;
  
  		private Object lastRecoveryMarker;
  
  		private boolean lastMessageSucceeded;
  
  		private int idleTaskExecutionCount = 0;
  
  		private volatile boolean idle = true;
  
  		public void run() {
  			synchronized (lifecycleMonitor) {
  				activeInvokerCount++;
  				lifecycleMonitor.notifyAll();
  			}
  			boolean messageReceived = false;
  			try {
  				if (maxMessagesPerTask < 0) {
  					messageReceived = executeOngoingLoop();
  				}
  				else {
  					int messageCount = 0;
  					while (isRunning() && messageCount < maxMessagesPerTask) {
  