JMdbFactory.java

package org.objectweb.jonas_ejb.container;


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 org.objectweb.jonas_ejb.deployment.api.MessageDrivenDesc;
import org.objectweb.jonas_ejb.deployment.api.MethodDesc;

import org.objectweb.jonas_jms.api.JmsManager;

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

public class JMdbFactory extends JFactory implements ServerSessionPool {

    private JmsManager jms = null;

    ConnectionConsumer cc = null;

    private List  sspool = new ArrayList();

    protected int instanceCount = 0;
    // initial value for pool size
    protected int minPoolSize = 0;
    // nb max of instances in pool
    protected int maxCacheSize = 0;

    protected ThreadPool threadpool = null;

    protected XATopicConnection tconn = null;

    protected XAQueueConnection qconn = null;

    public JMdbFactory(MessageDrivenDesc dd, JContainer cont, ThreadPool thp) {
        super(dd, cont);
        threadpool = thp;

        // 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()) {
                    TraceEjb.mdb.log(BasicLevel.DEBUG, "createDurableConnectionConsumer for "+ejbname);
                    cc  = tconn.createDurableConnectionConsumer(t, ejbname, selector, this, maxMessages);
                } else {
                    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);
                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
    // ---------------------------------------------------------------

    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);
                    }
                }
            }
        }
    }
    
    public int getPoolSize() {
        return sspool.size();
    }

    public void stop() {
        TraceEjb.mdb.log(BasicLevel.DEBUG, "");
        try {
            cc.close();
            if (tconn != null)
                tconn.close();
            if (qconn != null)
                qconn.close();
        } catch(javax.jms.JMSException e) {
            TraceEjb.logger.log(BasicLevel.WARN, "unregister: Cannot close Connection Consumer");
        }
    }

    public void sync() {
    }


    public JHome getHome() {
        return null;
    }

    public JLocalHome getLocalHome() {
        return null;
    }

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

    public ServerSession getServerSession() throws JMSException {
        TraceEjb.mdb.log(BasicLevel.DEBUG, "");

        return getNewInstance();
    }



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

    public void releaseServerSession(ServerSession ss) {
        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
    // ---------------------------------------------------------------

    public TimerService getTimerService() {
        if (myTimerService == null) {
            // TODO : Check that instance implements TimedObject ?
            myTimerService = new JTimerService(this);
        }
        return myTimerService;
    }

    public int getMinPoolSize() {
        return minPoolSize;
    }

    public int getMaxCacheSize() {
        return maxCacheSize;
    }

    public int getCacheSize() {
        return instanceCount;
    }

    public int getTransactionAttribute() {
        return ((MessageDrivenDesc)dd).getTxAttribute();
    }

    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();
                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;
            }
        }
    }

    public void reduceCache() {
        TraceEjb.swapper.log(BasicLevel.DEBUG, "");
        // 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());
        }

    }

    public void notifyTimeout(Timer timer) {
        TraceEjb.mdb.log(BasicLevel.DEBUG, "");

        // We need an instance from the pool to process the timeout.
        JMessageDrivenBean jmdb = null;
        try {
            jmdb = getNewInstance();
        } catch (JMSException e) {
            TraceEjb.logger.log(BasicLevel.ERROR, "exception:" + e);
            throw new EJBException("Cannot deliver the timeout: " + e);
        }

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

        // release the instance
        releaseServerSession(jmdb);
    }

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

    private JMessageDrivenBean getNewInstance() throws JMSException {
        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");
                }
            } else {
                TraceEjb.mdb.log(BasicLevel.DEBUG,"pool is empty");
                if (maxCacheSize == 0 ||  instanceCount < maxCacheSize) {
                    // 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");
                    }
                } else {
                    while (sspool.isEmpty()) {
                        TraceEjb.swapper.log(BasicLevel.DEBUG,"sspool.isEmpty() = true --> wait()");
                        try {
                            sspool.wait();
                            TraceEjb.swapper.log(BasicLevel.DEBUG,"sspool notified");
                        } catch (InterruptedException e) {
                            TraceEjb.swapper.log(BasicLevel.DEBUG, "sspool waiting interrupted");
                        } 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;
        }
    }

    private JMessageDrivenBean createNewInstance() throws Exception {
        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.
        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);
            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, threadpool);
        sess.setMessageListener((MessageListener)ss);

        // 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 {
            beanclass.getMethod("ejbCreate", (Class[]) null).invoke(mdb, (Object[]) null);
        } 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 {
            resetComponentContext(ctxsave);
        }

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

}

---