FailoverTransport.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 _ACTIVE_TRANSPORT_FAILOVER_FAILOVERTRANSPORT_H_
#define _ACTIVE_TRANSPORT_FAILOVER_FAILOVERTRANSPORT_H_

#include <activemq/util/Config.h>

#include <activemq/commands/Command.h>
#include <activemq/commands/ConnectionId.h>
#include <activemq/threads/TaskRunner.h>
#include <activemq/threads/CompositeTaskRunner.h>
#include <activemq/state/ConnectionStateTracker.h>
#include <activemq/transport/CompositeTransport.h>
#include <activemq/wireformat/WireFormat.h>

#include <decaf/util/List.h>
#include <decaf/util/Properties.h>
#include <decaf/net/URI.h>
#include <decaf/io/IOException.h>

namespace activemq {
namespace transport {
namespace failover {

    using namespace decaf::lang;
    using activemq::commands::Command;
    using activemq::commands::Response;

    class FailoverTransportListener;
    class BackupTransportPool;
    class FailoverTransportImpl;

    class AMQCPP_API FailoverTransport : public CompositeTransport,
                                         public activemq::threads::CompositeTask {
    private:

        friend class FailoverTransportListener;
        friend class BackupTransportPool;

        state::ConnectionStateTracker stateTracker;

        FailoverTransportImpl* impl;

    private:

        FailoverTransport(const FailoverTransport&);
        FailoverTransport& operator=(const FailoverTransport&);

    public:

        FailoverTransport();

        virtual ~FailoverTransport();

        void reconnect(bool rebalance);

        void add(bool rebalance, const std::string& uri);

    public: // CompositeTransport methods

        virtual void addURI(bool rebalance, const List<decaf::net::URI>& uris);

        virtual void removeURI(bool rebalance, const List<decaf::net::URI>& uris);

    public:

        virtual void start();

        virtual void stop();

        virtual void close();

        virtual void oneway(const Pointer<Command> command);

        virtual Pointer<FutureResponse> asyncRequest(const Pointer<Command> command,
                                                     const Pointer<ResponseCallback> responseCallback);

        virtual Pointer<Response> request(const Pointer<Command> command);

        virtual Pointer<Response> request(const Pointer<Command> command, unsigned int timeout);

        virtual Pointer<wireformat::WireFormat> getWireFormat() const;

        virtual void setWireFormat(const Pointer<wireformat::WireFormat> wireFormat AMQCPP_UNUSED) {}

        virtual void setTransportListener(TransportListener* listener);

        virtual TransportListener* getTransportListener() const;

        virtual bool isFaultTolerant() const {
            return true;
        }

        virtual bool isConnected() const;

        virtual bool isClosed() const;

        bool isInitialized() const;

        void setInitialized(bool value);

        virtual Transport* narrow(const std::type_info& typeId);

        virtual std::string getRemoteAddress() const;

        virtual void reconnect(const decaf::net::URI& uri);

        virtual void updateURIs(bool rebalance, const decaf::util::List<decaf::net::URI>& uris);

    public:

        virtual bool isPending() const;

        virtual bool iterate();

    public:

        long long getTimeout() const;

        void setTimeout(long long value);

        long long getInitialReconnectDelay() const;

        void setInitialReconnectDelay(long long value);

        long long getMaxReconnectDelay() const;

        void setMaxReconnectDelay(long long value);

        long long getBackOffMultiplier() const;

        void setBackOffMultiplier(long long value);

        bool isUseExponentialBackOff() const;

        void setUseExponentialBackOff(bool value);

        bool isRandomize() const;

        void setRandomize(bool value);

        int getMaxReconnectAttempts() const;

        void setMaxReconnectAttempts(int value);

        int getStartupMaxReconnectAttempts() const;

        void setStartupMaxReconnectAttempts(int value);

        long long getReconnectDelay() const;

        void setReconnectDelay(long long value);

        bool isBackup() const;

        void setBackup(bool value);

        int getBackupPoolSize() const;

        void setBackupPoolSize(int value);

        bool isTrackMessages() const;

        void setTrackMessages(bool value);

        bool isTrackTransactionProducers() const;

        void setTrackTransactionProducers(bool value);

        int getMaxCacheSize() const;

        void setMaxCacheSize(int value);

        int getMaxPullCacheSize() const;

        void setMaxPullCacheSize(int value);

        bool isReconnectSupported() const;

        void setReconnectSupported(bool value);

        bool isUpdateURIsSupported() const;

        void setUpdateURIsSupported(bool value);

        bool isRebalanceUpdateURIs() const;

        void setRebalanceUpdateURIs(bool rebalanceUpdateURIs);

        bool isPriorityBackup() const;

        void setPriorityBackup(bool priorityBackup);

        void setPriorityURIs(const std::string& priorityURIs);

        const decaf::util::List<decaf::net::URI>& getPriorityURIs() const;

        void setConnectionInterruptProcessingComplete(const Pointer<commands::ConnectionId> connectionId);

        bool isConnectedToPriority() const;

    protected:

        void restoreTransport(const Pointer<Transport> transport);

        void handleTransportFailure(const decaf::lang::Exception& error);

        void handleConnectionControl(const Pointer<Command> control);

    private:

        Pointer<Transport> createTransport(const decaf::net::URI& location) const;

        void processNewTransports(bool rebalance, std::string newTransports);

        void processResponse(const Pointer<Response> response);

    };

}}}

#endif /* _ACTIVE_TRANSPORT_FAILOVER_FAILOVERTRANSPORT_H_ */