/**
 * JOnAS: Java(TM) Open Application Server
 * Copyright (C) 1999-2004 Bull S.A.
 * Contact: jonas-team@ow2.org
 *
 * This library 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 any later version.
 *
 * This library 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 library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
 * USA
 *
 * --------------------------------------------------------------------------
 * $Id: JMdbFactory.java 15428 2008-10-07 11:20:29Z sauthieg $
 * --------------------------------------------------------------------------
 */

package org.ow2.jonas.lib.ejb21;

import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.ListIterator;

import javax.ejb.EJBException;
import javax.ejb.MessageDrivenBean;
import javax.ejb.MessageDrivenContext;
import javax.ejb.Timer;
import javax.ejb.TimerService;
import javax.jms.ConnectionConsumer;
import javax.jms.JMSException;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.ServerSession;
import javax.jms.ServerSessionPool;
import javax.jms.Session;
import javax.jms.Topic;
import javax.jms.XAQueueConnection;
import javax.jms.XAQueueConnectionFactory;
import javax.jms.XATopicConnection;
import javax.jms.XATopicConnectionFactory;
import javax.naming.Context;
import javax.transaction.Transaction;

import org.ow2.jonas.deployment.ejb.MessageDrivenDesc;
import org.ow2.jonas.deployment.ejb.MethodDesc;
import org.ow2.jonas.jms.JmsManager;



import org.objectweb.util.monolog.api.BasicLevel;

/**
 * This class is a factory for a Message Driven Bean
 * There is one such class per MDB class.
 * Contains all information related to the bean and set up all JMS environment for the bean
 * It manages a ServerSession pool to server MDB requests.
 * @author Philippe Coq, Philippe Durieux
 */
public class JMdbFactory extends JFactory implements ServerSessionPool {

    /**
     * JMS Manager
     */
    private JmsManager jms = null;

    /**
     * Connection Consumer for this message driven bean
     */
    private ConnectionConsumer cc = null;

    /**
     * pool of ServerSession objects
     */
    private List  sspool = new ArrayList();

    /**
     * number of instances in the pool
     */
    private int instanceCount = 0;

    /**
     * initial value for pool size
     */
    private int minPoolSize = 0;

    /**
     * nb max of instances in pool
     */
    private int maxCacheSize = 0;

    /**
     * JMS Topic Connection
     * always use XA Connections for transactions.
     */
    private XATopicConnection tconn = null;

    /**
     * JMS Queue Connection (Topic or Queue)
     * always use XA Connections for transactions.
     */
    private XAQueueConnection qconn = null;

    /**
     * Constructor
     * @param dd   Message Driven Descriptor
     * @param cont Container where this bean is defined
     */
    public JMdbFactory(MessageDrivenDesc dd, JContainer cont) {
        super(dd, cont);

        // Check if tx managed by the bean or the container
        txbeanmanaged = dd.isBeanManagedTransaction();

        // Check that JMS Service has been run in the server
        jms = cont.getJmsManager();
        if (jms == null) {
            TraceEjb.logger.log(BasicLevel.ERROR, "cannot deploy a message driven bean without the JMS Service");
            throw new EJBException("JMS Service must be run");
        }

        // Create a Connection Consumer, depending on Deployment Descriptor
        String selector = dd.getSelector();
        String dest = dd.getDestinationJndiName();

        // Get the number max of messages sent on a Session at one time.
        // There is a bug in Joram: If the number is greater than 1, Joram
        // will wait until this nb is reached!
        // We set the number max to 1 to work around this bug.
        // LATER: maxMessage = dd.getMaxMessages(), with this value configurable.
        int maxMessages = 1;

        if (dest == null) {
            throw new EJBException("The destination JNDI name is null in bean " + dd.getEjbName());
        }

        try {
            if (dd.isTopicDestination()) {
                // topic
                XATopicConnectionFactory tcf = jms.getXATopicConnectionFactory();
                tconn = tcf.createXATopicConnection();
                Topic t = jms.getTopic(dest);
                if (dd.isSubscriptionDurable()) {
                    if (TraceEjb.isDebugJms()) {
                        TraceEjb.mdb.log(BasicLevel.DEBUG, "createDurableConnectionConsumer for " + ejbname);
                    }
                    cc  = tconn.createDurableConnectionConsumer(t, ejbname, selector, this, maxMessages);
                } else {
                    if (TraceEjb.isDebugJms()) {
                        TraceEjb.mdb.log(BasicLevel.DEBUG, "createConnectionConsumer for " + dest);
                    }
                    cc  = tconn.createConnectionConsumer(t, selector, this, maxMessages);
                }
                tconn.start();
            } else {
                // queue
                XAQueueConnectionFactory qcf = jms.getXAQueueConnectionFactory();
                qconn = qcf.createXAQueueConnection();
                Queue q = jms.getQueue(dest);
                if (TraceEjb.isDebugJms()) {
                    TraceEjb.mdb.log(BasicLevel.DEBUG, "createConnectionConsumer for " + dest);
                }
                cc = qconn.createConnectionConsumer(q, selector, this, maxMessages);
                qconn.start();
            }
        } catch (Exception e) {
            throw new EJBException("Cannot create connection consumer in bean " + dd.getEjbName() + " :", e);
        }

        minPoolSize = dd.getPoolMin();
        maxCacheSize = dd.getCacheMax();
        if (TraceEjb.isDebugSwapper()) {
            TraceEjb.swapper.log(BasicLevel.DEBUG, " maxCacheSize = " + maxCacheSize
                                 + " minPoolSize = " + minPoolSize);
        }
    }

