org.springframework.jms.listener

Class AbstractPollingMessageListenerContainer

java.lang.Object

org.springframework.jms.support.JmsAccessor

org.springframework.jms.support.destination.JmsDestinationAccessor

org.springframework.jms.listener.AbstractJmsListeningContainer

org.springframework.jms.listener.AbstractMessageListenerContainer

org.springframework.jms.listener.AbstractPollingMessageListenerContainer

All Implemented Interfaces:

Aware, BeanNameAware, DisposableBean, InitializingBean, Lifecycle, Phased, SmartLifecycle

Direct Known Subclasses:

DefaultMessageListenerContainer

public abstract class AbstractPollingMessageListenerContainer
extends AbstractMessageListenerContainer

Base class for listener container implementations which are based on polling. Provides support for listener handling based on javax.jms.MessageConsumer, optionally participating in externally managed transactions.

This listener container variant is built for repeated polling attempts, each invoking the receiveAndExecute(java.lang.Object, Session, MessageConsumer) method. The MessageConsumer used may be reobtained fo reach attempt or cached inbetween attempts; this is up to the concrete implementation. The receive timeout for each attempt can be configured through the "receiveTimeout" property.

The underlying mechanism is based on standard JMS MessageConsumer handling, which is perfectly compatible with both native JMS and JMS in a J2EE environment. Neither the JMS MessageConsumer.setMessageListener facility nor the JMS ServerSessionPool facility is required. A further advantage of this approach is full control over the listening process, allowing for custom scaling and throttling and of concurrent message processing (which is up to concrete subclasses).

Message reception and listener execution can automatically be wrapped in transactions through passing a Spring PlatformTransactionManager into the "transactionManager" property. This will usually be a JtaTransactionManager in a J2EE enviroment, in combination with a JTA-aware JMS ConnectionFactory obtained from JNDI (check your J2EE server's documentation).

This base class does not assume any specific mechanism for asynchronous execution of polling invokers. Check out DefaultMessageListenerContainer for a concrete implementation which is based on Spring's TaskExecutor abstraction, including dynamic scaling of concurrent consumers and automatic self recovery.

Since:

2.0.3

Author:

Juergen Hoeller

See Also:

createListenerConsumer(Session), receiveAndExecute(java.lang.Object, Session, MessageConsumer), setTransactionManager(org.springframework.transaction.PlatformTransactionManager)

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
private class	AbstractPollingMessageListenerContainer.MessageListenerContainerResourceFactory ResourceFactory implementation that delegates to this listener container's protected callback methods.

Nested classes/interfaces inherited from class org.springframework.jms.listener.AbstractJmsListeningContainer

AbstractJmsListeningContainer.SharedConnectionNotInitializedException

Field Summary

Fields

Modifier and Type	Field and Description
private java.lang.Boolean	commitAfterNoMessageReceived
static long	DEFAULT_RECEIVE_TIMEOUT The default receive timeout: 1000 ms = 1 second.
private boolean	pubSubNoLocal
private long	receiveTimeout
private boolean	sessionTransactedCalled
private AbstractPollingMessageListenerContainer.MessageListenerContainerResourceFactory	transactionalResourceFactory
private DefaultTransactionDefinition	transactionDefinition
private PlatformTransactionManager	transactionManager

Fields inherited from class org.springframework.jms.listener.AbstractJmsListeningContainer

lifecycleMonitor, sharedConnectionMonitor

Fields inherited from class org.springframework.jms.support.JmsAccessor

logger

Constructor Summary

Constructors

Constructor and Description
AbstractPollingMessageListenerContainer()

Method Summary

Modifier and Type	Method and Description
protected MessageConsumer	createConsumer(Session session, Destination destination) Create a JMS MessageConsumer for the given Session and Destination.
protected MessageConsumer	createListenerConsumer(Session session) Create a MessageConsumer for the given JMS Session, registering a MessageListener for the specified listener.
protected boolean	doReceiveAndExecute(java.lang.Object invoker, Session session, MessageConsumer consumer, TransactionStatus status) 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) Fetch an appropriate Connection from the given JmsResourceHolder.
protected Session	getSession(JmsResourceHolder holder) Fetch an appropriate Session from the given JmsResourceHolder.
protected PlatformTransactionManager	getTransactionManager() Return the Spring PlatformTransactionManager to use for transactional wrapping of message reception plus listener execution.
void	initialize() Initialize this container.
protected boolean	isPubSubNoLocal() Return whether to inhibit the delivery of messages published by its own connection.
protected boolean	isSessionLocallyTransacted(Session session) This implementation checks whether the Session is externally synchronized.
protected void	messageReceived(java.lang.Object invoker, Session session) Template method that gets called right when a new message has been received, before attempting to process it.
protected void	noMessageReceived(java.lang.Object invoker, Session session) Template method that gets called when no message has been received, before returning to the receive loop again.
protected boolean	receiveAndExecute(java.lang.Object invoker, Session session, MessageConsumer consumer) 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) Receive a message from the given consumer.
private void	rollbackOnException(TransactionStatus status, java.lang.Throwable ex) Perform a rollback, handling rollback exceptions properly.
void	setPubSubNoLocal(boolean pubSubNoLocal) Set whether to inhibit the delivery of messages published by its own connection.
void	setReceiveTimeout(long receiveTimeout) Set the timeout to use for receive calls, in milliseconds.
void	setSessionTransacted(boolean sessionTransacted) Set the transaction mode that is used when creating a JMS Session.
void	setTransactionManager(PlatformTransactionManager transactionManager) Specify the Spring PlatformTransactionManager to use for transactional wrapping of message reception plus listener execution.
void	setTransactionName(java.lang.String transactionName) Specify the transaction name to use for transactional wrapping.
void	setTransactionTimeout(int transactionTimeout) Specify the transaction timeout to use for transactional wrapping, in seconds.
protected boolean	shouldCommitAfterNoMessageReceived(Session session) Determine whether to trigger a commit after no message has been received.

