/*
* 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 java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Hashtable;
import javax.ejb.EJBException;
import javax.ejb.Timer;
import javax.ejb.TimerService;
import javax.ejb.TransactionAttribute;
import javax.ejb.TransactionAttributeType;
import javax.ejb.TransactionManagement;
import javax.ejb.TransactionManagementType;
import javax.jms.Connection;
import javax.jms.ConnectionConsumer;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.ServerSessionPool;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import javax.naming.Context;
import javax.naming.NamingException;
import javax.transaction.Transaction;
import org.jboss.annotation.ejb.AcknowledgementMode;
import org.jboss.annotation.ejb.Consumer;
import org.jboss.annotation.ejb.DefaultActivationSpecs;
import org.jboss.annotation.ejb.Durability;
import org.jboss.annotation.ejb.Local;
import org.jboss.annotation.ejb.MessageProperties;
import org.jboss.annotation.ejb.MessagePropertiesImpl;
import org.jboss.annotation.ejb.Producer;
import org.jboss.annotation.ejb.Producers;
import org.jboss.aop.AspectManager;
import org.jboss.aop.MethodInfo;
import org.jboss.aop.advice.Interceptor;
import org.jboss.aop.joinpoint.Invocation;
import org.jboss.aop.joinpoint.InvocationResponse;
import org.jboss.aop.joinpoint.MethodInvocation;
import org.jboss.aop.util.MethodHashing;
import org.jboss.aop.util.PayloadKey;
import org.jboss.deployment.DeploymentException;
import org.jboss.ejb.txtimer.TimedObjectInvoker;
import org.jboss.ejb3.Container;
import org.jboss.ejb3.EJBContainer;
import org.jboss.ejb3.EJBContainerInvocation;
import org.jboss.ejb3.Ejb3Module;
import org.jboss.ejb3.EjbTimerUtil;
import org.jboss.ejb3.ThreadLocalENCFactory;
import org.jboss.ejb3.Ejb3Deployment;
import org.jboss.ejb3.dd.EnterpriseBean;
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;

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

   protected TimerService timerService;
   protected ActivationSpec activationSpec = new ActivationSpec();
   protected boolean isContainerManagedTx;
   protected AcknowledgementMode acknowledgeMode;
   protected Connection connection;
   protected MDBConfig config;
   protected ExceptionListenerImpl exListener;
   protected DLQHandler dlqHandler;
   protected HashMap<Long, ConnectionConsumer> connectionConsumers = new HashMap<Long, ConnectionConsumer>();
   protected HashMap<Long, ServerSessionPool> serverSessionPools = new HashMap<Long, ServerSessionPool>();
   protected ArrayList<ProducerFactory> producers = new ArrayList<ProducerFactory>();

   /**
    * 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";


   /**
    * 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;
   public static final String CONSUMER_MESSAGE = "CONSUMER_MESSAGE";


   public ConsumerContainer(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;
         }
      }
   }


   /**
    * 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);

      try
      {
         innerCreate();
      }
      catch (final JMSException e)
      {
         log.error("FAILED INNER CREATE", 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();
      }
      timerService = EjbTimerUtil.getTimerService(this, this);

      registerProducers();
   }

   public Class[] getProducerInterfaces(Container container1)
   {
      Class beanClass = container1.getBeanClass();
      Class[] interfaces = beanClass.getInterfaces();
      if (interfaces.length == 0) throw new RuntimeException("Bean class must implement at least one interface: " + beanClass.getName());
      if (interfaces.length == 1)
      {
         return interfaces;
      }
      ArrayList localInterfaces = new ArrayList();
      for (int i = 0; i < interfaces.length; i++)
      {
         if (interfaces[i].isAnnotationPresent(Producer.class))
         {
            localInterfaces.add(interfaces[i]);
         }
      }
      Producer annotation = (Producer)resolveAnnotation(Producer.class);
      if (annotation != null)
      {
         Class producer = annotation.producer();
         if (producer != null)
            localInterfaces.add(producer);
      }
      
      Producers producersAnnotation = (Producers)resolveAnnotation(Producers.class);
      if (producersAnnotation != null)
      {
         for (Producer producerAnnotation : producersAnnotation.value())
         {
            Class producer = producerAnnotation.producer();
            if (producer != null)
               localInterfaces.add(producer);
         }
      }
      
      if (localInterfaces.size() == 0) return null;
      interfaces = (Class[]) localInterfaces.toArray(new Class[localInterfaces.size()]);
      return interfaces;
   }

   protected void registerProducers() throws Exception
   {
      Destination dest = (Destination) getInitialContext().lookup(config.getDestination());
      Class[] producers = getProducerInterfaces(this);
      MessageProperties props = (MessageProperties) resolveAnnotation(MessageProperties.class);
      if (props == null) props = new MessagePropertiesImpl();
      for (Class producer : producers)
      {
         log.debug("Producer: " + producer.getName());
         ProducerFactory producerFactory = null;
         if (producer.isAnnotationPresent(Local.class))
         {
            producerFactory = new LocalProducerFactory(this, producer, props, dest, getInitialContext(), initialContextProperties);
         }
         else
         {
            producerFactory = new RemoteProducerFactory(this, producer, props, dest, getInitialContext(), initialContextProperties);
         }
         this.producers.add(producerFactory);
         producerFactory.start();
      }
   }

   protected void unregisterProducers() throws Exception
   {
      for (ProducerFactory factory : producers)
      {
         factory.stop();
      }
   }

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

      Consumer Consumer = (Consumer) resolveAnnotation(Consumer.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;
      }


      // 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);

      registerQueueListeners(qConnection, queue);

   }

   private void registerQueueListeners(QueueConnection qConnection, Queue queue)
           throws NamingException, JMSException
   {
      MessageListener listener = new MessageListenerImpl(this);
      Class[] producers = getProducerInterfaces(this);
      HashSet visited = new HashSet();
      for (Class producer : producers)
      {
         for (Method method : producer.getMethods())
         {
            long hash = MethodHashing.calculateHash(method);
            if (visited.contains(hash)) continue;
            visited.add(hash);
            Method methodMessage = (Method) advisedMethods.get(hash);

            // set up the server session pool
            ServerSessionPool serverSessionPool = createSessionPool(queue,
                                                                    qConnection,
                                                                    true, // tx
                                                                    acknowledgeMode.ordinal(),
                                                                    listener,
                                                                    methodMessage);
            log.debug("Server session pool: " + serverSessionPool);
            serverSessionPools.put(hash, serverSessionPool);

            // create the connection consumer
            ConnectionConsumer connectionConsumer =
                    qConnection.createConnectionConsumer(queue,
                                                         config.getMessageSelector(),
                                                         serverSessionPool,
                                                         config.getMaxMessages());
            log.debug("Connection consumer: " + connectionConsumer);
            connectionConsumers.put(hash, 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);


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

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

      registerTopicListeners(tConnection, subscriptionDurability, topic);

   }

   private void registerTopicListeners(TopicConnection tConnection, Durability subscriptionDurability, Topic topic)
           throws NamingException, JMSException
   {
      MessageListener listener = new MessageListenerImpl(this);
      Class[] producers = getProducerInterfaces(this);
      HashSet visited = new HashSet();
      for (Class producer : producers)
      {
         for (Method method : producer.getMethods())
         {
            long hash = MethodHashing.calculateHash(method);
            if (visited.contains(hash)) continue;
            visited.add(hash);
            Method methodMessage = (Method) advisedMethods.get(hash);

            // set up the server session pool
            ServerSessionPool serverSessionPool = createSessionPool(topic,
                                                                    tConnection,
                                                                    true, // tx
                                                                    acknowledgeMode.ordinal(),
                                                                    listener,
                                                                    methodMessage);

            serverSessionPools.put(hash, serverSessionPool);
            // To be no-durable or durable
            ConnectionConsumer connectionConsumer = null;
            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());
            }
            connectionConsumers.put(hash, connectionConsumer);
            log.debug("Topic connectionConsumer set up");
         }
      }
   }

   /**
    * Create a server session pool for the given connection.
    *
    * @param dest         The destination to use
    * @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 javax.naming.NamingException Description of Exception
    */
   protected ServerSessionPool
           createSessionPool(final Destination dest,
                             final Connection connection,
                             final boolean isTransacted,
                             int ack,
                             final MessageListener listener,
                             Method method)
           throws NamingException, JMSException
   {
      ServerSessionPool pool;
      Context context = getInitialContext();

      try
      {
         // first lookup the factory
         log.debug("looking up session pool factory: " +
                   config.getServerSessionPoolFactoryJNDI());
         ServerSessionPoolFactory factory = (ServerSessionPoolFactory)
                 context.lookup(config.getServerSessionPoolFactoryJNDI());
         TransactionAttribute trans = (TransactionAttribute) resolveAnnotation(method, TransactionAttribute.class);
         TransactionAttributeType txType = trans == null ? TransactionAttributeType.REQUIRED : trans.value();

         boolean isNotSupportedTx = txType == TransactionAttributeType.NOT_SUPPORTED;

         // the create the pool
         // 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)
      {
         // 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 Topic)
            {
               destType = "javax.jms.Topic";
            }
            else if (dest instanceof 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();
      }
   }


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

      Consumer md = (Consumer) resolveAnnotation(Consumer.class);
      activationSpec.merge(md.activationConfig());
   }


   public Object localInvoke(Method method, Object[] args) throws Throwable
   {
      ClassLoader oldLoader = Thread.currentThread().getContextClassLoader();
      ThreadLocalENCFactory.push(enc);
      try
      {
         long hash = MethodHashing.calculateHash(method);
         MethodInfo info = (MethodInfo) methodInterceptors.get(hash);
         if (info == null)
         {
            throw new RuntimeException("Could not resolve beanClass method from proxy call");
         }
         Interceptor[] aspects1 = info.interceptors;
         EJBContainerInvocation nextInvocation = new EJBContainerInvocation(info, aspects1);
         nextInvocation.setAdvisor(this);
         nextInvocation.setArguments(args);
         return nextInvocation.invokeNext();
      }
      finally
      {
         Thread.currentThread().setContextClassLoader(oldLoader);
         ThreadLocalENCFactory.pop();
      }
   }

   public InvocationResponse dynamicInvoke(Invocation invocation) throws Throwable
   {
      ClassLoader oldLoader = Thread.currentThread().getContextClassLoader();
      EJBContainerInvocation newSi = null;
      ThreadLocalENCFactory.push(enc);
      try
      {
         Thread.currentThread().setContextClassLoader(classloader);
         MethodInvocation si = (MethodInvocation) invocation;
         MethodInfo info = (MethodInfo) methodInterceptors.get(si.getMethodHash());
         if (info == null)
         {
            throw new RuntimeException("Could not resolve beanClass method from proxy call");
         }
         Interceptor[] aspects = info.interceptors;
         newSi = new EJBContainerInvocation(info, aspects);
         newSi.setArguments(si.getArguments());
         newSi.setMetaData(si.getMetaData());
         newSi.setAdvisor(this);

         InvocationResponse response = new InvocationResponse(newSi.invokeNext());
         response.setContextInfo(newSi.getResponseContextInfo());
         return response;
      }
      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
   {
      unregisterProducers();
      EjbTimerUtil.removeTimerService(this);
      // Silence the exception listener
      if (exListener != null)
      {
         exListener.stop();
      }

      innerStop();

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

   }

   public void destroy() throws Exception
   {

      // close the connection consumer
      try
      {
         for (ConnectionConsumer consumer : connectionConsumers.values())
         {
            consumer.close();
         }
      }
      catch (Exception e)
      {
         log.error("Failed to close connection consumer", e);
      }

      // clear the server session pool (if it is clearable)
      try
      {
         for (ServerSessionPool ssp : serverSessionPools.values())
         {

            if (ssp instanceof StdServerSessionPool)
            {
               StdServerSessionPool p = (StdServerSessionPool) ssp;
               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);
      }

   }

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

      /**
       * Construct a <tt>MessageListenerImpl</tt> .
       *
       * @param invoker The container invoker. Must not be null.
       */
      MessageListenerImpl(final ConsumerContainer 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;
            }
            Invocation invocation = (Invocation) ((ObjectMessage) message).getObject();
            invocation.getMetaData().addMetaData(CONSUMER_MESSAGE, CONSUMER_MESSAGE, message, PayloadKey.TRANSIENT);
            invoker.dynamicInvoke(invocation);

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

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

      ExceptionListenerImpl(final ConsumerContainer 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.destroy();
               invoker.start();
               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");
         }
      }
   }
}
