/*

  * 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 java.util.LinkedList;

 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.ServerSessionPool;

 import javax.jms.Session;

 import javax.jms.Topic;

 import javax.jms.TopicPublisher;

 import org.apache.commons.logging.Log;

 import org.apache.commons.logging.LogFactory;

 /** <p><code>SimpleMessenger</code> is an implementation of

   * Messenger which uses a single JMS Session for sending

   * 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 SimpleMessenger extends MessengerSupport {

     /** Logger */

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

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

     private SessionFactory sessionFactory;

     /** pool of MessengerSession instances */

     private LinkedList pool = new LinkedList();

     /** thread local data for RPCs */    

     private ThreadLocal threadLocalData = new ThreadLocal();

     private static int count;

     public SimpleMessenger() {

     }

     /** 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 {

         while (! pool.isEmpty()) {

             MessengerSession session = (MessengerSession) pool.removeFirst();

             session.close();

         }

         getSessionFactory().close();

     }

     public Session getSession() throws JMSException {

          throw new UnsupportedOperationException("Not supported by this Messenger. Please use borrowSession() and returnSession() instead");

     }

     public Session getAsyncSession() throws JMSException {

         throw new UnsupportedOperationException("Not supported by this Messenger. Please use borrowSession() and returnSession() instead");

     }

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

         ThreadLocalData data = null;

         MessengerSession messengerSession = borrowMessengerSession();

         try {

             data = getThreadLocalData(messengerSession.getSession());

             Destination replyTo = data.destination;

             message.setJMSReplyTo(replyTo);

         }

         finally {

             returnMessengerSession(messengerSession);

         }

         log.info("Sending message to destination: " + destination);

         send(destination, message);

         messengerSession = borrowMessengerSession();

         try {    

 //            MessageProducer producer = messengerSession.getMessageProducer(destination);

 //            if (messengerSession.isTopic()) {

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

 //            }

 //            else {

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

 //            }

 // 

             log.info("Message sent - now waiting for a response...");

             MessageConsumer consumer = data.consumer;

             Message response = consumer.receive();

 //            Message response = null;

             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

                 data.clear();

             }

             return response;

         }

         finally {

             returnMessengerSession(messengerSession);

         }

     }

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

         MessengerSession messengerSession = borrowMessengerSession();

         try {

             ThreadLocalData data = getThreadLocalData(messengerSession.getSession());

             Destination replyTo = data.destination;

             message.setJMSReplyTo(replyTo);

             MessageProducer producer = messengerSession.getMessageProducer(destination);

             MessageConsumer consumer = data.consumer;

             if (messengerSession.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

                 data.clear();

             }

             return response;

         }

         finally {

             returnMessengerSession(messengerSession);

         }

      }

     /**

      * @return the local thread data

      */

     protected ThreadLocalData getThreadLocalData(Session session) throws JMSException {

         ThreadLocalData data = (ThreadLocalData) threadLocalData.get();

         if (data == null) {

             data = new ThreadLocalData();

             threadLocalData.set(data);

         }

         if (data.destination == null) {

             data.destination = createTemporaryDestination();

         }

         if (data.consumer == null) {

             data.consumer = this.createConsumer(data.destination);

         }

         return data;

     }

     // Implementation methods

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

     protected static class ThreadLocalData {

         public MessageConsumer consumer;

         public Destination destination;

         public void clear() throws JMSException {

             destination = null;

             consumer.close();

         }

     }

     protected boolean isTopic(Connection connection) throws JMSException {

         return getSessionFactory().isTopic();

     }

     protected boolean isTopic(ConnectionFactory factory) throws JMSException {

         return getSessionFactory().isTopic();

     }

     /**

      * Factory method to create a new MessengerSession

      */

     protected MessengerSession createMessengerSession() throws JMSException {

         MessengerSession answer =  new MessengerSession(this, getSessionFactory());

         if (log.isDebugEnabled()) {

             log.debug("Created MessengerSession: " + ++count + " value: " + answer);

         }

         return answer;

     }

     /** 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");

     }

     protected synchronized MessengerSession borrowMessengerSession() throws JMSException {

         MessengerSession answer = null;

         if (pool.isEmpty()) {

             answer = createMessengerSession();

         }

         else {

             answer = (MessengerSession) pool.removeFirst();

         }

         if (log.isDebugEnabled()) {

             log.debug("#### Borrowing messenger session: " + answer);

         }

         return answer;

     }

     protected synchronized void returnMessengerSession(MessengerSession session) {

         if (log.isDebugEnabled()) {

             log.debug("#### Returning messenger session: " + session);

         }

         pool.addLast(session);

     }

 }