src/main/activemq/cmsutil/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() throw() {}

            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() throw() {}

            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() throw() {}

            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() throw() {}

            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() throw() {}

            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() throw();

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