/*
* JBoss, Home of Professional Open Source
* Copyright 2005, JBoss Inc., and individual contributors as indicated
* by the @authors tag. See the copyright.txt in the distribution for a
* full listing of individual contributors.
*
* This 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 (at your option) any later version.
*
* This software 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 software; if not, write to the Free
* Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
* 02110-1301 USA, or see the FSF site: http://www.fsf.org.
*/
package org.jboss.ejb3.mdb;

import org.jboss.annotation.ejb.AcknowledgementMode;
import org.jboss.annotation.ejb.DefaultActivationSpecs;
import org.jboss.annotation.ejb.Durability;
import org.jboss.annotation.ejb.ResourceAdapter;
import org.jboss.aop.AspectManager;
import org.jboss.aop.MethodInfo;
import org.jboss.aop.advice.Interceptor;
import org.jboss.aop.util.MethodHashing;
import org.jboss.deployment.DeploymentException;
import org.jboss.ejb.txtimer.TimedObjectInvoker;
import org.jboss.ejb3.*;
import org.jboss.ejb3.mdb.inflow.JBossMessageEndpointFactory;
import org.jboss.ejb3.interceptor.InterceptorInfoRepository;
import org.jboss.ejb3.tx.TxUtil;
import org.jboss.jms.ConnectionFactoryHelper;
import org.jboss.jms.asf.ServerSessionPoolFactory;
import org.jboss.jms.asf.StdServerSessionPool;
import org.jboss.jms.jndi.JMSProviderAdapter;
import org.jboss.logging.Logger;
import org.jboss.metadata.ActivationConfigPropertyMetaData;

import javax.ejb.*;
import javax.ejb.Timer;
import javax.jms.*;
import javax.jms.Queue;
import javax.management.MBeanServer;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;
import javax.naming.Context;
import javax.naming.NamingException;
import javax.transaction.Transaction;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.*;

/**
 * Comment
 *
 * @author <a href="mailto:bill@jboss.org">Bill Burke</a>
 * @version $Revision$
 */
public class MDB extends EJBContainer implements TimedObjectInvoker
{
   private static final Logger log = Logger.getLogger(MDB.class);

   protected Method ON_MESSAGE;
   protected TimerService timerService;
   protected ActivationSpec activationSpec = new ActivationSpec();
   protected boolean isContainerManagedTx;
   protected AcknowledgementMode acknowledgeMode;
   protected boolean isNotSupportedTx;
   protected Connection connection;
   protected ConnectionConsumer connectionConsumer;
   protected ServerSessionPool serverSessionPool;
   protected MDBConfig config;
   protected ExceptionListenerImpl exListener;
   protected DLQHandler dlqHandler;
   protected Class messagingType = null;
   protected JBossMessageEndpointFactory messageEndpointFactory;

   /**
    * This is needed because API changed from JBoss 4.0.1sp1 to 4.0.2
    * TODO remove this after 4.0.2 is out.
    */
   protected Method getServerSessionPool;

   /**
    * Default destination type. Used when no message-driven-destination is given
    * in ejb-jar, and a lookup of destinationJNDI from jboss.xml is not
    * successfull. Default value: javax.jms.Topic.
    */
   protected final static String DEFAULT_DESTINATION_TYPE = "javax.jms.Topic";

   public MDB(String ejbName, AspectManager manager, ClassLoader cl, String beanClassName, Hashtable ctxProperties,
              InterceptorInfoRepository interceptorRepository, Ejb3Deployment deployment)
   {
      super(Ejb3Module.BASE_EJB3_JMX_NAME + ",name=" + ejbName, manager, cl, beanClassName, ejbName, ctxProperties, interceptorRepository, deployment);
      beanContextClass = MDBContext.class;
      Method[] methods = ServerSessionPoolFactory.class.getMethods();
      for (Method m : methods)
      {
         if (m.getName().equals("getServerSessionPool"))
         {
            getServerSessionPool = m;
            break;
         }
      }
   }

