/*
 * 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.
 */

#ifndef _ACTIVEMQ_CORE_ACTIVEMQCONNECTION_H_
#define _ACTIVEMQ_CORE_ACTIVEMQCONNECTION_H_

#include <cms/EnhancedConnection.h>
#include <activemq/util/Config.h>
#include <activemq/core/Dispatcher.h>
#include <activemq/commands/ActiveMQTempDestination.h>
#include <activemq/commands/ConnectionInfo.h>
#include <activemq/commands/ConsumerInfo.h>
#include <activemq/commands/SessionId.h>
#include <activemq/exceptions/ActiveMQException.h>
#include <activemq/transport/Transport.h>
#include <activemq/transport/TransportListener.h>
#include <activemq/threads/Scheduler.h>
#include <activemq/core/kernels/ActiveMQProducerKernel.h>
#include <activemq/core/kernels/ActiveMQSessionKernel.h>
#include <decaf/util/Properties.h>
#include <decaf/util/ArrayList.h>
#include <decaf/util/concurrent/atomic/AtomicBoolean.h>
#include <decaf/util/concurrent/ExecutorService.h>
#include <decaf/lang/exceptions/UnsupportedOperationException.h>
#include <decaf/lang/exceptions/NullPointerException.h>
#include <decaf/lang/exceptions/IllegalStateException.h>

#include <string>
#include <memory>

namespace activemq {
namespace core {

    using decaf::lang::Pointer;

    class ActiveMQSession;
    class ConnectionConfig;
    class PrefetchPolicy;
    class RedeliveryPolicy;

