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