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());
		}
		return createConsumer(session, destination);
}

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

 protected boolean doReceiveAndExecute(Object invoker,
    Session session,
    MessageConsumer consumer,
    TransactionStatus status) throws JMSException 
{
		Connection conToClose = null;
		Session sessionToClose = null;
		MessageConsumer consumerToClose = null;
		try {
			Session sessionToUse = session;
			boolean transactional = false;
			if (sessionToUse == null) {
				sessionToUse = ConnectionFactoryUtils.doGetTransactionalSession(
						getConnectionFactory(), this.transactionalResourceFactory, true);
				transactional = (sessionToUse != null);
			}
			if (sessionToUse == null) {
				Connection conToUse = null;
				if (sharedConnectionEnabled()) {
					conToUse = getSharedConnection();
				}
				else {
					conToUse = createConnection();
					conToClose = conToUse;
					conToUse.start();
				}
				sessionToUse = createSession(conToUse);
				sessionToClose = sessionToUse;
			}
			MessageConsumer consumerToUse = consumer;
			if (consumerToUse == null) {
				consumerToUse = createListenerConsumer(sessionToUse);
				consumerToClose = consumerToUse;
			}
			Message message = receiveMessage(consumerToUse);
			if (message != null) {
				if (logger.isDebugEnabled()) {
					logger.debug("Received message of type [" + message.getClass() + "] from consumer [" +
							consumerToUse + "] of " + (transactional ? "transactional " : "") + "session [" +
							sessionToUse + "]");
				}
				messageReceived(invoker, sessionToUse);
				boolean exposeResource = (!transactional && isExposeListenerSession() &&
						!TransactionSynchronizationManager.hasResource(getConnectionFactory()));
				if (exposeResource) {
					TransactionSynchronizationManager.bindResource(
							getConnectionFactory(), new LocallyExposedJmsResourceHolder(sessionToUse));
				}
				try {
					doExecuteListener(sessionToUse, message);
				}
				catch (Throwable ex) {
					if (status != null) {
						if (logger.isDebugEnabled()) {
							logger.debug("Rolling back transaction because of listener exception thrown: " + ex);
						}
						status.setRollbackOnly();
					}
					handleListenerException(ex);
					// Rethrow JMSException to indicate an infrastructure problem
					// that may have to trigger recovery...
					if (ex instanceof JMSException) {
						throw (JMSException) ex;
					}
				}
				finally {
					if (exposeResource) {
						TransactionSynchronizationManager.unbindResource(getConnectionFactory());
					}
				}
				return true;
			}
			else {
				if (logger.isTraceEnabled()) {
					logger.trace("Consumer [" + consumerToUse + "] of " + (transactional ? "transactional " : "") +
							"session [" + sessionToUse + "] did not receive a message");
				}
				noMessageReceived(invoker, sessionToUse);
				return false;
			}
		}
		finally {
			JmsUtils.closeMessageConsumer(consumerToClose);
			JmsUtils.closeSession(sessionToClose);
			ConnectionFactoryUtils.releaseConnection(conToClose, getConnectionFactory(), true);
		}
}

Actually execute the listener for a message received from the given consumer, fetching all requires resources and invoking the listener.

 protected Connection getConnection(JmsResourceHolder holder) 
{
		return holder.getConnection();
}

Fetch an appropriate Connection from the given JmsResourceHolder.

This implementation accepts any JMS 1.1 Connection.

 protected Session getSession(JmsResourceHolder holder) 
{
		return holder.getSession();
}

Fetch an appropriate Session from the given JmsResourceHolder.

This implementation accepts any JMS 1.1 Session.

 protected final PlatformTransactionManager getTransactionManager() 
{
		return this.transactionManager;
}

Return the Spring PlatformTransactionManager to use for transactional wrapping of message reception plus listener execution.

 public  void initialize() 
{
		// Set sessionTransacted=true in case of a non-JTA transaction manager.
		if (!this.sessionTransactedCalled &&
				this.transactionManager instanceof ResourceTransactionManager &&
				!TransactionSynchronizationUtils.sameResourceFactory(
						(ResourceTransactionManager) this.transactionManager, getConnectionFactory())) {
			super.setSessionTransacted(true);
		}
		// Use bean name as default transaction name.
		if (this.transactionDefinition.getName() == null) {
			this.transactionDefinition.setName(getBeanName());
		}
		// Proceed with superclass initialization.
		super.initialize();
}

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

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

 protected boolean isSessionLocallyTransacted(Session session) 
{
		if (!super.isSessionLocallyTransacted(session)) {
			return false;
		}
		JmsResourceHolder resourceHolder =
				(JmsResourceHolder) TransactionSynchronizationManager.getResource(getConnectionFactory());
		return (resourceHolder == null || resourceHolder instanceof LocallyExposedJmsResourceHolder ||
				!resourceHolder.containsSession(session));
}

