/*
* JBoss, Home of Professional Open Source
* Copyright 2005, JBoss Inc., and individual contributors as indicated
* by the @authors tag. See the copyright.txt in the distribution for a
* full listing of individual contributors.
*
* This is free software; you can redistribute it and/or modify it
* under the terms of the GNU Lesser General Public License as
* published by the Free Software Foundation; either version 2.1 of
* the License, or (at your option) any later version.
*
* This software is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this software; if not, write to the Free
* Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
* 02110-1301 USA, or see the FSF site: http://www.fsf.org.
*/
package org.jboss.resource.adapter.jms;

import java.util.HashSet;
import java.util.Iterator;

import javax.jms.ConnectionConsumer;
import javax.jms.ConnectionMetaData;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.IllegalStateException;
import javax.jms.JMSException;
import javax.jms.Queue;
import javax.jms.QueueSession;
import javax.jms.ServerSessionPool;
import javax.jms.Session;
import javax.jms.TemporaryQueue;
import javax.jms.TemporaryTopic;
import javax.jms.Topic;
import javax.jms.TopicSession;
import javax.naming.Reference;
import javax.resource.Referenceable;
import javax.resource.ResourceException;
import javax.resource.spi.ConnectionManager;
import javax.resource.spi.ManagedConnectionFactory;

import org.jboss.logging.Logger;

/**
 * Implements the JMS Connection API and produces {@link JmsSession} objects.
 *
 * @author  <a href="mailto:peter.antman@tim.se">Peter Antman</a>.
 * @author  <a href="mailto:jason@planet57.com">Jason Dillon</a>
 * @author <a href="mailto:adrian@jboss.com">Adrian Brock</a>
 * @version <tt>$Revision$</tt>
 */
