/***************************************
 *                                     *
 *  JBoss: The OpenSource J2EE WebOS   *
 *                                     *
 *  Distributable under LGPL license.  *
 *  See terms of license at gnu.org.   *
 *                                     *
 ***************************************/

package org.jboss.resource.adapter.jms;

import java.io.Serializable;
import java.util.HashSet;
import java.util.Iterator;

import javax.jms.BytesMessage;
import javax.jms.Destination;
import javax.jms.IllegalStateException;
import javax.jms.JMSException;
import javax.jms.MessageListener;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Queue;
import javax.jms.QueueBrowser;
import javax.jms.QueueReceiver;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.StreamMessage;
import javax.jms.TemporaryQueue;
import javax.jms.TemporaryTopic;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;
import javax.resource.spi.ConnectionEvent;

import org.jboss.logging.Logger;

/**
 * Adapts the JMS QueueSession and TopicSession API to a JmsManagedConnection.
 * 
 * <p>Created: Tue Apr 17 22:39:45 2001
 * 
 * @author <a href="mailto:peter.antman@tim.se">Peter Antman </a>.
 * @author <a href="mailto:jason@planet57.com">Jason Dillon </a>.
 * @version $Revision: 1.10.6.3 $
 */
public class JmsSession implements Session, QueueSession, TopicSession
{
   private static final Logger log = Logger.getLogger(JmsSession.class);

   /** The managed connection for this session. */
   private JmsManagedConnection mc; // = null;

   /** The connection request info */
   private JmsConnectionRequestInfo info;

   /** The session factory for this session */
   private JmsSessionFactory sf;
   
   /** The message consumers */
   private HashSet consumers = new HashSet();
   
   /** The message producers */
   private HashSet producers = new HashSet();
   
   /** Whether trace is enabled */
   private boolean trace = log.isTraceEnabled();
   
   /**
    * Construct a <tt>JmsSession</tt>.
    * 
    * @param mc The managed connection for this session.
    */
   public JmsSession(final JmsManagedConnection mc, JmsConnectionRequestInfo info)
   {
      this.mc = mc;
      this.info = info;
      if (trace)
         log.trace("new JmsSession " + this + " mc=" + mc + " cri=" + info);
   }

   public void setJmsSessionFactory(JmsSessionFactory sf)
   {
      this.sf = sf;
   }
   
   /**
    * Ensure that the session is opened.
    * 
    * @return The session
    * 
    * @throws IllegalStateException The session is closed
    */
   Session getSession() throws JMSException
   {
      // ensure that the connection is opened
      if (mc == null)
         throw new IllegalStateException("The session is closed");
      
      Session session = mc.getSession();
      if (trace)
         log.trace("getSession " + session + " for " + this);
      return session;
   }

   // ---- Session API

   public BytesMessage createBytesMessage() throws JMSException
   {
      Session session = getSession();
      if (trace)
         log.trace("createBytesMessage" + session);
      return session.createBytesMessage();
   }

   public MapMessage createMapMessage() throws JMSException
   {
      Session session = getSession();
      if (trace)
         log.trace("createMapMessage" + session);
      return session.createMapMessage();
   }

   public Message createMessage() throws JMSException
   {
      Session session = getSession();
      if (trace)
         log.trace("createMessage" + session);
      return session.createMessage();
   }

   public ObjectMessage createObjectMessage() throws JMSException
   {
      Session session = getSession();
      if (trace)
         log.trace("createObjectMessage" + session);
      return session.createObjectMessage();
   }

   public ObjectMessage createObjectMessage(Serializable object) throws JMSException
   {
      Session session = getSession();
      if (trace)
         log.trace("createObjectMessage(Object)" + session);
      return session.createObjectMessage(object);
   }

   public StreamMessage createStreamMessage() throws JMSException
   {
      Session session = getSession();
      if (trace)
         log.trace("createStreamMessage" + session);
      return session.createStreamMessage();
   }

   public TextMessage createTextMessage() throws JMSException
   {
      Session session = getSession();
      if (trace)
         log.trace("createTextMessage" + session);
      return session.createTextMessage();
   }

   public TextMessage createTextMessage(String string) throws JMSException
   {
      Session session = getSession();
      if (trace)
         log.trace("createTextMessage(String)" + session);
      return session.createTextMessage(string);
   }