   public void setMessageEndpointFactory(JBossMessageEndpointFactory messageEndpointFactory)
   {
      this.messageEndpointFactory = messageEndpointFactory;
   }

   public Class getMessagingType()
   {
      if (messagingType == null)
      {
         MessageDriven annotation = (MessageDriven) resolveAnnotation(MessageDriven.class);
         messagingType = annotation.messageListenerInterface();
         if (messagingType.getName().equals(Object.class.getName()))
         {
            ArrayList<Class> list = ProxyFactoryHelper.getBusinessInterfaces(clazz);
            if (list.size() > 1 || list.size() == 0) throw new RuntimeException("unable to determine messagingType interface for MDB");
            messagingType = list.get(0);
         }
      }

      return messagingType;
   }

   public String getResourceAdaptorName()
   {
      ResourceAdapter annotation = (ResourceAdapter) resolveAnnotation(ResourceAdapter.class);
      if (annotation == null) return null;
      return annotation.value();
   }

   public String getDestination()
   {
      String result = null;
      MessageDriven annotation = (MessageDriven) resolveAnnotation(MessageDriven.class);
      for (ActivationConfigProperty property : annotation.activationConfig())
      {
         if (property.propertyName().equals("destination"))
         {
            return property.propertyValue();
         }
      }

      return result;
   }

   public Map getActivationConfigProperties()
   {
      HashMap result = new HashMap();
      MessageDriven annotation = (MessageDriven) resolveAnnotation(MessageDriven.class);
      for (ActivationConfigProperty property : annotation.activationConfig())
      {
         if (!property.propertyName().equals("destination") &&
                 !property.propertyName().equals("resourceAdaptorName") &&
                 !property.propertyName().equals("messagingType"))
         {
            ActivationConfigPropertyMetaData metaData = new ActivationConfigPropertyMetaData();
            try
            {
               Field nameField = ActivationConfigPropertyMetaData.class.getDeclaredField("name");
               nameField.setAccessible(true);
               nameField.set(metaData, property.propertyName());
               Field valueField = ActivationConfigPropertyMetaData.class.getDeclaredField("value");
               valueField.setAccessible(true);
               valueField.set(metaData, property.propertyValue());
            }
            catch (Exception e)
            {
               throw new RuntimeException(e);
            }

            /*
             * Older versions don't have this
             * TODO revert to this after we ditch 4.0.3SP1 as supported EJB3 platform
            metaData.setName(property.propertyName());
            metaData.setValue(property.propertyValue());
            */
            result.put(property.propertyName(), metaData);
         }
      }

      return result;
   }

   /**
    * Initialize the container invoker. Sets up a connection, a server session
    * pool and a connection consumer for the configured destination.
    * <p/>
    * Any JMSExceptions produced while initializing will be assumed to be
    * caused due to JMS Provider failure.
    *
    * @throws Exception Failed to initalize.
    */
   public void start() throws Exception
   {
      super.start();
      exListener = new ExceptionListenerImpl(this);

      populateActivationSpec();

      config = MDBConfig.createMDBConfig(activationSpec);

      innerStart();

      timerService = EjbTimerUtil.getTimerService(this, this);

      startProxies();
   }

   protected void innerStart() throws Exception
   {
      try
      {
         innerCreate();
      }
      catch (final JMSException e)
      {
         //
         // start a thread up to handle recovering the connection. so we can
         // attach to the jms resources once they become available
         //
         new Thread("EJB3 MDB Create Recovery Thread")
         {
            public void run()
            {
               exListener.onException(e);
            }
         }.start();
         return;
      }

      if (dlqHandler != null)
      {
         dlqHandler.start();
      }

      if (connection != null)
      {
         connection.setExceptionListener(exListener);
         connection.start();
      }

   }

