src/main/activemq/core/ActiveMQSession.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_ACTIVEMQSESSION_H_
#define _ACTIVEMQ_CORE_ACTIVEMQSESSION_H_

#include <cms/Session.h>
#include <cms/ExceptionListener.h>

#include <activemq/util/Config.h>
#include <activemq/util/Usage.h>
#include <activemq/exceptions/ActiveMQException.h>
#include <activemq/core/ActiveMQTransactionContext.h>
#include <activemq/commands/ActiveMQTempDestination.h>
#include <activemq/commands/Response.h>
#include <activemq/commands/SessionInfo.h>
#include <activemq/commands/ConsumerInfo.h>
#include <activemq/commands/ConsumerId.h>
#include <activemq/commands/ProducerId.h>
#include <activemq/commands/TransactionId.h>
#include <activemq/core/Dispatcher.h>
#include <activemq/core/MessageDispatchChannel.h>
#include <activemq/util/LongSequenceGenerator.h>
#include <activemq/threads/Scheduler.h>

#include <decaf/lang/Pointer.h>
#include <decaf/util/StlMap.h>
#include <decaf/util/Properties.h>
#include <decaf/util/concurrent/atomic/AtomicBoolean.h>
#include <decaf/util/concurrent/CopyOnWriteArrayList.h>

#include <string>
#include <memory>

namespace activemq{
namespace core{

    using decaf::lang::Pointer;
    using decaf::util::concurrent::atomic::AtomicBoolean;

    class SessionConfig;
    class ActiveMQConnection;
    class ActiveMQConsumer;
    class ActiveMQMessage;
    class ActiveMQProducer;
    class ActiveMQConsumer;
    class ActiveMQSessionExecutor;

    class AMQCPP_API ActiveMQSession : public virtual cms::Session, public Dispatcher {
    private:

        typedef decaf::util::StlMap< Pointer<commands::ConsumerId>,
                                     ActiveMQConsumer*,
                                     commands::ConsumerId::COMPARATOR> ConsumersMap;

        friend class ActiveMQSessionExecutor;

    protected:

        SessionConfig* config;

        Pointer<commands::SessionInfo> sessionInfo;

        Pointer<ActiveMQTransactionContext> transaction;

        ActiveMQConnection* connection;

        ConsumersMap consumers;

        AtomicBoolean closed;

        std::auto_ptr<ActiveMQSessionExecutor> executor;

        cms::Session::AcknowledgeMode ackMode;

        util::LongSequenceGenerator producerIds;

        util::LongSequenceGenerator producerSequenceIds;

        util::LongSequenceGenerator consumerIds;

        long long lastDeliveredSequenceId;

    private:

        ActiveMQSession( const ActiveMQSession& );
        ActiveMQSession& operator= ( const ActiveMQSession& );

    public:

        ActiveMQSession( ActiveMQConnection* connection,
                         const Pointer<commands::SessionId>& id,
                         cms::Session::AcknowledgeMode ackMode,
                         const decaf::util::Properties& properties );

        virtual ~ActiveMQSession() throw();

        virtual void redispatch( MessageDispatchChannel& unconsumedMessages );

        void start();

        void stop();

        bool isStarted() const;

        virtual bool isAutoAcknowledge() const {
            return this->ackMode == cms::Session::AUTO_ACKNOWLEDGE;
        }

        virtual bool isDupsOkAcknowledge() const {
            return this->ackMode == cms::Session::DUPS_OK_ACKNOWLEDGE;
        }

        virtual bool isClientAcknowledge() const {
            return this->ackMode == cms::Session::CLIENT_ACKNOWLEDGE;
        }

        virtual bool isIndividualAcknowledge() const {
            return this->ackMode == cms::Session::INDIVIDUAL_ACKNOWLEDGE;
        }

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

    public:  // Methods from ActiveMQMessageDispatcher

        virtual void dispatch( const Pointer<MessageDispatch>& message );

    public:   // Implements Methods

        virtual void close();

        virtual void commit();

        virtual void rollback();

        virtual void recover();

