* Licensed to the Apache Software Foundation (ASF) under one or more

* contributor license agreements.  See the NOTICE file distributed with

* this work for additional information regarding copyright ownership.

* The ASF licenses this file to You under the Apache License, Version 2.0

* (the "License"); you may not use this file except in compliance with

* the License.  You may obtain a copy of the License at

*

*      http://www.apache.org/licenses/LICENSE-2.0

*

* Unless required by applicable law or agreed to in writing, software

* distributed under the License is distributed on an "AS IS" BASIS,

* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

* See the License for the specific language governing permissions and

* limitations under the License.

*/

    public static final String DEFAULT_USER = ActiveMQConnectionFactory.DEFAULT_USER;

    public static final String DEFAULT_PASSWORD = ActiveMQConnectionFactory.DEFAULT_PASSWORD;

    public static final String DEFAULT_BROKER_URL = ActiveMQConnectionFactory.DEFAULT_BROKER_URL;

    public static int DEFAULT_THREAD_POOL_SIZE = 1000;

    private static final Logger LOG = LoggerFactory.getLogger(ActiveMQConnection.class);

    public final ConcurrentHashMap<ActiveMQTempDestination, ActiveMQTempDestination> activeTempDestinations = new ConcurrentHashMap<ActiveMQTempDestination, ActiveMQTempDestination>();

    protected boolean dispatchAsync=true;

    protected boolean alwaysSessionAsync = true;

    private TaskRunnerFactory sessionTaskRunner;

    private final ThreadPoolExecutor executor;

    // Connection state variables

    private final ConnectionInfo info;

    private ExceptionListener exceptionListener;

    private ClientInternalExceptionListener clientInternalExceptionListener;

    private boolean clientIDSet;

    private boolean isConnectionInfoSentToBroker;

    private boolean userSpecifiedClientID;

    // Configuration options variables

    private ActiveMQPrefetchPolicy prefetchPolicy = new ActiveMQPrefetchPolicy();

    private BlobTransferPolicy blobTransferPolicy;

    private RedeliveryPolicyMap redeliveryPolicyMap;

    private MessageTransformer transformer;

    private boolean disableTimeStampsByDefault;

    private boolean optimizedMessageDispatch = true;

    private boolean copyMessageOnSend = true;

    private boolean useCompression;

    private boolean objectMessageSerializationDefered;

    private boolean useAsyncSend;

    private boolean optimizeAcknowledge;

    private long optimizeAcknowledgeTimeOut = 0;

    private long optimizedAckScheduledAckInterval = 0;

    private boolean nestedMapAndListEnabled = true;

    private boolean useRetroactiveConsumer;

    private boolean exclusiveConsumer;

    private boolean alwaysSyncSend;

    private int closeTimeout = 15000;

    private boolean watchTopicAdvisories = true;

    private long warnAboutUnstartedConnectionTimeout = 500L;

    private int sendTimeout =0;

    private boolean sendAcksAsync=true;

    private boolean checkForDuplicates = true;

    private boolean queueOnlyConnection = false;

    private boolean consumerExpiryCheckEnabled = true;

    private final Transport transport;

    private final IdGenerator clientIdGenerator;

    private final JMSStatsImpl factoryStats;

    private final JMSConnectionStatsImpl stats;

    private final AtomicBoolean started = new AtomicBoolean(false);

    private final AtomicBoolean closing = new AtomicBoolean(false);

    private final AtomicBoolean closed = new AtomicBoolean(false);

    private final AtomicBoolean transportFailed = new AtomicBoolean(false);

    private final CopyOnWriteArrayList<ActiveMQSession> sessions = new CopyOnWriteArrayList<ActiveMQSession>();

    private final CopyOnWriteArrayList<ActiveMQConnectionConsumer> connectionConsumers = new CopyOnWriteArrayList<ActiveMQConnectionConsumer>();

    private final CopyOnWriteArrayList<TransportListener> transportListeners = new CopyOnWriteArrayList<TransportListener>();

    // Stream are deprecated and will be removed in a later release.

    private final CopyOnWriteArrayList<ActiveMQInputStream> inputStreams = new CopyOnWriteArrayList<ActiveMQInputStream>();

    private final CopyOnWriteArrayList<ActiveMQOutputStream> outputStreams = new CopyOnWriteArrayList<ActiveMQOutputStream>();

    // Maps ConsumerIds to ActiveMQConsumer objects

    private final ConcurrentHashMap<ConsumerId, ActiveMQDispatcher> dispatchers = new ConcurrentHashMap<ConsumerId, ActiveMQDispatcher>();

    private final ConcurrentHashMap<ProducerId, ActiveMQMessageProducer> producers = new ConcurrentHashMap<ProducerId, ActiveMQMessageProducer>();

    private final LongSequenceGenerator sessionIdGenerator = new LongSequenceGenerator();

    private final SessionId connectionSessionId;

    private final LongSequenceGenerator consumerIdGenerator = new LongSequenceGenerator();

    private final LongSequenceGenerator producerIdGenerator = new LongSequenceGenerator();

    private final LongSequenceGenerator tempDestinationIdGenerator = new LongSequenceGenerator();

    private final LongSequenceGenerator localTransactionIdGenerator = new LongSequenceGenerator();

    private AdvisoryConsumer advisoryConsumer;

    private final CountDownLatch brokerInfoReceived = new CountDownLatch(1);

    private BrokerInfo brokerInfo;

    private IOException firstFailureError;

    private int producerWindowSize = ActiveMQConnectionFactory.DEFAULT_PRODUCER_WINDOW_SIZE;

    // Assume that protocol is the latest. Change to the actual protocol

    // version when a WireFormatInfo is received.

    private final AtomicInteger protocolVersion = new AtomicInteger(CommandTypes.PROTOCOL_VERSION);

    private final long timeCreated;

    private final ConnectionAudit connectionAudit = new ConnectionAudit();

    private DestinationSource destinationSource;

    private final Object ensureConnectionInfoSentMutex = new Object();

    private boolean useDedicatedTaskRunner;

    protected AtomicInteger transportInterruptionProcessingComplete = new AtomicInteger(0);

    private long consumerFailoverRedeliveryWaitPeriod;

    private Scheduler scheduler;

    private boolean messagePrioritySupported = true;

    private boolean transactedIndividualAck = false;

    private boolean nonBlockingRedelivery = false;

    private boolean rmIdFromConnectionId = false;

    private int maxThreadPoolSize = DEFAULT_THREAD_POOL_SIZE;

    private RejectedExecutionHandler rejectedTaskHandler = null;

    /**

     * Construct an <code>ActiveMQConnection</code>

     *

     * @param transport

     * @param factoryStats

     * @throws Exception

     */

    protected ActiveMQConnection(final Transport transport, IdGenerator clientIdGenerator, IdGenerator connectionIdGenerator, JMSStatsImpl factoryStats) throws Exception {

        this.transport = transport;

        this.clientIdGenerator = clientIdGenerator;

        this.factoryStats = factoryStats;

        // Configure a single threaded executor who's core thread can timeout if

        // idle

        executor = new ThreadPoolExecutor(1, 1, 5, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), new ThreadFactory() {

            @Override

            public Thread newThread(Runnable r) {

                Thread thread = new Thread(r, "ActiveMQ Connection Executor: " + transport);

                //Don't make these daemon threads - see https://issues.apache.org/jira/browse/AMQ-796

                //thread.setDaemon(true);

                return thread;

            }

        });

        // asyncConnectionThread.allowCoreThreadTimeOut(true);

        String uniqueId = connectionIdGenerator.generateId();

        this.info = new ConnectionInfo(new ConnectionId(uniqueId));

        this.info.setManageable(true);

        this.info.setFaultTolerant(transport.isFaultTolerant());

        this.connectionSessionId = new SessionId(info.getConnectionId(), -1);

        this.transport.setTransportListener(this);

        this.stats = new JMSConnectionStatsImpl(sessions, this instanceof XAConnection);

        this.factoryStats.addConnection(this);

        this.timeCreated = System.currentTimeMillis();

        this.connectionAudit.setCheckForDuplicates(transport.isFaultTolerant());

    }

    protected void setUserName(String userName) {

        this.info.setUserName(userName);

    }

    protected void setPassword(String password) {

        this.info.setPassword(password);

    }

    /**

     * A static helper method to create a new connection

     *

     * @return an ActiveMQConnection

     * @throws JMSException

     */

    public static ActiveMQConnection makeConnection() throws JMSException {

        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();

        return (ActiveMQConnection)factory.createConnection();

    }

    /**

     * A static helper method to create a new connection

     *

     * @param uri

     * @return and ActiveMQConnection

     * @throws JMSException

     */

    public static ActiveMQConnection makeConnection(String uri) throws JMSException, URISyntaxException {

        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(uri);

        return (ActiveMQConnection)factory.createConnection();

    }

    /**

     * A static helper method to create a new connection

     *

     * @param user

     * @param password

     * @param uri

     * @return an ActiveMQConnection

     * @throws JMSException

     */

    public static ActiveMQConnection makeConnection(String user, String password, String uri) throws JMSException, URISyntaxException {

        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(user, password, new URI(uri));

        return (ActiveMQConnection)factory.createConnection();

    }

    /**

     * @return a number unique for this connection

     */

    public JMSConnectionStatsImpl getConnectionStats() {

        return stats;

    }

    /**

     * Creates a <CODE>Session</CODE> object.

     *

     * @param transacted indicates whether the session is transacted

     * @param acknowledgeMode indicates whether the consumer or the client will

     *                acknowledge any messages it receives; ignored if the

     *                session is transacted. Legal values are

     *                <code>Session.AUTO_ACKNOWLEDGE</code>,

     *                <code>Session.CLIENT_ACKNOWLEDGE</code>, and

     *                <code>Session.DUPS_OK_ACKNOWLEDGE</code>.

     * @return a newly created session

     * @throws JMSException if the <CODE>Connection</CODE> object fails to

     *                 create a session due to some internal error or lack of

     *                 support for the specific transaction and acknowledgement

     *                 mode.

     * @see Session#AUTO_ACKNOWLEDGE

     * @see Session#CLIENT_ACKNOWLEDGE

     * @see Session#DUPS_OK_ACKNOWLEDGE

     * @since 1.1

     */

    @Override

    public Session createSession(boolean transacted, int acknowledgeMode) throws JMSException {

        checkClosedOrFailed();

        ensureConnectionInfoSent();

        if(!transacted) {

            if (acknowledgeMode==Session.SESSION_TRANSACTED) {

                throw new JMSException("acknowledgeMode SESSION_TRANSACTED cannot be used for an non-transacted Session");

            } else if (acknowledgeMode < Session.SESSION_TRANSACTED || acknowledgeMode > ActiveMQSession.MAX_ACK_CONSTANT) {

                throw new JMSException("invalid acknowledgeMode: " + acknowledgeMode + ". Valid values are Session.AUTO_ACKNOWLEDGE (1), " +

                        "Session.CLIENT_ACKNOWLEDGE (2), Session.DUPS_OK_ACKNOWLEDGE (3), ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE (4) or for transacted sessions Session.SESSION_TRANSACTED (0)");

            }

        }

        return new ActiveMQSession(this, getNextSessionId(), transacted ? Session.SESSION_TRANSACTED : (acknowledgeMode == Session.SESSION_TRANSACTED

            ? Session.AUTO_ACKNOWLEDGE : acknowledgeMode), isDispatchAsync(), isAlwaysSessionAsync());

    }

    /**

     * @return sessionId

     */

    protected SessionId getNextSessionId() {

        return new SessionId(info.getConnectionId(), sessionIdGenerator.getNextSequenceId());

    }

    /**

     * Gets the client identifier for this connection.

     * <P>

     * This value is specific to the JMS provider. It is either preconfigured by

     * an administrator in a <CODE> ConnectionFactory</CODE> object or assigned

     * dynamically by the application by calling the <code>setClientID</code>

     * method.

     *

     * @return the unique client identifier

     * @throws JMSException if the JMS provider fails to return the client ID

     *                 for this connection due to some internal error.

     */

    @Override

    public String getClientID() throws JMSException {

        checkClosedOrFailed();

        return this.info.getClientId();

    }

    /**

     * Sets the client identifier for this connection.

     * <P>

     * The preferred way to assign a JMS client's client identifier is for it to

     * be configured in a client-specific <CODE>ConnectionFactory</CODE>

     * object and transparently assigned to the <CODE>Connection</CODE> object

     * it creates.

     * <P>

     * Alternatively, a client can set a connection's client identifier using a

     * provider-specific value. The facility to set a connection's client

     * identifier explicitly is not a mechanism for overriding the identifier

     * that has been administratively configured. It is provided for the case

     * where no administratively specified identifier exists. If one does exist,

     * an attempt to change it by setting it must throw an

     * <CODE>IllegalStateException</CODE>. If a client sets the client

     * identifier explicitly, it must do so immediately after it creates the

     * connection and before any other action on the connection is taken. After

     * this point, setting the client identifier is a programming error that

     * should throw an <CODE>IllegalStateException</CODE>.

     * <P>

     * The purpose of the client identifier is to associate a connection and its

     * objects with a state maintained on behalf of the client by a provider.

     * The only such state identified by the JMS API is that required to support

     * durable subscriptions.

     * <P>

     * If another connection with the same <code>clientID</code> is already

     * running when this method is called, the JMS provider should detect the

     * duplicate ID and throw an <CODE>InvalidClientIDException</CODE>.

     *

     * @param newClientID the unique client identifier

     * @throws JMSException if the JMS provider fails to set the client ID for

     *                 this connection due to some internal error.

     * @throws javax.jms.InvalidClientIDException if the JMS client specifies an

     *                 invalid or duplicate client ID.

     * @throws javax.jms.IllegalStateException if the JMS client attempts to set

     *                 a connection's client ID at the wrong time or when it has

     *                 been administratively configured.

     */

    @Override

    public void setClientID(String newClientID) throws JMSException {

        checkClosedOrFailed();

        if (this.clientIDSet) {

            throw new IllegalStateException("The clientID has already been set");

        }

        if (this.isConnectionInfoSentToBroker) {

            throw new IllegalStateException("Setting clientID on a used Connection is not allowed");

        }

        this.info.setClientId(newClientID);

        this.userSpecifiedClientID = true;

        ensureConnectionInfoSent();

    }

    /**

     * Sets the default client id that the connection will use if explicitly not

     * set with the setClientId() call.

     */

    public void setDefaultClientID(String clientID) throws JMSException {

        this.info.setClientId(clientID);

        this.userSpecifiedClientID = true;

    }

    /**

     * Gets the metadata for this connection.

     *

     * @return the connection metadata

     * @throws JMSException if the JMS provider fails to get the connection

     *                 metadata for this connection.

     * @see javax.jms.ConnectionMetaData

     */

    @Override

    public ConnectionMetaData getMetaData() throws JMSException {

        checkClosedOrFailed();

        return ActiveMQConnectionMetaData.INSTANCE;

    }

    /**

     * Gets the <CODE>ExceptionListener</CODE> object for this connection. Not

     * every <CODE>Connection</CODE> has an <CODE>ExceptionListener</CODE>

     * associated with it.

     *

     * @return the <CODE>ExceptionListener</CODE> for this connection, or

     *         null, if no <CODE>ExceptionListener</CODE> is associated with

     *         this connection.

     * @throws JMSException if the JMS provider fails to get the

     *                 <CODE>ExceptionListener</CODE> for this connection.

     * @see javax.jms.Connection#setExceptionListener(ExceptionListener)

     */

    @Override

    public ExceptionListener getExceptionListener() throws JMSException {

        checkClosedOrFailed();

        return this.exceptionListener;

    }

    /**

     * Sets an exception listener for this connection.

     * <P>

     * If a JMS provider detects a serious problem with a connection, it informs

     * the connection's <CODE> ExceptionListener</CODE>, if one has been

     * registered. It does this by calling the listener's <CODE>onException

     * </CODE>

     * method, passing it a <CODE>JMSException</CODE> object describing the

     * problem.

     * <P>

     * An exception listener allows a client to be notified of a problem

     * asynchronously. Some connections only consume messages, so they would

     * have no other way to learn their connection has failed.

     * <P>

     * A connection serializes execution of its <CODE>ExceptionListener</CODE>.

     * <P>

     * A JMS provider should attempt to resolve connection problems itself

     * before it notifies the client of them.

     *

     * @param listener the exception listener

     * @throws JMSException if the JMS provider fails to set the exception

     *                 listener for this connection.

     */

    @Override

    public void setExceptionListener(ExceptionListener listener) throws JMSException {

        checkClosedOrFailed();

        this.exceptionListener = listener;

    }

    /**

     * Gets the <code>ClientInternalExceptionListener</code> object for this connection.

     * Not every <CODE>ActiveMQConnectionn</CODE> has a <CODE>ClientInternalExceptionListener</CODE>

     * associated with it.

     *

     * @return the listener or <code>null</code> if no listener is registered with the connection.

     */

    public ClientInternalExceptionListener getClientInternalExceptionListener() {

        return clientInternalExceptionListener;

    }

    /**

     * Sets a client internal exception listener for this connection.

     * The connection will notify the listener, if one has been registered, of exceptions thrown by container components

     * (e.g. an EJB container in case of Message Driven Beans) during asynchronous processing of a message.

     * It does this by calling the listener's <code>onException()</code> method passing it a <code>Throwable</code>

     * describing the problem.

     *

     * @param listener the exception listener

     */

    public void setClientInternalExceptionListener(ClientInternalExceptionListener listener) {

        this.clientInternalExceptionListener = listener;

    }

    /**

     * Starts (or restarts) a connection's delivery of incoming messages. A call

     * to <CODE>start</CODE> on a connection that has already been started is

     * ignored.

     *

     * @throws JMSException if the JMS provider fails to start message delivery

     *                 due to some internal error.

     * @see javax.jms.Connection#stop()

     */

    @Override

    public void start() throws JMSException {

        checkClosedOrFailed();

        ensureConnectionInfoSent();

        if (started.compareAndSet(false, true)) {

            for (Iterator<ActiveMQSession> i = sessions.iterator(); i.hasNext();) {

                ActiveMQSession session = i.next();

                session.start();

            }

        }

    }

    /**

     * Temporarily stops a connection's delivery of incoming messages. Delivery

     * can be restarted using the connection's <CODE>start</CODE> method. When

     * the connection is stopped, delivery to all the connection's message

     * consumers is inhibited: synchronous receives block, and messages are not

     * delivered to message listeners.

     * <P>

     * This call blocks until receives and/or message listeners in progress have

     * completed.

     * <P>

     * Stopping a connection has no effect on its ability to send messages. A

     * call to <CODE>stop</CODE> on a connection that has already been stopped

     * is ignored.

     * <P>

     * A call to <CODE>stop</CODE> must not return until delivery of messages

     * has paused. This means that a client can rely on the fact that none of

     * its message listeners will be called and that all threads of control

     * waiting for <CODE>receive</CODE> calls to return will not return with a

     * message until the connection is restarted. The receive timers for a

     * stopped connection continue to advance, so receives may time out while

     * the connection is stopped.

     * <P>

     * If message listeners are running when <CODE>stop</CODE> is invoked, the

     * <CODE>stop</CODE> call must wait until all of them have returned before

     * it may return. While these message listeners are completing, they must

     * have the full services of the connection available to them.

     *

     * @throws JMSException if the JMS provider fails to stop message delivery

     *                 due to some internal error.

     * @see javax.jms.Connection#start()

     */

    @Override

    public void stop() throws JMSException {

        doStop(true);

    }

    /**

     * @see #stop()

     * @param checkClosed <tt>true</tt> to check for already closed and throw {@link java.lang.IllegalStateException} if already closed,

     *                    <tt>false</tt> to skip this check

     * @throws JMSException if the JMS provider fails to stop message delivery due to some internal error.

     */

    void doStop(boolean checkClosed) throws JMSException {

        if (checkClosed) {

            checkClosedOrFailed();

        }

        if (started.compareAndSet(true, false)) {

            synchronized(sessions) {

                for (Iterator<ActiveMQSession> i = sessions.iterator(); i.hasNext();) {

                    ActiveMQSession s = i.next();

                    s.stop();

                }

            }

        }

    }

    /**

     * Closes the connection.

     * <P>

     * Since a provider typically allocates significant resources outside the

     * JVM on behalf of a connection, clients should close these resources when

     * they are not needed. Relying on garbage collection to eventually reclaim

     * these resources may not be timely enough.

     * <P>

     * There is no need to close the sessions, producers, and consumers of a

     * closed connection.

     * <P>

     * Closing a connection causes all temporary destinations to be deleted.

     * <P>

     * When this method is invoked, it should not return until message

     * processing has been shut down in an orderly fashion. This means that all

     * message listeners that may have been running have returned, and that all

     * pending receives have returned. A close terminates all pending message

     * receives on the connection's sessions' consumers. The receives may return

     * with a message or with null, depending on whether there was a message

     * available at the time of the close. If one or more of the connection's

     * sessions' message listeners is processing a message at the time when

     * connection <CODE>close</CODE> is invoked, all the facilities of the

     * connection and its sessions must remain available to those listeners

     * until they return control to the JMS provider.

     * <P>

     * Closing a connection causes any of its sessions' transactions in progress

     * to be rolled back. In the case where a session's work is coordinated by

     * an external transaction manager, a session's <CODE>commit</CODE> and

     * <CODE> rollback</CODE> methods are not used and the result of a closed

     * session's work is determined later by the transaction manager. Closing a

     * connection does NOT force an acknowledgment of client-acknowledged

     * sessions.

     * <P>

     * Invoking the <CODE>acknowledge</CODE> method of a received message from

     * a closed connection's session must throw an

     * <CODE>IllegalStateException</CODE>. Closing a closed connection must

     * NOT throw an exception.

     *

     * @throws JMSException if the JMS provider fails to close the connection

     *                 due to some internal error. For example, a failure to

     *                 release resources or to close a socket connection can

     *                 cause this exception to be thrown.

     */

    @Override

    public void close() throws JMSException {

        // Store the interrupted state and clear so that cleanup happens without

        // leaking connection resources.  Reset in finally to preserve state.

        boolean interrupted = Thread.interrupted();

        try {

            // If we were running, lets stop first.

            if (!closed.get() && !transportFailed.get()) {

                // do not fail if already closed as according to JMS spec we must not

                // throw exception if already closed

                doStop(false);

            }

            synchronized (this) {

                if (!closed.get()) {

                    closing.set(true);

                    if (destinationSource != null) {

                        destinationSource.stop();

                        destinationSource = null;

                    }

                    if (advisoryConsumer != null) {

                        advisoryConsumer.dispose();

                        advisoryConsumer = null;

                    }

                    Scheduler scheduler = this.scheduler;

                    if (scheduler != null) {

                        try {

                            scheduler.stop();

                        } catch (Exception e) {

                            JMSException ex =  JMSExceptionSupport.create(e);

                            throw ex;

                        }

                    }

                    long lastDeliveredSequenceId = 0;

                    for (Iterator<ActiveMQSession> i = this.sessions.iterator(); i.hasNext();) {

                        ActiveMQSession s = i.next();

                        s.dispose();

                        lastDeliveredSequenceId = Math.max(lastDeliveredSequenceId, s.getLastDeliveredSequenceId());

                    }

                    for (Iterator<ActiveMQConnectionConsumer> i = this.connectionConsumers.iterator(); i.hasNext();) {

                        ActiveMQConnectionConsumer c = i.next();

                        c.dispose();

                    }

                    // Stream are deprecated and will be removed in a later release.

                    for (Iterator<ActiveMQInputStream> i = this.inputStreams.iterator(); i.hasNext();) {

                        ActiveMQInputStream c = i.next();

                        c.dispose();

                    }

                    for (Iterator<ActiveMQOutputStream> i = this.outputStreams.iterator(); i.hasNext();) {

                        ActiveMQOutputStream c = i.next();

                        c.dispose();

                    }

                    this.activeTempDestinations.clear();

                    if (isConnectionInfoSentToBroker) {

                        // If we announced ourselves to the broker.. Try to let the broker

                        // know that the connection is being shutdown.

                        RemoveInfo removeCommand = info.createRemoveCommand();

                        removeCommand.setLastDeliveredSequenceId(lastDeliveredSequenceId);

                        try {

                            doSyncSendPacket(info.createRemoveCommand(), closeTimeout);

                        } catch (JMSException e) {

                            if (e.getCause() instanceof RequestTimedOutIOException) {

                                // expected

                            } else {

                                throw e;

                            }

                        }

                        doAsyncSendPacket(new ShutdownInfo());

                    }

                    started.set(false);

                    // TODO if we move the TaskRunnerFactory to the connection

                    // factory

                    // then we may need to call

                    // factory.onConnectionClose(this);

                    if (sessionTaskRunner != null) {

                        sessionTaskRunner.shutdown();

                    }

                    closed.set(true);

                    closing.set(false);

                }

            }

        } finally {

            try {

                if (executor != null) {

                    ThreadPoolUtils.shutdown(executor);

                }

            } catch (Throwable e) {

                LOG.warn("Error shutting down thread pool: " + executor + ". This exception will be ignored.", e);

            }

            ServiceSupport.dispose(this.transport);

            factoryStats.removeConnection(this);

            if (interrupted) {

                Thread.currentThread().interrupt();

            }

        }

    }

    /**

     * Tells the broker to terminate its VM. This can be used to cleanly

     * terminate a broker running in a standalone java process. Server must have

     * property enable.vm.shutdown=true defined to allow this to work.

     */

    // TODO : org.apache.activemq.message.BrokerAdminCommand not yet

    // implemented.

    /*