   public ObjectName getJmxName()
   {
      ObjectName jmxName = null;
      String jndiName = ProxyFactoryHelper.getLocalJndiName(this);
      // The name must be escaped since the jndiName may be arbitrary
      String name = org.jboss.ejb.Container.BASE_EJB_CONTAINER_NAME + ",jndiName=" + jndiName;
      try
      {
         jmxName = org.jboss.mx.util.ObjectNameConverter.convert(name);
      }
      catch (MalformedObjectNameException e)
      {
         e.printStackTrace();
         throw new RuntimeException("Failed to create ObjectName, msg=" + e.getMessage());
      }

      return jmxName;
   }

   protected void startProxies() throws Exception
   {
      if (messageEndpointFactory != null)
      {
         messageEndpointFactory.start();
      }
   }

   private void innerCreate() throws Exception
   {
      log.debug("Initializing");

      MessageDriven messageDriven = (MessageDriven) resolveAnnotation(MessageDriven.class);

      // Get the JMS provider
      // todo get rid of server module dependency
      JMSProviderAdapter adapter = getJMSProviderAdapter();
      log.debug("Provider adapter: " + adapter);

      setupDLQ(adapter);

      // Is container managed?
      TransactionManagement transactionManagement = (TransactionManagement) resolveAnnotation(TransactionManagement.class);
      if (transactionManagement == null)
         isContainerManagedTx = true;
      else
         isContainerManagedTx = transactionManagement.value() == TransactionManagementType.CONTAINER;
      String ackMode = config.getAcknowledgeMode();
      if (ackMode != null && (ackMode.equals("DUPS_OK_ACKNOWLEDGE")
              || ackMode.equals(AcknowledgementMode.AUTO_ACKNOWLEDGE.name())
              || ackMode.equals("Dups-ok-acknowledge")))
      {
         acknowledgeMode = AcknowledgementMode.DUPS_OK;
      }
      else
      {
         acknowledgeMode = AcknowledgementMode.AUTO_ACKNOWLEDGE;
      }

      TransactionAttribute trans = (TransactionAttribute) resolveAnnotation(getOnMessage(), TransactionAttribute.class);
      TransactionAttributeType txType = trans == null ? TransactionAttributeType.REQUIRED : trans.value();

      isNotSupportedTx = txType == TransactionAttributeType.NOT_SUPPORTED;

      // Connect to the JNDI server and get a reference to root context
      Context context = adapter.getInitialContext();
      log.debug("context: " + context);

      // if we can't get the root context then exit with an exception
      if (context == null)
      {
         throw new RuntimeException("Failed to get the root context");
      }

      String destinationType = config.getDestinationType();

      // Unfortunately the destination is optional, so if we do not have one
      // here we have to look it up if we have a destinationJNDI, else give it
      // a default.
      if (destinationType == null)
      {
         log.warn("No message-driven-destination given; using; guessing type");
         destinationType = getDestinationType(context, config.getDestination());
      }

      if ("javax.jms.Topic".equals(destinationType))
      {
         innerCreateTopic(context, adapter);

      }
      else if ("javax.jms.Queue".equals(destinationType))
      {
         innerCreateQueue(context, adapter);

      }
      else
         throw new DeploymentException("Unknown destination-type " + destinationType);

      log.debug("Initialized with config " + toString());

      context.close();
   }