   public boolean getTransacted() throws JMSException
   {
      getSession(); // check closed
      return info.isTransacted();
   }

   /**
    * Always throws an Exception.
    * 
    * @throws IllegalStateException Method not allowed.
    */
   public MessageListener getMessageListener() throws JMSException
   {
      throw new IllegalStateException("Method not allowed");
   }

   /**
    * Always throws an Exception.
    * 
    * @throws IllegalStateException Method not allowed.
    */
   public void setMessageListener(MessageListener listener) throws JMSException
   {
      throw new IllegalStateException("Method not allowed");
   }

   /**
    * Always throws an Error.
    * 
    * @throws Error Method not allowed.
    */
   public void run()
   {
      // should this really throw an Error?
      throw new Error("Method not allowed");
   }

   /**
    * Closes the session. Sends a ConnectionEvent.CONNECTION_CLOSED to the
    * managed connection.
    * 
    * @throws JMSException Failed to close session.
    */
   public void close() throws JMSException
   {
      sf.closeSession(this);
      closeSession();
   }

   // FIXME - is this really OK, probably not
   public void commit() throws JMSException
   {
      Session session = getSession();
      if (info.isTransacted() == false)
         throw new IllegalStateException("Session is not transacted");
      if (trace)
         log.trace("Commit session " + this);
      session.commit();
   }

   public void rollback() throws JMSException
   {
      Session session = getSession();
      if (info.isTransacted() == false)
         throw new IllegalStateException("Session is not transacted");
      if (trace)
         log.trace("Rollback session " + this);
      session.rollback();
   }

   public void recover() throws JMSException
   {
      Session session = getSession();
      if (info.isTransacted())
         throw new IllegalStateException("Session is transacted");
      if (trace)
         log.trace("Recover session " + this);
      session.recover();
   }

   // --- TopicSession API

   public Topic createTopic(String topicName) throws JMSException
   {
      Session session = getSession();
      if (trace)
         log.trace("createTopic " + session + " topicName=" + topicName);
      Topic result = session.createTopic(topicName);
      if (trace)
         log.trace("createdTopic " + session + " topic=" + result);
      return result;
   }

   public TopicSubscriber createSubscriber(Topic topic) throws JMSException
   {
      TopicSession session = getTopicSession();
      if (trace)
         log.trace("createSubscriber " + session + " topic=" + topic);
      TopicSubscriber result = session.createSubscriber(topic);
      result = new JmsTopicSubscriber(result, this);
      if (trace)
         log.trace("createdSubscriber " + session + " JmsTopicSubscriber=" + result);
      addConsumer(result);
      return result;
   }

   public TopicSubscriber createSubscriber(Topic topic, String messageSelector, boolean noLocal) throws JMSException
   {
      TopicSession session = getTopicSession();
      if (trace)
         log.trace("createSubscriber " + session + " topic=" + topic + " selector=" + messageSelector + " noLocal=" + noLocal);
      TopicSubscriber result = session.createSubscriber(topic, messageSelector, noLocal);
      result = new JmsTopicSubscriber(result, this);
      if (trace)
         log.trace("createdSubscriber " + session + " JmsTopicSubscriber=" + result);
      addConsumer(result);
      return result;
   }

   public TopicSubscriber createDurableSubscriber(Topic topic, String name) throws JMSException
   {
      TopicSession session = getTopicSession();
      if (trace)
         log.trace("createDurableSubscriber " + session + " topic=" + topic + " name=" + name);
      TopicSubscriber result = session.createDurableSubscriber(topic, name);
      result = new JmsTopicSubscriber(result, this);
      if (trace)
         log.trace("createdDurableSubscriber " + session + " JmsTopicSubscriber=" + result);
      addConsumer(result);
      return result;
   }

   public TopicSubscriber createDurableSubscriber(Topic topic, String name, String messageSelector, boolean noLocal)
         throws JMSException
   {
      Session session = getSession();
      if (trace)
         log.trace("createDurableSubscriber " + session + " topic=" + topic + " name=" + name + " selector=" + messageSelector + " noLocal=" + noLocal);
      TopicSubscriber result = session.createDurableSubscriber(topic, name, messageSelector, noLocal);
      result = new JmsTopicSubscriber(result, this);
      if (trace)
         log.trace("createdDurableSubscriber " + session + " JmsTopicSubscriber=" + result);
      addConsumer(result);
      return result;
   }

