src/main/activemq/wireformat/openwire/OpenWireFormat.h

Go to the documentation of this file.

#ifndef _ACTIVEMQ_WIREFORMAT_OPENWIRE_OPENWIREFORMAT_H_
#define _ACTIVEMQ_WIREFORMAT_OPENWIRE_OPENWIREFORMAT_H_

#include <activemq/util/Config.h>
#include <activemq/commands/WireFormatInfo.h>
#include <activemq/commands/DataStructure.h>
#include <activemq/wireformat/WireFormat.h>
#include <activemq/wireformat/openwire/utils/BooleanStream.h>
#include <decaf/lang/Pointer.h>
#include <decaf/util/Properties.h>
#include <decaf/util/concurrent/atomic/AtomicBoolean.h>
#include <decaf/lang/exceptions/IllegalStateException.h>
#include <decaf/lang/exceptions/IllegalArgumentException.h>
#include <memory>

namespace activemq{
namespace wireformat{
namespace openwire{

namespace marshal {
    class DataStreamMarshaller;
}

    using decaf::lang::Pointer;

    class AMQCPP_API OpenWireFormat : public wireformat::WireFormat {
    protected:

        // Declared here to make life easier.
        static const unsigned char NULL_TYPE;

        // V1 if the default version we start at.
        static const int DEFAULT_VERSION;

        // Defines the maximum supported openwire version
        static const int MAX_SUPPORTED_VERSION;

    private:

        // Configuration parameters
        decaf::util::Properties properties;

        // Preferred WireFormatInfo
        Pointer<commands::WireFormatInfo> preferedWireFormatInfo;

        // Marshalers
        std::vector< marshal::DataStreamMarshaller* > dataMarshallers;

        // Uniquely Generated ID, initialize in the Ctor
        std::string id;

        // Indicates when we are in the doUnmarshal call
        decaf::util::concurrent::atomic::AtomicBoolean receiving;

        // WireFormat Data
        int version;
        bool stackTraceEnabled;
        bool tcpNoDelayEnabled;
        bool cacheEnabled;
        int cacheSize;
        bool tightEncodingEnabled;
        bool sizePrefixDisabled;
        long long maxInactivityDuration;
        long long maxInactivityDurationInitialDelay;

    public:

        OpenWireFormat( const decaf::util::Properties& properties );

        virtual ~OpenWireFormat();

        virtual bool hasNegotiator() const {
            return true;
        }

        virtual Pointer<transport::Transport> createNegotiator(
            const Pointer<transport::Transport>& transport );

        void addMarshaller( marshal::DataStreamMarshaller* marshaler );

        virtual void marshal( const Pointer<commands::Command>& command,
                              const activemq::transport::Transport* transport,
                              decaf::io::DataOutputStream* out );

        virtual Pointer<commands::Command> unmarshal( const activemq::transport::Transport* transport,
                                                      decaf::io::DataInputStream* in );

    public:

        virtual int tightMarshalNestedObject1( commands::DataStructure* object,
                                               utils::BooleanStream* bs );

        void tightMarshalNestedObject2( commands::DataStructure* o,
                                        decaf::io::DataOutputStream* ds,
                                        utils::BooleanStream* bs );

        commands::DataStructure* tightUnmarshalNestedObject( decaf::io::DataInputStream* dis,
                                                             utils::BooleanStream* bs );

        commands::DataStructure* looseUnmarshalNestedObject( decaf::io::DataInputStream* dis );

        void looseMarshalNestedObject( commands::DataStructure* o,
                                       decaf::io::DataOutputStream* dataOut );

        void renegotiateWireFormat( const commands::WireFormatInfo& info );

        void setPreferedWireFormatInfo( const Pointer<commands::WireFormatInfo>& info );

        const Pointer<commands::WireFormatInfo>& getPreferedWireFormatInfo() const {
            return this->preferedWireFormatInfo;
        }

        bool isStackTraceEnabled() const {
            return stackTraceEnabled;
        }

        void setStackTraceEnabled( bool stackTraceEnabled ) {
            this->stackTraceEnabled = stackTraceEnabled;
        }

        bool isTcpNoDelayEnabled() const {
            return tcpNoDelayEnabled;
        }

        void setTcpNoDelayEnabled( bool tcpNoDelayEnabled ) {
            this->tcpNoDelayEnabled = tcpNoDelayEnabled;
        }

        int getVersion() const {
            return version;
        }

        void setVersion( int version );

        virtual bool inReceive() const {
            return this->receiving.get();
        }

        bool isCacheEnabled() const {
            return cacheEnabled;
        }

        void setCacheEnabled( bool cacheEnabled ) {
            this->cacheEnabled = cacheEnabled;
        }

        int getCacheSize() const {
            return cacheSize;
        }

        void setCacheSize( int value ) {
            this->cacheSize = value;
        }

        bool isTightEncodingEnabled() const {
            return tightEncodingEnabled;
        }

        void setTightEncodingEnabled( bool tightEncodingEnabled ) {
            this->tightEncodingEnabled = tightEncodingEnabled;
        }

        bool isSizePrefixDisabled() const {
            return sizePrefixDisabled;
        }

        void setSizePrefixDisabled( bool sizePrefixDisabled ) {
            this->sizePrefixDisabled = sizePrefixDisabled;
        }

        long long getMaxInactivityDuration() const {
            return this->maxInactivityDuration;
        }

        void setMaxInactivityDuration( long long value ) {
            this->maxInactivityDuration = value;
        }

        long long getMaxInactivityDurationInitialDelay() const {
            return this->maxInactivityDurationInitialDelay;
        }

        void setMaxInactivityDurationInitialDelay( long long value ) {
            this->maxInactivityDurationInitialDelay = value;
        }

    protected:

        commands::DataStructure* doUnmarshal( decaf::io::DataInputStream* dis );

        void destroyMarshalers();

    };

}}}

#endif /*_ACTIVEMQ_WIREFORMAT_OPENWIRE_OPENWIREFORMAT_H_*/