Methods inherited from class org.springframework.jms.listener.AbstractMessageListenerContainer

checkMessageListener, commitIfNecessary, doExecuteListener, doInvokeListener, doInvokeListener, executeListener, getDefaultSubscriptionName, getDestination, getDestinationDescription, getDestinationName, getDurableSubscriptionName, getExceptionListener, getMessageListener, getMessageSelector, handleListenerException, invokeErrorHandler, invokeExceptionListener, invokeListener, isAcceptMessagesWhileStopping, isExposeListenerSession, isSubscriptionDurable, rollbackIfNecessary, rollbackOnExceptionIfNecessary, setAcceptMessagesWhileStopping, setDestination, setDestinationName, setDurableSubscriptionName, setErrorHandler, setExceptionListener, setExposeListenerSession, setMessageListener, setMessageSelector, setSubscriptionDurable, validateConfiguration

Methods inherited from class org.springframework.jms.listener.AbstractJmsListeningContainer

afterPropertiesSet, createSharedConnection, destroy, doInitialize, doRescheduleTask, doShutdown, doStart, doStop, establishSharedConnection, getBeanName, getClientId, getPausedTaskCount, getPhase, getSharedConnection, isActive, isAutoStartup, isRunning, logRejectedTask, prepareSharedConnection, refreshSharedConnection, rescheduleTaskIfNecessary, resumePausedTasks, runningAllowed, setAutoStartup, setBeanName, setClientId, setPhase, sharedConnectionEnabled, shutdown, start, startSharedConnection, stop, stop, stopSharedConnection

Methods inherited from class org.springframework.jms.support.destination.JmsDestinationAccessor

getDestinationResolver, isPubSubDomain, resolveDestinationName, setDestinationResolver, setPubSubDomain

Methods inherited from class org.springframework.jms.support.JmsAccessor

convertJmsAccessException, createConnection, createSession, getConnectionFactory, getSessionAcknowledgeMode, isClientAcknowledge, isSessionTransacted, setConnectionFactory, setSessionAcknowledgeMode, setSessionAcknowledgeModeName

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

DEFAULT_RECEIVE_TIMEOUT

public static final long DEFAULT_RECEIVE_TIMEOUT

The default receive timeout: 1000 ms = 1 second.

See Also:

Constant Field Values

transactionalResourceFactory

private final AbstractPollingMessageListenerContainer.MessageListenerContainerResourceFactory transactionalResourceFactory

sessionTransactedCalled

private boolean sessionTransactedCalled

pubSubNoLocal

private boolean pubSubNoLocal

transactionManager

private PlatformTransactionManager transactionManager

transactionDefinition

private DefaultTransactionDefinition transactionDefinition

receiveTimeout

private long receiveTimeout

commitAfterNoMessageReceived

private volatile java.lang.Boolean commitAfterNoMessageReceived

Constructor Detail

AbstractPollingMessageListenerContainer

public AbstractPollingMessageListenerContainer()

Method Detail

setSessionTransacted

public void setSessionTransacted(boolean sessionTransacted)

Description copied from class: JmsAccessor

Set the transaction mode that is used when creating a JMS Session. Default is "false".

Note that within a JTA transaction, the parameters passed to create(Queue/Topic)Session(boolean transacted, int acknowledgeMode) method are not taken into account. Depending on the J2EE transaction context, the container makes its own decisions on these values. Analogously, these parameters are not taken into account within a locally managed transaction either, since the accessor operates on an existing JMS Session in this case.

Setting this flag to "true" will use a short local JMS transaction when running outside of a managed transaction, and a synchronized local JMS transaction in case of a managed transaction (other than an XA transaction) being present. The latter has the effect of a local JMS transaction being managed alongside the main transaction (which might be a native JDBC transaction), with the JMS transaction committing right after the main transaction.

Overrides:

setSessionTransacted in class JmsAccessor

See Also:

javax.jms.Connection#createSession(boolean, int)

setPubSubNoLocal

public void setPubSubNoLocal(boolean pubSubNoLocal)

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

See Also:

javax.jms.TopicSession#createSubscriber(javax.jms.Topic, String, boolean)

isPubSubNoLocal

protected boolean isPubSubNoLocal()

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

setTransactionManager

public void setTransactionManager(PlatformTransactionManager transactionManager)