    // ---------------------------------------------------------------
    // Specific BeanFactory implementation
    // ---------------------------------------------------------------

    /**
     * Init pool of instances
     */
    public void initInstancePool() {
        if (minPoolSize != 0) {
            TraceEjb.mdb.log(BasicLevel.INFO, "pre-allocate a set of " + minPoolSize
                    + " message driven bean  instances");
            // pre-allocate a set of ServerSession
            synchronized (sspool) {
                for (int i = 0; i < minPoolSize; i++) {
                    ServerSession ss = null;
                    try {
                        ss = createNewInstance();
                        sspool.add(ss);
                    } catch (Exception e) {
                        TraceEjb.mdb.log(BasicLevel.ERROR, "cannot init pool of instances ");
                        throw new EJBException("cannot init pool of instances ", e);
                    }
                }
            }
        }
    }

    /**
     * @return the size of the ServerSessionPool
     */
    public int getPoolSize() {
        return sspool.size();
    }

    /**
     * stop this EJB.
     * call ejbRemove on all MDB
     * close the connection consumer
     * Stop the threads and remove the beans
     */
    public void stop() {
        if (TraceEjb.isDebugJms()) {
            TraceEjb.mdb.log(BasicLevel.DEBUG, "");
        }
        try {
            cc.close();
            if (tconn != null) {
                tconn.close();
            }
            if (qconn != null) {
                qconn.close();
            }
        } catch (JMSException e) {
            TraceEjb.logger.log(BasicLevel.WARN, "unregister: Cannot close Connection Consumer");
        }
        stopContainer();
    }

    /**
     * synchronize bean instances if needed
     */
    public void syncDirty(boolean notused) {
    }


    /**
     * @return the home if exist
     */
    public JHome getHome() {
        return null;
    }

    /**
     * @return the local home if exist
     */
    public JLocalHome getLocalHome() {
        return null;
    }

    // ---------------------------------------------------------------
    // ServerSessionPool Implementation
    // ---------------------------------------------------------------

    /**
     * Returns a server session from the pool. If pool is empty, creates a new one.
     *
     * @return Returns a server session from the pool.
     * @exception    JMSException - if an application server fails to return a Server Session
     *   out of its server session pool.
     */
    public ServerSession getServerSession() throws JMSException {
        if (TraceEjb.isDebugJms()) {
            TraceEjb.mdb.log(BasicLevel.DEBUG, "");
        }

        return getNewInstance(true);
    }



    // ---------------------------------------------------------------
    // Other methods
    // ---------------------------------------------------------------

    /**
     * put the ServerSession back to the pool
     * @param ss The ServerSession
     */
    public void releaseServerSession(ServerSession ss) {
        if (TraceEjb.isDebugJms()) {
            TraceEjb.mdb.log(BasicLevel.DEBUG, "");
        }

        synchronized (sspool) {
            sspool.add(ss);
            if (TraceEjb.isDebugSwapper()) {
                    TraceEjb.swapper.log(BasicLevel.DEBUG, "notifyAll ");
            }
            sspool.notifyAll();
        }
        if (TraceEjb.isDebugJms()) {
            TraceEjb.mdb.log(BasicLevel.DEBUG, "nb instances " + getCacheSize());
        }

    }

    // ---------------------------------------------------------------
    // other public methods
    // ---------------------------------------------------------------

    /**
     * Obtains the TimerService associated for this Bean
     * @return a JTimerService instance.
     */
    public TimerService getTimerService() {
        if (myTimerService == null) {
            // TODO : Check that instance implements TimedObject ?
            myTimerService = new JTimerService(this);
        }
        return myTimerService;
    }

    /**
     * @return min pool size
     * for Jmx
     */
    public int getMinPoolSize() {
        return minPoolSize;
    }

