Also see:

setCacheLevel

Also see:

setCacheLevel

Also see:

setCacheLevel

Also see:

setCacheLevel

Also see:

setCacheLevel

 protected TaskExecutor createDefaultTaskExecutor() 
{
		String beanName = getBeanName();
		String threadNamePrefix = (beanName != null ? beanName + "-" : DEFAULT_THREAD_NAME_PREFIX);
		return new SimpleAsyncTaskExecutor(threadNamePrefix);
}

Create a default TaskExecutor. Called if no explicit TaskExecutor has been specified.

The default implementation builds a org.springframework.core.task.SimpleAsyncTaskExecutor with the specified bean name (or the class name, if no bean name specified) as thread name prefix.

 protected  void doInitialize() throws JMSException 
{
		synchronized (this.lifecycleMonitor) {
			for (int i = 0; i <  this.concurrentConsumers; i++) {
				scheduleNewInvoker();
			}
		}
}

Creates the specified number of concurrent consumers, in the form of a JMS Session plus associated MessageConsumer running in a separate thread.

 protected  void doRescheduleTask(Object task) 
{
		this.taskExecutor.execute((Runnable) task);
}

Re-executes the given task via this listener container's TaskExecutor.

 protected  void doShutdown() throws JMSException 
{
		logger.debug("Waiting for shutdown of message listener invokers");
		try {
			synchronized (this.lifecycleMonitor) {
				while (this.activeInvokerCount  > 0) {
					if (logger.isDebugEnabled()) {
						logger.debug("Still waiting for shutdown of " + this.activeInvokerCount +
								" message listener invokers");
					}
					this.lifecycleMonitor.wait();
				}
			}
		}
		catch (InterruptedException ex) {
			// Re-interrupt current thread, to allow other threads to react.
			Thread.currentThread().interrupt();
		}
}

Destroy the registered JMS Sessions and associated MessageConsumers.

 protected  void establishSharedConnection() 
{
		try {
			super.establishSharedConnection();
		}
		catch (Exception ex) {
			logger.debug("Could not establish shared JMS Connection - " +
					"leaving it up to asynchronous invokers to establish a Connection as soon as possible", ex);
		}
}

Overridden to accept a failure in the initial setup - leaving it up to the asynchronous invokers to establish the shared Connection on first access.

 public final int getActiveConsumerCount() 
{
		synchronized (this.lifecycleMonitor) {
			return this.activeInvokerCount;
		}
}

Return the number of currently active consumers.

This number will always be inbetween "concurrentConsumers" and "maxConcurrentConsumers", but might be lower than "scheduledConsumerCount". (in case of some consumers being scheduled but not executed at the moment).

 public int getCacheLevel() 
{
		return this.cacheLevel;
}

Return the level of caching that this listener container is allowed to apply.

 public final int getConcurrentConsumers() 
{
		synchronized (this.lifecycleMonitor) {
			return this.concurrentConsumers;
		}
}

Return the "concurrentConsumer" setting.

This returns the currently configured "concurrentConsumers" value; the number of currently scheduled/active consumers might differ.

 public int getIdleTaskExecutionLimit() 
{
		synchronized (this.lifecycleMonitor) {
			return this.idleTaskExecutionLimit;
		}
}

Return the limit for idle executions of a receive task.

 public final int getMaxConcurrentConsumers() 
{
		synchronized (this.lifecycleMonitor) {
			return this.maxConcurrentConsumers;
		}
}

Return the "maxConcurrentConsumer" setting.

This returns the currently configured "maxConcurrentConsumers" value; the number of currently scheduled/active consumers might differ.

 public int getMaxMessagesPerTask() 
{
		synchronized (this.lifecycleMonitor) {
			return this.maxMessagesPerTask;
		}
}

Return the maximum number of messages to process in one task.

 public final int getScheduledConsumerCount() 
{
		synchronized (this.lifecycleMonitor) {
			return this.scheduledInvokers.size();
		}
}

Return the number of currently scheduled consumers.

