/*
    * $Id: AbstractConnector.java 10713 2008-02-06 11:25:16Z holger $
    * --------------------------------------------------------------------------------------
    * Copyright (c) MuleSource, Inc.  All rights reserved.  http://www.mulesource.com
    *
    * The software in this package is published under the terms of the CPAL v1.0
    * license, a copy of which has been included with this distribution in the
    * LICENSE.txt file.
    */
  
  package org.mule.providers;
  
  import org.mule.MuleManager;
  import org.mule.MuleRuntimeException;
  import org.mule.config.ThreadingProfile;
  import org.mule.config.i18n.CoreMessages;
  import org.mule.impl.AlreadyInitialisedException;
  import org.mule.impl.DefaultExceptionStrategy;
  import org.mule.impl.ImmutableMuleEndpoint;
  import org.mule.impl.MuleSessionHandler;
  import org.mule.impl.internal.notifications.ConnectionNotification;
  import org.mule.providers.service.TransportFactory;
  import org.mule.providers.service.TransportServiceDescriptor;
  import org.mule.providers.service.TransportServiceException;
  import org.mule.routing.filters.WildcardFilter;
  import org.mule.umo.MessagingException;
  import org.mule.umo.UMOComponent;
  import org.mule.umo.UMOEvent;
  import org.mule.umo.UMOException;
  import org.mule.umo.UMOMessage;
  import org.mule.umo.endpoint.UMOEndpoint;
  import org.mule.umo.endpoint.UMOEndpointURI;
  import org.mule.umo.endpoint.UMOImmutableEndpoint;
  import org.mule.umo.lifecycle.DisposeException;
  import org.mule.umo.lifecycle.Initialisable;
  import org.mule.umo.lifecycle.InitialisationException;
  import org.mule.umo.manager.UMOServerNotification;
  import org.mule.umo.manager.UMOWorkManager;
  import org.mule.umo.provider.ConnectorException;
  import org.mule.umo.provider.DispatchException;
  import org.mule.umo.provider.UMOConnectable;
  import org.mule.umo.provider.UMOConnector;
  import org.mule.umo.provider.UMOMessageAdapter;
  import org.mule.umo.provider.UMOMessageDispatcher;
  import org.mule.umo.provider.UMOMessageDispatcherFactory;
  import org.mule.umo.provider.UMOMessageReceiver;
  import org.mule.umo.provider.UMOSessionHandler;
  import org.mule.umo.provider.UMOStreamMessageAdapter;
  import org.mule.umo.transformer.UMOTransformer;
  import org.mule.util.ClassUtils;
  import org.mule.util.CollectionUtils;
  import org.mule.util.ObjectNameHelper;
  import org.mule.util.ObjectUtils;
  import org.mule.util.PropertiesUtils;
  import org.mule.util.StringUtils;
  import org.mule.util.concurrent.NamedThreadFactory;
  import org.mule.util.concurrent.WaitableBoolean;
  
  import java.beans.ExceptionListener;
  import java.io.InputStream;
  import java.io.OutputStream;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.Properties;
  
  import javax.resource.spi.work.WorkEvent;
  import javax.resource.spi.work.WorkListener;
  
  import edu.emory.mathcs.backport.java.util.concurrent.ConcurrentHashMap;
  import edu.emory.mathcs.backport.java.util.concurrent.ConcurrentMap;
  import edu.emory.mathcs.backport.java.util.concurrent.ScheduledExecutorService;
  import edu.emory.mathcs.backport.java.util.concurrent.ScheduledThreadPoolExecutor;
  import edu.emory.mathcs.backport.java.util.concurrent.ThreadFactory;
  import edu.emory.mathcs.backport.java.util.concurrent.TimeUnit;
  import edu.emory.mathcs.backport.java.util.concurrent.atomic.AtomicBoolean;
  import edu.emory.mathcs.backport.java.util.concurrent.atomic.AtomicReference;
  
  import org.apache.commons.beanutils.BeanUtils;
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  import org.apache.commons.pool.KeyedPoolableObjectFactory;
  import org.apache.commons.pool.impl.GenericKeyedObjectPool;
  
  /**
   * <code>AbstractConnector</code> provides base functionality for all connectors
   * provided with Mule. Connectors are the mechanism used to connect to external
   * systems and protocols in order to send and receive data. <p/> The
   * <code>AbstractConnector</code> provides getter and setter methods for endpoint
   * name, transport name and protocol. It also provides methods to stop and start
   * connecotors and sets up a dispatcher threadpool which allows deriving connectors
   * the possibility to dispatch work to separate threads. This functionality is
   * controlled with the <i> doThreading</i> property on the threadingProfiles for
   * dispachers and receivers. The lifecycle for a connector is -
   * <ol>
   * <li>Create
  * <li>Initialise
  * <li>Connect
  * <li>Connect receivers
  * <li>Start
  * <li>Start Receivers
  * <li>Stop
  * <li>Stop Receivers
  * <li>Disconnect
  * <li>Disconnect Receivers
  * <li>Dispose
  * <li>Dispose Receivers
  * </ol>
  */
 public abstract class AbstractConnector
     implements UMOConnector, ExceptionListener, UMOConnectable, WorkListener
 {
     /**
      * Default number of concurrent transactional receivers.
      */
     public static final int DEFAULT_NUM_CONCURRENT_TX_RECEIVERS = 4;
 
     /**
      * logger used by this class
      */
     protected final Log logger = LogFactory.getLog(getClass());
 
     /**
      * Specifies if the endpoint started
      */
     protected final AtomicBoolean started = new AtomicBoolean(false);
 
     /**
      * True once the endpoint has been initialsed
      */
     protected final AtomicBoolean initialised = new AtomicBoolean(false);
 
     /**
      * The name that identifies the endpoint
      */
     protected volatile String name;
 
     /**
      * The exception strategy used by this connector
      */
     protected volatile ExceptionListener exceptionListener;
 
     /**
      * Determines in the connector is alive and well
      */
     protected final AtomicBoolean disposed = new AtomicBoolean(false);
 
     /**
      * Determines in connector has been told to dispose
      */
     protected final AtomicBoolean disposing = new AtomicBoolean(false);
 
     /**
      * Factory used to create dispatchers for this connector
      */
     protected volatile UMOMessageDispatcherFactory dispatcherFactory;
 
     /**
      * A pool of dispatchers for this connector, keyed by endpoint
      */
     protected final GenericKeyedObjectPool dispatchers = new GenericKeyedObjectPool();
 
     /**
      * The collection of listeners on this connector. Keyed by entrypoint
      */
     protected final ConcurrentMap receivers = new ConcurrentHashMap();
 
     /**
      * Defines the dispatcher threading profile
      */
     private volatile ThreadingProfile dispatcherThreadingProfile = MuleManager.getConfiguration()
         .getMessageDispatcherThreadingProfile();
 
     /**
      * Defines the receiver threading profile
      */
     private volatile ThreadingProfile receiverThreadingProfile = MuleManager.getConfiguration()
         .getMessageReceiverThreadingProfile();
 
     /**
      * @see {@link #isCreateMultipleTransactedReceivers()}
      */
     protected volatile boolean createMultipleTransactedReceivers = true;
 
     /**
      * @see {@link #getNumberOfConcurrentTransactedReceivers()}
      */
     protected volatile int numberOfConcurrentTransactedReceivers = DEFAULT_NUM_CONCURRENT_TX_RECEIVERS;
 
     /**
      * The service descriptor can define a default inbound transformer to be used on
      * an endpoint if no other is set
      */
     protected volatile UMOTransformer defaultInboundTransformer;
 
     /**
      * The service descriptor can define a default outbound transformer to be used on
      * an endpoint if no other is set
      */
     protected volatile UMOTransformer defaultOutboundTransformer;
 
     /**
      * For some connectors such as http, a response transformer is required or where
      * a replyTo needs a trnasformer
      */
     protected volatile UMOTransformer defaultResponseTransformer;
 
     protected volatile ConnectionStrategy connectionStrategy;
 
     protected final WaitableBoolean connected = new WaitableBoolean(false);
 
     protected final WaitableBoolean connecting = new WaitableBoolean(false);
 
     /**
      * If the connect method was called via the start method, this will be set so
      * that when the connector comes on line it will be started
      */
     protected final WaitableBoolean startOnConnect = new WaitableBoolean(false);
 
     /**
      * Whether to fire message notifications for every message that is sent or
      * received from this connector. Default is {@code false}.
      */
     private volatile boolean enableMessageEvents;
 
     private final List supportedProtocols;
 
     /**
      * A shared work manager for all receivers registered with this connector.
      */
     private final AtomicReference/*<UMOWorkManager>*/ receiverWorkManager = new AtomicReference();
 
     /**
      * A shared work manager for all dispatchers created for this connector.
      */
     private final AtomicReference/*<UMOWorkManager>*/ dispatcherWorkManager = new AtomicReference();
 
     /**
      * A generic scheduling service for tasks that need to be performed periodically.
      */
     private final AtomicReference/*<ScheduledExecutorService>*/ scheduler = new AtomicReference();
 
     /**
      * Holds the service configuration for this connector
      */
     protected volatile TransportServiceDescriptor serviceDescriptor;
 
     /**
      * The map of service overrides that can e used to extend the capabilities of the
      * connector
      */
     protected volatile Properties serviceOverrides;
 
     /**
      * The strategy used for reading and writing session information to and fromt he
      * transport
      */
     protected volatile UMOSessionHandler sessionHandler = new MuleSessionHandler();
 
     /** Constructs a new AbstractConnector. */
     public AbstractConnector()
     {
         super();
 
         // make sure we always have an exception strategy
         exceptionListener = new DefaultExceptionStrategy();
         connectionStrategy = MuleManager.getConfiguration().getConnectionStrategy();
         enableMessageEvents = MuleManager.getConfiguration().isEnableMessageEvents();
 
         // always add at least the default protocol
         supportedProtocols = new ArrayList();
         supportedProtocols.add(getProtocol().toLowerCase());
 
         // NOTE: testOnBorrow MUST be FALSE. this is a bit of a design bug in
         // commons-pool since validate is used for both activation and passivation,
         // but has no way of knowing which way it is going.
         dispatchers.setTestOnBorrow(false);
         dispatchers.setTestOnReturn(true);
     }
 
     /** {@inheritDoc} */
     public String getName()
     {
         return name;
     }
 
     /** {@inheritDoc} */
     public void setName(String newName)
     {
         if (newName == null)
         {
             throw new IllegalArgumentException(CoreMessages.objectIsNull("Connector name").toString());
         }
 
         if (logger.isDebugEnabled())
         {
             logger.debug("Set UMOConnector name to: " + newName);
         }
 
         name = newName;
     }
 
     /** {@inheritDoc} */
     public final synchronized void initialise() throws InitialisationException
     {
         if (initialised.get())
         {
             throw new AlreadyInitialisedException("Connector '" + getName() + "'", this);
         }
 
         if (logger.isInfoEnabled())
         {
             logger.info("Initialising: " + this);
         }
 
         // Initialise the structure of this connector
         this.initFromServiceDescriptor();
 
         // we clear out any registered dispatchers and receivers without resetting
         // the actual containers since this it might actually be a re-initialise
         // (e.g. as in JmsConnector)
         this.disposeDispatchers();
         this.disposeReceivers();
 
         this.doInitialise();
 
         if (exceptionListener instanceof Initialisable)
         {
             ((Initialisable) exceptionListener).initialise();
         }
 
         initialised.set(true);
     }
 
     /** {@inheritDoc} */
     public final synchronized void startConnector() throws UMOException
     {
         this.checkDisposed();
 
         if (!this.isStarted())
         {
             if (!this.isConnected())
             {
                 startOnConnect.set(true);
                 // Don't call getConnectionStrategy(), it clones the connection strategy.
                 // Connectors should have a single reconnection thread, unlike per receiver/dispatcher
                 connectionStrategy.connect(this);
                 // Only start once we are connected
                 return;
             }
 
             if (logger.isInfoEnabled())
             {
                 logger.info("Starting: " + this);
             }
 
             // the scheduler is recreated after stopConnector()
             ScheduledExecutorService currentScheduler = (ScheduledExecutorService) scheduler.get();
             if (currentScheduler == null || currentScheduler.isShutdown())
             {
                 scheduler.set(this.getScheduler());
             }
 
             this.doStart();
             started.set(true);
 
             if (receivers != null)
             {
                 for (Iterator iterator = receivers.values().iterator(); iterator.hasNext();)
                 {
                     UMOMessageReceiver mr = (UMOMessageReceiver) iterator.next();
                     if (logger.isDebugEnabled())
                     {
                         logger.debug("Starting receiver on endpoint: " + mr.getEndpoint().getEndpointURI());
                     }
                     mr.start();
                 }
             }
 
             if (logger.isInfoEnabled())
             {
                 logger.info("Started: " + this);
             }
         }
     }
 
     /** {@inheritDoc} */
     public boolean isStarted()
     {
         return started.get();
     }
 
     /** {@inheritDoc} */
     public final synchronized void stopConnector() throws UMOException
     {
         if (this.isDisposed())
         {
             return;
         }
 
         if (this.isStarted())
         {
             if (logger.isInfoEnabled())
             {
                 logger.info("Stopping: " + this);
             }
 
             // shutdown our scheduler service
             ((ScheduledExecutorService) scheduler.get()).shutdown();
 
             this.doStop();
             started.set(false);
 
             // Stop all the receivers on this connector (this will cause them to
             // disconnect too)
             if (receivers != null)
             {
                 for (Iterator iterator = receivers.values().iterator(); iterator.hasNext();)
                 {
                     UMOMessageReceiver mr = (UMOMessageReceiver) iterator.next();
                     if (logger.isDebugEnabled())
                     {
                         logger.debug("Stopping receiver on endpoint: " + mr.getEndpoint().getEndpointURI());
                     }
                     mr.stop();
                 }
             }
         }
         
         if (this.isConnected())
         {
             try
             {
                 this.disconnect();
             }
             catch (Exception e)
             {
                 // TODO MULE-863: What should we really do?
                 logger.error("Failed to disconnect: " + e.getMessage(), e);
             }
         }
 
         // make sure the scheduler is gone
         scheduler.set(null);
 
         // we do not need to stop the work managers because they do no harm (will just be idle)
         // and will be reused on restart without problems.
         
         this.initialised.set(false);
         // started=false already issued above right after doStop()
         if (logger.isInfoEnabled())
         {
             logger.info("Stopped: " + this);
         }
     }
 
     /** {@inheritDoc} */
     public final synchronized void dispose()
     {
         disposing.set(true);
 
         if (logger.isInfoEnabled())
         {
             logger.info("Disposing: " + this);
         }
 
         try
         {
             this.stopConnector();
         }
         catch (UMOException e)
         {
             logger.warn("Failed to stop during shutdown: " + e.getMessage(), e);
         }
 
         this.disposeReceivers();
         this.disposeDispatchers();
         this.disposeWorkManagers();
 
         this.doDispose();        
         disposed.set(true);
 
         if (logger.isInfoEnabled())
         {
             logger.info("Disposed: " + this);
         }
     }
 
     protected void disposeWorkManagers()
     {
         logger.debug("Disposing dispatcher work manager");
         UMOWorkManager workManager = (UMOWorkManager) dispatcherWorkManager.get();
         if (workManager != null)
         {
             workManager.dispose();
         }
         dispatcherWorkManager.set(null);
 
         logger.debug("Disposing receiver work manager");
         workManager = (UMOWorkManager) receiverWorkManager.get();
         if (workManager != null)
         {
             workManager.dispose();
         }
         receiverWorkManager.set(null);
     }
 
     protected void disposeReceivers()
     {
         if (receivers != null)
         {
             logger.debug("Disposing Receivers");
 
             for (Iterator iterator = receivers.values().iterator(); iterator.hasNext();)
             {
                 UMOMessageReceiver receiver = (UMOMessageReceiver) iterator.next();
 
                 try
                 {
                     this.destroyReceiver(receiver, receiver.getEndpoint());
                 }
                 catch (Throwable e)
                 {
                     // TODO MULE-863: What should we really do?
                     logger.error("Failed to destroy receiver: " + receiver, e);
                 }
             }
 
             receivers.clear();
             logger.debug("Receivers Disposed");
         }
     }
 
     protected void disposeDispatchers()
     {
         if (dispatchers != null)
         {
             logger.debug("Disposing Dispatchers");
             dispatchers.clear();
             logger.debug("Dispatchers Disposed");
         }
     }
 
     /** {@inheritDoc} */
     public boolean isDisposed()
     {
         return disposed.get();
     }
 
     /** {@inheritDoc} */
     public void handleException(Exception exception)
     {
         if (exceptionListener == null)
         {
             throw new MuleRuntimeException(
                 CoreMessages.exceptionOnConnectorNotExceptionListener(this.getName()), exception);
         }
         else
         {
             exceptionListener.exceptionThrown(exception);
         }
     }
 
     /** {@inheritDoc} */
     public void exceptionThrown(Exception e)
     {
         handleException(e);
     }
 
     /**
      * @return the ExceptionStrategy for this endpoint
      * @see ExceptionListener
      */
     public ExceptionListener getExceptionListener()
     {
         return exceptionListener;
     }
 
     /**
      * @param listener the ExceptionStrategy to use with this endpoint
      * @see ExceptionListener
      */
     public void setExceptionListener(ExceptionListener listener)
     {
         exceptionListener = listener;
     }
 
     /**
      * @return Returns the dispatcherFactory.
      */
     public UMOMessageDispatcherFactory getDispatcherFactory()
     {
         return dispatcherFactory;
     }
 
     /**
      * @param dispatcherFactory The dispatcherFactory to set.
      */
     public void setDispatcherFactory(UMOMessageDispatcherFactory dispatcherFactory)
     {
         KeyedPoolableObjectFactory poolFactory;
 
         if (dispatcherFactory instanceof KeyedPoolableObjectFactory)
         {
             poolFactory = (KeyedPoolableObjectFactory) dispatcherFactory;
         }
         else
         {
             // need to adapt the UMOMessageDispatcherFactory for use by commons-pool
             poolFactory = new KeyedPoolMessageDispatcherFactoryAdapter(dispatcherFactory);
         }
 
         this.dispatchers.setFactory(poolFactory);
 
         // we keep a reference to the unadapted factory, otherwise people might end
         // up with ClassCastExceptions on downcast to their implementation (sigh)
         this.dispatcherFactory = dispatcherFactory;
     }
 
     /**
      * Returns the maximum number of dispatchers that can be concurrently active per
      * endpoint.
      *
      * @return max. number of active dispatchers
      */
     public int getMaxDispatchersActive()
     {
         return this.dispatchers.getMaxActive();
     }
 
     /**
      * Configures the maximum number of dispatchers that can be concurrently active
      * per endpoint
      *
      * @param maxActive max. number of active dispatchers
      */
     public void setMaxDispatchersActive(int maxActive)
     {
         this.dispatchers.setMaxActive(maxActive);
         // adjust maxIdle in tandem to avoid thrashing
         this.dispatchers.setMaxIdle(maxActive);
     }
 
     private UMOMessageDispatcher getDispatcher(UMOImmutableEndpoint endpoint) throws UMOException
     {
         this.checkDisposed();
 
         if (endpoint == null)
         {
             throw new IllegalArgumentException("Endpoint must not be null");
         }
 
         if (!this.supportsProtocol(endpoint.getConnector().getProtocol()))
         {
             throw new IllegalArgumentException(
                 CoreMessages.connectorSchemeIncompatibleWithEndpointScheme(this.getProtocol(), 
                     endpoint.getEndpointURI().toString()).getMessage());
         }
 
         try
         {
             if (logger.isDebugEnabled())
             {
                 logger.debug("Borrowing a dispatcher for endpoint: " + endpoint.getEndpointURI());
             }
         }
         catch (Exception ex)
         {
             throw new ConnectorException(CoreMessages.connectorCausedError(), this, ex);
         }
 
         UMOMessageDispatcher dispatcher = null;
 
         try
         {
             dispatcher = (UMOMessageDispatcher) dispatchers.borrowObject(endpoint);
             return dispatcher;
         }
         catch (Exception ex)
         {
             throw new ConnectorException(CoreMessages.connectorCausedError(), this, ex);
         }
         finally
         {
             try
             {
                 if (logger.isDebugEnabled())
                 {
                     logger.debug("Borrowed dispatcher: " + ObjectUtils.toString(dispatcher, "null"));
                 }
             }
             catch (Exception ex)
             {
                 throw new ConnectorException(CoreMessages.connectorCausedError(), this, ex);
             }
         }
     }
 
     private void returnDispatcher(UMOImmutableEndpoint endpoint, UMOMessageDispatcher dispatcher)
     {
         if (endpoint != null && dispatcher != null)
         {
             try
             {
                 if (logger.isDebugEnabled())
                 {
                     logger.debug("Returning dispatcher for endpoint: " + endpoint.getEndpointURI() + " = "
                                     + ObjectUtils.toString(dispatcher, "null"));
                 }
             }
             catch (Exception ex)
             {
                 // TODO MULE-863: What should we really do?
                 // ignore - if the dispatcher is broken, it will likely get cleaned
                 // up by the factory
             }
             finally
             {
                 try
                 {
                     dispatchers.returnObject(endpoint, dispatcher);
                 }
                 catch (Exception ex)
                 {
                     // TODO MULE-863: What should we really do?
                     // returnObject() never throws; this will go away in commons-pool
                 }
             }
         }
     }
 
     protected void checkDisposed() throws DisposeException
     {
         if (this.isDisposed())
         {
             throw new DisposeException(CoreMessages.cannotUseDisposedConnector(), this);
         }
     }
 
     /** {@inheritDoc} */
     public UMOMessageReceiver registerListener(UMOComponent component, UMOEndpoint endpoint) throws Exception
     {
         if (endpoint == null)
         {
             throw new IllegalArgumentException("The endpoint cannot be null when registering a listener");
         }
 
         if (component == null)
         {
             throw new IllegalArgumentException("The component cannot be null when registering a listener");
         }
 
         UMOEndpointURI endpointUri = endpoint.getEndpointURI();
         if (endpointUri == null)
         {
             throw new ConnectorException(CoreMessages.endpointIsNullForListener(), this);
         }
 
         logger.info("Registering listener: " + component.getDescriptor().getName() + " on endpointUri: "
                         + endpointUri.toString());
 
         UMOMessageReceiver receiver = this.getReceiver(component, endpoint);
 
         if (receiver != null)
         {
             throw new ConnectorException(CoreMessages.listenerAlreadyRegistered(endpointUri), this);
         }
         else
         {
             receiver = this.createReceiver(component, endpoint);
             Object receiverKey = getReceiverKey(component, endpoint);
             receiver.setReceiverKey(receiverKey.toString());
             receivers.put(receiverKey, receiver);
             // receivers.put(getReceiverKey(component, endpoint), receiver);
         }
 
         return receiver;
     }
 
     /**
      * The method determines the key used to store the receiver against.
      *
      * @param component the component for which the endpoint is being registered
      * @param endpoint the endpoint being registered for the component
      * @return the key to store the newly created receiver against
      */
     protected Object getReceiverKey(UMOComponent component, UMOEndpoint endpoint)
     {
         return StringUtils.defaultIfEmpty(endpoint.getEndpointURI().getFilterAddress(), endpoint
             .getEndpointURI().getAddress());
     }
 
     /** {@inheritDoc} */
     public final void unregisterListener(UMOComponent component, UMOEndpoint endpoint) throws Exception
     {
         if (component == null)
         {
             throw new IllegalArgumentException(
                 "The component must not be null when you unregister a listener");
         }
 
         if (endpoint == null)
         {
             throw new IllegalArgumentException("The endpoint must not be null when you unregister a listener");
         }
 
         UMOEndpointURI endpointUri = endpoint.getEndpointURI();
         if (endpointUri == null)
         {
             throw new IllegalArgumentException(
                 "The endpointUri must not be null when you unregister a listener");
         }
 
         if (logger.isInfoEnabled())
         {
             logger.info("Removing listener on endpointUri: " + endpointUri);
         }
 
         if (receivers != null && !receivers.isEmpty())
         {
             UMOMessageReceiver receiver = (UMOMessageReceiver) receivers.remove(this.getReceiverKey(component,
                 endpoint));
             if (receiver != null)
             {
                 this.destroyReceiver(receiver, endpoint);
             }
         }
     }
 
     /**
      * Getter for property 'dispatcherThreadingProfile'.
      *
      * @return Value for property 'dispatcherThreadingProfile'.
      */
     public ThreadingProfile getDispatcherThreadingProfile()
     {
         return dispatcherThreadingProfile;
     }
 
     /**
      * Setter for property 'dispatcherThreadingProfile'.
      *
      * @param dispatcherThreadingProfile Value to set for property
      *            'dispatcherThreadingProfile'.
      */
     public void setDispatcherThreadingProfile(ThreadingProfile dispatcherThreadingProfile)
     {
         this.dispatcherThreadingProfile = dispatcherThreadingProfile;
     }
 
     /**
      * Getter for property 'receiverThreadingProfile'.
      *
      * @return Value for property 'receiverThreadingProfile'.
      */
     public ThreadingProfile getReceiverThreadingProfile()
     {
         return receiverThreadingProfile;
     }
 
     /**
      * Setter for property 'receiverThreadingProfile'.
      *
      * @param receiverThreadingProfile Value to set for property
      *            'receiverThreadingProfile'.
      */
     public void setReceiverThreadingProfile(ThreadingProfile receiverThreadingProfile)
     {
         this.receiverThreadingProfile = receiverThreadingProfile;
     }
 
     protected void destroyReceiver(UMOMessageReceiver receiver, UMOEndpoint endpoint) throws Exception
     {
         receiver.dispose();
     }
 
     protected abstract void doInitialise() throws InitialisationException;
 
     /**
      * Template method to perform any work when destroying the connectoe
      */
     protected abstract void doDispose();
 
     /**
      * Template method to perform any work when starting the connectoe
      *
      * @throws UMOException if the method fails
      */
     protected abstract void doStart() throws UMOException;
 
     /**
      * Template method to perform any work when stopping the connectoe
      *
      * @throws UMOException if the method fails
      */
     protected abstract void doStop() throws UMOException;
 
     /**
      * Getter for property 'defaultInboundTransformer'.
      *
      * @return Value for property 'defaultInboundTransformer'.
      */
     public UMOTransformer getDefaultInboundTransformer()
     {
         if (defaultInboundTransformer != null)
         {
             try
             {
                 return (UMOTransformer) defaultInboundTransformer.clone();
             }
             catch (CloneNotSupportedException e)
             {
                 // TODO MULE-863: What should we really do?
                 logger.error("Failed to clone default Inbound transformer");
             }
         }
 
         return null;
     }
 
     /**
      * Setter for property 'defaultInboundTransformer'.
      *
      * @param defaultInboundTransformer Value to set for property
      *            'defaultInboundTransformer'.
      */
     public void setDefaultInboundTransformer(UMOTransformer defaultInboundTransformer)
     {
         this.defaultInboundTransformer = defaultInboundTransformer;
     }
 
     /**
      * Getter for property 'defaultResponseTransformer'.
      *
      * @return Value for property 'defaultResponseTransformer'.
      */
     public UMOTransformer getDefaultResponseTransformer()
     {
         if (defaultResponseTransformer != null)
         {
             try
             {
                 return (UMOTransformer) defaultResponseTransformer.clone();
             }
             catch (CloneNotSupportedException e)
             {
                 // TODO MULE-863: What should we really do?
                 logger.error("Failed to clone default Outbound transformer");
             }
         }
 
         return null;
     }
 
     /**
      * Getter for property 'defaultOutboundTransformer'.
      *
      * @return Value for property 'defaultOutboundTransformer'.
      */
     public UMOTransformer getDefaultOutboundTransformer()
     {
         if (defaultOutboundTransformer != null)
         {
             try
             {
                 return (UMOTransformer) defaultOutboundTransformer.clone();
             }
             catch (CloneNotSupportedException e)
             {
                 // TODO MULE-863: What should we really do?
                 logger.error("Failed to clone default Outbound transformer");
             }
         }
 
         return null;
     }
 
     /**
      * Setter for property 'defaultOutboundTransformer'.
      *
      * @param defaultOutboundTransformer Value to set for property
      *            'defaultOutboundTransformer'.
      */
     public void setDefaultOutboundTransformer(UMOTransformer defaultOutboundTransformer)
     {
         this.defaultOutboundTransformer = defaultOutboundTransformer;
     }
 
     /**
      * Setter for property 'defaultResponseTransformer'.
      *
      * @param defaultResponseTransformer Value to set for property
      *            'defaultResponseTransformer'.
      */
     public void setDefaultResponseTransformer(UMOTransformer defaultResponseTransformer)
     {
         this.defaultResponseTransformer = defaultResponseTransformer;
     }
 
     /**
      * Getter for property 'replyToHandler'.
      *
      * @return Value for property 'replyToHandler'.
      */
     public ReplyToHandler getReplyToHandler()
     {
         return new DefaultReplyToHandler(defaultResponseTransformer);
     }
 
     /**
      * Fires a server notification to all registered
      * {@link org.mule.impl.internal.notifications.CustomNotificationListener}
      * eventManager.
      *
      * @param notification the notification to fire. This must be of type
      *            {@link org.mule.impl.internal.notifications.CustomNotification}
      *            otherwise an exception will be thrown.
      * @throws UnsupportedOperationException if the notification fired is not a
      *             {@link org.mule.impl.internal.notifications.CustomNotification}
      */
     public void fireNotification(UMOServerNotification notification)
     {
         MuleManager.getInstance().fireNotification(notification);
     }
 
     /**
      * Getter for property 'connectionStrategy'.
      *
      * @return Value for property 'connectionStrategy'.
      */
     public ConnectionStrategy getConnectionStrategy()
     {
         // not happy with this but each receiver needs its own instance
         // of the connection strategy and using a factory just introduces extra
         // implementation
         try
         {
             return (ConnectionStrategy) BeanUtils.cloneBean(connectionStrategy);
         }
         catch (Exception e)
         {
             throw new MuleRuntimeException(CoreMessages.failedToClone("connectionStrategy"), e);
         }
     }
 
     /**
      * Setter for property 'connectionStrategy'.
      *
      * @param connectionStrategy Value to set for property 'connectionStrategy'.
      */
     public void setConnectionStrategy(ConnectionStrategy connectionStrategy)
     {
         this.connectionStrategy = connectionStrategy;
     }
 
     /** {@inheritDoc} */
     public boolean isDisposing()
     {
         return disposing.get();
     }
 
     /** {@inheritDoc} */
     public boolean isRemoteSyncEnabled()
     {
         return false;
     }
 
     public UMOMessageReceiver getReceiver(UMOComponent component, UMOEndpoint endpoint)
     {
         return (UMOMessageReceiver) receivers.get(this.getReceiverKey(component, endpoint));
     }
 
     /**
      * Getter for property 'receivers'.
      *
      * @return Value for property 'receivers'.
      */
     public Map getReceivers()
     {
         return Collections.unmodifiableMap(receivers);
     }
 
     public UMOMessageReceiver lookupReceiver(String key)
     {
         if (key != null)
         {
             return (UMOMessageReceiver) receivers.get(key);
         }
         else
         {
             throw new IllegalArgumentException("Receiver key must not be null");
         }
     }
 
     public UMOMessageReceiver[] getReceivers(String wildcardExpression)
     {
         WildcardFilter filter = new WildcardFilter(wildcardExpression);
         filter.setCaseSensitive(false);
 
         List found = new ArrayList();
 
         for (Iterator iterator = receivers.entrySet().iterator(); iterator.hasNext();)
         {
             Map.Entry e = (Map.Entry) iterator.next();
             if (filter.accept(e.getKey()))
             {
                 found.add(e.getValue());
             }
         }
 
         return (UMOMessageReceiver[]) CollectionUtils.toArrayOfComponentType(found,
             UMOMessageReceiver.class);
     }
 
     /** {@inheritDoc} */
     public void connect() throws Exception
     {
         this.checkDisposed();
 
         if (connected.get())
         {
             return;
         }
 
         /*
             Until the recursive startConnector() -> connect() -> doConnect() -> connect()
             calls are unwound between a connector and connection strategy, this call has
             to be here, and not below (commented out currently). Otherwise, e.g. WebspherMQ
             goes into an endless reconnect thrashing loop, see MULE-1150 for more details.
         */
         try
         {
             if (connecting.get())
             {
                 this.doConnect();
             }
 
             if (connecting.compareAndSet(false, true))
             {
                 if (logger.isDebugEnabled())
                 {
                     logger.debug("Connecting: " + this);
                 }
 
                 connectionStrategy.connect(this);
 
                 logger.info("Connected: " + getConnectionDescription());
                 // This method calls itself so the connecting flag is set first, then
                 // the connection is made on the second call
                 return;
             }
 
 
             // see the explanation above
             //this.doConnect();
             connected.set(true);
             connecting.set(false);
 
             this.fireNotification(new ConnectionNotification(this, getConnectEventId(),
                 ConnectionNotification.CONNECTION_CONNECTED));
         }
         catch (Exception e)
         {
             connected.set(false);
             connecting.set(false);
 
             this.fireNotification(new ConnectionNotification(this, getConnectEventId(),
                 ConnectionNotification.CONNECTION_FAILED));
 
             if (e instanceof ConnectException || e instanceof FatalConnectException)
             {
                 // rethrow
                 throw e;
             }
             else
             {
                 throw new ConnectException(e, this);
             }
         }
 
         if (startOnConnect.get())
         {
             this.startConnector();
         }
         else
         {
             for (Iterator iterator = receivers.values().iterator(); iterator.hasNext();)
             {
                 UMOMessageReceiver receiver = (UMOMessageReceiver) iterator.next();
                 if (logger.isDebugEnabled())
                 {
                     logger.debug("Connecting receiver on endpoint: "
                                     + receiver.getEndpoint().getEndpointURI());
                 }
                 receiver.connect();
             }
         }
     }
 
     /** {@inheritDoc} */
     public void disconnect() throws Exception
     {
         startOnConnect.set(this.isStarted());
 
         this.fireNotification(new ConnectionNotification(this, getConnectEventId(),
             ConnectionNotification.CONNECTION_DISCONNECTED));
 
         connected.set(false);
 
         try
         {
             this.doDisconnect();
         }
         finally
         {
             this.stopConnector();
         }
 
         logger.info("Disconnected: " + this.getConnectionDescription());
     }
 
     /** {@inheritDoc} */
     public String getConnectionDescription()
     {
         return this.toString();
     }
 
     /** {@inheritDoc} */
     public final boolean isConnected()
     {
         return connected.get();
     }
 
     /**
      * Template method where any connections should be made for the connector
      *
      * @throws Exception
      */
     protected abstract void doConnect() throws Exception;
 
     /**
      * Template method where any connected resources used by the connector should be
      * disconnected
      *
      * @throws Exception
      */
     protected abstract void doDisconnect() throws Exception;
 
     /**
      * The resource id used when firing ConnectEvents from this connector
      *
      * @return the resource id used when firing ConnectEvents from this connector
      */
     protected String getConnectEventId()
     {
         return getName();
     }
 
     /**
      * For better throughput when using TransactedMessageReceivers this will enable a
      * number of concurrent receivers, based on the value returned by
      * {@link #getNumberOfConcurrentTransactedReceivers()}. This property is used by
      * transports that support transactions, specifically receivers that extend the
      * TransactedPollingMessageReceiver.
      *
      * @return true if multiple receivers will be enabled for this connection
      */
     public boolean isCreateMultipleTransactedReceivers()
     {
         return createMultipleTransactedReceivers;
     }
 
     /**
      * @see {@link #isCreateMultipleTransactedReceivers()}
      * @param createMultipleTransactedReceivers if true, multiple receivers will be
      *            created for this connection
      */
     public void setCreateMultipleTransactedReceivers(boolean createMultipleTransactedReceivers)
     {
         this.createMultipleTransactedReceivers = createMultipleTransactedReceivers;
     }
 
     /**
      * Returns the number of concurrent receivers that will be launched when
      * {@link #isCreateMultipleTransactedReceivers()} returns <code>true</code>.
      *
      * @see #DEFAULT_NUM_CONCURRENT_TX_RECEIVERS
      */
     public int getNumberOfConcurrentTransactedReceivers()
     {
         return numberOfConcurrentTransactedReceivers;
     }
 
     /**
      * @see {@link #getNumberOfConcurrentTransactedReceivers()}
      * @param count the number of concurrent transacted receivers to start
      */
     public void setNumberOfConcurrentTransactedReceivers(int count)
     {
         numberOfConcurrentTransactedReceivers = count;
     }
 
     /**
      * Whether to fire message notifications for every message that is sent or
      * received from this connector
      */
     public boolean isEnableMessageEvents()
     {
         return enableMessageEvents;
     }
 
     /**
      * Whether to fire message notifications for every message that is sent or
      * received from this connector
      *
      * @param enableMessageEvents
      */
     public void setEnableMessageEvents(boolean enableMessageEvents)
     {
         this.enableMessageEvents = enableMessageEvents;
     }
 
     /**
      * Registers other protocols