JMdbEndpointFactory.java

package org.objectweb.jonas_ejb.container;

import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.ArrayList;
import java.util.List;
import java.util.LinkedList;
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.naming.Context;
import javax.resource.spi.ActivationSpec;
import javax.resource.spi.UnavailableException;
import javax.resource.spi.endpoint.MessageEndpoint;
import javax.resource.spi.endpoint.MessageEndpointFactory;
import javax.transaction.SystemException;
import javax.transaction.xa.XAResource;

import org.objectweb.jonas_ejb.deployment.api.MessageDrivenDesc;
import org.objectweb.jonas_ejb.deployment.api.MethodDesc;
import org.objectweb.jonas_ejb.deployment.api.ActivationConfigPropertyDesc;

import org.objectweb.jonas.resource.Rar;
import org.objectweb.jotm.Current;

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

public class JMdbEndpointFactory extends JFactory implements MessageEndpointFactory {

    private List endpool = new ArrayList();

    protected int instanceCount = 0;

    protected int minPoolSize = 0;

    protected int maxCacheSize = 0;

    private ActivationSpec as = null;

    private String msglistenType = null;

    private Rar rar = null;

    private boolean activated = false;

    private static String JMESSAGEENDPOINT_CLASS = "org.objectweb.jonas_ejb.container.JMessageEndpointIntf";

    private static String MESSAGEENDPOINT_CLASS = "javax.resource.spi.endpoint.MessageEndpoint";

    private static String MESSAGEDRIVENCONTEXT_CLASS = "javax.ejb.MessageDrivenContext";

    public JMdbEndpointFactory(MessageDrivenDesc dd, JContainer cont, ActivationSpec as) {
        super(dd, cont);
        String dest = dd.getDestinationJndiName();
        // Set the AS properties and validate the required config properties
        List acpl = null;
        if (dd.getMdActivationConfigDesc() != null) {
            acpl = dd.getMdActivationConfigDesc().getActivationConfigPropertyList();
        }
        List jAcpl = null;
        if (dd.getJonasMdActivationConfigDesc() != null) {
            jAcpl = dd.getJonasMdActivationConfigDesc().getActivationConfigPropertyList();
        }
        mdbEFInit(dd, dest, acpl, jAcpl, as);
    }

    public JMdbEndpointFactory(MessageDrivenDesc dd, String destination, JContainer cont,
                               ActivationSpec as) {
        super(dd, cont);

        // Build Activation spec linked list from variables

        List jAcpl = new LinkedList();
        List acpl = null;
        if (dd.getMdActivationConfigDesc() != null) {
            acpl = dd.getMdActivationConfigDesc().getActivationConfigPropertyList();
        }
        buildACL(dd, jAcpl);

        mdbEFInit(dd, destination, acpl, jAcpl, as);
    }

    private void buildACL(MessageDrivenDesc dd, List jacl) {

        String [] aclNames = {"destination", "destinationType", "messageSelector",
                              "acknowledgeMode", "subscriptionDurability" };

        ActivationConfigPropertyDesc acp = null;

        acp = new ActivationConfigPropertyDesc();
        acp.setActivationConfigPropertyName("destination");
        acp.setActivationConfigPropertyValue(dd.getDestinationJndiName());
        jacl.add(acp);

        acp = new ActivationConfigPropertyDesc();
        acp.setActivationConfigPropertyName("destinationType");
        if (dd.isTopicDestination()) {
            acp.setActivationConfigPropertyValue("javax.jms.Topic");
        } else {
            acp.setActivationConfigPropertyValue("javax.jms.Queue");
        }
        jacl.add(acp);

        acp = new ActivationConfigPropertyDesc();
        acp.setActivationConfigPropertyName("messageSelector");
        acp.setActivationConfigPropertyValue(dd.getSelector());
        jacl.add(acp);

        acp = new ActivationConfigPropertyDesc();
        acp.setActivationConfigPropertyName("acknowledgeMode");
        if (dd.getAcknowledgeMode() == MessageDrivenDesc.AUTO_ACKNOWLEDGE) {
            acp.setActivationConfigPropertyValue("Auto-acknowledge");
        } else {
            acp.setActivationConfigPropertyValue("Dups-ok-acknowledge");
        }
        jacl.add(acp);

        acp = new ActivationConfigPropertyDesc();
        acp.setActivationConfigPropertyName("subscriptionDurability");
        if (dd.getSubscriptionDurability() == MessageDrivenDesc.SUBS_DURABLE) {
            acp.setActivationConfigPropertyValue("Durable");
        } else {
            acp.setActivationConfigPropertyValue("NonDurable");
        }
        jacl.add(acp);

        if (dd.getSubscriptionDurability() == MessageDrivenDesc.SUBS_DURABLE) {
            acp = new ActivationConfigPropertyDesc();
            acp.setActivationConfigPropertyName("subscriptionName");
            acp.setActivationConfigPropertyValue(dd.getEjbName());
            jacl.add(acp);
        }

        // Check existing list to determine if additional items are configured, then pass them along also
        List acpl = null;
        if (dd.getJonasMdActivationConfigDesc() != null) {
            acpl = dd.getJonasMdActivationConfigDesc().getActivationConfigPropertyList();

            String acpName = null;
            boolean found = false;
            for (int i = 0; i < acpl.size(); i++) {
                acp = (ActivationConfigPropertyDesc) acpl.get(i);
                acpName = acp.getActivationConfigPropertyName();
                found = false;
                for (int j = 0; j < aclNames.length; j++) {
                    if (acpName.equals(aclNames[j])) {
                        found = true;
                        break;
                    }
                }
                if (!found) {
                    jacl.add(acp);
                }
            }

        }
    }