This number will always be inbetween "concurrentConsumers" and "maxConcurrentConsumers", but might be higher than "activeConsumerCount" (in case of some consumers being scheduled but not executed at the moment).

 protected  void handleListenerSetupFailure(Throwable ex,
    boolean alreadyRecovered) 
{
		if (ex instanceof JMSException) {
			invokeExceptionListener((JMSException) ex);
		}
		if (ex instanceof SharedConnectionNotInitializedException) {
			if (!alreadyRecovered) {
				logger.debug("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 {
				StringBuffer msg = new StringBuffer();
				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.info(msg, ex);
				}
				else {
					logger.info(msg);
				}
			}
		}
}

Handle the given exception that arose during setup of a listener. Called for every such exception in every concurrent listener.

The default implementation logs the exception at error level if not recovered yet, and at debug level if already recovered. Can be overridden in subclasses.

 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();
}

 protected  void messageReceived(Object invoker,
    Session session) 
{
		((AsyncMessageListenerInvoker) invoker).setIdle(false);
		scheduleNewInvokerIfAppropriate();
}

Tries scheduling a new invoker, since we know messages are coming in...

 protected  void noMessageReceived(Object invoker,
    Session session) 
{
		((AsyncMessageListenerInvoker) invoker).setIdle(true);
}

Marks the affected invoker as idle.

 protected  void recoverAfterListenerSetupFailure() 
{
		refreshConnectionUntilSuccessful();
		refreshDestination();
}

Recover this listener container after a listener failed to set itself up, for example reestablishing the underlying Connection.

The default implementation delegates to DefaultMessageListenerContainer's recovery-capable #refreshConnectionUntilSuccessful() method, which will try to re-establish a Connection to the JMS provider both for the shared and the non-shared Connection case.

 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) {
				StringBuffer msg = new StringBuffer();
				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.info(msg, ex);
				}
				else if (logger.isInfoEnabled()) {
					logger.info(msg);
				}
			}
			sleepInbetweenRecoveryAttempts();
		}
}

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.

The default implementation retries until it successfully established a Connection, for as long as this message listener container is active. Applies the specified recovery interval between retries.

 protected  void refreshDestination() 
{
		String destName = getDestinationName();
		if (destName != null) {
			DestinationResolver destResolver = getDestinationResolver();
			if (destResolver instanceof CachingDestinationResolver) {
				((CachingDestinationResolver) destResolver).removeFromCache(destName);
			}
		}
}

Refresh the JMS destination that this listener container operates on.

Called after listener setup failure, assuming that a cached Destination object might have become invalid (a typical case on WebLogic JMS).

The default implementation removes the destination from a DestinationResolver's cache, in case of a CachingDestinationResolver.

 protected  void scheduleNewInvokerIfAppropriate() 
{
		if (isRunning()) {
			resumePausedTasks();
			synchronized (this.lifecycleMonitor) {
				if (this.scheduledInvokers.size() <  this.maxConcurrentConsumers && getIdleInvokerCount() == 0) {
					scheduleNewInvoker();
					if (logger.isDebugEnabled()) {
						logger.debug("Raised scheduled invoker count: " + this.scheduledInvokers.size());
					}
				}
			}
		}
}

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 this listener container does not currently have idle invokers that are waiting for new messages already.

Called once a message has been received, to scale up while processing the message in the invoker that originally received it.

 public  void setCacheLevel(int cacheLevel) 
{
		this.cacheLevel = cacheLevel;
}

Specify the level of caching that this listener container is allowed to apply.

Default is CACHE_NONE if an external transaction manager has been specified (to reobtain all resources freshly within the scope of the external transaction), and CACHE_CONSUMER else (operating with local JMS resources).

Some J2EE servers only register their JMS resources with an ongoing XA transaction in case of a freshly obtained JMS Connection and Session, which is why this listener container does by default not cache any of those. However, if you want to optimize for a specific server, consider switching this setting to at least CACHE_CONNECTION or CACHE_SESSION even in conjunction with an external transaction manager.

Currently known servers that absolutely require CACHE_NONE for XA transaction processing: JBoss 4. For any others, consider raising the cache level.

 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, in the form of the name of the corresponding constant: e.g. "CACHE_CONNECTION".

 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;
			}
		}
}