    /**
     * Concrete connection used for all connectors to the
     * ActiveMQ broker.
     *
     * @since 2.0
     */
    class AMQCPP_API ActiveMQConnection : public virtual cms::EnhancedConnection,
                                          public transport::TransportListener {
    private:

        ConnectionConfig* config;

        /**
         * The instance of ConnectionMetaData to return to clients.
         */
        std::auto_ptr<cms::ConnectionMetaData> connectionMetaData;

        /**
         * Indicates if this Connection is started
         */
        decaf::util::concurrent::atomic::AtomicBoolean started;

        /**
         * Indicates that this connection has been closed, it is no longer
         * usable after this becomes true
         */
        decaf::util::concurrent::atomic::AtomicBoolean closed;

        /**
         * Indicates that this connection has been closed, it is no longer
         * usable after this becomes true
         */
        decaf::util::concurrent::atomic::AtomicBoolean closing;

        /**
         * Indicates that this connection's Transport has failed.
         */
        decaf::util::concurrent::atomic::AtomicBoolean transportFailed;

    private:

        ActiveMQConnection(const ActiveMQConnection&);
        ActiveMQConnection& operator=(const ActiveMQConnection&);

    public:

        /**
         * Constructor
         *
         * @param transport
         *        The Transport requested for this connection to the Broker.
         * @param properties
         *        The Properties that were defined for this connection
         */
        ActiveMQConnection(const Pointer<transport::Transport> transport,
                           const Pointer<decaf::util::Properties> properties);

        virtual ~ActiveMQConnection();

        /**
         * Adds the session resources for the given session instance.
         *
         * @param session
         *      The session to be added to this connection.
         *
         * @throws CMSException if an error occurs while removing performing the operation.
         */
        virtual void addSession(Pointer<activemq::core::kernels::ActiveMQSessionKernel> session);

        /**
         * Removes the session resources for the given session instance.
         *
         * @param session
         *      The session to be unregistered from this connection.
         *
         * @throws CMSException if an error occurs while removing performing the operation.
         */
        virtual void removeSession(Pointer<activemq::core::kernels::ActiveMQSessionKernel> session);

        /**
         * Adds an active Producer to the Set of known producers.
         *
         * @param producer
         *      The Producer to add from the the known set.
         *
         * @throws CMSException if an error occurs while removing performing the operation.
         */
        virtual void addProducer(Pointer<kernels::ActiveMQProducerKernel> producer);

        /**
         * Removes an active Producer to the Set of known producers.
         * @param producerId - The ProducerId to remove from the the known set.
         * @throws CMSException if an error occurs while removing performing the operation.
         */
        virtual void removeProducer(const Pointer<commands::ProducerId>& producerId);

        /**
         * Adds a dispatcher for a consumer.
         * @param consumer - The consumer for which to register a dispatcher.
         * @param dispatcher - The dispatcher to handle incoming messages for the consumer.
         * @throws CMSException if an error occurs while removing performing the operation.
         */
        virtual void addDispatcher(const Pointer<commands::ConsumerId>& consumer, Dispatcher* dispatcher);

        /**
         * Removes the dispatcher for a consumer.
         * @param consumer - The consumer for which to remove the dispatcher.
         * @throws CMSException if an error occurs while removing performing the operation.
         */
        virtual void removeDispatcher(const Pointer<commands::ConsumerId>& consumer);

        /**
         * If supported sends a message pull request to the service provider asking
         * for the delivery of a new message.  This is used in the case where the
         * service provider has been configured with a zero prefetch or is only
         * capable of delivering messages on a pull basis.
         * @param consumer - the ConsumerInfo for the requesting Consumer.
         * @param timeout - the time that the client is willing to wait.
         *
         * @throws ActiveMQException if an error occurs while removing performing the operation.
         */
        virtual void sendPullRequest(const commands::ConsumerInfo* consumer, long long timeout);

        /**
         * Checks if this connection has been closed
         * @return true if the connection is closed
         */
        bool isClosed() const {
            return this->closed.get();
        }

        /**
         * Check if this connection has been started.
         * @return true if the start method has been called.
         */
        bool isStarted() const {
            return this->started.get();
        }

        /**
         * Checks if the Connection's Transport has failed
         * @return true if the Connection's Transport has failed.
         */
        bool isTransportFailed() const {
            return this->transportFailed.get();
        }

        /**
         * Requests that the Broker removes the given Destination.  Calling this
         * method implies that the client is finished with the Destination and that
         * no other messages will be sent or received for the given Destination.  The
         * Broker frees all resources it has associated with this Destination.
         *
         * @param destination
         *        The Destination the Broker will be requested to remove.
         *
         * @throws NullPointerException
         *         If the passed Destination is Null
         * @throws IllegalStateException
         *         If the connection is closed.
         * @throws UnsupportedOperationException
         *         If the wire format in use does not support this operation.
         * @throws ActiveMQException
         *         If any other error occurs during the attempt to destroy the destination.
         */
        virtual void destroyDestination(const commands::ActiveMQDestination* destination);

        /**
         * Requests that the Broker removes the given Destination.  Calling this
         * method implies that the client is finished with the Destination and that
         * no other messages will be sent or received for the given Destination.  The
         * Broker frees all resources it has associated with this Destination.
         *
         * @param destination
         *        The CMS Destination the Broker will be requested to remove.
         *
         * @throws NullPointerException
         *         If the passed Destination is Null
         * @throws IllegalStateException
         *         If the connection is closed.
         * @throws UnsupportedOperationException
         *         If the wire format in use does not support this operation.
         * @throws ActiveMQException
         *         If any other error occurs during the attempt to destroy the destination.
         */
        virtual void destroyDestination(const cms::Destination* destination);

        /**
         * Allows Consumers to check if an incoming Message is a Duplicate.
         *
         * @param dispatcher
         *      The Dispatcher that is checking the Message for Duplication.
         * @param message
         *      The Message that should be checked.
         *
         * @return true if the Message was seen before.
         */
        bool isDuplicate(Dispatcher* dispatcher, Pointer<commands::Message> message);

        /**
         * Mark message as received.
         *
         * @param dispatcher
         *      The Dispatcher instance that has received the Message.
         * @param message
         *      The Message that has been received.
         */
        void rollbackDuplicate(Dispatcher* dispatcher, Pointer<commands::Message> message);

        /**
         * Removes the Audit information stored for a given MessageConsumer
         *
         * @param dispatcher
         *      The Dispatcher instance that has received the Message.
         */
        void removeAuditedDispatcher(Dispatcher* dispatcher);

    public:   // Connection Interface Methods

        /**
         * {@inheritDoc}
         */
        virtual const cms::ConnectionMetaData* getMetaData() const {
            return connectionMetaData.get();
        }

        /**
         * {@inheritDoc}
         */
        virtual cms::Session* createSession();

        /**
         * {@inheritDoc}
         */
        virtual std::string getClientID() const;

        /**
         * {@inheritDoc}
         */
        virtual void setClientID(const std::string& clientID);

        /**
         * {@inheritDoc}
         */
        virtual cms::Session* createSession(cms::Session::AcknowledgeMode ackMode);

        /**
         * {@inheritDoc}
         */
        virtual void close();

        /**
         * {@inheritDoc}
         */
        virtual void start();

        /**
         * {@inheritDoc}
         */
        virtual void stop();

        /**
         * {@inheritDoc}
         */
        virtual cms::ExceptionListener* getExceptionListener() const;

        /**
         * {@inheritDoc}
         */
        virtual void setExceptionListener(cms::ExceptionListener* listener);

        /**
         * {@inheritDoc}
         */
        virtual void setMessageTransformer(cms::MessageTransformer* transformer);

        /**
         * {@inheritDoc}
         */
        virtual cms::MessageTransformer* getMessageTransformer() const;

        /**
         * {@inheritDoc}
         */
        virtual cms::DestinationSource* getDestinationSource();

    public:   // Configuration Options

        /**
         * Sets the username that should be used when creating a new connection
         * @param username string
         */
        void setUsername(const std::string& username);

        /**
         * Gets the username that this factory will use when creating a new
         * connection instance.
         * @return username string, "" for default credentials
         */
        const std::string& getUsername() const;

        /**
         * Sets the password that should be used when creating a new connection
         * @param password string
         */
        void setPassword(const std::string& password);

        /**
         * Gets the password that this factory will use when creating a new
         * connection instance.
         * @return password string, "" for default credentials
         */
        const std::string& getPassword() const;

        /**
         * Sets the Client Id.
         * @param clientId - The new clientId value.
         */
        void setDefaultClientId(const std::string& clientId);

        /**
         * Sets the Broker URL that should be used when creating a new
         * connection instance
         * @param brokerURL string
         */
        void setBrokerURL(const std::string& brokerURL);

        /**
         * Gets the Broker URL that this factory will use when creating a new
         * connection instance.
         * @return brokerURL string
         */
        const std::string& getBrokerURL() const;

        /**
         * Sets the PrefetchPolicy instance that this factory should use when it creates
         * new Connection instances.  The PrefetchPolicy passed becomes the property of the
         * factory and will be deleted when the factory is destroyed.
         *
         * @param policy
         *      The new PrefetchPolicy that the ConnectionFactory should clone for Connections.
         */
        void setPrefetchPolicy(PrefetchPolicy* policy);

        /**
         * Gets the pointer to the current PrefetchPolicy that is in use by this ConnectionFactory.
         *
         * @return a pointer to this objects PrefetchPolicy.
         */
        PrefetchPolicy* getPrefetchPolicy() const;

        /**
         * Sets the RedeliveryPolicy instance that this factory should use when it creates
         * new Connection instances.  The RedeliveryPolicy passed becomes the property of the
         * factory and will be deleted when the factory is destroyed.
         *
         * @param policy
         *      The new RedeliveryPolicy that the ConnectionFactory should clone for Connections.
         */
        void setRedeliveryPolicy(RedeliveryPolicy* policy);

        /**
         * Gets the pointer to the current RedeliveryPolicy that is in use by this ConnectionFactory.
         *
         * @return a pointer to this objects RedeliveryPolicy.
         */
        RedeliveryPolicy* getRedeliveryPolicy() const;

        /**
         * @return The value of the dispatch asynchronously option sent to the broker.
         */
        bool isDispatchAsync() const;

        /**
         * Should messages be dispatched synchronously or asynchronously from the producer
         * thread for non-durable topics in the broker? For fast consumers set this to false.
         * For slow consumers set it to true so that dispatching will not block fast consumers. .
         *
         * @param value
         *        The value of the dispatch asynchronously option sent to the broker.
         */
        void setDispatchAsync(bool value);

        /**
         * Gets if the Connection should always send things Synchronously.
         *
         * @return true if sends should always be Synchronous.
         */
        bool isAlwaysSyncSend() const;

        /**
         * Sets if the Connection should always send things Synchronously.
         * @param value
         *        true if sends should always be Synchronous.
         */
        void setAlwaysSyncSend(bool value);

        /**
         * Gets if the useAsyncSend option is set
         * @return true if on false if not.
         */
        bool isUseAsyncSend() const;

        /**
         * Sets the useAsyncSend option
         * @param value - true to activate, false to disable.
         */
        void setUseAsyncSend(bool value);

        /**
         * Gets if the Connection is configured for Message body compression.
         * @return if the Message body will be Compressed or not.
         */
        bool isUseCompression() const;

        /**
         * Sets whether Message body compression is enabled.
         *
         * @param value
         *      Boolean indicating if Message body compression is enabled.
         */
        void setUseCompression(bool value);

        /**
         * Sets the Compression level used when Message body compression is enabled, a
         * value of -1 causes the Compression Library to use the default setting which
         * is a balance of speed and compression.  The range of compression levels is
         * [0..9] where 0 indicates best speed and 9 indicates best compression.
         *
         * @param value
         *      A signed int value that controls the compression level.
         */
        void setCompressionLevel(int value);

        /**
         * Gets the currently configured Compression level for Message bodies.
         *
         * @return the int value of the current compression level.
         */
        int getCompressionLevel() const;

        /**
         * Gets the assigned send timeout for this Connector
         * @return the send timeout configured in the connection uri
         */
        unsigned int getSendTimeout() const;

        /**
         * Sets the send timeout to use when sending Message objects, this will
         * cause all messages to be sent using a Synchronous request is non-zero.
         * @param timeout - The time to wait for a response.
         */
        void setSendTimeout(unsigned int timeout);

        /**
         * Gets the assigned connect response timeout for this Connector
         * @return the connect response timeout configured in the connection uri
         */
        unsigned int getConnectResponseTimeout() const;

        /**
         * Sets the connect response timeout to use when sending Message objects, this will
         * protect clients using a Synchronous request in the case of the broker not responding
         * or missing the brokers response.
         * @param timeout - The time to wait for a connect response.
         */
        void setConnectResponseTimeout(unsigned int connectResponseTimeout);

        /**
         * Gets the assigned close timeout for this Connector
         * @return the close timeout configured in the connection uri
         */
        unsigned int getCloseTimeout() const;

        /**
         * Sets the close timeout to use when sending the disconnect request.
         * @param timeout - The time to wait for a close message.
         */
        void setCloseTimeout(unsigned int timeout);

        /**
         * Gets the configured producer window size for Producers that are created
         * from this connector.  This only applies if there is no send timeout and the
         * producer is able to send asynchronously.
         * @return size in bytes of messages that this producer can produce before
         *         it must block and wait for ProducerAck messages to free resources.
         */
        unsigned int getProducerWindowSize() const;

        /**
         * Sets the size in Bytes of messages that a producer can send before it is blocked
         * to await a ProducerAck from the broker that frees enough memory to allow another
         * message to be sent.
         * @param windowSize - The size in bytes of the Producers memory window.
         */
        void setProducerWindowSize(unsigned int windowSize);

        /**
         * @return true if the Connections that this factory creates should support the
         * message based priority settings.
         */
        bool isMessagePrioritySupported() const;

        /**
         * Set whether or not this factory should create Connection objects with the Message
         * priority support function enabled.
         *
         * @param value
         *      Boolean indicating if Message priority should be enabled.
         */
        void setMessagePrioritySupported(bool value);

        /**
         * Get the Next Temporary Destination Id
         * @return the next id in the sequence.
         */
        long long getNextTempDestinationId();

        /**
         * Get the Next Temporary Destination Id
         * @return the next id in the sequence.
         */
        long long getNextLocalTransactionId();

        /**
         * Is the Connection configured to watch for advisory messages to maintain state of
         * temporary destination create and destroy.
         *
         * @return true if the Connection will listen for temporary topic advisory messages.
         */
        bool isWatchTopicAdvisories() const;

        /**
         * Sets whether this Connection is listening for advisory messages regarding temporary
         * destination creation and deletion.
         *
         * @param value
         *      Boolean indicating if advisory message monitoring should be enabled.
         */
        void setWatchTopicAdvisories(bool value);

        /**
         * Get the audit depth for Messages for consumers when using a fault
         * tolerant transport.  The higher the value the more messages are checked
         * for duplication, and the larger the performance impact of duplicate
         * detection will be.
         *
         * @return the configured audit depth.
         */
        int getAuditDepth() const;

        /**
         * Set the audit depth for Messages for consumers when using a fault
         * tolerant transport.  The higher the value the more messages are checked
         * for duplication, and the larger the performance impact of duplicate
         * detection will be.
         *
         * @param auditDepth
         *      The configured audit depth.
         */
        void setAuditDepth(int auditDepth);

        /**
         * The number of Producers that will be audited.
         *
         * @return the configured number of producers to include in the audit.
         */
        int getAuditMaximumProducerNumber() const;

        /**
         * The number of Producers that will be audited.
         *
         * @param auditMaximumProducerNumber
         *      The configured number of producers to include in the audit.
         */
        void setAuditMaximumProducerNumber(int auditMaximumProducerNumber);

        /**
         * Gets the value of the configured Duplicate Message detection feature.
         *
         * When enabled and a fault tolerant transport is used (think failover) then
         * this feature will help to detect and filter duplicate messages that might
         * otherwise be delivered to a consumer after a connection failure.
         *
         * Disabling this can increase performance since no Message auditing will
         * occur.
         *
         * @return the checkForDuplicates value currently set.
         */
        bool isCheckForDuplicates() const;

        /**
         * Gets the value of the configured Duplicate Message detection feature.
         *
         * When enabled and a fault tolerant transport is used (think failover) then
         * this feature will help to detect and filter duplicate messages that might
         * otherwise be delivered to a consumer after a connection failure.
         *
         * Disabling this can increase performance since no Message auditing will
         * occur.
         *
         * @param checkForDuplicates
         *      The checkForDuplicates value to be configured.
         */
        void setCheckForDuplicates(bool checkForDuplicates);

        /**
         * when true, submit individual transacted acks immediately rather than with transaction
         * completion.  This allows the acks to represent delivery status which can be persisted on
         * rollback Used in conjunction with KahaDB set to Rewrite On Redelivery.
         *
         * @return true if this option is enabled.
         */
        bool isTransactedIndividualAck() const;

        /**
         * when true, submit individual transacted acks immediately rather than with transaction
         * completion.  This allows the acks to represent delivery status which can be persisted on
         * rollback Used in conjunction with KahaDB set to Rewrite On Redelivery.
         *
         * @param transactedIndividualAck
         *      The value to set.
         */
        void setTransactedIndividualAck(bool transactedIndividualAck);

        /**
         * Returns true if non-blocking redelivery of Messages is configured for Consumers
         * that are rolled back or recovered.
         *
         * @return true if non-blocking redelivery is enabled.
         */
        bool isNonBlockingRedelivery() const;

        /**
         * When true a MessageConsumer will not stop Message delivery before re-delivering Messages
         * from a rolled back transaction.  This implies that message order will not be preserved and
         * also will result in the TransactedIndividualAck option to be enabled.
         *
         * @param nonBlockingRedelivery
         *      The value to configure for non-blocking redelivery.
         */
        void setNonBlockingRedelivery(bool nonBlockingRedelivery);

        /**
         * Gets the delay period for a consumer redelivery.
         *
         * @return configured time delay in milliseconds.
         */
        long long getConsumerFailoverRedeliveryWaitPeriod() const;

        /**
         * Sets the delay period for a consumer redelivery.
         *
         * @param value
         *      The configured time delay in milliseconds.
         */
        void setConsumerFailoverRedeliveryWaitPeriod(long long value);

        /**
         * @return true if optimizeAcknowledge is enabled.
         */
        bool isOptimizeAcknowledge() const;

        /**
         * Sets if Consumers are configured to use Optimized Acknowledge by default.
         *
         * @param optimizeAcknowledge
         *      The optimizeAcknowledge mode to set.
         */
        void setOptimizeAcknowledge(bool optimizeAcknowledge);

        /**
         * Gets the time between optimized ack batches in milliseconds.
         *
         * @return time between optimized ack batches in Milliseconds.
         */
        long long getOptimizeAcknowledgeTimeOut() const;

        /**
         * The max time in milliseconds between optimized ack batches.
         *
         * @param optimizeAcknowledgeTimeOut
         *      The time in milliseconds for optimized ack batches.
         */
        void setOptimizeAcknowledgeTimeOut(long long optimizeAcknowledgeTimeOut);

        /**
         * Gets the configured time interval that is used to force all MessageConsumers that have
         * optimizedAcknowledge enabled to send an ack for any outstanding Message Acks.  By default
         * this value is set to zero meaning that the consumers will not do any background Message
         * acknowledgment.
         *
         * @return the scheduledOptimizedAckInterval
         */
        long long getOptimizedAckScheduledAckInterval() const;

        /**
         * Sets the amount of time between scheduled sends of any outstanding Message Acks for
         * consumers that have been configured with optimizeAcknowledge enabled.
         *
         * Time is given in Milliseconds.
         *
         * @param optimizedAckScheduledAckInterval
         *      The scheduledOptimizedAckInterval to use for new Consumers.
         */
        void setOptimizedAckScheduledAckInterval(long long optimizedAckScheduledAckInterval);

        /**
         * Should all created consumers be retroactive.
         *
         * @return true if consumer will be created with the retroactive flag set.
         */
        bool isUseRetroactiveConsumer() const;

        /**
         * Sets whether or not retroactive consumers are enabled. Retroactive
         * consumers allow non-durable topic subscribers to receive old messages
         * that were published before the non-durable subscriber started.
         *
         * @param useRetroactiveConsumer
         *      The value of this configuration option.
         */
        void setUseRetroactiveConsumer(bool useRetroactiveConsumer);

        /**
         * Should all created consumers be exclusive.
         *
         * @return true if consumer will be created with the exclusive flag set.
         */
        bool isExclusiveConsumer() const;

        /**
         * Enables or disables whether or not queue consumers should be exclusive or
         * not for example to preserve ordering when not using Message Groups.
         *
         * @param exclusiveConsumer
         *      The value of this configuration option.
         */
        void setExclusiveConsumer(bool exclusiveConsumer);

        /**
         * Returns whether Message acknowledgments are sent asynchronously meaning no
         * response is required from the broker before the ack completes.
         *
         * @return the sendAcksAsync configured value.
         */
        bool isSendAcksAsync() const;

        /**
         * Sets whether Message acknowledgments are sent asynchronously meaning no
         * response is required from the broker before the ack completes.
         *
         * @param sendAcksAsync
         *      The sendAcksAsync configuration value to set.
         */
        void setSendAcksAsync(bool sendAcksAsync);

        /**
         * @return Returns the alwaysSessionAsync configuration setting.
         */
        bool isAlwaysSessionAsync() const;

        /**
         * If this flag is not set then a separate thread is not used for dispatching messages
         * for each Session in the Connection. However, a separate thread is always used if there
         * is more than one session, or the session isn't in auto acknowledge or duplicates ok mode.
         * By default this value is set to true and session dispatch happens asynchronously.
         */
        void setAlwaysSessionAsync(bool alwaysSessionAsync);

        /**
         * @return true if the consumer will skip checking messages for expiration.
         */
        bool isConsumerExpiryCheckEnabled();

        /**
         * Configures whether this consumer will perform message expiration processing
         * on all incoming messages.  This feature is enabled by default.
         *
         * @param consumerExpiryCheckEnabled
         *      False if the default message expiration checks should be disabled.
         */
        void setConsumerExpiryCheckEnabled(bool consumerExpiryCheckEnabled);

        /**
         * @return the current connection's OpenWire protocol version.
         */
        int getProtocolVersion() const;

    public: // TransportListener

        /**
         * Adds a transport listener so that a client can be notified of events in
         * the underlying transport, client's are always notified after the event has
         * been processed by the Connection class.  Client's should ensure that the
         * registered listener does not block or take a long amount of time to execute
         * in order to not degrade performance of this Connection.
         *
         * @param transportListener
         *      The TransportListener instance to add to this Connection's set of listeners
         *      to notify of Transport events.
         */
        void addTransportListener(transport::TransportListener* transportListener);

        /**
         * Removes a registered TransportListener from the Connection's set of Transport
         * listeners, this listener will no longer receive any Transport related events.  The
         * caller is responsible for freeing the listener in all cases.
         *
         * @param transportListener
         *      The pointer to the TransportListener to remove from the set of listeners.
         */
        void removeTransportListener(transport::TransportListener* transportListener);

        /**
         * Event handler for the receipt of a non-response command from the
         * transport.
         * @param command the received command object.
         */
        virtual void onCommand(const Pointer<commands::Command> command);

        /**
         * Event handler for an exception from a command transport.
         * @param ex The exception.
         */
        virtual void onException(const decaf::lang::Exception& ex);

        /**
         * The transport has suffered an interruption from which it hopes to recover
         */
        virtual void transportInterrupted();

        /**
         * The transport has resumed after an interruption
         */
        virtual void transportResumed();

    public:

        /**
         * Gets the ConnectionInfo for this Object, if the Connection is not open
         * than this method throws an exception.
         *
         * @throws ActiveMQException if an error occurs while performing this operation.
         */
        const commands::ConnectionInfo& getConnectionInfo() const;

        /**
         * Gets the ConnectionId for this Object, if the Connection is not open
         * than this method throws an exception.
         *
         * @throws ActiveMQException if an error occurs while performing this operation.
         */
        const commands::ConnectionId& getConnectionId() const;

        /**
         * Gets a reference to this object's Transport instance.
         *
         * @return a reference to the Transport that is in use by this Connection.
         */
        transport::Transport& getTransport() const;

        /**
         * Gets a reference to the Connection objects built in Scheduler instance.
         *
         * @return a reference to a Scheduler instance owned by this Connection.
         */
        Pointer<threads::Scheduler> getScheduler() const;

        /**
         * Returns the Id of the Resource Manager that this client will use should
         * it be entered into an XA Transaction.
         *
         * @return a string containing the resource manager Id for XA Transactions.
         */
        std::string getResourceManagerId() const;

        /**
         * Clean up this connection object, reseting it back to a state that mirrors
         * what a newly created ActiveMQConnection object has.
         */
        void cleanup();

        /**
         * Sends a message without request that the broker send a response to indicate that
         * it was received.
         *
         * @param command
         *      The Command object to send to the Broker.
         *
         * @throws ActiveMQException if not currently connected, or if the operation
         *         fails for any reason.
         */
        void oneway(Pointer<commands::Command> command);

        /**
         * Sends a synchronous request and returns the response from the broker.  This
         * method converts any error responses it receives into an exception.
         *
         * @param command
         *      The Command object that is to be sent to the broker.
         * @param timeout
         *      The time in milliseconds to wait for a response, default is zero or infinite.
         *
         * @return a Pointer instance to the Response object sent from the Broker.
         *
         * @throws BrokerException if the response from the broker is of type ExceptionResponse.
         * @throws ActiveMQException if any other error occurs while sending the Command.
         */
        Pointer<commands::Response> syncRequest(Pointer<commands::Command> command, unsigned int timeout = 0);

        /**
         * Sends a synchronous request and returns the response from the broker.  This
         * method converts any error responses it receives into an exception.
         *
         * @param command
         *      The Command object that is to be sent to the broker.
         * @param onComplete
         *      Completion callback that will be notified on send success or failure.
         *
         * @throws BrokerException if the response from the broker is of type ExceptionResponse.
         * @throws ActiveMQException if any other error occurs while sending the Command.
         */
        void asyncRequest(Pointer<commands::Command> command, cms::AsyncCallback* onComplete);

        /**
         * Notify the exception listener
         * @param ex the exception to fire
         */
        virtual void fire(const exceptions::ActiveMQException& ex);

        /**
         * Indicates that a Connection resource that is processing the transportInterrupted
         * event has completed.
         */
        void setTransportInterruptionProcessingComplete();

        /**
         * Sets the pointer to the first exception that caused the Connection to become failed.
         *
         * @param error
         *      pointer to the exception instance that is to be the first failure error if the
         *      first error is already set this value is deleted.
         */
        void setFirstFailureError(decaf::lang::Exception* error);

        /**
         * Gets the pointer to the first exception that caused the Connection to become failed.
         *
         * @return pointer to an Exception instance or NULL if none is set.
         */
        decaf::lang::Exception* getFirstFailureError() const;

        /**
         * Event handler for dealing with async exceptions.
         *
         * @param ex
         *      The exception that caused the error condition.
         */
        void onAsyncException(const decaf::lang::Exception& ex);

        /**
         * Handles async client internal exceptions which don't usually affect the connection
         * itself.  These are reported but do not shutdown the Connection.
         *
         * @param error the exception that the problem
         */
        void onClientInternalException(const decaf::lang::Exception& ex);

        /**
         * Check for Closed State and Throw an exception if true.
         *
         * @throws CMSException if the Connection is closed.
         */
        void checkClosed() const;

        /**
         * Check for Closed State and Failed State and Throw an exception if either is true.
         *
         * @throws CMSException if the Connection is closed or failed.
         */
        void checkClosedOrFailed() const;

        /**
         * If its not been sent, then send the ConnectionInfo to the Broker.
         */
        void ensureConnectionInfoSent();

        /**
         * @return the ExecutorService used to run jobs for this Connection
         */
        decaf::util::concurrent::ExecutorService* getExecutor() const;

        /**
         * Adds the given Temporary Destination to this Connections collection of known
         * Temporary Destinations.
         *
         * @param destination
         *      The temporary destination that this connection should track.
         */
        void addTempDestination(Pointer<commands::ActiveMQTempDestination> destination);

        /**
         * Removes the given Temporary Destination to this Connections collection of known
         * Temporary Destinations.
         *
         * @param destination
         *      The temporary destination that this connection should stop tracking.
         */
        void removeTempDestination(Pointer<commands::ActiveMQTempDestination> destination);

        /**
         * Removes the given Temporary Destination to this Connections collection of known
         * Temporary Destinations.
         *
         * @param destination
         *      The temporary destination that this connection should remove from the Broker.
         *
         * @throws CMSException if the temporary destination is in use by an active Session.
         */
        void deleteTempDestination(Pointer<commands::ActiveMQTempDestination> destination);

        /**
         * Removes any TempDestinations that this connection has cached, ignoring any exceptions
         * generated because the destination is in use as they should not be removed.  This method
         * is useful for Connection pools that retain connection objects for long durations and
         * want to periodically purge old temporary destination instances this connection is tracking.
         */
        void cleanUpTempDestinations();

        /**
         * Determines whether the supplied Temporary Destination has already been deleted from the
         * Broker.  If watchTopicAdvisories is disabled this method will always return false.
         *
         * @return true if the temporary destination was deleted already.
         */
        bool isDeleted(Pointer<commands::ActiveMQTempDestination> destination) const;

        /**
         * Returns an ArrayList that contains a copy of all Sessions that are
         * currently active in the Connection
         *
         * @return an ArrayList of Sessions active in this connection.
         */
        decaf::util::ArrayList< Pointer<activemq::core::kernels::ActiveMQSessionKernel> > getSessions() const;

    protected:

        /**
         * @return the next available Session Id.
         */
        virtual Pointer<commands::SessionId> getNextSessionId();

        // Sends a oneway disconnect message to the broker.
        void disconnect(long long lastDeliveredSequenceId);

        // Waits for all Consumers to handle the Transport Interrupted event.
        void waitForTransportInterruptionProcessingToComplete();

        // Marks processing complete for a single caller when interruption processing completes.
        void signalInterruptionProcessingComplete();

        // Allow subclasses to access the original Properties object for this connection.
        const decaf::util::Properties& getProperties() const;

        // Process the WireFormatInfo command
        void onWireFormatInfo(Pointer<commands::Command> command);

        // Process the ControlCommand command
        void onControlCommand(Pointer<commands::Command> command);

        // Process the ConnectionControl command
        void onConnectionControl(Pointer<commands::Command> command);

        // Process the ConsumerControl command
        void onConsumerControl(Pointer<commands::Command> command);

    };

}}

#endif /*_ACTIVEMQ_CORE_ACTIVEMQCONNECTION_H_*/