   private void innerCreateQueue(Context context, JMSProviderAdapter adapter)
           throws Exception
   {
      log.debug("Got destination type Queue for " + ejbName);

      // Get the JNDI suffix of the destination
      String jndiSuffix = parseJndiSuffix(config.getDestination(), ejbName);
      log.debug("jndiSuffix: " + jndiSuffix);

      // create a queue connection
      Object qFactory = context.lookup(adapter.getQueueFactoryRef());
      QueueConnection qConnection = null;
      try
      {
         qConnection = ConnectionFactoryHelper.createQueueConnection(qFactory, config.getUser(), config.getPassword());
         connection = qConnection;
      }
      catch (ClassCastException e)
      {
         throw new DeploymentException("Expected a QueueConnection check your provider adaptor: "
                 + adapter.getQueueFactoryRef());
      }

      // Set the optional client id
      String clientId = config.getClientID();

      log.debug("Using client id: " + clientId);
      if (clientId != null && clientId.length() > 0)
         connection.setClientID(clientId);

      // lookup or create the destination queue
      Queue queue = null;
      try
      {
         // First we try the specified queue
         if (config.getDestination() != null)
            queue = (Queue) context.lookup(config.getDestination());
      }
      catch (NamingException e)
      {
         log.warn("Could not find the queue destination-jndi-name=" + config.getDestination());
      }
      catch (ClassCastException e)
      {
         throw new DeploymentException("Expected a Queue destination-jndi-name=" + config.getDestination());
      }

      // FIXME: This is not portable, only works for JBossMQ
      if (queue == null)
         queue = (Queue) createDestination(Queue.class,
                 context,
                 "queue/" + jndiSuffix,
                 jndiSuffix);


      if (queue != null)
      {
         // set up the server session pool
         serverSessionPool = createSessionPool(queue,
                 qConnection,
                 true, // tx
                 acknowledgeMode.ordinal(),
                 new MessageListenerImpl(this));
         log.debug("Server session pool: " + pool);

         // create the connection consumer
         connectionConsumer =
                 qConnection.createConnectionConsumer(queue,
                         config.getMessageSelector(),
                         serverSessionPool,
                         config.getMaxMessages());
         log.debug("Connection consumer: " + connectionConsumer);
      }
   }

   private void innerCreateTopic(Context context, JMSProviderAdapter adapter)
           throws Exception
   {
      log.debug("Got destination type Topic for " + ejbName);

      // Get the JNDI suffix of the destination
      String jndiSuffix = parseJndiSuffix(config.getDestination(), ejbName);
      log.debug("jndiSuffix: " + jndiSuffix);

      // create a topic connection
      Object factory = context.lookup(adapter.getTopicFactoryRef());
      TopicConnection tConnection = null;
      try
      {
         tConnection = ConnectionFactoryHelper.createTopicConnection(factory, config.getUser(), config.getPassword());
         connection = tConnection;
      }
      catch (ClassCastException e)
      {
         throw new DeploymentException("Expected a TopicConnection check your provider adaptor: "
                 + adapter.getTopicFactoryRef());
      }

      // Fix: ClientId must be set as the first method call after connection creation.
      // Fix: ClientId is necessary for durable subscriptions.

      String clientId = config.getClientID();

      log.debug("Using client id: " + clientId);
      if (clientId != null && clientId.length() > 0)
         connection.setClientID(clientId);

      // lookup or create the destination topic
      Topic topic = null;
      try
      {
         // First we try the specified topic
         if (config.getDestination() != null)
            topic = (Topic) context.lookup(config.getDestination());
      }
      catch (NamingException e)
      {
         log.warn("Could not find the topic destination-jndi-name=" + config.getDestination());
      }
      catch (ClassCastException e)
      {
         throw new DeploymentException("Expected a Topic destination-jndi-name=" + config.getDestination());
      }

      // FIXME: This is not portable, only works for JBossMQ
      if (topic == null)
         topic = (Topic) createDestination(Topic.class,
                 context,
                 "topic/" + jndiSuffix,
                 jndiSuffix);
      if (topic != null)
      {
         // set up the server session pool
         serverSessionPool = createSessionPool(topic,
                 tConnection,
                 true, // tx
                 acknowledgeMode.ordinal(),
                 new MessageListenerImpl(this));

         String durability = config.getDurability();
         Durability subscriptionDurability;

         if (durability != null && durability.equals("Durable"))
         {
            subscriptionDurability = Durability.DURABLE;
         }
         else
         {
            subscriptionDurability = Durability.NON_DURABLE;
         }

         // To be no-durable or durable
         if (subscriptionDurability == Durability.NON_DURABLE)
         {
            // Create non durable
            connectionConsumer =
                    tConnection.createConnectionConsumer(topic,
                            config.getMessageSelector(),
                            serverSessionPool,
                            config.getMaxMessages());
         }
         else
         {
            // Durable subscription
            String durableName = config.getSubscriptionName();

            connectionConsumer =
                    tConnection.createDurableConnectionConsumer(topic,
                            durableName,
                            config.getMessageSelector(),
                            serverSessionPool,
                            config.getMaxMessages());
         }
         log.debug("Topic connectionConsumer set up");
      }
   }

