Java > Open Source Codes > org > springframework > jms > core > JmsTemplate


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.core;
18
19 import javax.jms.Connection ;
20 import javax.jms.ConnectionFactory ;
21 import javax.jms.DeliveryMode ;
22 import javax.jms.Destination ;
23 import javax.jms.JMSException ;
24 import javax.jms.Message ;
25 import javax.jms.MessageConsumer ;
26 import javax.jms.MessageProducer ;
27 import javax.jms.Session ;
28
29 import org.springframework.jms.JmsException;
30 import org.springframework.jms.connection.ConnectionFactoryUtils;
31 import org.springframework.jms.connection.JmsResourceHolder;
32 import org.springframework.jms.support.JmsUtils;
33 import org.springframework.jms.support.converter.MessageConverter;
34 import org.springframework.jms.support.converter.SimpleMessageConverter;
35 import org.springframework.jms.support.destination.JmsDestinationAccessor;
36 import org.springframework.transaction.support.TransactionSynchronizationManager;
37 import org.springframework.util.Assert;
38
39 /**
40  * Helper class that simplifies synchronous JMS access code.
41  *
42  * <p>This class requires a JMS 1.1+ provider, because it builds on the
43  * domain-independent API. <b>Use the {@link JmsTemplate102 JmsTemplate102}
44  * subclass for JMS 1.0.2 providers.</b>
45  *
46  * <p>If you want to use dynamic destination creation, you must specify
47  * the type of JMS destination to create, using the "pubSubDomain" property.
48  * For other operations, this is not necessary, in contrast to when working
49  * with JmsTemplate102. Point-to-Point (Queues) is the default domain.
50  *
51  * <p>Default settings for JMS Sessions are "not transacted" and "auto-acknowledge".
52  * As defined by the J2EE specification, the transaction and acknowledgement
53  * parameters are ignored when a JMS Session is created inside an active
54  * transaction, no matter if a JTA transaction or a Spring-managed transaction.
55  * To configure them for native JMS usage, specify appropriate values for
56  * the "sessionTransacted" and "sessionAcknowledgeMode" bean properties.
57  *
58  * <p>This template uses a
59  * {@link org.springframework.jms.support.destination.DynamicDestinationResolver}
60  * and a {@link org.springframework.jms.support.converter.SimpleMessageConverter}
61  * as default strategies for resolving a destination name or converting a message,
62  * respectively. These defaults can be overridden through the "destinationResolver"
63  * and "messageConverter" bean properties.
64  *
65  * @author Mark Pollack
66  * @author Juergen Hoeller
67  * @since 1.1
68  * @see #setConnectionFactory
69  * @see #setPubSubDomain
70  * @see #setDestinationResolver
71  * @see #setMessageConverter
72  * @see JmsTemplate102
73  * @see javax.jms.MessageProducer
74  * @see javax.jms.MessageConsumer
75  */
76 public class JmsTemplate extends JmsDestinationAccessor implements JmsOperations {
77
78     /**
79      * Default timeout for receive operations:
80      * -1 indicates a blocking receive without timeout.
81      */
82     public static final long DEFAULT_RECEIVE_TIMEOUT = -1;
83
84
85     private final JmsTemplateResourceFactory transactionalResourceFactory =
86             new JmsTemplateResourceFactory();
87
88
89     private Object defaultDestination;
90
91     private MessageConverter messageConverter;
92
93
94     private boolean messageIdEnabled = true;
95
96     private boolean messageTimestampEnabled = true;
97
98     private boolean pubSubNoLocal = false;
99
100     private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
101
102
103     private boolean explicitQosEnabled = false;
104
105     private int deliveryMode = Message.DEFAULT_DELIVERY_MODE;
106
107     private int priority = Message.DEFAULT_PRIORITY;
108
109     private long timeToLive = Message.DEFAULT_TIME_TO_LIVE;
110
111
112     /**
113      * Create a new JmsTemplate for bean-style usage.
114      * <p>Note: The ConnectionFactory has to be set before using the instance.
115      * This constructor can be used to prepare a JmsTemplate via a BeanFactory,
116      * typically setting the ConnectionFactory via setConnectionFactory.
117      * @see #setConnectionFactory
118      */
119     public JmsTemplate() {
120         initDefaultStrategies();
121     }
122
123     /**
124      * Create a new JmsTemplate, given a ConnectionFactory.
125      * @param connectionFactory the ConnectionFactory to obtain Connections from
126      */
127     public JmsTemplate(ConnectionFactory connectionFactory) {
128         this();
129         setConnectionFactory(connectionFactory);
130         afterPropertiesSet();
131     }
132
133     /**
134      * Initialize the default implementations for the template's strategies:
135      * DynamicDestinationResolver and SimpleMessageConverter.
136      * @see #setDestinationResolver
137      * @see #setMessageConverter
138      * @see org.springframework.jms.support.destination.DynamicDestinationResolver
139      * @see org.springframework.jms.support.converter.SimpleMessageConverter
140      */
141     protected void initDefaultStrategies() {
142         setMessageConverter(new SimpleMessageConverter());
143     }
144
145
146     /**
147      * Set the destination to be used on send/receive operations that do not
148      * have a destination parameter.
149      * <p>Alternatively, specify a "defaultDestinationName", to be
150      * dynamically resolved via the DestinationResolver.
151      * @see #send(MessageCreator)
152      * @see #convertAndSend(Object)
153      * @see #convertAndSend(Object, MessagePostProcessor)
154      * @see #setDefaultDestinationName(String)
155      */
156     public void setDefaultDestination(Destination destination) {
157         this.defaultDestination = destination;
158     }
159
160     /**
161      * Return the destination to be used on send/receive operations that do not
162      * have a destination parameter.
163      */
164     public Destination getDefaultDestination() {
165         return (this.defaultDestination instanceof Destination ? (Destination ) this.defaultDestination : null);
166     }
167
168     /**
169      * Set the destination name to be used on send/receive operations that
170      * do not have a destination parameter. The specified name will be
171      * dynamically resolved via the DestinationResolver.
172      * <p>Alternatively, specify a JMS Destination object as "defaultDestination".
173      * @see #send(MessageCreator)
174      * @see #convertAndSend(Object)
175      * @see #convertAndSend(Object, MessagePostProcessor)
176      * @see #setDestinationResolver
177      * @see #setDefaultDestination(javax.jms.Destination)
178      */
179     public void setDefaultDestinationName(String destinationName) {
180         this.defaultDestination = destinationName;
181     }
182
183     /**
184      * Return the destination name to be used on send/receive operations that
185      * do not have a destination parameter.
186      */
187     public String getDefaultDestinationName() {
188         return (this.defaultDestination instanceof String ? (String ) this.defaultDestination : null);
189     }
190
191     /**
192      * Set the message converter for this template. Used to resolve
193      * Object parameters to convertAndSend methods and Object results
194      * from receiveAndConvert methods.
195      * <p>The default converter is a SimpleMessageConverter, which is able
196      * to handle BytesMessages, TextMessages and ObjectMessages.
197      * @see #convertAndSend
198      * @see #receiveAndConvert
199      * @see org.springframework.jms.support.converter.SimpleMessageConverter
200      */
201     public void setMessageConverter(MessageConverter messageConverter) {
202         this.messageConverter = messageConverter;
203     }
204
205     /**
206      * Return the message converter for this template.
207      */
208     public MessageConverter getMessageConverter() {
209         return this.messageConverter;
210     }
211
212
213     /**
214      * Set whether message IDs are enabled. Default is "true".
215      * <p>This is only a hint to the JMS producer.
216      * See the JMS javadocs for details.
217      * @see javax.jms.MessageProducer#setDisableMessageID
218      */
219     public void setMessageIdEnabled(boolean messageIdEnabled) {
220         this.messageIdEnabled = messageIdEnabled;
221     }
222
223     /**
224      * Return whether message IDs are enabled.
225      */
226     public boolean isMessageIdEnabled() {
227         return this.messageIdEnabled;
228     }
229
230     /**
231      * Set whether message timestamps are enabled. Default is "true".
232      * <p>This is only a hint to the JMS producer.
233      * See the JMS javadocs for details.
234      * @see javax.jms.MessageProducer#setDisableMessageTimestamp
235      */
236     public void setMessageTimestampEnabled(boolean messageTimestampEnabled) {
237         this.messageTimestampEnabled = messageTimestampEnabled;
238     }
239
240     /**
241      * Return whether message timestamps are enabled.
242      */
243     public boolean isMessageTimestampEnabled() {
244         return this.messageTimestampEnabled;
245     }
246
247     /**
248      * Set whether to inhibit the delivery of messages published by its own connection.
249      * Default is "false".
250      * @see javax.jms.TopicSession#createSubscriber(javax.jms.Topic, String, boolean)
251      */
252     public void setPubSubNoLocal(boolean pubSubNoLocal) {
253         this.pubSubNoLocal = pubSubNoLocal;
254     }
255
256     /**
257      * Return whether to inhibit the delivery of messages published by its own connection.
258      */
259     public boolean isPubSubNoLocal() {
260         return this.pubSubNoLocal;
261     }
262
263     /**
264      * Set the timeout to use for receive calls.
265      * The default is -1, which means no timeout.
266      * @see javax.jms.MessageConsumer#receive(long)
267      * @see javax.jms.MessageConsumer#receive()
268      */
269     public void setReceiveTimeout(long receiveTimeout) {
270         this.receiveTimeout = receiveTimeout;
271     }
272
273     /**
274      * Return the timeout to use for receive calls.
275      */
276     public long getReceiveTimeout() {
277         return this.receiveTimeout;
278     }
279
280
281     /**
282      * Set if the QOS values (deliveryMode, priority, timeToLive)
283      * should be used for sending a message.
284      * @see #setDeliveryMode
285      * @see #setPriority
286      * @see #setTimeToLive
287      */
288     public void setExplicitQosEnabled(boolean explicitQosEnabled) {
289         this.explicitQosEnabled = explicitQosEnabled;
290     }
291
292     /**
293      * If "true", then the values of deliveryMode, priority, and timeToLive
294      * will be used when sending a message. Otherwise, the default values,
295      * that may be set administratively, will be used.
296      * @return true if overriding default values of QOS parameters
297      * (deliveryMode, priority, and timeToLive)
298      * @see #setDeliveryMode
299      * @see #setPriority
300      * @see #setTimeToLive
301      */
302     public boolean isExplicitQosEnabled() {
303         return this.explicitQosEnabled;
304     }
305
306     /**
307      * Set whether message delivery should be persistent or non-persistent,
308      * specified as boolean value ("true" or "false"). This will set the delivery
309      * mode accordingly, to either "PERSISTENT" (1) or "NON_PERSISTENT" (2).
310      * <p>Default it "true" aka delivery mode "PERSISTENT".
311      * @see #setDeliveryMode(int)
312      * @see javax.jms.DeliveryMode#PERSISTENT
313      * @see javax.jms.DeliveryMode#NON_PERSISTENT
314      */
315     public void setDeliveryPersistent(boolean deliveryPersistent) {
316         this.deliveryMode = (deliveryPersistent ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT);
317     }
318
319     /**
320      * Set the delivery mode to use when sending a message.
321      * Default is the Message default: "PERSISTENT".
322      * <p>Since a default value may be defined administratively,
323      * this is only used when "isExplicitQosEnabled" equals "true".
324      * @param deliveryMode the delivery mode to use
325      * @see #isExplicitQosEnabled
326      * @see javax.jms.DeliveryMode#PERSISTENT
327      * @see javax.jms.DeliveryMode#NON_PERSISTENT
328      * @see javax.jms.Message#DEFAULT_DELIVERY_MODE
329      * @see javax.jms.MessageProducer#send(javax.jms.Message, int, int, long)
330      */
331     public void setDeliveryMode(int deliveryMode) {
332         this.deliveryMode = deliveryMode;
333     }
334
335     /**
336      * Return the delivery mode to use when sending a message.
337      */
338     public int getDeliveryMode() {
339         return this.deliveryMode;
340     }
341
342     /**
343      * Set the priority of a message when sending.
344      * <p>Since a default value may be defined administratively,
345      * this is only used when "isExplicitQosEnabled" equals "true".
346      * @see #isExplicitQosEnabled
347      * @see javax.jms.Message#DEFAULT_PRIORITY
348      * @see javax.jms.MessageProducer#send(javax.jms.Message, int, int, long)
349      */
350     public void setPriority(int priority) {
351         this.priority = priority;
352     }
353
354     /**
355      * Return the priority of a message when sending.
356      */
357     public int getPriority() {
358         return this.priority;
359     }
360
361     /**
362      * Set the time-to-live of the message when sending.
363      * <p>Since a default value may be defined administratively,
364      * this is only used when "isExplicitQosEnabled" equals "true".
365      * @param timeToLive the message's lifetime (in milliseconds)
366      * @see #isExplicitQosEnabled
367      * @see javax.jms.Message#DEFAULT_TIME_TO_LIVE
368      * @see javax.jms.MessageProducer#send(javax.jms.Message, int, int, long)
369      */
370     public void setTimeToLive(long timeToLive) {
371         this.timeToLive = timeToLive;
372     }
373
374     /**
375      * Return the time-to-live of the message when sending.
376      */
377     public long getTimeToLive() {
378         return this.timeToLive;
379     }
380
381
382     private void checkDefaultDestination() throws IllegalStateException {
383         if (this.defaultDestination == null) {
384             throw new IllegalStateException (
385                     "No defaultDestination or defaultDestinationName specified. Check configuration of JmsTemplate.");
386         }
387     }
388
389     private void checkMessageConverter() throws IllegalStateException {
390         if (getMessageConverter() == null) {
391             throw new IllegalStateException ("No messageConverter registered. Check configuration of JmsTemplate.");
392         }
393     }
394
395
396     /**
397      * Execute the action specified by the given action object within a
398      * JMS Session. Generalized version of <code>execute(SessionCallback)</code>,
399      * allowing the JMS Connection to be started on the fly.
400      * <p>Use <code>execute(SessionCallback)</code> for the general case.
401      * Starting the JMS Connection is just necessary for receiving messages,
402      * which is preferably achieved through the <code>receive</code> methods.
403      * @param action callback object that exposes the session
404      * @return the result object from working with the session
405      * @throws JmsException if there is any problem
406      * @see #execute(SessionCallback)
407      * @see #receive
408      */
409     public Object execute(SessionCallback action, boolean startConnection) throws JmsException {
410         Assert.notNull(action, "Callback object must not be null");
411
412         Connection conToClose = null;
413         Session sessionToClose = null;
414         try {
415             Session sessionToUse = ConnectionFactoryUtils.doGetTransactionalSession(
416                     getConnectionFactory(), this.transactionalResourceFactory);
417             if (sessionToUse == null) {
418                 conToClose = createConnection();
419                 sessionToClose = createSession(conToClose);
420                 if (startConnection) {
421                     conToClose.start();
422                 }
423                 sessionToUse = sessionToClose;
424             }
425             if (logger.isDebugEnabled()) {
426                 logger.debug("Executing callback on JMS Session [" + sessionToUse + "]");
427             }
428             return action.doInJms(sessionToUse);
429         }
430         catch (JMSException ex) {
431             throw convertJmsAccessException(ex);
432         }
433         finally {
434             JmsUtils.closeSession(sessionToClose);
435             ConnectionFactoryUtils.releaseConnection(conToClose, getConnectionFactory(), startConnection);
436         }
437     }
438
439     public Object execute(SessionCallback action) throws JmsException {
440         return execute(action, false);
441     }
442
443     public Object execute(final ProducerCallback action) throws JmsException {
444         Assert.notNull(action, "Callback object must not be null");
445
446         return execute(new SessionCallback() {
447             public Object doInJms(Session session) throws JMSException {
448                 MessageProducer producer = createProducer(session, null);
449                 try {
450                     return action.doInJms(session, producer);
451                 }
452                 finally {
453                     JmsUtils.closeMessageProducer(producer);
454                 }
455             }
456         }, false);
457     }
458
459
460     //-------------------------------------------------------------------------
461 // Convenience methods for sending messages
462 //-------------------------------------------------------------------------
463
464     public void send(MessageCreator messageCreator) throws JmsException {
465         checkDefaultDestination();
466         if (getDefaultDestination() != null) {
467             send(getDefaultDestination(), messageCreator);
468         }
469         else {
470             send(getDefaultDestinationName(), messageCreator);
471         }
472     }
473
474     public void send(final Destination destination, final MessageCreator messageCreator) throws JmsException {
475         execute(new SessionCallback() {
476             public Object doInJms(Session session) throws JMSException {
477                 doSend(session, destination, messageCreator);
478                 return null;
479             }
480         }, false);
481     }
482
483     public void send(final String destinationName, final MessageCreator messageCreator) throws JmsException {
484         execute(new SessionCallback() {
485             public Object doInJms(Session session) throws JMSException {
486                 Destination destination = resolveDestinationName(session, destinationName);
487                 doSend(session, destination, messageCreator);
488                 return null;
489             }
490         }, false);
491     }
492
493     /**
494      * Send the given JMS message.
495      * @param session the JMS Session to operate on
496      * @param destination the JMS Destination to send to
497      * @param messageCreator callback to create a JMS Message
498      * @throws JMSException if thrown by JMS API methods
499      */
500     protected void doSend(Session session, Destination destination, MessageCreator messageCreator)
501             throws JMSException {
502
503         Assert.notNull(messageCreator, "MessageCreator must not be null");
504
505         MessageProducer producer = createProducer(session, destination);
506         try {
507             Message message = messageCreator.createMessage(session);
508             if (logger.isDebugEnabled()) {
509                 logger.debug("Sending created message [" + message + "]");
510             }
511             doSend(producer, message);
512             // Check commit - avoid commit call within a JTA transaction.
513 if (session.getTransacted() && isSessionLocallyTransacted(session)) {
514                 // Transacted session created by this template -> commit.
515 JmsUtils.commitIfNecessary(session);
516             }
517         }
518         finally {
519             JmsUtils.closeMessageProducer(producer);
520         }
521     }
522
523     /**
524      * Actually send the given JMS message.
525      * @param producer the JMS MessageProducer to send with
526      * @param message the JMS Message to send
527      * @throws JMSException if thrown by JMS API methods
528      */
529     protected void doSend(MessageProducer producer, Message message) throws JMSException {
530         if (isExplicitQosEnabled()) {
531             producer.send(message, getDeliveryMode(), getPriority(), getTimeToLive());
532         }
533         else {
534             producer.send(message);
535         }
536     }
537
538
539     //-------------------------------------------------------------------------
540 // Convenience methods for sending auto-converted messages
541 //-------------------------------------------------------------------------
542
543     public void convertAndSend(Object message) throws JmsException {
544         checkDefaultDestination();
545         if (getDefaultDestination() != null) {
546             convertAndSend(getDefaultDestination(), message);
547         }
548         else {
549             convertAndSend(getDefaultDestinationName(), message);
550         }
551     }
552
553     public void convertAndSend(Destination destination, final Object message) throws JmsException {
554         checkMessageConverter();
555         send(destination, new MessageCreator() {
556             public Message createMessage(Session session) throws JMSException {
557                 return getMessageConverter().toMessage(message, session);
558             }
559         });
560     }
561
562     public void convertAndSend(String destinationName, final Object message) throws JmsException {
563         checkMessageConverter();
564         send(destinationName, new MessageCreator() {
565             public Message createMessage(Session session) throws JMSException {
566                 return getMessageConverter().toMessage(message, session);
567             }
568         });
569     }
570
571     public void convertAndSend(Object message, MessagePostProcessor postProcessor) throws JmsException {
572         checkDefaultDestination();
573         if (getDefaultDestination() != null) {
574             convertAndSend(getDefaultDestination(), message, postProcessor);
575         }
576         else {
577             convertAndSend(getDefaultDestinationName(), message, postProcessor);
578         }
579     }
580
581     public void convertAndSend(
582             Destination destination, final Object message, final MessagePostProcessor postProcessor)
583             throws JmsException {
584
585         checkMessageConverter();
586         send(destination, new MessageCreator() {
587             public Message createMessage(Session session) throws JMSException {
588                 Message msg = getMessageConverter().toMessage(message, session);
589                 return postProcessor.postProcessMessage(msg);
590             }
591         });
592     }
593
594     public void convertAndSend(
595             String destinationName, final Object message, final MessagePostProcessor postProcessor)
596         throws JmsException {
597
598         checkMessageConverter();
599         send(destinationName, new MessageCreator() {
600             public Message createMessage(Session session) throws JMSException {
601                 Message msg = getMessageConverter().toMessage(message, session);
602                 return postProcessor.postProcessMessage(msg);
603             }
604         });
605     }
606
607
608     //-------------------------------------------------------------------------
609 // Convenience methods for receiving messages
610 //-------------------------------------------------------------------------
611
612     public Message receive() throws JmsException {
613         checkDefaultDestination();
614         if (getDefaultDestination() != null) {
615             return receive(getDefaultDestination());
616         }
617         else {
618             return receive(getDefaultDestinationName());
619         }
620     }
621
622     public Message receive(final Destination destination) throws JmsException {
623         return (Message ) execute(new SessionCallback() {
624             public Object doInJms(Session session) throws JMSException {
625                 return doReceive(session, destination, null);
626             }
627         }, true);
628     }
629
630     public Message receive(final String destinationName) throws JmsException {
631         return (Message ) execute(new SessionCallback() {
632             public Object doInJms(Session session) throws JMSException {
633                 Destination destination = resolveDestinationName(session, destinationName);
634                 return doReceive(session, destination, null);
635             }
636         }, true);
637     }
638
639     public Message receiveSelected(String messageSelector) throws JmsException {
640         checkDefaultDestination();
641         if (getDefaultDestination() != null) {
642             return receiveSelected(getDefaultDestination(), messageSelector);
643         }
644         else {
645             return receiveSelected(getDefaultDestinationName(), messageSelector);
646         }
647     }
648
649     public Message receiveSelected(final Destination destination, final String messageSelector) throws JmsException {
650         return (Message ) execute(new SessionCallback() {
651             public Object doInJms(Session session) throws JMSException {
652                 return doReceive(session, destination, messageSelector);
653             }
654         }, true);
655     }
656
657     public Message receiveSelected(final String destinationName, final String messageSelector) throws JmsException {
658         return (Message ) execute(new SessionCallback() {
659             public Object doInJms(Session session) throws JMSException {
660                 Destination destination = resolveDestinationName(session, destinationName);
661                 return doReceive(session, destination, messageSelector);
662             }
663         }, true);
664     }
665
666     /**
667      * Receive a JMS message.
668      * @param session the JMS Session to operate on
669      * @param destination the JMS Destination to receive from
670      * @param messageSelector the message selector for this consumer (can be <code>null</code>)
671      * @throws JMSException if thrown by JMS API methods
672      */
673     protected Message doReceive(Session session, Destination destination, String messageSelector)
674             throws JMSException {
675
676         return doReceive(session, createConsumer(session, destination, messageSelector));
677     }
678
679     /**
680      * Actually receive a JMS message.
681      * @param session the JMS Session to operate on
682      * @param consumer the JMS MessageConsumer to send with
683      * @return the JMS Message received, or <code>null</code> if none
684      * @throws JMSException if thrown by JMS API methods
685      */
686     protected Message doReceive(Session session, MessageConsumer consumer) throws JMSException {
687         try {
688             // Use transaction timeout (if available).
689 long timeout = getReceiveTimeout();
690             JmsResourceHolder resourceHolder =
691                     (JmsResourceHolder) TransactionSynchronizationManager.getResource(getConnectionFactory());
692             if (resourceHolder != null && resourceHolder.hasTimeout()) {
693                 timeout = resourceHolder.getTimeToLiveInMillis();
694             }
695             Message message = (timeout >= 0) ?
696                     consumer.receive(timeout) : consumer.receive();
697             if (session.getTransacted()) {
698                 // Commit necessary - but avoid commit call within a JTA transaction.
699 if (isSessionLocallyTransacted(session)) {
700                     // Transacted session created by this template -> commit.
701 JmsUtils.commitIfNecessary(session);
702                 }
703             }
704             else if (isClientAcknowledge(session)) {
705                 // Manually acknowledge message, if any.
706 if (message != null) {
707                     message.acknowledge();
708                 }
709             }
710             return message;
711         }
712         finally {
713             JmsUtils.closeMessageConsumer(consumer);
714         }
715     }
716
717
718     //-------------------------------------------------------------------------
719 // Convenience methods for receiving auto-converted messages
720 //-------------------------------------------------------------------------
721
722     public Object receiveAndConvert() throws JmsException {
723         checkMessageConverter();
724         return doConvertFromMessage(receive());
725     }
726
727     public Object receiveAndConvert(Destination destination) throws JmsException {
728         checkMessageConverter();
729         return doConvertFromMessage(receive(destination));
730     }
731
732     public Object receiveAndConvert(String destinationName) throws JmsException {
733         checkMessageConverter();
734         return doConvertFromMessage(receive(destinationName));
735     }
736
737     public Object receiveSelectedAndConvert(String messageSelector) throws JmsException {
738         checkMessageConverter();
739         return doConvertFromMessage(receiveSelected(messageSelector));
740     }
741
742     public Object receiveSelectedAndConvert(Destination destination, String messageSelector) throws JmsException {
743         checkMessageConverter();
744         return doConvertFromMessage(receiveSelected(destination, messageSelector));
745     }
746
747     public Object receiveSelectedAndConvert(String destinationName, String messageSelector) throws JmsException {
748         checkMessageConverter();
749         return doConvertFromMessage(receiveSelected(destinationName, messageSelector));
750     }
751
752     /**
753      * Extract the content from the given JMS message.
754      * @param message the JMS Message to convert (can be <code>null</code>)
755      * @return the content of the message, or <code>null</code> if none
756      */
757     protected Object doConvertFromMessage(Message message) {
758         if (message != null) {
759             try {
760                 return getMessageConverter().fromMessage(message);
761             }
762             catch (JMSException ex) {
763                 throw convertJmsAccessException(ex);
764             }
765         }
766         return null;
767     }
768
769
770     //-------------------------------------------------------------------------
771 // JMS 1.1 factory methods, potentially overridden for JMS 1.0.2
772 //-------------------------------------------------------------------------
773
774     /**
775      * Fetch an appropriate Connection from the given JmsResourceHolder.
776      * <p>This implementation accepts any JMS 1.1 Connection.
777      * @param holder the JmsResourceHolder
778      * @return an appropriate Connection fetched from the holder,
779      * or <code>null</code> if none found
780      */
781     protected Connection getConnection(JmsResourceHolder holder) {
782         return holder.getConnection();
783     }
784
785     /**
786      * Fetch an appropriate Session from the given JmsResourceHolder.
787      * <p>This implementation accepts any JMS 1.1 Session.
788      * @param holder the JmsResourceHolder
789      * @return an appropriate Session fetched from the holder,
790      * or <code>null</code> if none found
791      */
792     protected Session getSession(JmsResourceHolder holder) {
793         return holder.getSession();
794     }
795
796     /**
797      * Check whether the given Session is locally transacted, that is, whether
798      * its transaction is managed by this listener container's Session handling
799      * and not by an external transaction coordinator.
800      * <p>Note: The Session's own transacted flag will already have been checked
801      * before. This method is about finding out whether the Session's transaction
802      * is local or externally coordinated.
803      * @param session the Session to check
804      * @return whether the given Session is locally transacted
805      * @see #isSessionTransacted()
806      * @see org.springframework.jms.connection.ConnectionFactoryUtils#isSessionTransactional
807      */
808     protected boolean isSessionLocallyTransacted(Session session) {
809         return isSessionTransacted() &&
810                 !ConnectionFactoryUtils.isSessionTransactional(session, getConnectionFactory());
811     }
812
813     /**
814      * Create a JMS MessageProducer for the given Session and Destination,
815      * configuring it to disable message ids and/or timestamps (if necessary).
816      * <p>Delegates to <code>doCreateProducer</code> for creation of the raw
817      * JMS MessageProducer, which needs to be specific to JMS 1.1 or 1.0.2.
818      * @param session the JMS Session to create a MessageProducer for
819      * @param destination the JMS Destination to create a MessageProducer for
820      * @return the new JMS MessageProducer
821      * @throws JMSException if thrown by JMS API methods
822      * @see #doCreateProducer
823      * @see #setMessageIdEnabled
824      * @see #setMessageTimestampEnabled
825      */
826     protected MessageProducer createProducer(Session session, Destination destination) throws JMSException {
827         MessageProducer producer = doCreateProducer(session, destination);
828         if (!isMessageIdEnabled()) {
829             producer.setDisableMessageID(true);
830         }
831         if (!isMessageTimestampEnabled()) {
832             producer.setDisableMessageTimestamp(true);
833         }
834         return producer;
835     }
836
837     /**
838      * Create a raw JMS MessageProducer for the given Session and Destination.
839      * <p>This implementation uses JMS 1.1 API.
840      * @param session the JMS Session to create a MessageProducer for
841      * @param destination the JMS Destination to create a MessageProducer for
842      * @return the new JMS MessageProducer
843      * @throws JMSException if thrown by JMS API methods
844      */
845     protected MessageProducer doCreateProducer(Session session, Destination destination) throws JMSException {
846         return session.createProducer(destination);
847     }
848
849     /**
850      * Create a JMS MessageConsumer for the given Session and Destination.
851      * <p>This implementation uses JMS 1.1 API.
852      * @param session the JMS Session to create a MessageConsumer for
853      * @param destination the JMS Destination to create a MessageConsumer for
854      * @param messageSelector the message selector for this consumer (can be <code>null</code>)
855      * @return the new JMS MessageConsumer
856      * @throws JMSException if thrown by JMS API methods
857      */
858     protected MessageConsumer createConsumer(Session session, Destination destination, String messageSelector)
859             throws JMSException {
860
861         // Only pass in the NoLocal flag in case of a Topic:
862 // Some JMS providers, such as WebSphere MQ 6.0, throw IllegalStateException
863 // in case of the NoLocal flag being specified for a Queue.
864 if (isPubSubDomain()) {
865             return session.createConsumer(destination, messageSelector, isPubSubNoLocal());
866         }
867         else {
868             return session.createConsumer(destination, messageSelector);
869         }
870     }
871
872
873     /**
874      * ResourceFactory implementation that delegates to this template's protected callback methods.
875      */
876     private class JmsTemplateResourceFactory implements ConnectionFactoryUtils.ResourceFactory {
877
878         public Connection getConnection(JmsResourceHolder holder) {
879             return JmsTemplate.this.getConnection(holder);
880         }
881
882         public Session getSession(JmsResourceHolder holder) {
883             return JmsTemplate.this.getSession(holder);
884         }
885
886         public Connection createConnection() throws JMSException {
887             return JmsTemplate.this.createConnection();
888         }
889
890         public Session createSession(Connection con) throws JMSException {
891             return JmsTemplate.this.createSession(con);
892         }
893
894         public boolean isSynchedLocalTransactionAllowed() {
895             return JmsTemplate.this.isSessionTransacted();
896         }
897     }
898
899 }
900

A to Z: JavaDoc & Examples

---

Daily Java News & Articles

---

Open Source Projects

---

Open Source Codes

---

Free Computer Books

---

Remove Frame

---

Popular Tags