protected MessageConsumer createConsumer(Session session,
    Destination destination) throws JMSException 
{
		// Only pass in the NoLocal flag in case of a Topic:
		// Some JMS providers, such as WebSphere MQ 6.0, throw IllegalStateException
		// in case of the NoLocal flag being specified for a Queue.
		if (isPubSubDomain()) {
			if (isSubscriptionDurable() && destination instanceof Topic) {
				return session.createDurableSubscriber(
						(Topic) destination, getDurableSubscriptionName(), getMessageSelector(), isPubSubNoLocal());
			}
			else {
				return session.createConsumer(destination, getMessageSelector(), isPubSubNoLocal());
			}
		}
		else {
			return session.createConsumer(destination, getMessageSelector());
		}
}

Create a JMS MessageConsumer for the given Session and Destination.

This implementation uses JMS 1.1 API.

 protected MessageConsumer createListenerConsumer(Session session) throws JMSException 
{
		Destination destination = getDestination();
		if (destination == null) {
			destination = resolveDestinationName(session, getDestinationName());
		}
		MessageConsumer consumer = createConsumer(session, destination);
		if (this.taskExecutor != null) {
			consumer.setMessageListener(new MessageListener() {
				public void onMessage(final Message message) {
					taskExecutor.execute(new Runnable() {
						public void run() {
							processMessage(message, session);
						}
					});
				}
			});
		}
		else {
			consumer.setMessageListener(new MessageListener() {
				public void onMessage(Message message) {
					processMessage(message, session);
				}
			});
		}
		return consumer;
}

Create a MessageConsumer for the given JMS Session, registering a MessageListener for the specified listener.

 protected  void doInitialize() throws JMSException 
{
		establishSharedConnection();
		initializeConsumers();
}

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

 protected  void doShutdown() throws JMSException 
{
		logger.debug("Closing JMS MessageConsumers");
		for (Iterator it = this.consumers.iterator(); it.hasNext();) {
			MessageConsumer consumer = (MessageConsumer) it.next();
			JmsUtils.closeMessageConsumer(consumer);
		}
		logger.debug("Closing JMS Sessions");
		for (Iterator it = this.sessions.iterator(); it.hasNext();) {
			Session session = (Session) it.next();
			JmsUtils.closeSession(session);
		}
}

Destroy the registered JMS Sessions and associated MessageConsumers.

 protected  void doStart() throws JMSException 
{
		super.doStart();
		initializeConsumers();
}

Re-initializes this container's JMS message consumers, if not initialized already.

 protected  void initializeConsumers() throws JMSException 
{
		// Register Sessions and MessageConsumers.
		synchronized (this.consumersMonitor) {
			if (this.consumers == null) {
				this.sessions = new HashSet(this.concurrentConsumers);
				this.consumers = new HashSet(this.concurrentConsumers);
				Connection con = getSharedConnection();
				for (int i = 0; i <  this.concurrentConsumers; i++) {
					Session session = createSession(con);
					MessageConsumer consumer = createListenerConsumer(session);
					this.sessions.add(session);
					this.consumers.add(consumer);
				}
			}
		}
}

Initialize the JMS Sessions and MessageConsumers for this container.

 protected boolean isPubSubNoLocal() 
{
		return this.pubSubNoLocal;
}

Return whether to inhibit the delivery of messages published by its own connection.

 public  void onException(JMSException ex) 
{
		// First invoke the user-specific ExceptionListener, if any.
		invokeExceptionListener(ex);
		// Now try to recover the shared Connection and all consumers...
		if (logger.isInfoEnabled()) {
			logger.info("Trying to recover from JMS Connection exception: " + ex);
		}
		try {
			synchronized (this.consumersMonitor) {
				this.sessions = null;
				this.consumers = null;
			}
			refreshSharedConnection();
			initializeConsumers();
			logger.info("Successfully refreshed JMS Connection");
		}
		catch (JMSException recoverEx) {
			logger.debug("Failed to recover JMS Connection", recoverEx);
			logger.error("Encountered non-recoverable JMSException", ex);
		}
}

JMS ExceptionListener implementation, invoked by the JMS provider in case of connection failures. Re-initializes this listener container's shared connection and its sessions and consumers.

 protected  void prepareSharedConnection(Connection connection) throws JMSException 
{
		super.prepareSharedConnection(connection);
		connection.setExceptionListener(this);
}

Registers this listener container as JMS ExceptionListener on the shared connection.

 protected  void processMessage(Message message,
    Session session) 
{
		boolean exposeResource = isExposeListenerSession();
		if (exposeResource) {
			TransactionSynchronizationManager.bindResource(
					getConnectionFactory(), new LocallyExposedJmsResourceHolder(session));
		}
		try {
			executeListener(session, message);
		}
		finally {
			if (exposeResource) {
				TransactionSynchronizationManager.unbindResource(getConnectionFactory());
			}
		}
}

Process a message received from the provider.

Executes the listener, exposing the current JMS Session as thread-bound resource (if "exposeListenerSession" is "true").

 public  void setConcurrentConsumers(int concurrentConsumers) 
{
		Assert.isTrue(concurrentConsumers  > 0, "'concurrentConsumers' value must be at least 1 (one)");
		this.concurrentConsumers = concurrentConsumers;
}

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

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.

 public  void setPubSubNoLocal(boolean pubSubNoLocal) 
{
	                    	 
		this.pubSubNoLocal = pubSubNoLocal;
}

Set whether to inhibit the delivery of messages published by its own connection. Default is "false".

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

Set the Spring TaskExecutor to use for executing the listener once a message has been received by the provider.

Default is none, that is, to run in the JMS provider's own receive thread, blocking the provider's receive endpoint while executing the listener.

Specify a TaskExecutor for executing the listener in a different thread, rather than blocking the JMS provider, usually integrating with an existing thread pool. This allows to keep the number of concurrent consumers low (1) while still processing messages concurrently (decoupled from receiving!).

NOTE: Specifying a TaskExecutor for listener execution affects acknowledgement semantics. Messages will then always get acknowledged before listener execution, with the underlying Session immediately reused for receiving the next message. Using this in combination with a transacted session or with client acknowledgement will lead to unspecified results!

NOTE: Concurrent listener execution via a TaskExecutor will lead to concurrent processing of messages that have been received by the same underlying Session. As a consequence, it is not recommended to use this setting with a SessionAwareMessageListener , at least not if the latter performs actual work on the given Session. A standard javax.jms.MessageListener will work fine, in general.

 protected final boolean sharedConnectionEnabled() 
{
		return true;
}

Always use a shared JMS Connection.

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