Specify the number of concurrent consumers to create. Default is 1.

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.

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.

Do not raise the number of concurrent consumers for a topic. This would lead to concurrent consumption of the same message, which is hardly ever desirable.

This setting can be modified at runtime, for example through JMX.

 public  void setIdleTaskExecutionLimit(int idleTaskExecutionLimit) 
{
		Assert.isTrue(idleTaskExecutionLimit  > 0, "'idleTaskExecutionLimit' must be 1 or higher");
		synchronized (this.lifecycleMonitor) {
			this.idleTaskExecutionLimit = idleTaskExecutionLimit;
		}
}

Specify the limit for idle executions of a receive 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.

Default is 1, closing idle resources early once a task didn't receive a message. This applies to dynamic scheduling only; see the "maxConcurrentConsumers" setting. The minimum number of consumers (see "concurrentConsumers" ) will be kept around until shutdown in any case.

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).

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).

This setting can be modified at runtime, for example through JMX.

 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);
		}
}

Specify the maximum number of concurrent consumers to create. Default is 1.

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.

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.

Do not raise the number of concurrent consumers for a topic. This would lead to concurrent consumption of the same message, which is hardly ever desirable.

This setting can be modified at runtime, for example through JMX.

 public  void setMaxMessagesPerTask(int maxMessagesPerTask) 
{
		Assert.isTrue(maxMessagesPerTask != 0, "'maxMessagesPerTask' must not be 0");
		synchronized (this.lifecycleMonitor) {
			this.maxMessagesPerTask = maxMessagesPerTask;
		}
}

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 "receiveTimeout" property).

Default is unlimited (-1) in case of a standard TaskExecutor, reusing the original invoker threads until shutdown (at the expense of limited dynamic scheduling).

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.

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.

This setting can be modified at runtime, for example through JMX.

 public  void setRecoveryInterval(long recoveryInterval) 
{
		this.recoveryInterval = recoveryInterval;
}

Specify the interval between recovery attempts, in milliseconds. The default is 5000 ms, that is, 5 seconds.

 public  void setTaskExecutor(TaskExecutor taskExecutor) 
{
	                                                                                             	 
		this.taskExecutor = taskExecutor;
}

Set the Spring TaskExecutor to use for running the listener threads.

Default is a org.springframework.core.task.SimpleAsyncTaskExecutor , starting up a number of new threads, according to the specified number of concurrent consumers.

Specify an alternative 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.

 protected final boolean sharedConnectionEnabled() 
{
		return (getCacheLevel()  >= CACHE_CONNECTION);
}

Use a shared JMS Connection depending on the "cacheLevel" setting.

 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();
			}
		}
}

Sleep according to the specified recovery interval. Called inbetween recovery attempts.

 public  void start() throws JmsException 
{
		synchronized (this.lifecycleMonitor) {
			this.stopCallback = null;
		}
		super.start();
}

Overridden to reset the stop callback, if any.

 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 Connection.start(), relying on listeners to perform appropriate recovery.

 public  void stop(Runnable callback) throws JmsException 
{
		synchronized (this.lifecycleMonitor) {
			this.stopCallback = callback;
		}
		stop();
}

Stop this listener container, invoking the specific callback once all listener processing has actually stopped.

Note: Further stop(runnable) calls (before processing has actually stopped) will override the specified callback. Only the latest specified callback will be invoked.

If a subsequent #start() call restarts the listener container before it has fully stopped, the callback will not get invoked at all.

 protected  void stopSharedConnection() 
{
		try {
			super.stopSharedConnection();
		}
		catch (Exception ex) {
			logger.debug("Connection stop failed - relying on listeners to perform recovery after restart", ex);
		}
}

This implementations proceeds even after an exception thrown from Connection.stop(), relying on listeners to perform appropriate recovery after a restart.

 protected  void validateConfiguration() 
{
		super.validateConfiguration();
		synchronized (this.lifecycleMonitor) {
			if (isSubscriptionDurable() && this.concurrentConsumers != 1) {
				throw new IllegalArgumentException("Only 1 concurrent consumer supported for durable subscription");
			}
		}
}