/*
 * Copyright 2001, 2002,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.axis.transport.jms;

import org.apache.axis.AxisFault;
import org.apache.axis.components.jms.JMSVendorAdapter;
import org.apache.axis.components.logger.LogFactory;
import org.apache.axis.utils.Messages;
import org.apache.commons.logging.Log;

import java.util.HashMap;
import java.util.Iterator;

/**
 * JMSConnectorManager manages a pool of connectors and works with the
 * vendor adapters to support the reuse of JMS connections.
 *
 * @author Ray Chun (rchun@sonicsoftware.com)
 */
public class JMSConnectorManager
{
    protected static Log log =
            LogFactory.getLog(JMSConnectorManager.class.getName());

    private static JMSConnectorManager s_instance = new JMSConnectorManager();

    private static HashMap vendorConnectorPools = new HashMap();
    private int DEFAULT_WAIT_FOR_SHUTDOWN = 90000; // 1.5 minutes

    private JMSConnectorManager()
    {
    }

    public static JMSConnectorManager getInstance()
    {
        return s_instance;
    }

    /**
     * Returns the pool of JMSConnectors for a particular vendor
     */
    public ShareableObjectPool getVendorPool(String vendorId)
    {
        return (ShareableObjectPool)vendorConnectorPools.get(vendorId);
    }

    /**
     * Retrieves a JMSConnector that satisfies the provided connector criteria
     */
    public JMSConnector getConnector(HashMap connectorProperties,
                                     HashMap connectionFactoryProperties,
                                     String username,
                                     String password,
                                     JMSVendorAdapter vendorAdapter)
        throws AxisFault
    {
        JMSConnector connector = null;

        try
        {
            // check for a vendor-specific pool, and create if necessary
            ShareableObjectPool vendorConnectors = getVendorPool(vendorAdapter.getVendorId());
            if (vendorConnectors == null)
            {
                synchronized (vendorConnectorPools)
                {
                    vendorConnectors = getVendorPool(vendorAdapter.getVendorId());
                    if (vendorConnectors == null)
                    {
                        vendorConnectors = new ShareableObjectPool();
                        vendorConnectorPools.put(vendorAdapter.getVendorId(), vendorConnectors);
                    }
                }
            }

            // look for a matching JMSConnector among existing connectors
            synchronized (vendorConnectors)
            {
                try
                {

                    connector = JMSConnectorFactory.matchConnector(vendorConnectors.getElements(),
                                                                   connectorProperties,
                                                                   connectionFactoryProperties,
                                                                   username,
                                                                   password,
                                                                   vendorAdapter);
                }
                catch (Exception e) {} // ignore. a new connector will be created if no match is found

                if (connector == null)
                {
                        connector = JMSConnectorFactory.createClientConnector(connectorProperties,
                                                                              connectionFactoryProperties,
                                                                              username,
                                                                              password,
                                                                              vendorAdapter);
                        connector.start();
                }
            }
        }
        catch (Exception e)
        {
            log.error(Messages.getMessage("cannotConnectError"), e);

            if(e instanceof AxisFault)
                throw (AxisFault)e;
            throw new AxisFault("cannotConnect", e);
        }

        return connector;
    }

    /**
     * Closes JMSConnectors in all pools
     */
    void closeAllConnectors()
    {
        if (log.isDebugEnabled()) {
            log.debug("Enter: JMSConnectorManager::closeAllConnectors");
        }

        synchronized (vendorConnectorPools)
        {
            Iterator iter = vendorConnectorPools.values().iterator();
            while (iter.hasNext())
            {
                // close all connectors in the vendor pool
                ShareableObjectPool pool = (ShareableObjectPool)iter.next();
                synchronized (pool)
                {
                    java.util.Iterator connectors = pool.getElements().iterator();
                    while (connectors.hasNext())
                    {
                        JMSConnector conn = (JMSConnector)connectors.next();
                        try
                        {
                            // shutdown automatically decrements the ref count of a connector before closing it
                            // call reserve() to simulate the checkout
                            reserve(conn);
                            closeConnector(conn);
                        }
                        catch (Exception e) {} // ignore. the connector is already being deactivated
                    }
                }
            }
        }

        if (log.isDebugEnabled()) {
            log.debug("Exit: JMSConnectorManager::closeAllConnectors");
        }
    }

