public abstract class AbstractMessageListenerContainer extends AbstractJmsListeningContainer implements MessageListenerContainer
MessageListener
or Spring's
SessionAwareMessageListener
for actual message processing.
Usually holds a single JMS Connection
that all listeners are supposed
to be registered on, which is the standard JMS way of managing listener sessions.
Can alternatively also be used with a fresh Connection per listener, for Java EE
style XA-aware JMS messaging. The actual registration process is up to concrete
subclasses.
NOTE: The default behavior of this message listener container is to
never propagate an exception thrown by a message listener up to the JMS
provider. Instead, it will log any such exception at the error level.
This means that from the perspective of the attendant JMS provider no such
listener will ever fail. However, if error handling is necessary, then
any implementation of the ErrorHandler
strategy may be provided to
the setErrorHandler(ErrorHandler)
method. Note that JMSExceptions
will be passed to the ErrorHandler in addition to (but after) being
passed to an ExceptionListener
, if one has been provided.
The listener container offers the following message acknowledgment options:
DefaultMessageListenerContainer
,
it means automatic message acknowledgment before listener execution, with
no redelivery in case of an exception and no redelivery in case of other listener
execution interruptions either. For SimpleMessageListenerContainer
,
it means automatic message acknowledgment after listener execution, with
no redelivery in case of a user exception thrown but potential redelivery in case
of the JVM dying during listener execution. In order to consistently arrange for
redelivery with any container variant, consider "CLIENT_ACKNOWLEDGE" mode or -
preferably - setting "sessionTransacted" to "true" instead.
DefaultMessageListenerContainer
)
or shortly after (SimpleMessageListenerContainer
) listener execution;
no redelivery in case of a user exception thrown but potential redelivery in case
of the JVM dying during listener execution. In order to consistently arrange for
redelivery with any container variant, consider "CLIENT_ACKNOWLEDGE" mode or -
preferably - setting "sessionTransacted" to "true" instead.
There are two solutions to the duplicate message processing problem:
DefaultMessageListenerContainer
, through specifying
an external "transactionManager" (typically a
JtaTransactionManager
, with
a corresponding XA-aware JMS ConnectionFactory
passed in
as "connectionFactory").
Recommendations:
JtaTransactionManager
as
"transactionManager" for a fully XA-aware JMS provider - typically when
running on a Java EE server, but also for other environments with a JTA
transaction manager present. This will give full "exactly-once" guarantees
without custom duplicate message checks, at the price of additional
runtime processing overhead.
Note that the "sessionTransacted" flag is strongly recommended over
JmsTransactionManager
, provided
that transactions do not need to be managed externally. As a consequence,
set the transaction manager only if you are using JTA or if you need to
synchronize with custom external transaction arrangements.
setMessageListener(java.lang.Object)
,
MessageListener
,
SessionAwareMessageListener
,
handleListenerException(java.lang.Throwable)
,
DefaultMessageListenerContainer
,
SimpleMessageListenerContainer
,
JmsMessageEndpointManager
AbstractJmsListeningContainer.SharedConnectionNotInitializedException
lifecycleMonitor, sharedConnectionMonitor
RECEIVE_TIMEOUT_INDEFINITE_WAIT, RECEIVE_TIMEOUT_NO_WAIT
logger
DEFAULT_PHASE
Constructor and Description |
---|
AbstractMessageListenerContainer() |
Modifier and Type | Method and Description |
---|---|
protected void |
checkMessageListener(Object messageListener)
Check the given message listener, throwing an exception
if it does not correspond to a supported listener type.
|
protected void |
commitIfNecessary(Session session,
Message message)
Perform a commit or message acknowledgement, as appropriate.
|
protected MessageConsumer |
createConsumer(Session session,
Destination destination)
Create a JMS MessageConsumer for the given Session and Destination.
|
protected void |
doExecuteListener(Session session,
Message message)
Execute the specified listener,
committing or rolling back the transaction afterwards (if necessary).
|
protected void |
doInvokeListener(MessageListener listener,
Message message)
Invoke the specified listener as standard JMS MessageListener.
|
protected void |
doInvokeListener(SessionAwareMessageListener listener,
Session session,
Message message)
Invoke the specified listener as Spring SessionAwareMessageListener,
exposing a new JMS Session (potentially with its own transaction)
to the listener if demanded.
|
protected void |
executeListener(Session session,
Message message)
Execute the specified listener,
committing or rolling back the transaction afterwards (if necessary).
|
protected String |
getDefaultSubscriptionName(Object messageListener)
Determine the default subscription name for the given message listener.
|
Destination |
getDestination()
Return the destination to receive messages from.
|
protected String |
getDestinationDescription()
Return a descriptive String for this container's JMS destination
(never
null ). |
String |
getDestinationName()
Return the name of the destination to receive messages from.
|
String |
getDurableSubscriptionName()
Return the name of a durable subscription to create, if any.
|
ErrorHandler |
getErrorHandler()
Return the ErrorHandler to be invoked in case of any uncaught exceptions thrown
while processing a Message.
|
ExceptionListener |
getExceptionListener()
Return the JMS ExceptionListener to notify in case of a JMSException thrown
by the registered message listener or the invocation infrastructure, if any.
|
MessageConverter |
getMessageConverter()
Return the
MessageConverter that can be used to
convert Message , if any. |
Object |
getMessageListener()
Return the message listener object to register.
|
String |
getMessageSelector()
Return the JMS message selector expression (or
null if none). |
QosSettings |
getReplyQosSettings()
Return the
QosSettings to use when sending a reply,
or null if the broker's defaults should be used. |
String |
getSubscriptionName()
Return the name of a subscription to create, if any.
|
protected void |
handleListenerException(Throwable ex)
Handle the given exception that arose during listener execution.
|
protected void |
invokeErrorHandler(Throwable ex)
Invoke the registered ErrorHandler, if any.
|
protected void |
invokeExceptionListener(JMSException ex)
Invoke the registered JMS ExceptionListener, if any.
|
protected void |
invokeListener(Session session,
Message message)
Invoke the specified listener: either as standard JMS MessageListener
or (preferably) as Spring SessionAwareMessageListener.
|
boolean |
isAcceptMessagesWhileStopping()
Return whether to accept received messages while the listener container
in the process of stopping.
|
boolean |
isExposeListenerSession()
Return whether to expose the listener JMS
Session to a
registered SessionAwareMessageListener . |
boolean |
isPubSubNoLocal()
Return whether to inhibit the delivery of messages published by its own connection.
|
boolean |
isReplyPubSubDomain()
Return whether the Publish/Subscribe domain (
Topics ) is used
for replies. |
protected boolean |
isSessionLocallyTransacted(Session session)
Check whether the given Session is locally transacted, that is, whether
its transaction is managed by this listener container's Session handling
and not by an external transaction coordinator.
|
boolean |
isSubscriptionDurable()
Return whether to make the subscription durable.
|
boolean |
isSubscriptionShared()
Return whether to make the subscription shared.
|
protected void |
rollbackIfNecessary(Session session)
Perform a rollback, if appropriate.
|
protected void |
rollbackOnExceptionIfNecessary(Session session,
Throwable ex)
Perform a rollback, handling rollback exceptions properly.
|
void |
setAcceptMessagesWhileStopping(boolean acceptMessagesWhileStopping)
Set whether to accept received messages while the listener container
in the process of stopping.
|
abstract void |
setConcurrency(String concurrency)
Specify concurrency limits.
|
void |
setDestination(Destination destination)
Set the destination to receive messages from.
|
void |
setDestinationName(String destinationName)
Set the name of the destination to receive messages from.
|
void |
setDurableSubscriptionName(String durableSubscriptionName)
Set the name of a durable subscription to create.
|
void |
setErrorHandler(ErrorHandler errorHandler)
Set the ErrorHandler to be invoked in case of any uncaught exceptions thrown
while processing a Message.
|
void |
setExceptionListener(ExceptionListener exceptionListener)
Set the JMS ExceptionListener to notify in case of a JMSException thrown
by the registered message listener or the invocation infrastructure.
|
void |
setExposeListenerSession(boolean exposeListenerSession)
Set whether to expose the listener JMS Session to a registered
SessionAwareMessageListener as well as to
JmsTemplate calls. |
void |
setMessageConverter(MessageConverter messageConverter)
Set the
MessageConverter strategy for converting JMS Messages. |
void |
setMessageListener(Object messageListener)
Set the message listener implementation to register.
|
void |
setMessageSelector(String messageSelector)
Set the JMS message selector expression (or
null if none). |
void |
setPubSubNoLocal(boolean pubSubNoLocal)
Set whether to inhibit the delivery of messages published by its own connection.
|
void |
setReplyPubSubDomain(boolean replyPubSubDomain)
Configure the reply destination type.
|
void |
setReplyQosSettings(QosSettings replyQosSettings)
Configure the
QosSettings to use when sending a reply. |
void |
setSubscriptionDurable(boolean subscriptionDurable)
Set whether to make the subscription durable.
|
void |
setSubscriptionName(String subscriptionName)
Set the name of a subscription to create.
|
void |
setSubscriptionShared(boolean subscriptionShared)
Set whether to make the subscription shared.
|
void |
setupMessageListener(Object messageListener)
Setup the message listener to use.
|
protected void |
validateConfiguration()
Validate the configuration of this container.
|
afterPropertiesSet, createSharedConnection, destroy, doInitialize, doRescheduleTask, doShutdown, doStart, doStop, establishSharedConnection, getBeanName, getClientId, getPausedTaskCount, getPhase, getSharedConnection, initialize, isActive, isAutoStartup, isRunning, logRejectedTask, prepareSharedConnection, refreshSharedConnection, rescheduleTaskIfNecessary, resumePausedTasks, runningAllowed, setAutoStartup, setBeanName, setClientId, setPhase, sharedConnectionEnabled, shutdown, start, startSharedConnection, stop, stopSharedConnection
getDestinationResolver, isPubSubDomain, receiveFromConsumer, resolveDestinationName, setDestinationResolver, setPubSubDomain
convertJmsAccessException, createConnection, createSession, getConnectionFactory, getSessionAcknowledgeMode, isClientAcknowledge, isSessionTransacted, obtainConnectionFactory, setConnectionFactory, setSessionAcknowledgeMode, setSessionAcknowledgeModeName, setSessionTransacted
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait
getDestinationResolver, isPubSubDomain
getPhase, isAutoStartup, stop
public abstract void setConcurrency(String concurrency)
public void setDestination(@Nullable Destination destination)
Alternatively, specify a "destinationName", to be dynamically
resolved via the DestinationResolver
.
Note: The destination may be replaced at runtime, with the listener container picking up the new destination immediately (works e.g. with DefaultMessageListenerContainer, as long as the cache level is less than CACHE_CONSUMER). However, this is considered advanced usage; use it with care!
setDestinationName(String)
@Nullable public Destination getDestination()
null
if the configured destination is not an actual Destination
type;
c.f. when the destination is a String
.public void setDestinationName(@Nullable String destinationName)
The specified name will be dynamically resolved via the configured
destination resolver
.
Alternatively, specify a JMS Destination
object as "destination".
Note: The destination may be replaced at runtime, with the listener container picking up the new destination immediately (works e.g. with DefaultMessageListenerContainer, as long as the cache level is less than CACHE_CONSUMER). However, this is considered advanced usage; use it with care!
setDestination(javax.jms.Destination)
@Nullable public String getDestinationName()
null
if the configured destination is not a
String
type; c.f. when
it is an actual Destination
.protected String getDestinationDescription()
null
).public void setMessageSelector(@Nullable String messageSelector)
null
if none).
Default is none.
See the JMS specification for a detailed definition of selector expressions.
Note: The message selector may be replaced at runtime, with the listener container picking up the new selector value immediately (works e.g. with DefaultMessageListenerContainer, as long as the cache level is less than CACHE_CONSUMER). However, this is considered advanced usage; use it with care!
@Nullable public String getMessageSelector()
null
if none).public void setMessageListener(@Nullable Object messageListener)
MessageListener
object
or a Spring SessionAwareMessageListener
object.
Note: The message listener may be replaced at runtime, with the listener container picking up the new listener object immediately (works e.g. with DefaultMessageListenerContainer, as long as the cache level is less than CACHE_CONSUMER). However, this is considered advanced usage; use it with care!
IllegalArgumentException
- if the supplied listener is not a
MessageListener
or a SessionAwareMessageListener
MessageListener
,
SessionAwareMessageListener
@Nullable public Object getMessageListener()
protected void checkMessageListener(@Nullable Object messageListener)
By default, only a standard JMS MessageListener
object or a
Spring SessionAwareMessageListener
object will be accepted.
messageListener
- the message listener object to checkIllegalArgumentException
- if the supplied listener is not a
MessageListener
or a SessionAwareMessageListener
MessageListener
,
SessionAwareMessageListener
protected String getDefaultSubscriptionName(Object messageListener)
messageListener
- the message listener object to checkSubscriptionNameProvider
public void setSubscriptionDurable(boolean subscriptionDurable)
Default is "false". Set this to "true" to register a durable subscription, typically in combination with a "subscriptionName" value (unless your message listener class name is good enough as subscription name).
Only makes sense when listening to a topic (pub-sub domain), therefore this method switches the "pubSubDomain" flag as well.
public boolean isSubscriptionDurable()
public void setSubscriptionShared(boolean subscriptionShared)
Default is "false". Set this to "true" to register a shared subscription, typically in combination with a "subscriptionName" value (unless your message listener class name is good enough as subscription name). Note that shared subscriptions may also be durable, so this flag can (and often will) be combined with "subscriptionDurable" as well.
Only makes sense when listening to a topic (pub-sub domain), therefore this method switches the "pubSubDomain" flag as well.
Requires a JMS 2.0 compatible message broker.
public boolean isSubscriptionShared()
public void setSubscriptionName(@Nullable String subscriptionName)
The subscription name needs to be unique within this client's JMS client id. Default is the class name of the specified message listener.
Note: Only 1 concurrent consumer (which is the default of this message listener container) is allowed for each subscription, except for a shared subscription (which requires JMS 2.0).
@Nullable public String getSubscriptionName()
public void setDurableSubscriptionName(@Nullable String durableSubscriptionName)
The durable subscription name needs to be unique within this client's JMS client id. Default is the class name of the specified message listener.
Note: Only 1 concurrent consumer (which is the default of this message listener container) is allowed for each durable subscription, except for a shared durable subscription (which requires JMS 2.0).
@Nullable public String getDurableSubscriptionName()
public void setPubSubNoLocal(boolean pubSubNoLocal)
Session.createConsumer(javax.jms.Destination, String, boolean)
public boolean isPubSubNoLocal()
public void setReplyPubSubDomain(boolean replyPubSubDomain)
pubSubDomain
value is used (see JmsDestinationAccessor.isPubSubDomain()
.
This setting primarily indicates what type of destination to resolve if dynamic destinations are enabled.
replyPubSubDomain
- "true" for the Publish/Subscribe domain (Topics
),
"false" for the Point-to-Point domain (Queues
)JmsDestinationAccessor.setDestinationResolver(org.springframework.jms.support.destination.DestinationResolver)
public boolean isReplyPubSubDomain()
Topics
) is used
for replies. Otherwise, the Point-to-Point domain (Queues
)
is used.isReplyPubSubDomain
in interface MessageListenerContainer
public void setReplyQosSettings(@Nullable QosSettings replyQosSettings)
QosSettings
to use when sending a reply. Can be set to
null
to indicate that the broker's defaults should be used.replyQosSettings
- the QoS settings to use when sending a reply or null
to use the default vas.@Nullable public QosSettings getReplyQosSettings()
MessageListenerContainer
QosSettings
to use when sending a reply,
or null
if the broker's defaults should be used.getReplyQosSettings
in interface MessageListenerContainer
public void setMessageConverter(@Nullable MessageConverter messageConverter)
MessageConverter
strategy for converting JMS Messages.@Nullable public MessageConverter getMessageConverter()
MessageListenerContainer
MessageConverter
that can be used to
convert Message
, if any.getMessageConverter
in interface MessageListenerContainer
public void setExceptionListener(@Nullable ExceptionListener exceptionListener)
@Nullable public ExceptionListener getExceptionListener()
public void setErrorHandler(@Nullable ErrorHandler errorHandler)
By default, there will be no ErrorHandler so that error-level logging is the only result.
@Nullable public ErrorHandler getErrorHandler()
public void setExposeListenerSession(boolean exposeListenerSession)
SessionAwareMessageListener
as well as to
JmsTemplate
calls.
Default is "true", reusing the listener's Session
.
Turn this off to expose a fresh JMS Session fetched from the same
underlying JMS Connection
instead, which might be necessary
on some JMS providers.
Note that Sessions managed by an external transaction manager will
always get exposed to JmsTemplate
calls. So in terms of JmsTemplate exposure, this setting only affects
locally transacted Sessions.
SessionAwareMessageListener
public boolean isExposeListenerSession()
Session
to a
registered SessionAwareMessageListener
.public void setAcceptMessagesWhileStopping(boolean acceptMessagesWhileStopping)
Default is "false", rejecting such messages through aborting the receive attempt. Switch this flag on to fully process such messages even in the stopping phase, with the drawback that even newly sent messages might still get processed (if coming in before all receive timeouts have expired).
NOTE: Aborting receive attempts for such incoming messages might lead to the provider's retry count decreasing for the affected messages. If you have a high number of concurrent consumers, make sure that the number of retries is higher than the number of consumers, to be on the safe side for all potential stopping scenarios.
public boolean isAcceptMessagesWhileStopping()
protected void validateConfiguration()
AbstractJmsListeningContainer
The default implementation is empty. To be overridden in subclasses.
validateConfiguration
in class AbstractJmsListeningContainer
public void setupMessageListener(Object messageListener)
MessageListenerContainer
IllegalArgumentException
if that message listener type is not supported.setupMessageListener
in interface MessageListenerContainer
protected void executeListener(Session session, Message message)
session
- the JMS Session to operate onmessage
- the received JMS MessageinvokeListener(javax.jms.Session, javax.jms.Message)
,
commitIfNecessary(javax.jms.Session, javax.jms.Message)
,
rollbackOnExceptionIfNecessary(javax.jms.Session, java.lang.Throwable)
,
handleListenerException(java.lang.Throwable)
protected void doExecuteListener(Session session, Message message) throws JMSException
session
- the JMS Session to operate onmessage
- the received JMS MessageJMSException
- if thrown by JMS API methodsinvokeListener(javax.jms.Session, javax.jms.Message)
,
commitIfNecessary(javax.jms.Session, javax.jms.Message)
,
rollbackOnExceptionIfNecessary(javax.jms.Session, java.lang.Throwable)
,
JmsAccessor.convertJmsAccessException(javax.jms.JMSException)
protected void invokeListener(Session session, Message message) throws JMSException
session
- the JMS Session to operate onmessage
- the received JMS MessageJMSException
- if thrown by JMS API methodssetMessageListener(java.lang.Object)
protected void doInvokeListener(SessionAwareMessageListener listener, Session session, Message message) throws JMSException
listener
- the Spring SessionAwareMessageListener to invokesession
- the JMS Session to operate onmessage
- the received JMS MessageJMSException
- if thrown by JMS API methodsSessionAwareMessageListener
,
setExposeListenerSession(boolean)
protected void doInvokeListener(MessageListener listener, Message message) throws JMSException
Default implementation performs a plain invocation of the
onMessage
method.
listener
- the JMS MessageListener to invokemessage
- the received JMS MessageJMSException
- if thrown by JMS API methodsMessageListener.onMessage(javax.jms.Message)
protected void commitIfNecessary(Session session, @Nullable Message message) throws JMSException
session
- the JMS Session to commitmessage
- the Message to acknowledgeJMSException
- in case of commit failureprotected void rollbackIfNecessary(Session session) throws JMSException
session
- the JMS Session to rollbackJMSException
- in case of a rollback errorprotected void rollbackOnExceptionIfNecessary(Session session, Throwable ex) throws JMSException
session
- the JMS Session to rollbackex
- the thrown application exception or errorJMSException
- in case of a rollback errorprotected boolean isSessionLocallyTransacted(Session session)
Note: The Session's own transacted flag will already have been checked before. This method is about finding out whether the Session's transaction is local or externally coordinated.
session
- the Session to checkJmsAccessor.isSessionTransacted()
,
ConnectionFactoryUtils.isSessionTransactional(javax.jms.Session, javax.jms.ConnectionFactory)
protected MessageConsumer createConsumer(Session session, Destination destination) throws JMSException
This implementation uses JMS 1.1 API.
session
- the JMS Session to create a MessageConsumer fordestination
- the JMS Destination to create a MessageConsumer forJMSException
- if thrown by JMS API methodsprotected void handleListenerException(Throwable ex)
The default implementation logs the exception at warn level, not propagating it to the JMS provider — assuming that all handling of acknowledgement and/or transactions is done by this listener container. This can be overridden in subclasses.
ex
- the exception to handleprotected void invokeExceptionListener(JMSException ex)
ex
- the exception that arose during JMS processingsetExceptionListener(javax.jms.ExceptionListener)
protected void invokeErrorHandler(Throwable ex)
ex
- the uncaught error that arose during JMS processing.setErrorHandler(org.springframework.util.ErrorHandler)