
package org.mule.modules.amazon.adapters;

import org.apache.commons.pool.KeyedPoolableObjectFactory;
import org.apache.commons.pool.impl.GenericKeyedObjectPool;
import org.mule.api.Capabilities;
import org.mule.api.Capability;
import org.mule.api.ConnectionManager;
import org.mule.api.MuleContext;
import org.mule.api.construct.FlowConstruct;
import org.mule.api.context.MuleContextAware;
import org.mule.api.lifecycle.Disposable;
import org.mule.api.lifecycle.Initialisable;
import org.mule.api.lifecycle.Startable;
import org.mule.api.lifecycle.Stoppable;
import org.mule.config.PoolingProfile;
import org.mule.modules.amazon.SQSConnector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * A {@code SQSConnectorConnectionManager} is a wrapper around {@link SQSConnector } that adds connection management capabilities to the pojo.
 * 
 */
public class SQSConnectorConnectionManager
    implements Capabilities, ConnectionManager<SQSConnectorConnectionManager.ConnectionKey, SQSConnectorLifecycleAdapter> , MuleContextAware, Initialisable
{

    /**
     * 
     */
    private String queueName;
    private String accessKey;
    private String secretAccessKey;
    private static Logger logger = LoggerFactory.getLogger(SQSConnectorConnectionManager.class);
    /**
     * Mule Context
     * 
     */
    private MuleContext muleContext;
    /**
     * Flow construct
     * 
     */
    private FlowConstruct flowConstruct;
    /**
     * Connector Pool
     * 
     */
    private GenericKeyedObjectPool connectionPool;
    protected PoolingProfile connectionPoolingProfile;

    /**
     * Sets accessKey
     * 
     * @param value Value to set
     */
    public void setAccessKey(String value) {
        this.accessKey = value;
    }

    /**
     * Retrieves accessKey
     * 
     */
    public String getAccessKey() {
        return this.accessKey;
    }

    /**
     * Sets secretAccessKey
     * 
     * @param value Value to set
     */
    public void setSecretAccessKey(String value) {
        this.secretAccessKey = value;
    }

    /**
     * Retrieves secretAccessKey
     * 
     */
    public String getSecretAccessKey() {
        return this.secretAccessKey;
    }

    /**
     * Sets connectionPoolingProfile
     * 
     * @param value Value to set
     */
    public void setConnectionPoolingProfile(PoolingProfile value) {
        this.connectionPoolingProfile = value;
    }

    /**
     * Retrieves connectionPoolingProfile
     * 
     */
    public PoolingProfile getConnectionPoolingProfile() {
        return this.connectionPoolingProfile;
    }

    /**
     * Sets queueName
     * 
     * @param value Value to set
     */
    public void setQueueName(String value) {
        this.queueName = value;
    }

    /**
     * Retrieves queueName
     * 
     */
    public String getQueueName() {
        return this.queueName;
    }

    /**
     * Sets flow construct
     * 
     * @param flowConstruct Flow construct to set
     */
    public void setFlowConstruct(FlowConstruct flowConstruct) {
        this.flowConstruct = flowConstruct;
    }

    /**
     * Set the Mule context
     * 
     * @param context Mule context to set
     */
    public void setMuleContext(MuleContext context) {
        this.muleContext = context;
    }

    public void initialise() {
        GenericKeyedObjectPool.Config config = new GenericKeyedObjectPool.Config();
        if (connectionPoolingProfile!= null) {
            config.maxIdle = connectionPoolingProfile.getMaxIdle();
            config.maxActive = connectionPoolingProfile.getMaxActive();
            config.maxWait = connectionPoolingProfile.getMaxWait();
            config.whenExhaustedAction = ((byte) connectionPoolingProfile.getExhaustedAction());
        }
        connectionPool = new GenericKeyedObjectPool(new SQSConnectorConnectionManager.ConnectionFactory(this), config);
    }

    public SQSConnectorLifecycleAdapter acquireConnection(SQSConnectorConnectionManager.ConnectionKey key)
        throws Exception
    {
        return ((SQSConnectorLifecycleAdapter) connectionPool.borrowObject(key));
    }

    public void releaseConnection(SQSConnectorConnectionManager.ConnectionKey key, SQSConnectorLifecycleAdapter connection)
        throws Exception
    {
        connectionPool.returnObject(key, connection);
    }

    public void destroyConnection(SQSConnectorConnectionManager.ConnectionKey key, SQSConnectorLifecycleAdapter connection)
        throws Exception
    {
        connectionPool.invalidateObject(key, connection);
    }

    /**
     * Returns true if this module implements such capability
     * 
     */
    public boolean isCapableOf(Capability capability) {
        if (capability == Capability.LIFECYCLE_CAPABLE) {
            return true;
        }
        if (capability == Capability.CONNECTION_MANAGEMENT_CAPABLE) {
            return true;
        }
        return false;
    }

    private static class ConnectionFactory
        implements KeyedPoolableObjectFactory
    {

        private SQSConnectorConnectionManager connectionManager;

        public ConnectionFactory(SQSConnectorConnectionManager connectionManager) {
            this.connectionManager = connectionManager;
        }

        public Object makeObject(Object key)
            throws Exception
        {
            if (!(key instanceof SQSConnectorConnectionManager.ConnectionKey)) {
                throw new RuntimeException("Invalid key type");
            }
            SQSConnectorLifecycleAdapter connector = new SQSConnectorLifecycleAdapter();
            connector.setAccessKey(connectionManager.getAccessKey());
            connector.setSecretAccessKey(connectionManager.getSecretAccessKey());
            if (connector instanceof Initialisable) {
                connector.initialise();
            }
            if (connector instanceof Startable) {
                connector.start();
            }
            return connector;
        }

        public void destroyObject(Object key, Object obj)
            throws Exception
        {
            if (!(key instanceof SQSConnectorConnectionManager.ConnectionKey)) {
                throw new RuntimeException("Invalid key type");
            }
            if (!(obj instanceof SQSConnectorLifecycleAdapter)) {
                throw new RuntimeException("Invalid connector type");
            }
            try {
                ((SQSConnectorLifecycleAdapter) obj).disconnect();
            } catch (Exception e) {
                throw e;
            } finally {
                if (((SQSConnectorLifecycleAdapter) obj) instanceof Stoppable) {
                    ((SQSConnectorLifecycleAdapter) obj).stop();
                }
                if (((SQSConnectorLifecycleAdapter) obj) instanceof Disposable) {
                    ((SQSConnectorLifecycleAdapter) obj).dispose();
                }
            }
        }

        public boolean validateObject(Object key, Object obj) {
            if (!(obj instanceof SQSConnectorLifecycleAdapter)) {
                throw new RuntimeException("Invalid connector type");
            }
            try {
                return ((SQSConnectorLifecycleAdapter) obj).isConnected();
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
                return false;
            }
        }

        public void activateObject(Object key, Object obj)
            throws Exception
        {
            if (!(key instanceof SQSConnectorConnectionManager.ConnectionKey)) {
                throw new RuntimeException("Invalid key type");
            }
            if (!(obj instanceof SQSConnectorLifecycleAdapter)) {
                throw new RuntimeException("Invalid connector type");
            }
            try {
                if (!((SQSConnectorLifecycleAdapter) obj).isConnected()) {
                    ((SQSConnectorLifecycleAdapter) obj).connect(((SQSConnectorConnectionManager.ConnectionKey) key).getQueueName());
                }
            } catch (Exception e) {
                throw e;
            }
        }

        public void passivateObject(Object key, Object obj)
            throws Exception
        {
        }

    }


    /**
     * A tuple of connection parameters
     * 
     */
    public static class ConnectionKey {

        /**
         * 
         */
        private String queueName;

        public ConnectionKey(String queueName) {
            this.queueName = queueName;
        }

        /**
         * Sets queueName
         * 
         * @param value Value to set
         */
        public void setQueueName(String value) {
            this.queueName = value;
        }

        /**
         * Retrieves queueName
         * 
         */
        public String getQueueName() {
            return this.queueName;
        }

        public int hashCode() {
            int hash = 1;
            hash = ((hash* 31)+ this.queueName.hashCode());
            return hash;
        }

        public boolean equals(Object obj) {
            return ((obj instanceof SQSConnectorConnectionManager.ConnectionKey)&&(this.queueName == ((SQSConnectorConnectionManager.ConnectionKey) obj).queueName));
        }

    }

}