   /**
    * Create a server session pool for the given connection.
    *
    * @param connection   The connection to use.
    * @param minSession   The minumum number of sessions
    * @param maxSession   The maximum number of sessions.
    * @param keepAlive    The time to keep sessions alive
    * @param isTransacted True if the sessions are transacted.
    * @param ack          The session acknowledgement mode.
    * @param listener     The message listener.
    * @return A server session pool.
    * @throws javax.jms.JMSException
    * @throws NamingException        Description of Exception
    */
   protected ServerSessionPool
           createSessionPool(final Destination dest,
                             final Connection connection,
                             final boolean isTransacted,
                             int ack,
                             final MessageListener listener)
           throws NamingException, JMSException
   {
      ServerSessionPool pool;
      Context context = EJB3Util.getInitialContext(initialContextProperties);

      try
      {
         // first lookup the factory
         log.debug("JNDI Hashtable: " + initialContextProperties);
         ServerSessionPoolFactory factory = (ServerSessionPoolFactory)
                 context.lookup(config.getServerSessionPoolFactoryJNDI());

         // To work with JBoss 4.0.1SP1 and JBoss 4.0.2
         // TODO remove this once JBoss 4.0.2 comes out
         if (getServerSessionPool.getParameterTypes().length == 9)
         {
            // the create the pool
            try
            {
               pool = (ServerSessionPool) getServerSessionPool.invoke(factory, dest, connection, config.getMinPoolSize(), config.getMaxPoolSize(), config.getKeepAlive(), isTransacted, ack, !isContainerManagedTx || isNotSupportedTx, listener);
            }
            catch (IllegalAccessException e)
            {
               throw new RuntimeException(e);
            }
            catch (InvocationTargetException e)
            {
               throw new RuntimeException(e);
            }
         }
         else
         {
            // the create the pool
            try
            {
               pool = (ServerSessionPool) getServerSessionPool.invoke(factory, connection, config.getMinPoolSize(), config.getMaxPoolSize(), config.getKeepAlive(), isTransacted, ack, !isContainerManagedTx || isNotSupportedTx, listener);
            }
            catch (IllegalAccessException e)
            {
               throw new RuntimeException(e);
            }
            catch (InvocationTargetException e)
            {
               throw new RuntimeException(e);
            }
         }
      }
      finally
      {
         context.close();
      }

      return pool;
   }

   /**
    * Create and or lookup a JMS destination.
    *
    * @param type       Either javax.jms.Queue or javax.jms.Topic.
    * @param ctx        The naming context to lookup destinations from.
    * @param jndiName   The name to use when looking up destinations.
    * @param jndiSuffix The name to use when creating destinations.
    * @return The destination.
    * @throws IllegalArgumentException Type is not Queue or Topic.
    * @throws Exception                Description of Exception
    */
   protected Destination createDestination(final Class type,
                                           final Context ctx,
                                           final String jndiName,
                                           final String jndiSuffix)
           throws Exception
   {
      try
      {
         // first try to look it up
         return (Destination) ctx.lookup(jndiName);
      }
      catch (NamingException e)
      {
         // is JMS?
         if (getDestination() == null)
         {
            return null;
         }
         else
         {
            // if the lookup failes, the try to create it
            log.warn("destination not found: " + jndiName + " reason: " + e);
            log.warn("creating a new temporary destination: " + jndiName);

            //
            // jason: we should do away with this...
            //
            // attempt to create the destination (note, this is very
            // very, very unportable).
            //

            MBeanServer server = org.jboss.mx.util.MBeanServerLocator.locateJBoss();

            String methodName;
            if (type == Topic.class)
            {
               methodName = "createTopic";
            }
            else if (type == Queue.class)
            {
               methodName = "createQueue";
            }
            else
            {
               // type was not a Topic or Queue, bad user
               throw new IllegalArgumentException
                       ("Expected javax.jms.Queue or javax.jms.Topic: " + type);
            }

            // invoke the server to create the destination
            server.invoke(new ObjectName("jboss.mq:service=DestinationManager"),
                    methodName,
                    new Object[]{jndiSuffix},
                    new String[]{"java.lang.String"});

            // try to look it up again
            return (Destination) ctx.lookup(jndiName);
         }
      }
   }