    /**
     * @return max cache size
     * for Jmx
     */
    public int getMaxCacheSize() {
        return maxCacheSize;
    }

    /**
     * @return current cache size ( = nb of instance created)
     * for Jmx
     */
    public int getCacheSize() {
        return instanceCount;
    }

    /**
     * @return the Transaction Attribute
     */
    public int getTransactionAttribute() {
        return ((MessageDrivenDesc) dd).getTxAttribute();
    }

    /**
     * For Message Driven Beans, only 2 cases are possible.
     * @param rctx The Request Context
     */
    public void checkTransaction(RequestCtx rctx) {
        if (rctx.txAttr == MethodDesc.TX_REQUIRED) {
            try {
                if (tm.getTransaction() != null) {
                    // This should not occur (DEBUG)
                    TraceEjb.logger.log(BasicLevel.ERROR, "Transaction already opened by this thread.");
                    TraceEjb.logger.log(BasicLevel.ERROR, "Transaction status = " + tm.getStatus());
                    TraceEjb.logger.log(BasicLevel.ERROR, "Transaction = " + tm.getTransaction());
                    Thread.dumpStack();
                    return;
                }
                tm.begin();
                rctx.mustCommit = true;
                rctx.currTx = tm.getTransaction();
                if (TraceEjb.isDebugTx()) {
                    TraceEjb.tx.log(BasicLevel.DEBUG, "Transaction started: " + rctx.currTx);
                }
            } catch (Exception e) {
                // No exception raised in case of MDB
                TraceEjb.logger.log(BasicLevel.ERROR, "cannot start tx", e);
                return;
            }
        }
    }

    /**
     * Reduce number of instances in memory in the free list
     * we reduce to the minPoolSize
     */
    public void reduceCache() {
        // reduce the pool to the minPoolSize
        int poolsz = minPoolSize;
        synchronized (sspool) {
            if (TraceEjb.isDebugSwapper()) {
                TraceEjb.swapper.log(BasicLevel.DEBUG, "try to reduce " + sspool.size()
                                     + " to " + poolsz);
            }
            while (sspool.size() > poolsz) {
                ListIterator i = sspool.listIterator();
                if (i.hasNext()) {
                    i.next();
                    i.remove();
                    instanceCount--;
                }
            }
        }
        if (TraceEjb.isDebugSwapper()) {
            TraceEjb.swapper.log(BasicLevel.DEBUG, "cacheSize= " + getCacheSize());
        }

    }

    /**
     * Notify a timeout for this bean
     * @param timer timer whose expiration caused this notification.
     */
    public void notifyTimeout(Timer timer) {
        if (stopped) {
            TraceEjb.mdb.log(BasicLevel.WARN, "Container stopped");
            return;
        }
        if (TraceEjb.isDebugJms()) {
            TraceEjb.mdb.log(BasicLevel.DEBUG, "");
        }

        // We need an instance from the pool to process the timeout.
        JMessageDrivenBean jmdb = null;
        jmdb = getNewInstance(false);

        // deliver the timeout to the bean
        jmdb.deliverTimeout(timer);

        // release the instance
        releaseServerSession(jmdb);
    }

    // ---------------------------------------------------------------
    // private methods
    // ---------------------------------------------------------------

    /**
     * Returns a new instance of the bean. Try to get one from the pool,
     * and create a new one if the pool is empty.
     * @param canwait true if can wait if max-cache-size reached.
     * @return Returns a JMessageDrivenBean instance from the pool.
     */
    private JMessageDrivenBean getNewInstance(boolean canwait) {
        if (TraceEjb.isDebugJms()) {
            TraceEjb.mdb.log(BasicLevel.DEBUG, "");
        }

        // try to get one from the Pool
        JMessageDrivenBean ss = null;

        // try to find a free context in the pool
        synchronized (sspool) {
            if (!sspool.isEmpty()) {
                try {
                    ss = (JMessageDrivenBean) sspool.remove(0);
                    return ss;
                } catch (IndexOutOfBoundsException ex) {
                    // This should never happen
                    TraceEjb.logger.log(BasicLevel.ERROR, "exception:" + ex);
                    throw new EJBException("Cannot get an instance from the pool", ex);
                }
            } else {
                if (TraceEjb.isDebugJms()) {
                    TraceEjb.mdb.log(BasicLevel.DEBUG, "pool is empty");
                }
                if (maxCacheSize == 0 ||  instanceCount < maxCacheSize || !canwait) {
                    // Pool is empty creates the ServerSession object
                    try {
                        ss = createNewInstance();
                    } catch (Exception e) {
                        TraceEjb.logger.log(BasicLevel.ERROR, "exception:" + e);
                        throw new EJBException("Cannot create a new instance", e);
                    }
                } else {
                    while (sspool.isEmpty()) {
                        if (TraceEjb.isDebugSwapper()) {
                            TraceEjb.swapper.log(BasicLevel.DEBUG, "sspool.isEmpty() = true --> wait()");
                        }
                        try {
                            sspool.wait();
                            if (TraceEjb.isDebugSwapper()) {
                                TraceEjb.swapper.log(BasicLevel.DEBUG, "sspool notified");
                            }
                        } catch (InterruptedException e) {
                            if (TraceEjb.isDebugSwapper()) {
                                TraceEjb.swapper.log(BasicLevel.DEBUG, "sspool waiting interrupted", e);
                            }
                        } catch (Exception e) {
                            throw new EJBException("synchronization pb", e);
                        }
                    }
                    try {
                        ListIterator i = sspool.listIterator();
                        if (i.hasNext()) {
                            ss = (JMessageDrivenBean) i.next();
                            i.remove();
                        }
                        return ss;
                    } catch (IndexOutOfBoundsException ex) {
                        // pool is empty
                    }
                }

            }
            if (TraceEjb.isDebugSwapper()) {
                TraceEjb.swapper.log(BasicLevel.DEBUG, "nb instances " + getCacheSize());
            }
            return ss;
        }
    }