   public TopicPublisher createPublisher(Topic topic) throws JMSException
   {
      TopicSession session = getTopicSession();
      if (trace)
         log.trace("createPublisher " + session + " topic=" + topic);
      TopicPublisher result = session.createPublisher(topic);
      if (trace)
         log.trace("createdPublisher " + session + " publisher=" + result);
      addProducer(result);
      return result;
   }

   public TemporaryTopic createTemporaryTopic() throws JMSException
   {
      Session session = getSession();
      if (trace)
         log.trace("createTemporaryTopic " + session);
      TemporaryTopic temp = session.createTemporaryTopic();
      if (trace)
         log.trace("createdTemporaryTopic " + session + " temp=" + temp);
      sf.addTemporaryTopic(temp);
      return temp;
   }

   public void unsubscribe(String name) throws JMSException
   {
      Session session = getSession();
      if (trace)
         log.trace("unsubscribe " + session + " name=" + name);
      session.unsubscribe(name);
   }

   //--- QueueSession API

   public QueueBrowser createBrowser(Queue queue) throws JMSException
   {
      Session session = getSession();
      if (trace)
         log.trace("createBrowser " + session + " queue=" + queue);
      QueueBrowser result = session.createBrowser(queue);
      if (trace)
         log.trace("createdBrowser " + session + " browser=" + result);
      return result;
   }

   public QueueBrowser createBrowser(Queue queue, String messageSelector) throws JMSException
   {
      Session session = getSession();
      if (trace)
         log.trace("createBrowser " + session + " queue=" + queue + " selector=" + messageSelector);
      QueueBrowser result = session.createBrowser(queue, messageSelector);
      if (trace)
         log.trace("createdBrowser " + session + " browser=" + result);
      return result;
   }

   public Queue createQueue(String queueName) throws JMSException
   {
      Session session = getSession();
      if (trace)
         log.trace("createQueue " + session + " queueName=" + queueName);
      Queue result = session.createQueue(queueName);
      if (trace)
         log.trace("createdQueue " + session + " queue=" + result);
      return result;
   }

   public QueueReceiver createReceiver(Queue queue) throws JMSException
   {
      QueueSession session = getQueueSession();
      if (trace)
         log.trace("createReceiver " + session + " queue=" + queue);
      QueueReceiver result = session.createReceiver(queue);
      result = new JmsQueueReceiver(result, this);
      if (trace)
         log.trace("createdReceiver " + session + " receiver=" + result);
      addConsumer(result);
      return result;
   }

   public QueueReceiver createReceiver(Queue queue, String messageSelector) throws JMSException
   {
      QueueSession session = getQueueSession();
      if (trace)
         log.trace("createReceiver " + session + " queue=" + queue + " selector=" + messageSelector);
      QueueReceiver result = session.createReceiver(queue, messageSelector);
      result = new JmsQueueReceiver(result, this);
      if (trace)
         log.trace("createdReceiver " + session + " receiver=" + result);
      addConsumer(result);
      return result;
   }

   public QueueSender createSender(Queue queue) throws JMSException
   {
      QueueSession session = getQueueSession();
      if (trace)
         log.trace("createSender " + session + " queue=" + queue);
      QueueSender result = session.createSender(queue);
      if (trace)
         log.trace("createdSender " + session + " sender=" + result);
      addProducer(result);
      return result;
   }

   public TemporaryQueue createTemporaryQueue() throws JMSException
   {
      Session session = getSession();
      if (trace)
         log.trace("createTemporaryQueue " + session);
      TemporaryQueue temp = session.createTemporaryQueue();
      if (trace)
         log.trace("createdTemporaryQueue " + session + " temp=" + temp);
      sf.addTemporaryQueue(temp);
      return temp;
   }

   // -- JMS 1.1

   public MessageConsumer createConsumer(Destination destination) throws JMSException
   {
      Session session = getSession();
      if (trace)
         log.trace("createConsumer " + session + " dest=" + destination);
      MessageConsumer result = session.createConsumer(destination);
      result = new JmsMessageConsumer(result, this);
      if (trace)
         log.trace("createdConsumer " + session + " consumer=" + result);
      addConsumer(result);
      return result;
   }

