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

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

        ConnectionConfig* config;

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

        decaf::util::concurrent::atomic::AtomicBoolean started;

        decaf::util::concurrent::atomic::AtomicBoolean closed;

        decaf::util::concurrent::atomic::AtomicBoolean closing;

        decaf::util::concurrent::atomic::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();

        virtual void addSession(Pointer<activemq::core::kernels::ActiveMQSessionKernel> session);

        virtual void removeSession(Pointer<activemq::core::kernels::ActiveMQSessionKernel> session);

        virtual void addProducer(Pointer<kernels::ActiveMQProducerKernel> 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);

        virtual void setMessageTransformer(cms::MessageTransformer* transformer);

        virtual cms::MessageTransformer* getMessageTransformer() const;

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

        bool isWatchTopicAdvisories() const;

        void setWatchTopicAdvisories(bool value);

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

        void asyncRequest(Pointer<commands::Command> command, cms::AsyncCallback* onComplete);

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

        void setTransportInterruptionProcessingComplete();

        void setFirstFailureError(decaf::lang::Exception* error);

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

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

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

        void checkClosed() const;

        void checkClosedOrFailed() const;

        void ensureConnectionInfoSent();

        decaf::util::concurrent::ExecutorService* getExecutor() const;

        void addTempDestination(Pointer<commands::ActiveMQTempDestination> destination);

        void removeTempDestination(Pointer<commands::ActiveMQTempDestination> destination);

        void deleteTempDestination(Pointer<commands::ActiveMQTempDestination> destination);

        void cleanUpTempDestinations();

        bool isDeleted(Pointer<commands::ActiveMQTempDestination> destination) const;

    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;

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