ActiveMQSessionKernel.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_KERNELS_ACTIVEMQSESSIONKERNEL_H_
#define _ACTIVEMQ_CORE_KERNELS_ACTIVEMQSESSIONKERNEL_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/core/kernels/ActiveMQConsumerKernel.h>
#include <activemq/core/kernels/ActiveMQProducerKernel.h>
#include <activemq/commands/ActiveMQTempDestination.h>
#include <activemq/commands/Response.h>
#include <activemq/commands/MessageAck.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/ArrayList.h>
#include <decaf/util/Properties.h>
#include <decaf/util/concurrent/atomic/AtomicBoolean.h>
#include <decaf/util/concurrent/atomic/AtomicInteger.h>

#include <string>
#include <memory>

namespace activemq {
namespace core {

    class ActiveMQConnection;
    class ActiveMQConsumer;
    class ActiveMQProducer;
    class ActiveMQSessionExecutor;

namespace kernels {

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

    class SessionConfig;

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

        friend class activemq::core::ActiveMQSessionExecutor;

    protected:

        SessionConfig* config;

        Pointer<commands::SessionInfo> sessionInfo;

        Pointer<ActiveMQTransactionContext> transaction;

        ActiveMQConnection* connection;

        AtomicBoolean closed;

        std::auto_ptr<ActiveMQSessionExecutor> executor;

        cms::Session::AcknowledgeMode ackMode;

        util::LongSequenceGenerator producerIds;

        util::LongSequenceGenerator producerSequenceIds;

        util::LongSequenceGenerator consumerIds;

        long long lastDeliveredSequenceId;

    private:

        ActiveMQSessionKernel(const ActiveMQSessionKernel&);
        ActiveMQSessionKernel& operator=(const ActiveMQSessionKernel&);

    public:

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

        virtual ~ActiveMQSessionKernel();

        virtual void redispatch(MessageDispatchChannel& unconsumedMessages);

        virtual void start();

        virtual 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:   // ActiveMQSessionKernel specific Methods

        void send(kernels::ActiveMQProducerKernel* producer, Pointer<commands::ActiveMQDestination> destination,
                  cms::Message* message, int deliveryMode, int priority, long long timeToLive,
                  util::MemoryUsage* producerWindow, long long sendTimeout, cms::AsyncCallback* onComplete);

        cms::ExceptionListener* getExceptionListener();

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

        virtual cms::MessageTransformer* getMessageTransformer() const;

        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(Pointer<ActiveMQConsumerKernel> consumer);

        void removeConsumer(Pointer<ActiveMQConsumerKernel> consumer);

        void addProducer(Pointer<ActiveMQProducerKernel> producer);

        void removeProducer(Pointer<ActiveMQProducerKernel> producer);

        virtual void doStartTransaction();

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

        void acknowledge();

        void deliverAcks();

        void clearMessagesInProgress(
            decaf::lang::Pointer<decaf::util::concurrent::atomic::AtomicInteger> transportsInterrupted);

        void wakeup();

        Pointer<commands::ConsumerId> getNextConsumerId();

        Pointer<commands::ProducerId> getNextProducerId();

        void doClose();

        void dispose();

        void setPrefetchSize(Pointer<commands::ConsumerId> id, int prefetch);

        void close(Pointer<commands::ConsumerId> id);

        bool isInUse(Pointer<commands::ActiveMQDestination> destination);

        Pointer<ActiveMQProducerKernel> lookupProducerKernel(Pointer<commands::ProducerId> id);

        Pointer<ActiveMQConsumerKernel> lookupConsumerKernel(Pointer<commands::ConsumerId> id);

        bool iterateConsumers();

        void checkMessageListener() const;

        virtual int getHashCode() const;

        void sendAck(decaf::lang::Pointer<commands::MessageAck> ack, bool async = false);

        bool isSessionAsyncDispatch() const;

        void setSessionAsyncDispatch(bool sessionAsyncDispatch);

        decaf::util::ArrayList< Pointer<ActiveMQConsumerKernel> > getConsumers() const;

   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.
       // @return a unique Temporary Destination name
       std::string createTemporaryDestinationName();

    };

}}}

#endif /* _ACTIVEMQ_CORE_KERNELS_ACTIVEMQSESSIONKERNEL_H_ */