CmsTemplate.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_CMSUTIL_CMSTEMPLATE_H_
#define _ACTIVEMQ_CMSUTIL_CMSTEMPLATE_H_

#include <activemq/util/Config.h>
#include <activemq/cmsutil/CmsDestinationAccessor.h>
#include <activemq/cmsutil/SessionCallback.h>
#include <activemq/cmsutil/ProducerCallback.h>
#include <activemq/cmsutil/SessionPool.h>
#include <cms/ConnectionFactory.h>
#include <cms/DeliveryMode.h>
#include <string>

namespace activemq {
namespace cmsutil {

    // Forward declarations.
    class MessageCreator;

    class AMQCPP_API CmsTemplate : public CmsDestinationAccessor {
    public:

        static const long long RECEIVE_TIMEOUT_NO_WAIT;

        static const long long RECEIVE_TIMEOUT_INDEFINITE_WAIT;

        static const int DEFAULT_PRIORITY;

        static const long long DEFAULT_TIME_TO_LIVE;

    public:

        class ProducerExecutor;
        friend class ProducerExecutor;
        class ProducerExecutor : public SessionCallback {
        protected:

            ProducerCallback* action;
            CmsTemplate* parent;
            cms::Destination* destination;

        private:

            ProducerExecutor(const ProducerExecutor&);
            ProducerExecutor& operator=(const ProducerExecutor&);

        public:

            ProducerExecutor(ProducerCallback* action,
                             CmsTemplate* parent,
                             cms::Destination* destination)
                : SessionCallback(), action(action), parent(parent), destination(destination) {
            }

            virtual ~ProducerExecutor() {}

            virtual void doInCms(cms::Session* session);

            virtual cms::Destination* getDestination(cms::Session* session AMQCPP_UNUSED) {
                return destination;
            }
        };

        class ResolveProducerExecutor;
        friend class ResolveProducerExecutor;
        class ResolveProducerExecutor : public ProducerExecutor {
        private:

            std::string destinationName;

        private:

            ResolveProducerExecutor(const ResolveProducerExecutor&);
            ResolveProducerExecutor& operator=(const ResolveProducerExecutor&);

        public:

            ResolveProducerExecutor(ProducerCallback* action,
                                    CmsTemplate* parent,
                                    const std::string& destinationName)
                : ProducerExecutor(action, parent, NULL), destinationName(destinationName) {
            }

            virtual ~ResolveProducerExecutor() {}

            virtual cms::Destination* getDestination(cms::Session* session);
        };

        class SendExecutor;
        friend class SendExecutor;
        class SendExecutor : public ProducerCallback {
        private:

            MessageCreator* messageCreator;
            CmsTemplate* parent;

        private:

            SendExecutor(const SendExecutor&);
            SendExecutor& operator=(const SendExecutor&);

        public:

            SendExecutor(MessageCreator* messageCreator, CmsTemplate* parent) :
                ProducerCallback(), messageCreator(messageCreator), parent(parent) {
            }

            virtual ~SendExecutor() {
            }

            virtual void doInCms(cms::Session* session, cms::MessageProducer* producer) {
                parent->doSend(session, producer, messageCreator);
            }
        };

        class ReceiveExecutor;
        friend class ReceiveExecutor;
        class ReceiveExecutor : public SessionCallback {
        protected:

            cms::Destination* destination;
            std::string selector;
            bool noLocal;
            cms::Message* message;
            CmsTemplate* parent;

        private:

            ReceiveExecutor(const ReceiveExecutor&);
            ReceiveExecutor& operator=(const ReceiveExecutor&);

        public:

            ReceiveExecutor(CmsTemplate* parent, cms::Destination* destination,
                            const std::string& selector, bool noLocal) :
                SessionCallback(), destination(destination), selector(selector), noLocal(noLocal), message(NULL), parent(parent) {
            }

            virtual ~ReceiveExecutor() {}

            virtual void doInCms(cms::Session* session);

            virtual cms::Destination* getDestination(cms::Session* session AMQCPP_UNUSED) {
                return destination;
            }

            cms::Message* getMessage() {
                return message;
            }
        };

        class ResolveReceiveExecutor;
        friend class ResolveReceiveExecutor;
        class ResolveReceiveExecutor : public ReceiveExecutor {
        private:

            std::string destinationName;

        private:

            ResolveReceiveExecutor(const ResolveReceiveExecutor&);
            ResolveReceiveExecutor& operator=(const ResolveReceiveExecutor&);

        public:

            ResolveReceiveExecutor(CmsTemplate* parent, const std::string& selector,
                                   bool noLocal, const std::string& destinationName) :
                ReceiveExecutor(parent, NULL, selector, noLocal), destinationName(destinationName) {
            }

            virtual ~ResolveReceiveExecutor() {}

            virtual cms::Destination* getDestination(cms::Session* session);
        };

    private:

        static const int NUM_SESSION_POOLS = (int)cms::Session::SESSION_TRANSACTED + 1;

        cms::Connection* connection;

        SessionPool* sessionPools[NUM_SESSION_POOLS];

        cms::Destination* defaultDestination;