   /**
    * Parse the JNDI suffix from the given JNDI name.
    *
    * @param jndiname     The JNDI name used to lookup the destination.
    * @param defautSuffix Description of Parameter
    * @return The parsed suffix or the defaultSuffix
    */
   protected String parseJndiSuffix(final String jndiname,
                                    final String defautSuffix)
   {
      // jndiSuffix is merely the name that the user has given the MDB.
      // since the jndi name contains the message type I have to split
      // at the "/" if there is no slash then I use the entire jndi name...
      String jndiSuffix = "";

      if (jndiname != null)
      {
         int indexOfSlash = jndiname.indexOf("/");
         if (indexOfSlash != -1)
         {
            jndiSuffix = jndiname.substring(indexOfSlash + 1);
         }
         else
         {
            jndiSuffix = jndiname;
         }
      }
      else
      {
         // if the jndi name from jboss.xml is null then lets use the ejbName
         jndiSuffix = defautSuffix;
      }

      return jndiSuffix;
   }

   /**
    * Return the JMSProviderAdapter that should be used.
    *
    * @return The JMSProviderAdapter to use.
    */
   protected JMSProviderAdapter getJMSProviderAdapter()
           throws NamingException
   {
      Context context = getInitialContext();
      //todo make this pluggable
      String providerAdapterJNDI = config.getProviderAdapterJNDI();
      try
      {
         log.debug("Looking up provider adapter: " + providerAdapterJNDI);
         return (JMSProviderAdapter) context.lookup(providerAdapterJNDI);
      }
      finally
      {
         context.close();
      }
   }

   /**
    * Try to get a destination type by looking up the destination JNDI, or
    * provide a default if there is not destinationJNDI or if it is not possible
    * to lookup.
    *
    * @param ctx             The naming context to lookup destinations from.
    * @param destinationJNDI The name to use when looking up destinations.
    * @return The destination type, either derived from destinationJDNI or
    *         DEFAULT_DESTINATION_TYPE
    */
   protected String getDestinationType(Context ctx, String destinationJNDI)
   {
      String destType = null;

      if (destinationJNDI != null)
      {
         try
         {
            Destination dest = (Destination) ctx.lookup(destinationJNDI);
            if (dest instanceof javax.jms.Topic)
            {
               destType = "javax.jms.Topic";
            }
            else if (dest instanceof javax.jms.Queue)
            {
               destType = "javax.jms.Queue";
            }
         }
         catch (NamingException ex)
         {
            log.debug("Could not do heristic lookup of destination ", ex);
         }

      }
      if (destType == null)
      {
         log.warn("Could not determine destination type, defaults to: " +
                 DEFAULT_DESTINATION_TYPE);

         destType = DEFAULT_DESTINATION_TYPE;
      }

      return destType;
   }

   private void setupDLQ(JMSProviderAdapter adapter) throws Exception
   {
      // Set up Dead Letter Queue handler
      if (config.isUseDLQ())
      {
         dlqHandler = new DLQHandler(adapter, config);
         dlqHandler.create();
      }
   }