    private void mdbEFInit(MessageDrivenDesc dd, String dest, List acpl,
                           List jAcpl, ActivationSpec aSpec) {

        String methName = "mdbEFInit: ";
        String ejbName = dd.getEjbName();

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

        // Set ActivationSpec
        as = aSpec;

        minPoolSize = dd.getPoolMin();
        maxCacheSize = dd.getCacheMax();
        trace(BasicLevel.DEBUG, methName + "maxCacheSize = " + maxCacheSize + " minPoolSize = " + minPoolSize);

        // Get the associated Resource Adapter and messagelistenerType
        rar = Rar.getRar(dest);
        if (rar == null) {
            trace(BasicLevel.ERROR, methName + "cannot retrieve associated Resource Adapter ");
            throw new EJBException("cannot retrieve associated Resource Adapter ");
        }
        msglistenType = rar.getInterface(dest);

        try {
            rar.configureAS(as, acpl, jAcpl, dest, ejbName);
        } catch (Exception ex) {
            trace(BasicLevel.ERROR, methName + "cannot configure activationspec " + ex);
            ex.printStackTrace();
            throw new EJBException("cannot configure activationspec ", ex);
        }

        // pre-allocate a set of Endpoints
        synchronized (endpool) {
            for (int i = 0; i < minPoolSize; i++) {
                JMessageEndpoint ep = null;
                try {
                    ep = createNewInstance();
                    endpool.add(ep);
                } catch (Exception e) {
                    trace(BasicLevel.ERROR, methName + "cannot init pool of instances ");
                    throw new EJBException("cannot init pool of instances ", e);
                }
            }
        }
        if (minPoolSize != 0) {
            trace(BasicLevel.INFO, methName + "pre-allocate a set of " + minPoolSize
                                 + " message driven bean  instances");
        }

        // Call to register an XAResource with JOTM
        try {
            ActivationSpec [] asArray = new ActivationSpec[1];
            asArray[0] = as;
            XAResource [] xar = rar.getResourceAdapter().getXAResources(asArray);
            if (xar != null && xar.length > 0) {
                ((Current) tm).getTransactionRecovery().
                           registerResourceManager(ejbName+msglistenType,
                                                   xar[0], "", null);
            }
        } catch (Exception ex) {
            trace(BasicLevel.ERROR, ex.getMessage(), ex);
        }

        activated = true;
        // EndpointActivation
        try {
            rar.getResourceAdapter().endpointActivation(this, as);
        } catch (Exception ex) {
            activated = false;
            trace(BasicLevel.ERROR, methName + "cannot activate endpoint " + ex);
            ex.printStackTrace();
            throw new EJBException("cannot activate endpoint ", ex);
        }
    }

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

    public void initInstancePool() {
    }
    
    public int getPoolSize() {
        return endpool.size();
    }