    /**
     * Closes JMS connectors that match the specified endpoint address
     */
    void closeMatchingJMSConnectors(HashMap connectorProps, HashMap cfProps,
                                    String username, String password,
                                    JMSVendorAdapter vendorAdapter)
    {
        if (log.isDebugEnabled()) {
            log.debug("Enter: JMSConnectorManager::closeMatchingJMSConnectors");
        }

        try
        {
            String vendorId = vendorAdapter.getVendorId();

            // get the vendor-specific pool of connectors
            ShareableObjectPool vendorConnectors = null;
            synchronized (vendorConnectorPools)
            {
                vendorConnectors = getVendorPool(vendorId);
            }

            // it's possible that there is no pool for that vendor
            if (vendorConnectors == null)
                return;

            synchronized (vendorConnectors)
            {
                // close any matched connectors
                JMSConnector connector = null;
                while ((vendorConnectors.size() > 0) &&
                       (connector = JMSConnectorFactory.matchConnector(vendorConnectors.getElements(),
                                                                       connectorProps,
                                                                       cfProps,
                                                                       username,
                                                                       password,
                                                                       vendorAdapter)) != null)
                {
                    closeConnector(connector);
                }
            }
        }
        catch (Exception e)
        {
            log.warn(Messages.getMessage("failedJMSConnectorShutdown"), e);
        }

        if (log.isDebugEnabled()) {
            log.debug("Exit: JMSConnectorManager::closeMatchingJMSConnectors");
        }
    }

    private void closeConnector(JMSConnector conn)
    {
        conn.stop();
        conn.shutdown();
    }

    /**
     * Adds a JMSConnector to the appropriate vendor pool
     */
    public void addConnectorToPool(JMSConnector conn)
    {
        if (log.isDebugEnabled()) {
            log.debug("Enter: JMSConnectorManager::addConnectorToPool");
        }

        ShareableObjectPool vendorConnectors = null;
        synchronized (vendorConnectorPools)
        {
            String vendorId = conn.getVendorAdapter().getVendorId();
            vendorConnectors = getVendorPool(vendorId);
            // it's possible the pool does not yet exist (if, for example, the connector
            // is created before invoking the call/JMSTransport, as is the case with
            // SimpleJMSListener)
            if (vendorConnectors == null)
            {
                vendorConnectors = new ShareableObjectPool();
                vendorConnectorPools.put(vendorId, vendorConnectors);
            }
        }

        synchronized (vendorConnectors)
        {
            vendorConnectors.addObject(conn);
        }

        if (log.isDebugEnabled()) {
            log.debug("Exit: JMSConnectorManager::addConnectorToPool");
        }
    }

    /**
     * Removes a JMSConnector from the appropriate vendor pool
     */
    public void removeConnectorFromPool(JMSConnector conn)
    {
        if (log.isDebugEnabled()) {
            log.debug("Enter: JMSConnectorManager::removeConnectorFromPool");
        }

        ShareableObjectPool vendorConnectors = null;
        synchronized (vendorConnectorPools)
        {
            vendorConnectors = getVendorPool(conn.getVendorAdapter().getVendorId());
        }
        if (vendorConnectors == null)
            return;

        synchronized (vendorConnectors)
        {
            // first release, to decrement the ref count (it is automatically incremented when
            // the connector is matched)
            vendorConnectors.release(conn);
            vendorConnectors.removeObject(conn);
        }

        if (log.isDebugEnabled()) {
            log.debug("Exit: JMSConnectorManager::removeConnectorFromPool");
        }
    }

