Java > Open Source Codes > org > springframework > jms > listener > AbstractMessageListenerContainer


1 /*
2  * Copyright 2002-2007 the original author or authors.
3  *
4  * Licensed under the Apache License, Version 2.0 (the "License");
5  * you may not use this file except in compliance with the License.
6  * You may obtain a copy of the License at
7  *
8  * http://www.apache.org/licenses/LICENSE-2.0
9  *
10  * Unless required by applicable law or agreed to in writing, software
11  * distributed under the License is distributed on an "AS IS" BASIS,
12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13  * See the License for the specific language governing permissions and
14  * limitations under the License.
15  */
16
17 package org.springframework.jms.listener;
18
19 import javax.jms.Connection ;
20 import javax.jms.Destination ;
21 import javax.jms.ExceptionListener ;
22 import javax.jms.JMSException ;
23 import javax.jms.Message ;
24 import javax.jms.MessageListener ;
25 import javax.jms.Queue ;
26 import javax.jms.Session ;
27 import javax.jms.Topic ;
28
29 import org.springframework.jms.support.JmsUtils;
30 import org.springframework.util.Assert;
31
32 /**
33  * Abstract base class for message listener containers. Can either host
34  * a standard JMS {@link javax.jms.MessageListener} or a Spring-specific
35  * {@link SessionAwareMessageListener}.
36  *
37  * <p>Usually holds a single JMS {@link Connection} that all listeners are
38  * supposed to be registered on, which is the standard JMS way of managing
39  * listeners. Can alternatively also be used with a fresh Connection per
40  * listener, for J2EE-style XA-aware JMS messaging. The actual registration
41  * process is up to concrete subclasses.
42  *
43  * <p><b>NOTE:</b> The default behavior of this message listener container
44  * is to <b>never</b> propagate an exception thrown by a message listener up to
45  * the JMS provider. Instead, it will log any such exception at the error level.
46  * This means that from the perspective of the attendant JMS provider no such
47  * listener will ever fail.
48  *
49  * <p>The listener container offers the following message acknowledgment options:
50  * <ul>
51  * <li>"sessionAcknowledgeMode" set to "AUTO_ACKNOWLEDGE" (default):
52  * Automatic message acknowledgment <i>before</i> listener execution;
53  * no redelivery in case of exception thrown.
54  * <li>"sessionAcknowledgeMode" set to "CLIENT_ACKNOWLEDGE":
55  * Automatic message acknowledgment <i>after</i> successful listener execution;
56  * no redelivery in case of exception thrown.
57  * <li>"sessionAcknowledgeMode" set to "DUPS_OK_ACKNOWLEDGE":
58  * <i>Lazy</i> message acknowledgment during or after listener execution;
59  * <i>potential redelivery</i> in case of exception thrown.
60  * <li>"sessionTransacted" set to "true":
61  * Transactional acknowledgment after successful listener execution;
62  * <i>guaranteed redelivery</i> in case of exception thrown.
63  * </ul>
64  * The exact behavior might vary according to the concrete listener container
65  * and JMS provider used.
66  *
67  * <p>Note that there is a corner case when using "sessionTransacted",
68  * where the listener might have returned successfully but the server
69  * died before acknowledging the message. As a consequence, a message
70  * <i>might get redelivered even after successful processing</i> -
71  * potentially leading to duplicate processing of the message.
72  * This violates "exactly-once" semantics, at least potentially.
73  *
74  * <p>There are two solutions to the duplicate processing problem:
75  * <ul>
76  * <li>Either add <i>duplicate message detection</i> to your listener, in the
77  * form of a business entity existence check or a protocol table check. This
78  * usually just needs to be done in case of the JMSRedelivered flag being
79  * set on the incoming message (else just process straightforwardly).
80  * <li>Or wrap the <i>entire processing with an XA transaction</i>, covering the
81  * reception of the message as well as the execution of the message listener.
82  * This is only supported by {@link DefaultMessageListenerContainer}, through
83  * specifying a "transactionManager" (typically a
84  * {@link org.springframework.transaction.jta.JtaTransactionManager}, with
85  * a corresponding XA-aware JMS {@link javax.jms.ConnectionFactory} passed in as
86  * "connectionFactory").
87  * </ul>
88  * Note that XA transaction coordination adds significant runtime overhead,
89  * so it might be feasible to avoid it unless absolutely necessary.
90  *
91  * <p><b>Recommendations:</b>
92  * <ul>
93  * <li>The general recommendation is to set "sessionTransacted" to "true",
94  * typically in combination with local database transactions triggered by the
95  * listener implementation, through Spring's standard transaction facilities.
96  * This will work nicely in Tomcat or in a standalone environment, often
97  * combined with custom duplicate message detection (if it is unacceptable
98  * to ever process the same message twice).
99  * <li>Alternatively, specify a
100  * {@link org.springframework.transaction.jta.JtaTransactionManager} as
101  * "transactionManager" for a fully XA-aware JMS provider - typically when
102  * running on a J2EE server, but also for other environments with a JTA
103  * transaction manager present. This will give full "exactly-once" guarantees
104  * without custom duplicate message checks, at the price of additional
105  * runtime processing overhead.
106  * </ul>
107  *
108  * <p>Note that it is also possible to specify a
109  * {@link org.springframework.jms.connection.JmsTransactionManager} as external
110  * "transactionManager", providing fully synchronized Spring transactions based
111  * on local JMS transactions. The effect is similar to "sessionTransacted" set
112  * to "true", the difference being that this external transaction management
113  * will also affect independent JMS access code within the service layer
114  * (e.g. based on {@link org.springframework.jms.core.JmsTemplate} or
115  * {@link org.springframework.jms.connection.TransactionAwareConnectionFactoryProxy}),
116  * not just direct JMS Session usage in a {@link SessionAwareMessageListener}.
117  *
118  * @author Juergen Hoeller
119  * @since 2.0
120  * @see #setMessageListener
121  * @see javax.jms.MessageListener
122  * @see SessionAwareMessageListener
123  * @see #handleListenerException
124  * @see DefaultMessageListenerContainer
125  * @see SimpleMessageListenerContainer
126  * @see org.springframework.jms.listener.serversession.ServerSessionMessageListenerContainer
127  */
128 public abstract class AbstractMessageListenerContainer extends AbstractJmsListeningContainer {
129
130     private Object destination;
131
132     private String messageSelector;
133
134     private Object messageListener;
135
136     private boolean subscriptionDurable = false;
137
138     private String durableSubscriptionName;
139
140     private ExceptionListener exceptionListener;
141
142     private boolean exposeListenerSession = true;
143
144     private boolean acceptMessagesWhileStopping = false;
145
146
147     /**
148      * Set the destination to receive messages from.
149      * <p>Alternatively, specify a "destinationName", to be dynamically
150      * resolved via the {@link org.springframework.jms.support.destination.DestinationResolver}.
151      * @see #setDestinationName(String)
152      */
153     public void setDestination(Destination destination) {
154         Assert.notNull(destination, "'destination' must not be null");
155         this.destination = destination;
156         if (destination instanceof Topic && !(destination instanceof Queue )) {
157             // Clearly a Topic: let's se the "pubSubDomain" flag.
158 setPubSubDomain(true);
159         }
160     }
161
162     /**
163      * Return the destination to receive messages from. Will be <code>null</code>
164      * if the configured destination is not an actual {@link Destination} type;
165      * c.f. {@link #setDestinationName(String) when the destination is a String}.
166      */
167     protected Destination getDestination() {
168         return (this.destination instanceof Destination ? (Destination ) this.destination : null);
169     }
170
171     /**
172      * Set the name of the destination to receive messages from.
173      * <p>The specified name will be dynamically resolved via the configured
174      * {@link #setDestinationResolver(org.springframework.jms.support.destination.DestinationResolver) destination resolver}.
175      * <p>Alternatively, specify a JMS {@link Destination} object as "destination".
176      * @param destinationName the desired destination (can be <code>null</code>)
177      * @see #setDestination(javax.jms.Destination)
178      */
179     public void setDestinationName(String destinationName) {
180         Assert.notNull(destinationName, "'destinationName' must not be null");
181         this.destination = destinationName;
182     }
183
184     /**
185      * Return the name of the destination to receive messages from.
186      * Will be <code>null</code> if the configured destination is not a
187      * {@link String} type; c.f. {@link #setDestination(Destination) when
188      * it is an actual Destination}.
189      */
190     protected String getDestinationName() {
191         return (this.destination instanceof String ? (String ) this.destination : null);
192     }
193
194     /**
195      * Set the JMS message selector expression (or <code>null</code> if none).
196      * Default is none.
197      * <p>See the JMS specification for a detailed definition of selector expressions.
198      */
199     public void setMessageSelector(String messageSelector) {
200         this.messageSelector = messageSelector;
201     }
202
203     /**
204      * Return the JMS message selector expression (or <code>null</code> if none).
205      */
206     protected String getMessageSelector() {
207         return this.messageSelector;
208     }
209
210
211     /**
212      * Set the message listener implementation to register.
213      * This can be either a standard JMS {@link MessageListener} object
214      * or a Spring {@link SessionAwareMessageListener} object.
215      * @throws IllegalArgumentException if the supplied listener is not a
216      * {@link MessageListener} or a {@link SessionAwareMessageListener}
217      * @see javax.jms.MessageListener
218      * @see SessionAwareMessageListener
219      */
220     public void setMessageListener(Object messageListener) {
221         checkMessageListener(messageListener);
222         this.messageListener = messageListener;
223         if (this.durableSubscriptionName == null) {
224             // Use message listener class name as default name for a durable subscription.
225 this.durableSubscriptionName = messageListener.getClass().getName();
226         }
227     }
228
229     /**
230      * Check the given message listener, throwing an exception
231      * if it does not correspond to a supported listener type.
232      * <p>By default, only a standard JMS {@link MessageListener} object or a
233      * Spring {@link SessionAwareMessageListener} object will be accepted.
234      * @param messageListener the message listener object to check
235      * @throws IllegalArgumentException if the supplied listener is not a
236      * {@link MessageListener} or a {@link SessionAwareMessageListener}
237      * @see javax.jms.MessageListener
238      * @see SessionAwareMessageListener
239      */
240     protected void checkMessageListener(Object messageListener) {
241         if (!(messageListener instanceof MessageListener ||
242                 messageListener instanceof SessionAwareMessageListener)) {
243             throw new IllegalArgumentException (
244                     "messageListener needs to be of type [" + MessageListener .class.getName() +
245                     "] or [" + SessionAwareMessageListener.class.getName() + "]");
246         }
247     }
248
249     /**
250      * Return the message listener object to register.
251      */
252     protected Object getMessageListener() {
253         return this.messageListener;
254     }
255
256     /**
257      * Set whether to make the subscription durable. The durable subscription name
258      * to be used can be specified through the "durableSubscriptionName" property.
259      * <p>Default is "false". Set this to "true" to register a durable subscription,
260      * typically in combination with a "durableSubscriptionName" value (unless
261      * your message listener class name is good enough as subscription name).
262      * <p>Only makes sense when listening to a topic (pub-sub domain).
263      * @see #setDurableSubscriptionName
264      */
265     public void setSubscriptionDurable(boolean subscriptionDurable) {
266         this.subscriptionDurable = subscriptionDurable;
267     }
268
269     /**
270      * Return whether to make the subscription durable.
271      */
272     protected boolean isSubscriptionDurable() {
273         return this.subscriptionDurable;
274     }
275
276     /**
277      * Set the name of a durable subscription to create. To be applied in case
278      * of a topic (pub-sub domain) with subscription durability activated.
279      * <p>The durable subscription name needs to be unique within this client's
280      * JMS client id. Default is the class name of the specified message listener.
281      * <p>Note: Only 1 concurrent consumer (which is the default of this
282      * message listener container) is allowed for each durable subscription.
283      * @see #setSubscriptionDurable
284      * @see #setClientId
285      * @see #setMessageListener
286      */
287     public void setDurableSubscriptionName(String durableSubscriptionName) {
288         Assert.notNull(durableSubscriptionName, "'durableSubscriptionName' must not be null");
289         this.durableSubscriptionName = durableSubscriptionName;
290     }
291
292     /**
293      * Return the name of a durable subscription to create, if any.
294      */
295     protected String getDurableSubscriptionName() {
296         return this.durableSubscriptionName;
297     }
298
299     /**
300      * Set the JMS ExceptionListener to notify in case of a JMSException thrown
301      * by the registered message listener or the invocation infrastructure.
302      */
303     public void setExceptionListener(ExceptionListener exceptionListener) {
304         this.exceptionListener = exceptionListener;
305     }
306
307     /**
308      * Return the JMS ExceptionListener to notify in case of a JMSException thrown
309      * by the registered message listener or the invocation infrastructure, if any.
310      */
311     protected ExceptionListener getExceptionListener() {
312         return this.exceptionListener;
313     }
314
315     /**
316      * Set whether to expose the listener JMS Session to a registered
317      * {@link SessionAwareMessageListener}. Default is "true", reusing
318      * the listener's {@link Session}.
319      * <p>Turn this off to expose a fresh JMS Session fetched from the same
320      * underlying JMS {@link Connection} instead, which might be necessary
321      * on some JMS providers.
322      * @see SessionAwareMessageListener
323      */
324     public void setExposeListenerSession(boolean exposeListenerSession) {
325         this.exposeListenerSession = exposeListenerSession;
326     }
327
328     /**
329      * Return whether to expose the listener JMS {@link Session} to a
330      * registered {@link SessionAwareMessageListener}.
331      */
332     protected boolean isExposeListenerSession() {
333         return this.exposeListenerSession;
334     }
335
336     /**
337      * Set whether to accept received messages while the listener container
338      * in the process of stopping.
339      * <p>Default is "false", rejecting such messages through aborting the
340      * receive attempt. Switch this flag on to fully process such messages
341      * even in the stopping phase, with the drawback that even newly sent
342      * messages might still get processed (if coming in before all receive
343      * timeouts have expired).
344      * <p><b>NOTE:</b> Aborting receive attempts for such incoming messages
345      * might lead to the provider's retry count decreasing for the affected
346      * messages. If you have a high number of concurrent consumers, make sure
347      * that the number of retries is higher than the number of consumers,
348      * to be on the safe side for all potential stopping scenarios.
349      */
350     public void setAcceptMessagesWhileStopping(boolean acceptMessagesWhileStopping) {
351         this.acceptMessagesWhileStopping = acceptMessagesWhileStopping;
352     }
353
354     /**
355      * Return whether to accept received messages while the listener container
356      * in the process of stopping.
357      */
358     protected boolean isAcceptMessagesWhileStopping() {
359         return this.acceptMessagesWhileStopping;
360     }
361
362     protected void validateConfiguration() {
363         if (this.destination == null) {
364             throw new IllegalArgumentException ("Property 'destination' or 'destinationName' is required");
365         }
366         if (this.messageListener == null) {
367             throw new IllegalArgumentException ("Property 'messageListener' is required");
368         }
369         if (isSubscriptionDurable() && !isPubSubDomain()) {
370             throw new IllegalArgumentException ("A durable subscription requires a topic (pub-sub domain)");
371         }
372     }
373
374
375     //-------------------------------------------------------------------------
376 // Template methods for listener execution
377 //-------------------------------------------------------------------------
378
379     /**
380      * Execute the specified listener,
381      * committing or rolling back the transaction afterwards (if necessary).
382      * @param session the JMS Session to operate on
383      * @param message the received JMS Message
384      * @see #invokeListener
385      * @see #commitIfNecessary
386      * @see #rollbackOnExceptionIfNecessary
387      * @see #handleListenerException
388      */
389     protected void executeListener(Session session, Message message) {
390         try {
391             doExecuteListener(session, message);
392         }
393         catch (Throwable ex) {
394             handleListenerException(ex);
395         }
396     }
397
398     /**
399      * Execute the specified listener,
400      * committing or rolling back the transaction afterwards (if necessary).
401      * @param session the JMS Session to operate on
402      * @param message the received JMS Message
403      * @throws JMSException if thrown by JMS API methods
404      * @see #invokeListener
405      * @see #commitIfNecessary
406      * @see #rollbackOnExceptionIfNecessary
407      * @see #convertJmsAccessException
408      */
409     protected void doExecuteListener(Session session, Message message) throws JMSException {
410         if (!isAcceptMessagesWhileStopping() && !isRunning()) {
411             if (logger.isWarnEnabled()) {
412                 logger.warn("Rejecting received message because of the listener container " +
413                         "having been stopped in the meantime: " + message);
414             }
415             rollbackIfNecessary(session);
416             throw new MessageRejectedWhileStoppingException();
417         }
418         try {
419             invokeListener(session, message);
420         }
421         catch (JMSException ex) {
422             rollbackOnExceptionIfNecessary(session, ex);
423             throw ex;
424         }
425         catch (RuntimeException ex) {
426             rollbackOnExceptionIfNecessary(session, ex);
427             throw ex;
428         }
429         catch (Error err) {
430             rollbackOnExceptionIfNecessary(session, err);
431             throw err;
432         }
433         commitIfNecessary(session, message);
434     }
435
436     /**
437      * Invoke the specified listener: either as standard JMS MessageListener
438      * or (preferably) as Spring SessionAwareMessageListener.
439      * @param session the JMS Session to operate on
440      * @param message the received JMS Message
441      * @throws JMSException if thrown by JMS API methods
442      * @see #setMessageListener
443      */
444     protected void invokeListener(Session session, Message message) throws JMSException {
445         if (getMessageListener() instanceof SessionAwareMessageListener) {
446             doInvokeListener((SessionAwareMessageListener) getMessageListener(), session, message);
447         }
448         else if (getMessageListener() instanceof MessageListener ) {
449             doInvokeListener((MessageListener ) getMessageListener(), message);
450         }
451         else {
452             throw new IllegalArgumentException ("Only MessageListener and SessionAwareMessageListener supported");
453         }
454     }
455
456     /**
457      * Invoke the specified listener as Spring SessionAwareMessageListener,
458      * exposing a new JMS Session (potentially with its own transaction)
459      * to the listener if demanded.
460      * @param listener the Spring SessionAwareMessageListener to invoke
461      * @param session the JMS Session to operate on
462      * @param message the received JMS Message
463      * @throws JMSException if thrown by JMS API methods
464      * @see SessionAwareMessageListener
465      * @see #setExposeListenerSession
466      */
467     protected void doInvokeListener(SessionAwareMessageListener listener, Session session, Message message)
468             throws JMSException {
469
470         Connection conToClose = null;
471         Session sessionToClose = null;
472         try {
473             Session sessionToUse = session;
474             if (!isExposeListenerSession()) {
475                 // We need to expose a separate Session.
476 conToClose = createConnection();
477                 sessionToClose = createSession(conToClose);
478                 sessionToUse = sessionToClose;
479             }
480             // Actually invoke the message listener...
481 if (logger.isDebugEnabled()) {
482                 logger.debug("Invoking listener with message of type [" + message.getClass() +
483                         "] and session [" + sessionToUse + "]");
484             }
485             listener.onMessage(message, sessionToUse);
486             // Clean up specially exposed Session, if any.
487 if (sessionToUse != session) {
488                 if (sessionToUse.getTransacted() && isSessionLocallyTransacted(sessionToUse)) {
489                     // Transacted session created by this container -> commit.
490 JmsUtils.commitIfNecessary(sessionToUse);
491                 }
492             }
493         }
494         finally {
495             JmsUtils.closeSession(sessionToClose);
496             JmsUtils.closeConnection(conToClose);
497         }
498     }
499
500     /**
501      * Invoke the specified listener as standard JMS MessageListener.
502      * <p>Default implementation performs a plain invocation of the
503      * <code>onMessage</code> method.
504      * @param listener the JMS MessageListener to invoke
505      * @param message the received JMS Message
506      * @throws JMSException if thrown by JMS API methods
507      * @see javax.jms.MessageListener#onMessage
508      */
509     protected void doInvokeListener(MessageListener listener, Message message) throws JMSException {
510         listener.onMessage(message);
511     }
512
513     /**
514      * Perform a commit or message acknowledgement, as appropriate.
515      * @param session the JMS Session to commit
516      * @param message the Message to acknowledge
517      * @throws javax.jms.JMSException in case of commit failure
518      */
519     protected void commitIfNecessary(Session session, Message message) throws JMSException {
520         // Commit session or acknowledge message.
521 if (session.getTransacted()) {
522             // Commit necessary - but avoid commit call within a JTA transaction.
523 if (isSessionLocallyTransacted(session)) {
524                 // Transacted session created by this container -> commit.
525 JmsUtils.commitIfNecessary(session);
526             }
527         }
528         else if (isClientAcknowledge(session)) {
529             message.acknowledge();
530         }
531     }
532
533     /**
534      * Perform a rollback, if appropriate.
535      * @param session the JMS Session to rollback
536      * @throws javax.jms.JMSException in case of a rollback error
537      */
538     protected void rollbackIfNecessary(Session session) throws JMSException {
539         if (session.getTransacted() && isSessionLocallyTransacted(session)) {
540             // Transacted session created by this container -> rollback.
541 JmsUtils.rollbackIfNecessary(session);
542         }
543     }
544
545     /**
546      * Perform a rollback, handling rollback exceptions properly.
547      * @param session the JMS Session to rollback
548      * @param ex the thrown application exception or error
549      * @throws javax.jms.JMSException in case of a rollback error
550      */
551     protected void rollbackOnExceptionIfNecessary(Session session, Throwable ex) throws JMSException {
552         try {
553             if (session.getTransacted() && isSessionLocallyTransacted(session)) {
554                 // Transacted session created by this container -> rollback.
555 if (logger.isDebugEnabled()) {
556                     logger.debug("Initiating transaction rollback on application exception", ex);
557                 }
558                 JmsUtils.rollbackIfNecessary(session);
559             }
560         }
561         catch (IllegalStateException ex2) {
562             logger.debug("Could not roll back because Session already closed", ex2);
563         }
564         catch (JMSException ex2) {
565             logger.error("Application exception overridden by rollback exception", ex);
566             throw ex2;
567         }
568         catch (RuntimeException ex2) {
569             logger.error("Application exception overridden by rollback exception", ex);
570             throw ex2;
571         }
572         catch (Error err) {
573             logger.error("Application exception overridden by rollback error", ex);
574             throw err;
575         }
576     }
577
578     /**
579      * Check whether the given Session is locally transacted, that is, whether
580      * its transaction is managed by this listener container's Session handling
581      * and not by an external transaction coordinator.
582      * <p>Note: The Session's own transacted flag will already have been checked
583      * before. This method is about finding out whether the Session's transaction
584      * is local or externally coordinated.
585      * @param session the Session to check
586      * @return whether the given Session is locally transacted
587      * @see #isSessionTransacted()
588      * @see org.springframework.jms.connection.ConnectionFactoryUtils#isSessionTransactional
589      */
590     protected boolean isSessionLocallyTransacted(Session session) {
591         return isSessionTransacted();
592     }
593
594     /**
595      * Handle the given exception that arose during listener execution.
596      * <p>The default implementation logs the exception at error level,
597      * not propagating it to the JMS provider - assuming that all handling of
598      * acknowledgement and/or transactions is done by this listener container.
599      * This can be overridden in subclasses.
600      * @param ex the exception to handle
601      */
602     protected void handleListenerException(Throwable ex) {
603         if (ex instanceof MessageRejectedWhileStoppingException) {
604             // Internal exception - has been handled before.
605 return;
606         }
607         if (ex instanceof JMSException ) {
608             invokeExceptionListener((JMSException ) ex);
609         }
610         if (isActive()) {
611             // Regular case: failed while active.
612 // Log at error level.
613 logger.warn("Execution of JMS message listener failed", ex);
614         }
615         else {
616             // Rare case: listener thread failed after container shutdown.
617 // Log at debug level, to avoid spamming the shutdown log.
618 logger.debug("Listener exception after container shutdown", ex);
619         }
620     }
621
622     /**
623      * Invoke the registered JMS ExceptionListener, if any.
624      * @param ex the exception that arose during JMS processing
625      * @see #setExceptionListener
626      */
627     protected void invokeExceptionListener(JMSException ex) {
628         ExceptionListener exceptionListener = getExceptionListener();
629         if (exceptionListener != null) {
630             exceptionListener.onException(ex);
631         }
632     }
633
634
635     /**
636      * Internal exception class that indicates a rejected message on shutdown.
637      * Used to trigger a rollback for an external transaction manager in that case.
638      */
639     private static class MessageRejectedWhileStoppingException extends RuntimeException {
640
641     }
642
643 }
644

A to Z: JavaDoc & Examples

---

Daily Java News & Articles

---

Open Source Projects

---

Open Source Codes

---

Free Computer Books

---

Remove Frame

---

Popular Tags