JMessageEndpointProxy.java

package org.objectweb.jonas_ejb.container;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;

import javax.ejb.MessageDrivenBean;
import javax.ejb.EJBException;
import javax.resource.ResourceException;
import javax.resource.spi.IllegalStateException;
import javax.transaction.xa.XAResource;

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

import org.objectweb.transaction.jta.TransactionManager;
import org.objectweb.util.monolog.api.BasicLevel;

public class JMessageEndpointProxy implements InvocationHandler {

    protected JMdbEndpointFactory bf = null;

    protected JMessageEndpoint ep = null;

    protected MessageDrivenBean mdb = null;

    protected TransactionManager tm = null;

    boolean b4Delivery = false;

    int msgCount = 0;

    // RequestCtx associated to the thread
    // Thread specific data
    private transient static ThreadLocal requestCtx = new ThreadLocal();
    
    public JMessageEndpointProxy(JMdbEndpointFactory bf, MessageDrivenBean mdb, JMessageEndpoint ep) {
        this.bf = bf;
        this.mdb = mdb;
        this.ep = ep;
        // keep these locally for efficiency.
        tm = bf.getTransactionManager();
    }

    // ------------------------------------------------------------------
    // InvocationHandler implementation
    // ------------------------------------------------------------------

    public Object invoke(Object obj, Method method, Object[] aobj) throws Throwable, NoSuchMethodException,
            ResourceException, Exception {
        RequestCtx rctx = null;
        String methodName = method.getName();
        Object ret = null;
        Throwable invokeEx = null;

        TraceEjb.mdb.log(BasicLevel.DEBUG, "Calling " + methodName + " on " + this);
        if (ep.released) {
            throw new IllegalStateException("Endpoint is in a released state and must be reactivated before using");
        }
        // Object methods
        if ("equals".equals(methodName)) {
            Object obj1 = null;
            if (Proxy.isProxyClass(aobj[0].getClass())) {
                obj1 = (Object) Proxy.getInvocationHandler(aobj[0]);
            } else {
                obj1 = aobj[0];
            }
            ret = new Boolean(this.equals(obj1));
        } else if ("hashCode".equals(methodName)) {
            ret = new Integer(this.hashCode());
        } else if ("toString".equals(methodName)) {
            ret = this.toString();
        } else if ("afterDelivery".equals(methodName)) {
            if (!b4Delivery) {
                throw new IllegalStateException(methodName + " called w/o call to beforeDelivery");
            }
            b4Delivery = false;
            msgCount = 0;
            rctx = (RequestCtx) requestCtx.get();
            try {
                if (rctx.mustCommit && ep.getXAResource() != null) {
                    rctx.currTx.delistResource(ep.getXAResource(), XAResource.TMSUCCESS);
                }
                bf.postInvoke(rctx);
            } catch (Exception e) {
                TraceEjb.logger.log(BasicLevel.ERROR, "exception on postInvoke: ", e);
                throw new RuntimeException(e);
            }
        } else if ("beforeDelivery".equals(methodName)) {
            
            TraceEjb.mdb.log(BasicLevel.DEBUG, "beforeDelivery called");
            if (b4Delivery) {
                throw new IllegalStateException(methodName + " called w/o call to afterDelivery");
            }
            if (bf.isTxBeanManaged()) {
                throw new IllegalStateException(methodName + " cannot be called when using bean managed transactions");
            }
            if (tm.getTransaction() != null) {
                throw new IllegalStateException(methodName + " cannot be called when using an imported transaction");
            }
            msgCount = 0;
            // retrieve intented method name passed in parameter
            Object obj1 = null;
            if (Proxy.isProxyClass(aobj[0].getClass())) {
                obj1 = (Object) Proxy.getInvocationHandler(aobj[0]);
            } else {
                obj1 = aobj[0];
            }
            Method intentedTargetMethod = (Method) obj1;
            String intentedTargetMethodName = intentedTargetMethod.getName();

            TraceEjb.mdb.log(BasicLevel.DEBUG, "intentedTargetMethodName=" + intentedTargetMethodName);
            try {
                TraceEjb.mdb.log(BasicLevel.DEBUG, "before preInvoke");
                rctx = bf.preInvoke(getTxAttr(intentedTargetMethod));
                // JCA 1.5 ?12.5.6 mentions that beforeDelivery and afterDelivery
                // must be called from a single thread of control.
                // so the RequestCtx is stored in thread specific data for next
                // use in afterDelivery call and intended method call.
                requestCtx.set(rctx);
                b4Delivery = true;
                if (rctx.mustCommit && ep.getXAResource() != null) {

                    rctx.currTx.enlistResource(ep.getXAResource());
                    TraceEjb.mdb.log(BasicLevel.DEBUG, "enlistResource Ok");
                }
            } catch (Exception e) {
                TraceEjb.logger.log(BasicLevel.ERROR, "preInvoke failed: ", e);
                throw new RuntimeException(e);
            }
            TraceEjb.mdb.log(BasicLevel.DEBUG, "beforeDelivery ended");
        } else if ("release".equals(methodName)) {
            bf.releaseEndpoint(ep);
        } else {
            msgCount++;

            if (!b4Delivery) {
                boolean isImportedTx = tm.getTransaction() != null;
                rctx = bf.preInvoke(getTxAttr(method));
                // JCA 1.5 ?12.5.9
                // if txRequired && imported tx, use the source managed Tx
                // and ignore the XAResource
                if (rctx.mustCommit) {
                    if (!isImportedTx) {

                        if (ep.getXAResource() != null) {
                            rctx.currTx.enlistResource(ep.getXAResource());
                        }
                    } else {
                        rctx.mustCommit = false;
                    }
                }
                bf.checkSecurity(null);
            } else if (msgCount > 1) {
                throw new IllegalStateException("Unable to deliver multiple messages");
            } else {
                rctx = (RequestCtx) requestCtx.get();
            }
            try {
                TraceEjb.mdb.log(BasicLevel.DEBUG, "Before invoke");
                ret = method.invoke(mdb, aobj);
                TraceEjb.mdb.log(BasicLevel.DEBUG, "After invoke");
            } catch (InvocationTargetException ite) {
                Throwable t = ite.getTargetException();
                if (t instanceof RuntimeException) {
                    if (rctx != null) {
                        rctx.sysExc = new EJBException((RuntimeException) t);
                    }
                } else {
                    if (rctx != null) {
                        rctx.sysExc = t;
                    }
                }
                invokeEx = rctx.sysExc;
                TraceEjb.logger.log(BasicLevel.ERROR, "error thrown by an enterprise Bean", t);
            } catch (Throwable ex) {
                //ex.printStackTrace();
                if (rctx != null) {
                    rctx.sysExc = ex;
                }
                invokeEx = ex;

                TraceEjb.logger.log(BasicLevel.ERROR, "error thrown by an enterprise Bean", ex);
            } finally {
                if (!b4Delivery) {
                    try {
                        if (!bf.isTxBeanManaged() && rctx.mustCommit && ep.getXAResource() != null) {
                            rctx.currTx.delistResource(ep.getXAResource(), XAResource.TMSUCCESS);
                        }
                        bf.postInvoke(rctx);
                    } catch (Exception e) {
                        TraceEjb.logger.log(BasicLevel.ERROR, "exception on postInvoke: ", e);
                        if (invokeEx == null) {
                            invokeEx = e;
                        }
                    }
                } else {
                    // Reset message count if not using after/before delivery
                    msgCount = 0;
                }
            }
        }
        if (invokeEx != null) {
            TraceEjb.logger.log(BasicLevel.ERROR, "Exception raised: ", invokeEx);
            throw invokeEx;
        }
        TraceEjb.mdb.log(BasicLevel.DEBUG, "ret="+ret);
        return ret;
    }

    private int getTxAttr(Method m) {
        int ret = MethodDesc.TX_NOT_SUPPORTED;
        try {
            ret = bf.isDeliveryTransacted(m) ? MethodDesc.TX_REQUIRED : MethodDesc.TX_NOT_SUPPORTED;
        } catch (NoSuchMethodException nsme) {
        }
        return ret;
    }
}

---