/*

  * Copyright 1999-2004 The Apache Software Foundation.

  * 

  * Licensed under the Apache License, Version 2.0 (the "License");

  * you may not use this file except in compliance with the License.

  * You may obtain a copy of the License at

  * 

  *      http://www.apache.org/licenses/LICENSE-2.0

  * 

  * Unless required by applicable law or agreed to in writing, software

  * distributed under the License is distributed on an "AS IS" BASIS,

  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

  * See the License for the specific language governing permissions and

  * limitations under the License.

  */

 package org.apache.commons.messenger;

 import javax.jms.Connection;

 import javax.jms.ConnectionFactory;

 import javax.jms.Destination;

 import javax.jms.JMSException;

 import javax.jms.Message;

 import javax.jms.MessageConsumer;

 import javax.jms.MessageListener;

 import javax.jms.MessageProducer;

 import javax.jms.Queue;

 import javax.jms.QueueSender;

 import javax.jms.QueueSession;

 import javax.jms.ServerSessionPool;

 import javax.jms.Session;

 import javax.jms.Topic;

 import javax.jms.TopicPublisher;

 import javax.jms.TopicSession;

 import javax.naming.Context;

 import org.apache.commons.logging.Log;

 import org.apache.commons.logging.LogFactory;

 /** <p><code>DefaultMessenger</code> is the default implementation of

   * Messenger which uses a {@link ThreadLocal} variable

   * to keep the JMS Session that should be used for a given calling thread.</p>

   *

   * @author <a href="mailto:jstrachan@apache.org">James Strachan</a>

   * @version $Revision: 155459 $

   */

 public class DefaultMessenger extends MessengerSupport {

     private static final boolean SHARE_CONNECTION = true;

     /** Logger */

     private static final Log log = LogFactory.getLog(DefaultMessenger.class);

     /** the MessengerSession for each thread */

     private ThreadLocal messengerSessionPool = new ThreadLocal();

     /** the SessionFactory used to create new JMS sessions */

     private SessionFactory sessionFactory;

     public DefaultMessenger() {

     }

     /** Returns the SessionFactory used to create new JMS sessions */

     public SessionFactory getSessionFactory() throws JMSException {

         if (sessionFactory == null) {

             sessionFactory = createSessionFactory();

         }

         return sessionFactory;

     }

     /** Sets the SessionFactory used to create new JMS sessions */

     public void setSessionFactory(SessionFactory sessionFactory) {

         this.sessionFactory = sessionFactory;

     }

     public Connection getConnection() throws JMSException {

         return getSessionFactory().getConnection();

     }

     public ServerSessionPool createServerSessionPool(MessageListener messageListener, int maxThreads)

         throws JMSException {

         return getSessionFactory().createServerSessionPool(messageListener, maxThreads);

     }

     public synchronized void close() throws JMSException {

         /** note that only the current session is terminated */

         MessengerSession session = getMessengerSession();

         // clear all the pools...

         messengerSessionPool = new ThreadLocal();

         session.close();

         getSessionFactory().close();

     }

     public Session getSession() throws JMSException {

         return getMessengerSession().getSession();

     }

     public Session getAsyncSession() throws JMSException {

         return getMessengerSession().getListenerSession();

     }

     public Message call(Destination destination, Message message) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         try {

             Destination replyTo = getReplyToDestination();

             message.setJMSReplyTo(replyTo);

             // NOTE - we could consider adding a correlation ID per request so that we can ignore

             // any cruft or old messages that are sent onto our inbound queue.

             //

             // Though that does mean that we must then rely on the inbound message having

             // the right correlationID. Though at least this strategy would mean

             // that we could have a single consumer on a temporary queue for all threads

             // and use correlation IDs to dispatch to the corrent thread

             //

             // Maybe this should be a configurable strategy

             MessageProducer producer = session.getMessageProducer(destination);

             MessageConsumer consumer = getReplyToConsumer();

             if (session.isTopic()) {

                 ((TopicPublisher) producer).publish((Topic) destination, message);

             }

             else {

                 ((QueueSender) producer).send((Queue) destination, message);

             }

             Message response = consumer.receive();

             if (response == null) {

                 // we could have timed out so lets trash the temporary destination

                 // so that the next call() method will use a new destination to avoid

                 // the response for this call() coming back on later call() invokcations

                 clearReplyToDestination();

             }

             return response;

         }

         finally {

             returnMessengerSession(session);

         }

     }

     public Message call(Destination destination, Message message, long timeoutMillis) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         try {

             Destination replyTo = getReplyToDestination();

             message.setJMSReplyTo(replyTo);

             // NOTE - we could consider adding a correlation ID per request so that we can ignore

             // any cruft or old messages that are sent onto our inbound queue.

             //

             // Though that does mean that we must then rely on the inbound message having

             // the right correlationID. Though at least this strategy would mean

             // that we could have a single consumer on a temporary queue for all threads

             // and use correlation IDs to dispatch to the corrent thread

             //

             // Maybe this should be a configurable strategy

             MessageProducer producer = session.getMessageProducer(destination);

             MessageConsumer consumer = getReplyToConsumer();

             if (session.isTopic()) {

                 ((TopicPublisher) producer).publish((Topic) destination, message);

             }

             else {

                 ((QueueSender) producer).send((Queue) destination, message);

             }

             Message response = consumer.receive(timeoutMillis);

             if (response == null) {

                 // we could have timed out so lets trash the temporary destination

                 // so that the next call() method will use a new destination to avoid

                 // the response for this call() coming back on later call() invokcations

                 clearReplyToDestination();

             }

             return response;

         }

         finally {

             returnMessengerSession(session);

         }

     }

     // Implementation methods

     //-------------------------------------------------------------------------

     protected boolean isTopic(Connection connection) throws JMSException {

         return getSessionFactory().isTopic();

     }

     protected boolean isTopic(ConnectionFactory factory) throws JMSException {

         return getSessionFactory().isTopic();

     }

     /**

      * @return the MessageConsumer for this threads temporary destination

      * which is cached for the duration of this process.

      */

     protected MessageConsumer getReplyToConsumer() throws JMSException {

         MessengerSession messengerSession = getMessengerSession();

         MessageConsumer consumer = messengerSession.getReplyToConsumer();

         synchronized ( messengerSession ) {

             if (consumer == null) {

                 consumer = createMessageConsumer(messengerSession, messengerSession.getSession(), messengerSession.getReplyToDestination());

                 messengerSession.setReplyToConsumer(consumer);

             }

         }

         return consumer;

     }

     /**

      * Clears the temporary destination used to receive reply-to messages

      * which will lazily force a new destination and consumer to be created next

      * time a call() method is invoked.

      */

     protected void clearReplyToDestination() throws JMSException {

         MessengerSession messengerSession = getMessengerSession();

         messengerSession.setReplyToDestination(null);

         MessageConsumer consumer = messengerSession.getReplyToConsumer();

         if (consumer != null) {

             messengerSession.setReplyToConsumer(null);

             // ensure that everything is nullified first before we close

             // just in case an exception occurs

             consumer.close();

         }

     }

     protected Destination getReplyToDestination() throws JMSException {

         return getMessengerSession().getReplyToDestination();

     }

     protected MessengerSession getMessengerSession() throws JMSException {

         return borrowMessengerSession();

     }

     protected MessengerSession borrowMessengerSession() throws JMSException {

         MessengerSession answer = (MessengerSession) messengerSessionPool.get();

         if (answer == null) {

             answer = createMessengerSession();

             messengerSessionPool.set(answer);

         }

         return answer;

     }

     protected void returnMessengerSession(MessengerSession session) {

     }

     /**

      * Factory method to create a new MessengerSession

      */

     protected MessengerSession createMessengerSession() throws JMSException {

         return new MessengerSession(this, getSessionFactory());

     }

     /** Factory method to create a SessionFactory.

       * Derived classes could override this method to create the SessionFactory

       * from a well known place

       */

     protected SessionFactory createSessionFactory() throws JMSException {

         throw new JMSException("No SessionFactory configured for this Messenger. Cannot create a new JMS Session");

     }

     public Queue getQueue(QueueSession session, String subject) throws JMSException {

         // XXXX: might want to cache

         Context ctx = null;

         JNDISessionFactory factory = null;

         Queue queue = null;

         if (isJndiDestinations()) {

             try {

                 factory = (JNDISessionFactory) getSessionFactory();

                 ctx = factory.getContext();

                 queue = (Queue) ctx.lookup(subject);

             }

             catch (Exception e) {

                 log.error("Unable to lookup subject: " + subject + ". Exception: " + e, e);

             }

         }

         else {

             // XXXX: might want to cache

             queue = session.createQueue(subject);

         }

         return queue;

     }

     public Topic getTopic(TopicSession session, String subject) throws JMSException {

         // XXXX: might want to cache

         Context ctx = null;

         JNDISessionFactory factory = null;

         Topic topic = null;

         if (isJndiDestinations()) {

             try {

                 factory = (JNDISessionFactory) getSessionFactory();

                 ctx = factory.getContext();

                 topic = (Topic) ctx.lookup(subject);

             }

             catch (Exception e) {

                 log.error("Unable to lookup subject: " + subject + ". Exception: " + e, e);

             }

         }

         else {

             topic = session.createTopic(subject);

         }

         return topic;

     }

 }