ConnectionStateTracker.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_STATE_CONNECTIONSTATETRACKER_H_
#define _ACTIVEMQ_STATE_CONNECTIONSTATETRACKER_H_

#include <activemq/util/Config.h>
#include <activemq/commands/ConnectionId.h>
#include <activemq/exceptions/ActiveMQException.h>
#include <activemq/state/CommandVisitorAdapter.h>
#include <activemq/state/ConnectionState.h>
#include <activemq/state/ConsumerState.h>
#include <activemq/state/ProducerState.h>
#include <activemq/state/SessionState.h>
#include <activemq/state/TransactionState.h>
#include <activemq/state/Tracked.h>
#include <activemq/transport/Transport.h>

#include <decaf/util/concurrent/ConcurrentStlMap.h>
#include <decaf/lang/Pointer.h>

namespace activemq {
namespace state {

    class RemoveTransactionAction;
    using decaf::lang::Pointer;
    using decaf::util::concurrent::ConcurrentStlMap;

    class AMQCPP_API ConnectionStateTracker: public CommandVisitorAdapter {
    private:

        const Pointer<Tracked> TRACKED_RESPONSE_MARKER;

        ConcurrentStlMap<Pointer<ConnectionId>, Pointer<ConnectionState>, ConnectionId::COMPARATOR> connectionStates;

        // TODO - The Map doesn't have a way to automatically remove the eldest Entry
        //        Either we need to implement something similar to LinkedHashMap or find
        //        some other way of tracking the eldest entry into the map and removing it
        //        if the cache size is exceeded.
        ConcurrentStlMap<Pointer<MessageId>, Pointer<Command>, MessageId::COMPARATOR> messageCache;

        ConcurrentStlMap<std::string, Pointer<Command> > messagePullCache;

        bool trackTransactions;
        bool restoreSessions;
        bool restoreConsumers;
        bool restoreProducers;
        bool restoreTransaction;
        bool trackMessages;
        bool trackTransactionProducers;
        int maxCacheSize;
        int currentCacheSize;

        friend class RemoveTransactionAction;

    public:

        ConnectionStateTracker();

        virtual ~ConnectionStateTracker();

        Pointer<Tracked> track(Pointer<Command> command);

        void trackBack(Pointer<Command> command);

        void restore(Pointer<transport::Transport> transport);

        void connectionInterruptProcessingComplete(
            transport::Transport* transport, Pointer<ConnectionId> connectionId);

        void transportInterrupted();

        virtual Pointer<Command> processDestinationInfo(DestinationInfo* info);

        virtual Pointer<Command> processRemoveDestination(DestinationInfo* info);

        virtual Pointer<Command> processProducerInfo(ProducerInfo* info);

        virtual Pointer<Command> processRemoveProducer(ProducerId* id);

        virtual Pointer<Command> processConsumerInfo(ConsumerInfo* info);

        virtual Pointer<Command> processRemoveConsumer(ConsumerId* id);

        virtual Pointer<Command> processSessionInfo(SessionInfo* info);

        virtual Pointer<Command> processRemoveSession(SessionId* id);

        virtual Pointer<Command> processConnectionInfo(ConnectionInfo* info);

        virtual Pointer<Command> processRemoveConnection(ConnectionId* id);

        virtual Pointer<Command> processMessage(Message* message);

        virtual Pointer<Command> processBeginTransaction(TransactionInfo* info);

        virtual Pointer<Command> processPrepareTransaction(TransactionInfo* info);

        virtual Pointer<Command> processCommitTransactionOnePhase(TransactionInfo* info);

        virtual Pointer<Command> processCommitTransactionTwoPhase(TransactionInfo* info);

        virtual Pointer<Command> processRollbackTransaction(TransactionInfo* info);

        virtual Pointer<Command> processEndTransaction(TransactionInfo* info);

        virtual Pointer<Command> processMessagePull(MessagePull* pull);

        bool isRestoreConsumers() const {
            return this->restoreConsumers;
        }

        void setRestoreConsumers(bool restoreConsumers) {
            this->restoreConsumers = restoreConsumers;
        }

        bool isRestoreProducers() const {
            return this->restoreProducers;
        }

        void setRestoreProducers(bool restoreProducers) {
            this->restoreProducers = restoreProducers;
        }

        bool isRestoreSessions() const {
            return this->restoreSessions;
        }

        void setRestoreSessions(bool restoreSessions) {
            this->restoreSessions = restoreSessions;
        }

        bool isTrackTransactions() const {
            return this->trackTransactions;
        }

        void setTrackTransactions(bool trackTransactions) {
            this->trackTransactions = trackTransactions;
        }

        bool isRestoreTransaction() const {
            return this->restoreTransaction;
        }

        void setRestoreTransaction(bool restoreTransaction) {
            this->restoreTransaction = restoreTransaction;
        }

        bool isTrackMessages() const {
            return this->trackMessages;
        }

        void setTrackMessages(bool trackMessages) {
            this->trackMessages = trackMessages;
        }

        int getMaxCacheSize() const {
            return this->maxCacheSize;
        }

        void setMaxCacheSize(int maxCacheSize) {
            this->maxCacheSize = maxCacheSize;
        }

        bool isTrackTransactionProducers() const {
            return this->trackTransactionProducers;
        }

        void setTrackTransactionProducers(bool trackTransactionProducers) {
            this->trackTransactionProducers = trackTransactionProducers;
        }

    private:

        void doRestoreTransactions(Pointer<transport::Transport> transport, Pointer<ConnectionState> connectionState);

        void doRestoreSessions(Pointer<transport::Transport> transport, Pointer<ConnectionState> connectionState);

        void doRestoreConsumers(Pointer<transport::Transport> transport, Pointer<SessionState> sessionState);

        void doRestoreProducers(Pointer<transport::Transport> transport, Pointer<SessionState> sessionState);

        void doRestoreTempDestinations(Pointer<transport::Transport> transport, Pointer<ConnectionState> connectionState);

    };

}}

#endif /*_ACTIVEMQ_STATE_CONNECTIONSTATETRACKER_H_*/