    /**
     * Create a new instance of the bean
     * @throws Exception if MDB instanciation is not possible
     * @return Returns a new JMessageDrivenBean instance.
     */
    private JMessageDrivenBean createNewInstance() throws Exception {
        if (TraceEjb.isDebugJms()) {
            TraceEjb.mdb.log(BasicLevel.DEBUG, "");
        }
        Session sess = null;
        JMessageDrivenBean ss = null;
        MessageDrivenDesc mdd = (MessageDrivenDesc) dd;
        if (tconn != null) {
            if (mdd.isRequired()) {
                sess = tconn.createXATopicSession();
            } else {
                sess = tconn.createTopicSession(false, mdd.getAcknowledgeMode());
            }
        } else if (qconn != null) {
            if (mdd.isRequired()) {
                sess = qconn.createXAQueueSession();
            } else {
                sess = qconn.createQueueSession(false, mdd.getAcknowledgeMode());
            }
        } else {
            TraceEjb.mdb.log(BasicLevel.ERROR, "connection not initialized");
            throw new Exception("JMS connection not initialized");
        }

        // Set ContextClassLoader with the ejbclassloader.
        // This is necessary in case ejbCreate calls another bean in the same jar.
        ClassLoader old = Thread.currentThread().getContextClassLoader();
        Thread.currentThread().setContextClassLoader(myClassLoader());

        // Creates the new instance
        MessageDrivenBean mdb = null;
        try {
            mdb  = (MessageDrivenBean) beanclass.newInstance();
        } catch (Exception e) {
            TraceEjb.logger.log(BasicLevel.ERROR, "failed to create instance:", e);
            resetToOldClassLoader(old);
            throw new EJBException("Container failed to create instance of Message Driven Bean", e);
        }

        // Instanciates a new JMessageDrivenBean object
        // and set it as the MessageListener for this Session.
        ss = new JMessageDrivenBean(this, sess, mdb, wm);
        try {
            sess.setMessageListener((MessageListener) ss);
        } catch (JMSException je) {
            resetToOldClassLoader(old);
            throw je;
        }

        // starts the bean instance: setMessageDrivenContext() + ejbCreate()
        // see EJB spec. 2.0 page 322.
        // Both operations must be called with the correct ComponentContext
        Context ctxsave = setComponentContext();
        mdb.setMessageDrivenContext((MessageDrivenContext) ss);
        try {
            Method m = beanclass.getMethod("ejbCreate", (Class[]) null);
            boolean bm = m.isAccessible();
            if( !bm) {
                m.setAccessible(true);
            }
            m.invoke(mdb, (Object[]) null);
            m.setAccessible(bm);
        } catch (Exception e) {
            TraceEjb.logger.log(BasicLevel.ERROR, "cannot call ejbCreate on message driven bean instance ", e);
            throw new EJBException(" Container fails to call ejbCreate on message driven bean instance", e);
        } finally {
            resetToOldClassLoader(old);
            resetComponentContext(ctxsave);
        }

        synchronized (sspool) {
            instanceCount++;
        }
        return ss;
    }

    /**
     * Reset currentThread context ClassLoader to a given ClassLoader
     * @param old Old ClassLoader to reuse
     */
    private void resetToOldClassLoader(ClassLoader old) {
        if (old != null) {
            Thread.currentThread().setContextClassLoader(old);
        }
    }

    /*
     * Make sense only for entities
     */
    public void storeInstances(Transaction tx) {
        // unused
    }
}
