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