   public MessageConsumer createConsumer(Destination destination, String messageSelector) throws JMSException
   {
      Session session = getSession();
      if (trace)
         log.trace("createConsumer " + session + " dest=" + destination + " messageSelector=" + messageSelector);
      MessageConsumer result = session.createConsumer(destination, messageSelector);
      result = new JmsMessageConsumer(result, this);
      if (trace)
         log.trace("createdConsumer " + session + " consumer=" + result);
      addConsumer(result);
      return result;
   }

   public MessageConsumer createConsumer(Destination destination, String messageSelector, boolean noLocal)
         throws JMSException
   {
      Session session = getSession();
      if (trace)
         log.trace("createConsumer " + session + " dest=" + destination + " messageSelector=" + messageSelector + " noLocal=" + noLocal);
      MessageConsumer result = session.createConsumer(destination, messageSelector, noLocal);
      result = new JmsMessageConsumer(result, this);
      if (trace)
         log.trace("createdConsumer " + session + " consumer=" + result);
      addConsumer(result);
      return result;
   }

   public MessageProducer createProducer(Destination destination) throws JMSException
   {
      Session session = getSession();
      if (trace)
         log.trace("createProducer " + session + " dest=" + destination);
      MessageProducer result = getSession().createProducer(destination);
      if (trace)
         log.trace("createdProducer " + session + " producer=" + result);
      addProducer(result);
      return result;
   }

   public int getAcknowledgeMode() throws JMSException
   {
      getSession(); // check closed
      return info.getAcknowledgeMode();
   }

   // --- JmsManagedConnection api

   void setManagedConnection(final JmsManagedConnection mc)
   {
      if (this.mc != null)
         this.mc.removeHandle(this);
      this.mc = mc;
   }

   void destroy()
   {
      mc = null;
   }

   void start() throws JMSException
   {
      if (mc != null)
         mc.start();
   }

   void stop() throws JMSException
   {
      if (mc != null)
         mc.stop();
   }

   void checkStrict() throws JMSException
   {
      if (mc != null && mc.getManagedConnectionFactory().isStrict())
         throw new IllegalStateException(JmsSessionFactory.ISE);
   }
   
   void closeSession() throws JMSException
   {
      if (mc != null)
      {
         log.trace("Closing session");

         try
         {
            mc.stop();
         }
         catch (Throwable t)
         {
            log.trace("Error stopping managed connection", t);
         }
         
         synchronized (consumers)
         {
            for (Iterator i = consumers.iterator(); i.hasNext();)
            {
               JmsMessageConsumer consumer = (JmsMessageConsumer) i.next();
               try
               {
                  consumer.closeConsumer();
               }
               catch (Throwable t)
               {
                  log.trace("Error closing consumer", t);
               }
               i.remove();
            }
         }

         synchronized (producers)
         {
            for (Iterator i = producers.iterator(); i.hasNext();)
            {
               MessageProducer producer = (MessageProducer) i.next();
               try
               {
                  producer.close();
               }
               catch (Throwable t)
               {
                  log.trace("Error closing producer", t);
               }
               i.remove();
            }
         }
         
         mc.removeHandle(this);
         ConnectionEvent ev = new ConnectionEvent(mc, ConnectionEvent.CONNECTION_CLOSED);
         ev.setConnectionHandle(this);
         mc.sendEvent(ev);
         mc = null;
      }
   }
   
   void addConsumer(MessageConsumer consumer)
   {
      synchronized (consumers)
      {
         consumers.add(consumer);
      }
   }
   
   void removeConsumer(MessageConsumer consumer)
   {
      synchronized (consumers)
      {
         consumers.remove(consumer);
      }
   }
   
   void addProducer(MessageProducer producer)
   {
      synchronized (producers)
      {
         producers.add(producer);
      }
   }
   
   void removeProducer(MessageProducer producer)
   {
      synchronized (producers)
      {
         producers.remove(producer);
      }
   }
   
   QueueSession getQueueSession() throws JMSException
   {
      return (QueueSession) getSession();
   }
   
   TopicSession getTopicSession() throws JMSException
   {
      return (TopicSession) getSession();
   }
}