        virtual cms::MessageConsumer* createConsumer( const cms::Destination* destination );

        virtual cms::MessageConsumer* createConsumer( const cms::Destination* destination,
                                                      const std::string& selector );

        virtual cms::MessageConsumer* createConsumer( const cms::Destination* destination,
                                                      const std::string& selector,
                                                      bool noLocal );

        virtual cms::MessageConsumer* createDurableConsumer( const cms::Topic* destination,
                                                             const std::string& name,
                                                             const std::string& selector,
                                                             bool noLocal = false );

        virtual cms::MessageProducer* createProducer( const cms::Destination* destination );

        virtual cms::QueueBrowser* createBrowser( const cms::Queue* queue );

        virtual cms::QueueBrowser* createBrowser( const cms::Queue* queue, const std::string& selector );

        virtual cms::Queue* createQueue( const std::string& queueName );

        virtual cms::Topic* createTopic( const std::string& topicName );

        virtual cms::TemporaryQueue* createTemporaryQueue();

        virtual cms::TemporaryTopic* createTemporaryTopic();

        virtual cms::Message* createMessage();

        virtual cms::BytesMessage* createBytesMessage();

        virtual cms::BytesMessage* createBytesMessage( const unsigned char* bytes, int bytesSize );

        virtual cms::StreamMessage* createStreamMessage();

        virtual cms::TextMessage* createTextMessage();

        virtual cms::TextMessage* createTextMessage( const std::string& text );

        virtual cms::MapMessage* createMapMessage();

        virtual cms::Session::AcknowledgeMode getAcknowledgeMode() const;

        virtual bool isTransacted() const;

        virtual void unsubscribe( const std::string& name );

   public:   // ActiveMQSession specific Methods

        void send( cms::Message* message, ActiveMQProducer* producer, util::Usage* usage );

        cms::ExceptionListener* getExceptionListener();

        const commands::SessionInfo& getSessionInfo() const {
            this->checkClosed();
            return *( this->sessionInfo );
        }

        const commands::SessionId& getSessionId() const {
            this->checkClosed();
            return *( this->sessionInfo->getSessionId() );
        }

        ActiveMQConnection* getConnection() const {
            return this->connection;
        }

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

        long long getLastDeliveredSequenceId() const {
            return this->lastDeliveredSequenceId;
        }

        void setLastDeliveredSequenceId( long long value ) {
            this->lastDeliveredSequenceId = value;
        }

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

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

        void addConsumer( ActiveMQConsumer* consumer );

        void removeConsumer( const Pointer<commands::ConsumerId>& consumerId );

        void addProducer( ActiveMQProducer* producer );

        void removeProducer( ActiveMQProducer* producer );

        virtual void doStartTransaction();

        Pointer<ActiveMQTransactionContext> getTransactionContext() {
            return this->transaction;
        }

        void acknowledge();

        void deliverAcks();

        void clearMessagesInProgress();

        void wakeup();

        Pointer<commands::ConsumerId> getNextConsumerId();

        Pointer<commands::ProducerId> getNextProducerId();

        void doClose();

        void dispose();

   private:

       long long getNextProducerSequenceId() {
           return this->producerSequenceIds.getNextSequenceId();
       }

       // Checks for the closed state and throws if so.
       void checkClosed() const;

       // Send the Destination Creation Request to the Broker, alerting it
       // that we've created a new Temporary Destination.
       // @param tempDestination - The new Temporary Destination
       void createTemporaryDestination( commands::ActiveMQTempDestination* tempDestination );

       // Send the Destination Destruction Request to the Broker, alerting
       // it that we've removed an existing Temporary Destination.
       // @param tempDestination - The Temporary Destination to remove
       void destroyTemporaryDestination( commands::ActiveMQTempDestination* tempDestination );

       // Creates a new Temporary Destination name using the connection id
       // and a rolling count.
       // @returns a unique Temporary Destination name
       std::string createTemporaryDestinationName();

    };

}}

#endif /*_ACTIVEMQ_CORE_ACTIVEMQSESSION_H_*/