This implementation checks whether the Session is externally synchronized. In this case, the Session is not locally transacted, despite the listener container's "sessionTransacted" flag being set to "true".

 protected  void messageReceived(Object invoker,
    Session session) 
{

}

Template method that gets called right when a new message has been received, before attempting to process it. Allows subclasses to react to the event of an actual incoming message, for example adapting their consumer count.

 protected  void noMessageReceived(Object invoker,
    Session session) 
{

}

Template method that gets called right no message has been received, before attempting to process it. Allows subclasses to react to the event of an actual incoming message, for example marking .

 protected boolean receiveAndExecute(Object invoker,
    Session session,
    MessageConsumer consumer) throws JMSException 
{
		if (this.transactionManager != null) {
			// Execute receive within transaction.
			TransactionStatus status = this.transactionManager.getTransaction(this.transactionDefinition);
			boolean messageReceived = true;
			try {
				messageReceived = doReceiveAndExecute(invoker, session, consumer, status);
			}
			catch (JMSException ex) {
				rollbackOnException(status, ex);
				throw ex;
			}
			catch (RuntimeException ex) {
				rollbackOnException(status, ex);
				throw ex;
			}
			catch (Error err) {
				rollbackOnException(status, err);
				throw err;
			}
			this.transactionManager.commit(status);
			return messageReceived;
		}
		else {
			// Execute receive outside of transaction.
			return doReceiveAndExecute(invoker, session, consumer, null);
		}
}

Execute the listener for a message received from the given consumer, wrapping the entire operation in an external transaction if demanded.

 protected Message receiveMessage(MessageConsumer consumer) throws JMSException 
{
		return (this.receiveTimeout <  0 ? consumer.receive() : consumer.receive(this.receiveTimeout));
}

Receive a message from the given consumer.

 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 setReceiveTimeout(long receiveTimeout) 
{
		this.receiveTimeout = receiveTimeout;
}

Set the timeout to use for receive calls, in milliseconds. The default is 1000 ms, that is, 1 second.

NOTE: This value needs to be smaller than the transaction timeout used by the transaction manager (in the appropriate unit, of course). -1 indicates no timeout at all; however, this is only feasible if not running within a transaction manager.

 public  void setSessionTransacted(boolean sessionTransacted) 
{
	    
		super.setSessionTransacted(sessionTransacted);
		this.sessionTransactedCalled = true;
}

 public  void setTransactionManager(PlatformTransactionManager transactionManager) 
{
		this.transactionManager = transactionManager;
}

Specify the Spring org.springframework.transaction.PlatformTransactionManager to use for transactional wrapping of message reception plus listener execution.

Default is none, not performing any transactional wrapping. If specified, this will usually be a Spring org.springframework.transaction.jta.JtaTransactionManager or one of its subclasses, in combination with a JTA-aware ConnectionFactory that this message listener container obtains its Connections from.

Note: Consider the use of local JMS transactions instead. Simply switch the "sessionTransacted" flag to "true" in order to use a locally transacted JMS Session for the entire receive processing, including any Session operations performed by a SessionAwareMessageListener (e.g. sending a response message). Alternatively, a org.springframework.jms.connection.JmsTransactionManager may be used for fully synchronized Spring transactions based on local JMS transactions. Check AbstractMessageListenerContainer 's javadoc for a discussion of transaction choices and message redelivery scenarios.

 public  void setTransactionName(String transactionName) 
{
		this.transactionDefinition.setName(transactionName);
}

Specify the transaction name to use for transactional wrapping. Default is the bean name of this listener container, if any.

 public  void setTransactionTimeout(int transactionTimeout) 
{
		this.transactionDefinition.setTimeout(transactionTimeout);
}

Specify the transaction timeout to use for transactional wrapping, in seconds. Default is none, using the transaction manager's default timeout.