    public void stop() {
        String methName = "stop: ";
        trace(BasicLevel.DEBUG, methName);
        try {
            // Deactivate the endpoint
            rar.getResourceAdapter().endpointDeactivation(this, as);
            // Loop thru the endpool and release them
            synchronized (endpool) {
                trace(BasicLevel.DEBUG, methName + "stopping " + this);
                JMessageEndpoint ep = null;
                while (endpool.size() > 0) {
                    ep = (JMessageEndpoint) endpool.remove(0);
                    instanceCount--;
                    try {
                        ep.mdb.ejbRemove();
                    } catch (Exception e) {
                        trace(BasicLevel.ERROR, methName + "Cannot remove mdb: "
                               + ep.mdb);
                    }
                }
            }
        } catch (Exception e) {
            trace(BasicLevel.WARN, methName + "Cannot deactivate the endpoint");
        }
    }

    public void sync() {
    }

    public JHome getHome() {
        return null;
    }

    public JLocalHome getLocalHome() {
        return null;
    }

    // ---------------------------------------------------------------
    // MessageEndpointFactory implementation
    // ---------------------------------------------------------------

    public MessageEndpoint createEndpoint(XAResource xaResource) throws UnavailableException {
        String methName = "createEndpoint: ";
        trace(BasicLevel.DEBUG, methName);

        if (!activated) {
            trace(BasicLevel.ERROR, methName + "mdb is not usuable either initial deployment or undeployment ");
            throw new UnavailableException("mdb is not usuable either initial deployment or undeployment ");
        }

        JMessageEndpoint ep = null;
        try {
            ep = getNewInstance(xaResource);
        } catch (Exception ex) {
            trace(BasicLevel.ERROR, methName + "cannot create an endpoint ");
            throw new UnavailableException("cannot create an endpoint ", ex);
        }
        return (MessageEndpoint) ep.mep;
    }

    public boolean isDeliveryTransacted(Method method) throws NoSuchMethodException {
        int txAttribute = 0;
        try {
            txAttribute = dd.getMethodDesc(method).getTxAttribute();
        } catch (Exception ex) {
            trace(BasicLevel.ERROR, "isDeliveryTransacted: No such method exists. " + method.getName());
            throw new NoSuchMethodException("No such method exists. " + method.getName());
        }
        return txAttribute == MethodDesc.TX_REQUIRED ? true : false;
    }

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

    public JMessageEndpoint getMessageEndpoint() throws Exception {
        trace(BasicLevel.DEBUG, "getMessageEndpoint: ");

        return getNewInstance(null);
    }

    public void releaseEndpoint(JMessageEndpoint ep) {
        String methName = "releaseEndpoint: ";
        trace(BasicLevel.DEBUG, methName + ep);

        ep.setReleasedState(true);
        synchronized (endpool) {
            endpool.add(ep);
            trace(BasicLevel.DEBUG, methName + "notifyAll " );
            endpool.notifyAll();
        }
        trace(BasicLevel.DEBUG, methName + "nb instances " + getCacheSize());
        trace(BasicLevel.DEBUG, methName + "nb free cached instances " + getPoolSize());

    }