    /**
     * Performs a non-exclusive checkout of the JMSConnector
     */
    public void reserve(JMSConnector connector) throws Exception
    {
        ShareableObjectPool pool = null;
        synchronized (vendorConnectorPools)
        {
            pool = getVendorPool(connector.getVendorAdapter().getVendorId());
        }
        if (pool != null)
            pool.reserve(connector);
    }

    /**
     * Performs a non-exclusive checkin of the JMSConnector
     */
    public void release(JMSConnector connector)
    {
        ShareableObjectPool pool = null;
        synchronized (vendorConnectorPools)
        {
            pool = getVendorPool(connector.getVendorAdapter().getVendorId());
        }
        if (pool != null)
            pool.release(connector);
    }

    /**
     * A simple non-blocking pool impl for objects that can be shared.
     * Only a ref count is necessary to prevent collisions at shutdown.
     * Todo: max size, cleanup stale connections
     */
    public class ShareableObjectPool
    {
        // maps object to ref count wrapper
        private java.util.HashMap m_elements;

        // holds objects which should no longer be leased (pending removal)
        private java.util.HashMap m_expiring;

        private int m_numElements = 0;

        public ShareableObjectPool()
        {
            m_elements = new java.util.HashMap();
            m_expiring = new java.util.HashMap();
        }

        /**
         * Adds the object to the pool, if not already added
         */
        public void addObject(Object obj)
        {
            ReferenceCountedObject ref = new ReferenceCountedObject(obj);
            synchronized (m_elements)
            {
                if (!m_elements.containsKey(obj) && !m_expiring.containsKey(obj))
                    m_elements.put(obj, ref);
            }
        }

        /**
         * Removes the object from the pool.  If the object is reserved,
         * waits the specified time before forcibly removing
         * Todo: check expirations with the next request instead of holding up the current request
         */
        public void removeObject(Object obj, long waitTime)
        {
            ReferenceCountedObject ref = null;
            synchronized (m_elements)
            {
                ref = (ReferenceCountedObject)m_elements.get(obj);
                if (ref == null)
                    return;

                m_elements.remove(obj);

                if (ref.count() == 0)
                    return;
                else
                    // mark the object for expiration
                    m_expiring.put(obj, ref);
            }

            // connector is now marked for expiration. wait for the ref count to drop to zero
            long expiration = System.currentTimeMillis() + waitTime;
            while (ref.count() > 0)
            {
                try
                {
                    Thread.sleep(5000);
                }
                catch (InterruptedException e) {} // ignore
                if (System.currentTimeMillis() > expiration)
                    break;
            }

            // also clear from the expiring list
            m_expiring.remove(obj);
        }

        public void removeObject(Object obj)
        {
            removeObject(obj, DEFAULT_WAIT_FOR_SHUTDOWN);
        }

        /**
         * Marks the connector as in use by incrementing the connector's reference count
         */
        public void reserve(Object obj) throws Exception
        {
            synchronized (m_elements)
            {
                if (m_expiring.containsKey(obj))
                    throw new Exception("resourceUnavailable");

                ReferenceCountedObject ref = (ReferenceCountedObject)m_elements.get(obj);
                ref.increment();
            }
        }

        /**
         * Decrements the connector's reference count
         */
        public void release(Object obj)
        {
            synchronized (m_elements)
            {
                ReferenceCountedObject ref = (ReferenceCountedObject)m_elements.get(obj);
                ref.decrement();
            }
        }

        public synchronized java.util.Set getElements()
        {
            return m_elements.keySet();
        }

        public synchronized int size()
        {
            return m_elements.size();
        }

        /**
         * Wrapper to track the use count of an object
         */
        public class ReferenceCountedObject
        {
            private Object m_object;
            private int m_refCount;

            public ReferenceCountedObject(Object obj)
            {
                m_object = obj;
                m_refCount = 0;
            }

            public synchronized void increment()
            {
                m_refCount++;
            }

            public synchronized void decrement()
            {
                if (m_refCount > 0)
                    m_refCount--;
            }

            public synchronized int count()
            {
                return m_refCount;
            }
        }
    }
}