   /**
    * Initialize the ON_MESSAGE reference.
    */
   protected Method getOnMessage()
   {
      if (ON_MESSAGE != null) return ON_MESSAGE;
      try
      {
         final Class arg = Message.class;
         if (javax.jms.MessageListener.class.isAssignableFrom(clazz))
         {
            ON_MESSAGE = clazz.getMethod("onMessage", new Class[]{arg});
         }
         else
         {
            Class messagingTypeClass = getMessagingType();
            Method[] methods = messagingTypeClass.getDeclaredMethods();
            ON_MESSAGE = methods[0];
         }
      }
      catch (Exception e)
      {
         e.printStackTrace();
         throw new ExceptionInInitializerError(e);
      }

      return ON_MESSAGE;
   }

   private void populateActivationSpec()
   {
      DefaultActivationSpecs defaultSpecs = (DefaultActivationSpecs) resolveAnnotation(DefaultActivationSpecs.class);
      if (defaultSpecs != null)
      {
         activationSpec.merge(defaultSpecs.value());
      }

      MessageDriven md = (MessageDriven) resolveAnnotation(MessageDriven.class);

      activationSpec.merge(md.activationConfig());
   }

   public Object localInvoke(Method method, Object[] args) throws Throwable
   {
      MethodInfo info = getMethodInfo(method);
      if (info == null)
      {
         throw new RuntimeException("Could not resolve beanClass method from proxy call: " + method.toString());
      }
      return localInvoke(info, args);

   }

   public MethodInfo getMethodInfo(Method method)
   {
      long hash = MethodHashing.calculateHash(method);
      MethodInfo info = (MethodInfo) methodInterceptors.get(hash);
      return info;
   }

   public Object localInvoke(MethodInfo info, Object[] args) throws Throwable
   {
      ClassLoader oldLoader = Thread.currentThread().getContextClassLoader();
      ThreadLocalENCFactory.push(enc);
      try
      {
         Interceptor[] aspects = info.interceptors;
         EJBContainerInvocation nextInvocation = new EJBContainerInvocation(info, aspects);
         nextInvocation.setAdvisor(this);
         nextInvocation.setArguments(args);
         return nextInvocation.invokeNext();
      }
      finally
      {
         Thread.currentThread().setContextClassLoader(oldLoader);
         ThreadLocalENCFactory.pop();
      }
   }

   public TimerService getTimerService()
   {
      return timerService;
   }

   public void callTimeout(Timer timer) throws Exception
   {
      Method timeout = callbackHandler.getTimeoutCallback();
      if (timeout == null) throw new EJBException("No method has been annotated with @Timeout");
      Object[] args = {timer};
      try
      {
         localInvoke(timeout, args);
      }
      catch (Throwable throwable)
      {
         if (throwable instanceof Exception) throw (Exception) throwable;
         throw new RuntimeException(throwable);
      }
   }

   /**
    * Stop done from inside, we should not stop the exceptionListener in inner
    * stop.
    */
   protected void innerStop()
   {
      try
      {
         if (connection != null)
         {
            connection.setExceptionListener(null);
            log.debug("unset exception listener");
         }
      }
      catch (Exception e)
      {
         log.error("Could not set ExceptionListener to null", e);
      }

      // Stop the connection
      try
      {
         if (connection != null)
         {
            connection.stop();
            log.debug("connection stopped");
         }
      }
      catch (Exception e)
      {
         log.error("Could not stop JMS connection", e);
      }
   }


   public void stop() throws Exception
   {
      EjbTimerUtil.removeTimerService(this);
      // Silence the exception listener
      if (exListener != null)
      {
         exListener.stop();
      }

      innerStop();

      if (dlqHandler != null)
      {
         dlqHandler.stop();
      }

      stopProxies();
   }

   protected void stopProxies() throws Exception
   {
      if (messageEndpointFactory != null)
      {
         messageEndpointFactory.stop();
      }
   }