    // ---------------------------------------------------------------
    // 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) {
        String methName = "checkTransaction: ";
        if (rctx.txAttr == MethodDesc.TX_REQUIRED) {
            try {
                if (txbeanmanaged) {
                    if (tm.getTransaction() == null) {
                        trace(BasicLevel.ERROR, methName + "No transaction and need one");
                        return;
                    }
                } else {
                    if (tm.getTransaction() == null) {
                        tm.begin();
                    }
                }
                rctx.mustCommit = true;
                rctx.currTx = tm.getTransaction();
            } catch (Exception e) {
                // No exception raised in case of MDB
                trace(BasicLevel.ERROR, methName + "cannot start tx:" + e);
                return;
            }
        } else {
            if (rctx.txAttr != MethodDesc.TX_NOT_SUPPORTED) {
                trace(BasicLevel.ERROR, methName + "Bad transaction attribute: " + rctx.txAttr);
            }
            try {
                rctx.currTx = tm.getTransaction();
                if (rctx.currTx != null) {
                    trace(BasicLevel.DEBUG, methName + "Suspending client tx");
                    rctx.clientTx = tm.suspend();
                    rctx.currTx = null;
                }
            } catch (SystemException e) {
                trace(BasicLevel.ERROR, methName + "cannot suspend transaction:\n", e);
                return;
            }
        }
    }

    public void reduceCache() {
        String methName = "reduceCache: ";
        trace(BasicLevel.DEBUG, methName);
        // reduce the pool to the minPoolSize
        int poolsz = minPoolSize;
        synchronized (endpool) {
            trace(BasicLevel.DEBUG, methName + "try to reduce " + endpool.size() + " to " + poolsz);
            while (endpool.size() > poolsz) {
                ListIterator i = endpool.listIterator();
                if (i.hasNext()) {
                    i.next();
                    i.remove();
                    instanceCount--;
                }
            }
        }
        trace(BasicLevel.DEBUG, methName + "cacheSize= " + getCacheSize());

    }

    public void notifyTimeout(Timer timer) {
        String methName = "notifyTimeout: ";
        trace(BasicLevel.DEBUG, methName);

        // We need an instance from the pool to process the timeout.
        JMessageEndpoint ep = null;
        try {
            ep = getNewInstance(null);
        } catch (Exception e) {
            trace(BasicLevel.ERROR, methName + "exception:" + e);
            throw new EJBException("Cannot deliver the timeout: " + e);
        }

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

        // release the instance
        releaseEndpoint(ep);
    }

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

    private JMessageEndpoint getNewInstance(XAResource xaResource) throws Exception {
        String methName = "getNewInstance: ";
        trace(BasicLevel.DEBUG, methName + "Factory: " + this + " XAResource: " + xaResource);

        // try to get one from the Pool
        JMessageEndpoint ep = null;

        // try to find a free entry in the pool
        synchronized (endpool) {
            if (!endpool.isEmpty()) {
                try {
                    ep = (JMessageEndpoint) endpool.remove(0);
                } catch (Exception ex) {
                    // This should never happen
                    trace(BasicLevel.ERROR, methName + "Exception:" + ex);
                    throw new EJBException("Cannot get an instance from the pool");
                }
            } else {
                trace(BasicLevel.DEBUG, methName + "pool is empty");
                if (maxCacheSize == 0 || instanceCount < maxCacheSize) {
                    // Pool has free slots, create a new MessageEndpoint object
                    try {
                        ep = createNewInstance();
                    } catch (Exception e) {
                        trace(BasicLevel.ERROR, methName + "exception:" + e);
                        throw new EJBException("Cannot create a new instance");
                    }
                } else {
                    while (endpool.isEmpty()) {
                        trace(BasicLevel.DEBUG, methName + "endpool.isEmpty() = true --> wait()");
                        try {
                            endpool.wait();
                            trace(BasicLevel.DEBUG, methName + "endpool notified");
                        } catch (InterruptedException e) {
                            trace(BasicLevel.DEBUG, methName + "endpool waiting interrupted");
                        } catch (Exception e) {
                            throw new EJBException("synchronization pb", e);
                        }
                    }
                    try {
                        ep = (JMessageEndpoint) endpool.remove(0);
                    } catch (Exception ex) {
                        // This should never happen
                        trace(BasicLevel.ERROR, methName + "Exception:" + ex);
                        throw new EJBException("Cannot get an instance from the pool");
                    }
                }

            }
            trace(BasicLevel.DEBUG, methName + "nb instances " + getCacheSize());
            ep.setXAResource(xaResource);
            ep.setReleasedState(false);
            trace(BasicLevel.DEBUG, methName + "Returning " + ep);
            return ep;
        }
    }

    private JMessageEndpoint createNewInstance() throws Exception {
        String methName = "createNewInstance: ";
        trace(BasicLevel.DEBUG, methName);
        JMessageEndpointProxy epProxy = null;
        MessageEndpoint ep = null;
        JMessageEndpoint jep = null;
        MessageDrivenDesc mdd = (MessageDrivenDesc) dd;
        ClassLoader cls = myClassLoader();

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

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

        // Instanciates a new JMessageEndpoint object
        jep = new JMessageEndpoint(this, mdb);
        epProxy = new JMessageEndpointProxy(this, mdb, jep);
        Class endpointClass = cls.loadClass(MESSAGEENDPOINT_CLASS);
        Class msgListenerClass = cls.loadClass(msglistenType);
        ep = (MessageEndpoint) Proxy.newProxyInstance(cls, new Class[] {endpointClass, msgListenerClass}, epProxy);

        jep.setProxy(ep);
        // 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) jep);
        try {
            beanclass.getMethod("ejbCreate", (Class[]) null).invoke(mdb, (Object[]) null);
        } catch (Exception e) {
            trace(BasicLevel.ERROR, methName + "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 (endpool) {
            instanceCount++;
        }
        return jep;
    }

    
    private void trace(int level, String str, Exception ex) {
        if (TraceEjb.isDebugJms()) {
            TraceEjb.mdb.log(level, str, ex);
        }
    }
    private void trace(int level, String str) {
        if (TraceEjb.isDebugJms()) {
            TraceEjb.mdb.log(level, str);
        }
    }
}

---