        std::string defaultDestinationName;

        bool messageIdEnabled;

        bool messageTimestampEnabled;

        bool noLocal;

        long long receiveTimeout;

        bool explicitQosEnabled;

        int deliveryMode;

        int priority;

        long long timeToLive;

        bool initialized;

    private:

        CmsTemplate(const CmsTemplate&);
        CmsTemplate& operator=(const CmsTemplate&);

    public:

        CmsTemplate();
        CmsTemplate(cms::ConnectionFactory* connectionFactory);

        virtual ~CmsTemplate();

        virtual void setDefaultDestination(cms::Destination* defaultDestination) {
            this->defaultDestination = defaultDestination;
        }

        virtual const cms::Destination* getDefaultDestination() const {
            return this->defaultDestination;
        }

        virtual cms::Destination* getDefaultDestination() {
            return this->defaultDestination;
        }

        virtual void setDefaultDestinationName(const std::string& defaultDestinationName) {
            if (defaultDestinationName != this->defaultDestinationName) {
                this->defaultDestination = NULL;
                this->defaultDestinationName = defaultDestinationName;
            }
        }

        virtual const std::string getDefaultDestinationName() const {
            return this->defaultDestinationName;
        }

        virtual void setPubSubDomain(bool pubSubDomain) {
            if (pubSubDomain != isPubSubDomain()) {
                this->defaultDestination = NULL;
                CmsDestinationAccessor::setPubSubDomain(pubSubDomain);
            }
        }

        virtual void setMessageIdEnabled(bool messageIdEnabled) {
            this->messageIdEnabled = messageIdEnabled;
        }

        virtual bool isMessageIdEnabled() const {
            return this->messageIdEnabled;
        }

        virtual void setMessageTimestampEnabled(bool messageTimestampEnabled) {
            this->messageTimestampEnabled = messageTimestampEnabled;
        }

        virtual bool isMessageTimestampEnabled() const {
            return this->messageTimestampEnabled;
        }

        virtual void setNoLocal(bool noLocal) {
            this->noLocal = noLocal;
        }

        virtual bool isNoLocal() const {
            return this->noLocal;
        }

        virtual void setReceiveTimeout(long long receiveTimeout) {
            this->receiveTimeout = receiveTimeout;
        }

        virtual long long getReceiveTimeout() const {
            return this->receiveTimeout;
        }

        virtual void setExplicitQosEnabled(bool explicitQosEnabled) {
            this->explicitQosEnabled = explicitQosEnabled;
        }

        virtual bool isExplicitQosEnabled() const {
            return this->explicitQosEnabled;
        }

        virtual void setDeliveryPersistent(bool deliveryPersistent) {
            this->deliveryMode = (deliveryPersistent ? 0 : 1);
        }

        virtual void setDeliveryMode(int deliveryMode) {
            this->deliveryMode = deliveryMode;
        }

        virtual int getDeliveryMode() const {
            return this->deliveryMode;
        }

        virtual void setPriority(int priority) {
            this->priority = priority;
        }

        virtual int getPriority() const {
            return this->priority;
        }

        virtual void setTimeToLive(long long timeToLive) {
            this->timeToLive = timeToLive;
        }

        virtual long long getTimeToLive() const {
            return this->timeToLive;
        }

        virtual void execute(SessionCallback* action);

        virtual void execute(ProducerCallback* action);

        virtual void execute(cms::Destination* dest, ProducerCallback* action);

        virtual void execute(const std::string& destinationName, ProducerCallback* action);

        virtual void send(MessageCreator* messageCreator);

        virtual void send(cms::Destination* dest, MessageCreator* messageCreator);

        virtual void send(const std::string& destinationName, MessageCreator* messageCreator);

        virtual cms::Message* receive();

        virtual cms::Message* receive(cms::Destination* destination);

        virtual cms::Message* receive(const std::string& destinationName);

        virtual cms::Message* receiveSelected(const std::string& selector);

        virtual cms::Message* receiveSelected(cms::Destination* destination, const std::string& selector);

        virtual cms::Message* receiveSelected(const std::string& destinationName, const std::string& selector);

    protected:

        void init();

        void destroy();

    private:

        void initDefaults();

        void createSessionPools();

        void destroySessionPools();

        void checkDefaultDestination();

        cms::Connection* getConnection();

        PooledSession* takeSession();

        void returnSession(PooledSession*& session);

        cms::MessageProducer* createProducer(cms::Session* session, cms::Destination* dest);

        void destroyProducer(cms::MessageProducer*& producer);

        cms::MessageConsumer* createConsumer(cms::Session* session, cms::Destination* dest,
                                             const std::string& selector, bool noLocal);

        void destroyConsumer(cms::MessageConsumer*& consumer);

        void destroyMessage(cms::Message*& message);

        void doSend(cms::Session* session, cms::MessageProducer* producer, MessageCreator* messageCreator);

        cms::Message* doReceive(cms::MessageConsumer* consumer);

        cms::Destination* resolveDefaultDestination(cms::Session* session);

    };

}}

#endif /*_ACTIVEMQ_CMSUTIL_CMSTEMPLATE_H_*/