src/main/activemq/core/ActiveMQConnection.h

Go to the documentation of this file.

/*
 * 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/Connection.h>
#include <activemq/util/Config.h>
#include <activemq/core/ActiveMQConnectionMetaData.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 <decaf/util/Properties.h>
#include <decaf/util/concurrent/atomic/AtomicBoolean.h>
#include <decaf/util/concurrent/CopyOnWriteArrayList.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;
    using decaf::util::concurrent::atomic::AtomicBoolean;

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

    class AMQCPP_API ActiveMQConnection : public virtual cms::Connection,
                                          public transport::TransportListener {
    private:

        ConnectionConfig* config;

        std::auto_ptr<cms::ConnectionMetaData> connectionMetaData;

        AtomicBoolean started;

        AtomicBoolean closed;

        AtomicBoolean closing;

        AtomicBoolean transportFailed;

    private:

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

    public:

        ActiveMQConnection( const Pointer<transport::Transport>& transport,
                            const Pointer<decaf::util::Properties>& properties );

        virtual ~ActiveMQConnection() throw();

        virtual void addSession( ActiveMQSession* session );

        virtual void removeSession( ActiveMQSession* session );

        virtual void addProducer( ActiveMQProducer* producer );

        virtual void removeProducer( const Pointer<commands::ProducerId>& producerId );

        virtual void addDispatcher( const Pointer<commands::ConsumerId>& consumer, Dispatcher* dispatcher );

        virtual void removeDispatcher( const Pointer<commands::ConsumerId>& consumer );

        virtual void sendPullRequest( const commands::ConsumerInfo* consumer, long long timeout );

        bool isClosed() const {
            return this->closed.get();
        }

        bool isStarted() const {
            return this->started.get();
        }

        bool isTransportFailed() const {
            return this->transportFailed.get();
        }

        virtual void destroyDestination( const commands::ActiveMQDestination* destination );

        virtual void destroyDestination( const cms::Destination* destination );

    public:   // Connection Interface Methods

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

        virtual cms::Session* createSession();

        virtual std::string getClientID() const;

        virtual void setClientID( const std::string& clientID );

        virtual cms::Session* createSession( cms::Session::AcknowledgeMode ackMode );

        virtual void close();

        virtual void start();

        virtual void stop();

        virtual cms::ExceptionListener* getExceptionListener() const;

        virtual void setExceptionListener( cms::ExceptionListener* listener );

    public:   // Configuration Options

        void setUsername( const std::string& username );

        const std::string& getUsername() const;

        void setPassword( const std::string& password );

        const std::string& getPassword() const;

        void setDefaultClientId( const std::string& clientId );

        void setBrokerURL( const std::string& brokerURL );

        const std::string& getBrokerURL() const;

        void setPrefetchPolicy( PrefetchPolicy* policy );

        PrefetchPolicy* getPrefetchPolicy() const;

        void setRedeliveryPolicy( RedeliveryPolicy* policy );

        RedeliveryPolicy* getRedeliveryPolicy() const;

        bool isDispatchAsync() const;

        void setDispatchAsync( bool value );

        bool isAlwaysSyncSend() const;

        void setAlwaysSyncSend( bool value );

        bool isUseAsyncSend() const;

        void setUseAsyncSend( bool value );

        bool isUseCompression() const;

        void setUseCompression( bool value );

        void setCompressionLevel( int value );

        int getCompressionLevel() const;

        unsigned int getSendTimeout() const;

        void setSendTimeout( unsigned int timeout );

        unsigned int getCloseTimeout() const;

        void setCloseTimeout( unsigned int timeout );

        unsigned int getProducerWindowSize() const;

        void setProducerWindowSize( unsigned int windowSize );

        bool isMessagePrioritySupported() const;

        void setMessagePrioritySupported( bool value );

        long long getNextTempDestinationId();

        long long getNextLocalTransactionId();

    public: // TransportListener

        void addTransportListener( transport::TransportListener* transportListener );

        void removeTransportListener( transport::TransportListener* transportListener );

        virtual void onCommand( const Pointer<commands::Command>& command );

        virtual void onException( const decaf::lang::Exception& ex );

        virtual void transportInterrupted();

        virtual void transportResumed();

    public:

        const commands::ConnectionInfo& getConnectionInfo() const;

        const commands::ConnectionId& getConnectionId() const;

        transport::Transport& getTransport() const;

        Pointer<threads::Scheduler> getScheduler() const;

        std::string getResourceManagerId() const;

        void cleanup();

        void oneway( Pointer<commands::Command> command );

        Pointer<commands::Response> syncRequest( Pointer<commands::Command> command, unsigned int timeout = 0 );

        virtual void fire( const exceptions::ActiveMQException& ex );

        void setTransportInterruptionProcessingComplete();

        decaf::lang::Exception* getFirstFailureError() const;

        void onAsyncException( const decaf::lang::Exception& ex );

        void checkClosed() const;

        void checkClosedOrFailed() const;

        void ensureConnectionInfoSent();

    protected:

        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;

    };

}}

#endif /*_ACTIVEMQ_CORE_ACTIVEMQCONNECTION_H_*/