public class JmsSessionFactoryImpl
   implements JmsSessionFactory, Referenceable
{
   private static final Logger log = Logger.getLogger(JmsSessionFactoryImpl.class);

   /** Are we closed? */
   private boolean closed = false;

   /** Whether trace is enabled */
   private boolean trace = log.isTraceEnabled();
   
   private Reference reference;

   // Used from JmsConnectionFactory
   private String userName;
   private String password;
   private String clientID;
   private int type;

   /* Whether we are started */
   private boolean started = false;
   
   /** JmsRa own factory */
   private JmsManagedConnectionFactory mcf;

   /** Hook to the appserver */
   private ConnectionManager cm;

   /** The sessions */
   private HashSet sessions = new HashSet();

   /** The temporary queues */
   private HashSet tempQueues = new HashSet();

   /** The temporary topics */
   private HashSet tempTopics = new HashSet();
   
   public JmsSessionFactoryImpl(final ManagedConnectionFactory mcf,
                                final ConnectionManager cm,
                                final int type)
   {
      this.mcf = (JmsManagedConnectionFactory) mcf;
      this.cm = cm;
      
      if (cm == null)
         // This is standalone usage, no appserver
         this.cm = new JmsConnectionManager();
      else 
         this.cm = cm;

      this.type = type;

      if (trace)
         log.trace("mcf=" + mcf + ", cm=" + cm + ", type=" + type);
   }

   public void setReference(final Reference reference)
   {
      this.reference = reference;
   }
    
   public Reference getReference()
   {
      return reference;
   }
    
   // --- API for JmsConnectionFactoryImpl
   
   public void setUserName(final String name) 
   {
      userName = name;
   }
    
   public void setPassword(final String password) 
   {
      this.password = password;
   }

   //---- QueueConnection ---
   
   public QueueSession createQueueSession(final boolean transacted, 
                                          final int acknowledgeMode) 
      throws JMSException
   {
      checkClosed();
      if (type == JmsConnectionFactory.TOPIC)
         throw new IllegalStateException("Can not get a queue session from a topic connection");
      return allocateConnection(transacted, acknowledgeMode, type);
   }
    
   public ConnectionConsumer createConnectionConsumer
      (Queue queue,
       String messageSelector,
       ServerSessionPool sessionPool,
       int maxMessages) 
      throws JMSException 
   {
      throw new IllegalStateException(ISE);
   }
    
   //--- TopicConnection ---
    
   public TopicSession createTopicSession(final boolean transacted, 
                                          final int acknowledgeMode) 
      throws JMSException
   { 
      checkClosed();
      if (type == JmsConnectionFactory.QUEUE)
         throw new IllegalStateException("Can not get a topic session from a queue connection");
      return allocateConnection(transacted, acknowledgeMode, type);
   }

   public ConnectionConsumer createConnectionConsumer
      (Topic topic,
       String messageSelector,
       ServerSessionPool sessionPool,
       int maxMessages) 
      throws JMSException 
   {
      throw new IllegalStateException(ISE);
   }		       

   public ConnectionConsumer createDurableConnectionConsumer(
      Topic topic, 
      String subscriptionName,
      String messageSelector,
      ServerSessionPool sessionPool, 
      int maxMessages) 
      throws JMSException
   {
      throw new IllegalStateException(ISE);
   }
   
   //--- All the Connection methods
   
   public String getClientID() throws JMSException
   {
      checkClosed();
      return clientID;
   }
    
   public void setClientID(String cID) throws JMSException
   {
      if (mcf.isStrict())
         throw new IllegalStateException(ISE);
      
      checkClosed();
      if (clientID != null)
         throw new IllegalStateException("Cannot change client id");
      clientID = cID;
   }
    
   public ConnectionMetaData getMetaData() throws JMSException
   {
      checkClosed();
      return mcf.getMetaData();
   }
    
   public ExceptionListener getExceptionListener() throws JMSException
   {
      throw new IllegalStateException(ISE);
   }
    
   public void setExceptionListener(ExceptionListener listener)
      throws JMSException
   {
      throw new IllegalStateException(ISE);
   }
    
   public void start() throws JMSException
   {
      checkClosed();
      if (trace)
         log.trace("start() " + this);
      synchronized (sessions)
      {
         if (started)
            return;
         started = true;
         for (Iterator i = sessions.iterator(); i.hasNext();)
         {
            JmsSession session = (JmsSession) i.next();
            session.start();
         }
      }
   }
    
   public void stop() throws JMSException
   {
      if (mcf.isStrict())
         throw new IllegalStateException(ISE);
      checkClosed();
      if (trace)
         log.trace("stop() " + this);
      synchronized (sessions)
      {
         if (started == false)
            return;
         started = true;
         for (Iterator i = sessions.iterator(); i.hasNext();)
         {
            JmsSession session = (JmsSession) i.next();
            session.stop();
         }
      }
   }

   public void close() throws JMSException
   {
      if (closed)
         return;
      closed = true;

      if (trace)
         log.trace("close() " + this);
      
      synchronized (sessions)
      {
         for (Iterator i = sessions.iterator(); i.hasNext();)
         {
            JmsSession session = (JmsSession) i.next();
            try
            {
               session.closeSession();
            }
            catch (Throwable t)
            {
               log.trace("Error closing session", t);
            }
            i.remove();
         }
      }
      
      synchronized (tempQueues)
      {
         for (Iterator i = tempQueues.iterator(); i.hasNext();)
         {
            TemporaryQueue temp = (TemporaryQueue) i.next();
            try
            {
               if (trace)
                  log.trace("Closing temporary queue " + temp + " for " + this);
               temp.delete();
            }
            catch (Throwable t)
            {
               log.trace("Error deleting temporary queue", t);
            }
            i.remove();
         }
      }
      
      synchronized (tempTopics)
      {
         for (Iterator i = tempTopics.iterator(); i.hasNext();)
         {
            TemporaryTopic temp = (TemporaryTopic) i.next();
            try
            {
               if (trace)
                  log.trace("Closing temporary topic " + temp + " for " + this);
               temp.delete();
            }
            catch (Throwable t)
            {
               log.trace("Error deleting temporary queue", t);
            }
            i.remove();
         }
      }
   }

   public void closeSession(JmsSession session) throws JMSException
   {
      synchronized (sessions)
      {
         sessions.remove(session);
      }
   }
   
   public void addTemporaryQueue(TemporaryQueue temp)
   {
      synchronized(tempQueues)
      {
         tempQueues.add(temp);
      }
   }
   
   public void addTemporaryTopic(TemporaryTopic temp)
   {
      synchronized(tempTopics)
      {
         tempTopics.add(temp);
      }
   }
   
   // -- JMS 1.1

   public ConnectionConsumer createConnectionConsumer(Destination destination, ServerSessionPool pool, int maxMessages) throws JMSException
   {
      throw new IllegalStateException(ISE);
   }

   public ConnectionConsumer createConnectionConsumer(Destination destination, String name, ServerSessionPool pool, int maxMessages) throws JMSException
   {
      throw new IllegalStateException(ISE);
   }

   public Session createSession(boolean transacted, int acknowledgeMode)
      throws JMSException
   {
      checkClosed();
      return allocateConnection(transacted, acknowledgeMode, type);
   }

   protected JmsSession allocateConnection(boolean transacted, int acknowledgeMode, int sessionType) throws JMSException
   {
      try
      {
         synchronized (sessions)
         {
            if (mcf.isStrict() && sessions.isEmpty() == false)
               throw new IllegalStateException("Only allowed one session per connection. See the J2EE spec, e.g. J2EE1.4 Section 6.6");
            if (transacted)
               acknowledgeMode = Session.SESSION_TRANSACTED;
            JmsConnectionRequestInfo info = new JmsConnectionRequestInfo(transacted, acknowledgeMode, sessionType);
            info.setUserName(userName);
            info.setPassword(password);
            info.setClientID(clientID);

            if (trace)
               log.trace("Allocating session for " + this + " with request info=" + info);
            JmsSession session = (JmsSession) cm.allocateConnection(mcf, info);
            if (trace)
               log.trace("Allocated  " + this + " session=" + session);
            session.setJmsSessionFactory(this);
            if (started)
               session.start();
            sessions.add(session);
            return session;
         }
      }
      catch (ResourceException e)
      {
         log.error("could not create session", e);
         
         JMSException je = new JMSException
            ("Could not create a session: " + e);
         je.setLinkedException(e);
         throw je;
      }               
   }  

   protected void checkClosed() throws IllegalStateException
   {
      if (closed)
         throw new IllegalStateException("The connection is closed");
   }
}