Specify the Spring 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 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 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.

See Also:

JtaTransactionManager, JmsTransactionManager

getTransactionManager

protected final PlatformTransactionManager getTransactionManager()

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

setTransactionName

public void setTransactionName(java.lang.String transactionName)

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

See Also:

TransactionDefinition.getName()

setTransactionTimeout

public void setTransactionTimeout(int transactionTimeout)

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

See Also:

TransactionDefinition.getTimeout(), setReceiveTimeout(long)

setReceiveTimeout

public void setReceiveTimeout(long 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.

See Also:

javax.jms.MessageConsumer#receive(long), javax.jms.MessageConsumer#receive(), setTransactionTimeout(int)

initialize

public void initialize()

Description copied from class: AbstractJmsListeningContainer

Initialize this container.

Creates a JMS Connection, starts the javax.jms.Connection (if "autoStartup" hasn't been turned off), and calls AbstractJmsListeningContainer.doInitialize().

Overrides:

initialize in class AbstractJmsListeningContainer

createListenerConsumer

protected MessageConsumer createListenerConsumer(Session session)
                                          throws JMSException

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

Parameters:

session - the JMS Session to work on

Returns:

the MessageConsumer

Throws:

javax.jms.JMSException - if thrown by JMS methods

JMSException

See Also:

receiveAndExecute(java.lang.Object, Session, MessageConsumer)

receiveAndExecute

protected boolean receiveAndExecute(java.lang.Object invoker,
                                    Session session,
                                    MessageConsumer consumer)
                             throws JMSException

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

Parameters:

session - the JMS Session to work on

consumer - the MessageConsumer to work on

Returns:

whether a message has been received

Throws:

JMSException - if thrown by JMS methods

See Also:

doReceiveAndExecute(java.lang.Object, Session, MessageConsumer, org.springframework.transaction.TransactionStatus)

doReceiveAndExecute

protected boolean doReceiveAndExecute(java.lang.Object invoker,
                                      Session session,
                                      MessageConsumer consumer,
                                      TransactionStatus status)
                               throws JMSException

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

Parameters:

session - the JMS Session to work on

consumer - the MessageConsumer to work on

status - the TransactionStatus (may be null)

Returns:

whether a message has been received

Throws:

JMSException - if thrown by JMS methods

See Also:

#doExecuteListener(javax.jms.Session, javax.jms.Message)

isSessionLocallyTransacted

protected boolean isSessionLocallyTransacted(Session 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".

Overrides:

isSessionLocallyTransacted in class AbstractMessageListenerContainer

Parameters:

session - the Session to check

Returns:

whether the given Session is locally transacted

See Also:

JmsResourceHolder

shouldCommitAfterNoMessageReceived

protected boolean shouldCommitAfterNoMessageReceived(Session session)

Determine whether to trigger a commit after no message has been received. This is a good idea on any JMS provider other than Tibco, which is what this default implementation checks for.

Parameters:

session - the current JMS Session which received no message

Returns:

whether to call AbstractMessageListenerContainer.commitIfNecessary(Session, Message) on the given Session

rollbackOnException

private void rollbackOnException(TransactionStatus status,
                                 java.lang.Throwable ex)

Perform a rollback, handling rollback exceptions properly.

Parameters:

status - object representing the transaction

ex - the thrown listener exception or error

receiveMessage

protected Message receiveMessage(MessageConsumer consumer)
                          throws JMSException

Receive a message from the given consumer.

Parameters:

consumer - the MessageConsumer to use

Returns:

the Message, or null if none

Throws:

JMSException - if thrown by JMS methods

messageReceived

protected void messageReceived(java.lang.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.

Parameters:

invoker - the invoker object (passed through)

session - the receiving JMS Session

noMessageReceived

protected void noMessageReceived(java.lang.Object invoker,
                                 Session session)

Template method that gets called when no message has been received, before returning to the receive loop again. Allows subclasses to react to the event of no incoming message, for example marking the invoker as idle.

Parameters:

invoker - the invoker object (passed through)

session - the receiving JMS Session

getConnection

protected Connection getConnection(JmsResourceHolder holder)

Fetch an appropriate Connection from the given JmsResourceHolder.

This implementation accepts any JMS 1.1 Connection.

Parameters:

holder - the JmsResourceHolder

Returns:

an appropriate Connection fetched from the holder, or null if none found

getSession

protected Session getSession(JmsResourceHolder holder)

Fetch an appropriate Session from the given JmsResourceHolder.

This implementation accepts any JMS 1.1 Session.

Parameters:

holder - the JmsResourceHolder

Returns:

an appropriate Session fetched from the holder, or null if none found

createConsumer

protected MessageConsumer createConsumer(Session session,
                                         Destination destination)
                                  throws JMSException

Create a JMS MessageConsumer for the given Session and Destination.

This implementation uses JMS 1.1 API.

Parameters:

session - the JMS Session to create a MessageConsumer for

destination - the JMS Destination to create a MessageConsumer for

Returns:

the new JMS MessageConsumer

Throws:

javax.jms.JMSException - if thrown by JMS API methods

JMSException