   public void innerDestroy() throws Exception
   {

      // close the connection consumer
      try
      {
         if (connectionConsumer != null)
         {
            connectionConsumer.close();
         }
      }
      catch (Exception e)
      {
         log.error("Failed to close connection consumer", e);
      }

      // clear the server session pool (if it is clearable)
      try
      {
         if (serverSessionPool instanceof StdServerSessionPool)
         {
            StdServerSessionPool p = (StdServerSessionPool) serverSessionPool;
            p.clear();
         }
      }
      catch (Exception e)
      {
         log.error("Failed to clear session pool", e);
      }

      // close the connection
      if (connection != null)
      {
         try
         {
            connection.close();
            connection = null;
         }
         catch (Exception e)
         {
            log.error("Failed to close connection", e);
         }
      }

      // Take down DLQ
      try
      {
         if (dlqHandler != null)
         {
            dlqHandler.destroy();
            dlqHandler = null;
         }
      }
      catch (Exception e)
      {
         log.error("Failed to close the dlq handler", e);
      }

   }

   public void destroy() throws Exception
   {
      innerDestroy();
   }

   /**
    * An implementation of MessageListener that passes messages on to the
    * container invoker.
    */
   class MessageListenerImpl
           implements MessageListener
   {
      /**
       * The container invoker.
       */
      MDB invoker;
      // = null;

      /**
       * Construct a <tt>MessageListenerImpl</tt> .
       *
       * @param invoker The container invoker. Must not be null.
       */
      MessageListenerImpl(final MDB invoker)
      {
         // assert invoker != null;

         this.invoker = invoker;
      }

      /**
       * Process a message.
       *
       * @param message The message to process.
       */
      public void onMessage(final Message message)
      {
         // assert message != null;
         if (log.isTraceEnabled())
         {
            log.trace("processing message: " + message);
         }

         // Invoke, shuld we catch any Exceptions??
         try
         {
            Transaction tx = TxUtil.getTransactionManager().getTransaction();

            // DLQHandling
            if (config.isUseDLQ() && // Is Dead Letter Queue used at all
                    message.getJMSRedelivered() && // Was message resent
                    dlqHandler.handleRedeliveredMessage(message, tx)) //Did the DLQ handler take care of the message
            {
               // Message will be placed on Dead Letter Queue,
               // if redelivered to many times
               return;
            }
            invoker.localInvoke(ON_MESSAGE,
                    new Object[]{message});

         }
         catch (RuntimeException e)
         {
            throw e;
         }
         catch (Throwable e)
         {
            log.error("Exception in JMSCI message listener", e);
         }
      }
   }

   /**
    * ExceptionListener for failover handling.
    */
   class ExceptionListenerImpl
           implements ExceptionListener
   {
      MDB invoker;
      Thread currentThread;
      boolean notStoped = true;

      ExceptionListenerImpl(final MDB invoker)
      {
         this.invoker = invoker;
      }

      /**
       * #Description of the Method
       *
       * @param ex Description of Parameter
       */
      public void onException(JMSException ex)
      {
         currentThread = Thread.currentThread();

         log.warn("JMS provider failure detected: ", ex);
         boolean tryIt = true;
         while (tryIt && notStoped)
         {
            log.info("Trying to reconnect to JMS provider");
            try
            {
               try
               {
                  Thread.sleep(config.getReconnectInterval());
               }
               catch (InterruptedException ie)
               {
                  tryIt = false;
                  return;
               }

               // Reboot container
               invoker.innerStop();
               invoker.innerDestroy();
               invoker.innerStart();
               tryIt = false;

               log.info("Reconnected to JMS provider");
            }
            catch (Exception e)
            {
               log.error("Reconnect failed: JMS provider failure detected:", e);
            }
         }
         currentThread = null;
      }

      void stop()
      {
         log.debug("Stop requested");

         notStoped = false;
         if (currentThread != null)
         {
            currentThread.interrupt();
            log.debug("Current thread interrupted");
         }
      }
   }
}
