Message.hpp

Go to the documentation of this file.

//
// Copyright (c) 2010-2020 60East Technologies Inc., All Rights Reserved.
//
// This computer software is owned by 60East Technologies Inc. and is
// protected by U.S. copyright laws and other laws and by international
// treaties.  This computer software is furnished by 60East Technologies
// Inc. pursuant to a written license agreement and may be used, copied,
// transmitted, and stored only in accordance with the terms of such
// license agreement and with the inclusion of the above copyright notice.
// This computer software or any other copies thereof may not be provided
// or otherwise made available to any other person.
//
// U.S. Government Restricted Rights.  This computer software: (a) was
// developed at private expense and is in all respects the proprietary
// information of 60East Technologies Inc.; (b) was not developed with
// government funds; (c) is a trade secret of 60East Technologies Inc.
// for all purposes of the Freedom of Information Act; and (d) is a
// commercial item and thus, pursuant to Section 12.212 of the Federal
// Acquisition Regulations (FAR) and DFAR Supplement Section 227.7202,
// Government's use, duplication or disclosure of the computer software
// is subject to the restrictions set forth by 60East Technologies Inc..
//
#ifndef __AMPS_MESSAGE_HPP__
#define __AMPS_MESSAGE_HPP__
#include "util.hpp"
#include "constants.hpp"
#include "amps_generated.h"
#include "Field.hpp"
#include <stdio.h>
#ifdef _WIN32
#if (_MSC_VER >= 1400) // VS2005 or higher
#define AMPS_snprintf sprintf_s
#define AMPS_snprintf_amps_uint64_t(buf,sz,val) sprintf_s(buf,sz,"%I64u",val)
#ifdef _WIN64
#define AMPS_snprintf_sizet(buf,sz,val) sprintf_s(buf,sz,"%lu",val)
#else
#define AMPS_snprintf_sizet(buf,sz,val) sprintf_s(buf,sz,"%u",val)
#endif
#else // VS2003 or older
#define AMPS_snprintf _snprintf
#ifdef _WIN64
#define AMPS_snprintf_sizet(buf,sz,val) _sprintf(buf,sz,"%lu",val)
#else
#define AMPS_snprintf_sizet(buf,sz,val) _sprintf(buf,sz,"%u",val)
#endif
#endif
#else
#define AMPS_snprintf snprintf
#ifdef __x86_64__
#define AMPS_snprintf_amps_uint64_t(buf,sz,val) snprintf(buf,sz,"%lu",val)
#define AMPS_snprintf_sizet(buf,sz,val) snprintf(buf,sz,"%lu",val)
#else
#define AMPS_snprintf_amps_uint64_t(buf,sz,val) snprintf(buf,sz,"%llu",val)
#define AMPS_snprintf_sizet(buf,sz,val) snprintf(buf,sz,"%u",val)
#endif
#endif
#define AMPS_UNSET_INDEX    (size_t)-1
#define AMPS_UNSET_SEQUENCE (amps_uint64_t)-1

// Macros for determing what parts of TR1 we can depend on
#if defined(__GXX_EXPERIMENTAL_CXX0X__) || (_MSC_VER >= 1600)
#define AMPS_USE_FUNCTIONAL 1
#endif

#if (_MSC_VER >= 1600) || ( (__GNUC__ >= 4) && (__GNUC_MINOR__) >=5 )
#define AMPS_USE_LAMBDAS 1
#endif

#ifdef AMPS_USE_FUNCTIONAL
#include <functional>
#endif

#include <algorithm>


#define AMPS_OPTIONS_NONE ""
#define AMPS_OPTIONS_LIVE "live,"
#define AMPS_OPTIONS_OOF "oof,"
#define AMPS_OPTIONS_REPLACE "replace,"
#define AMPS_OPTIONS_NOEMPTIES "no_empties,"
#define AMPS_OPTIONS_SENDKEYS "send_keys,"
#define AMPS_OPTIONS_TIMESTAMP "timestamp,"
#define AMPS_OPTIONS_NOSOWKEY "no_sowkey,"
#define AMPS_OPTIONS_CANCEL "cancel,"
#define AMPS_OPTIONS_RESUME "resume,"
#define AMPS_OPTIONS_PAUSE "pause,"
#define AMPS_OPTIONS_FULLY_DURABLE "fully_durable,"
#define AMPS_OPTIONS_EXPIRE "expire,"
#define AMPS_OPTIONS_TOPN(x) "top_n=##x,"
#define AMPS_OPTIONS_MAXBACKLOG(x) "max_backlog=##x,"
#define AMPS_OPTIONS_RATE(x) "rate=##x,"

namespace AMPS
{
typedef void* amps_subscription_handle;

class ClientImpl;

class MessageImpl : public RefBody
{
private:
    amps_handle _message;
    Mutex _lock;
    bool _owner;
    mutable bool _isIgnoreAutoAck;
    size_t _bookmarkSeqNo;
    amps_subscription_handle _subscription;
    ClientImpl* _clientImpl;
public:
    MessageImpl(amps_handle message_, bool owner_ = false,
                bool ignoreAutoAck_ = false, size_t bookmarkSeqNo_ = 0,
                amps_subscription_handle subscription_ = NULL,
                ClientImpl* clientImpl_=NULL)
        : _message(message_), _owner(owner_), _isIgnoreAutoAck(ignoreAutoAck_)
        , _bookmarkSeqNo(bookmarkSeqNo_)
        , _subscription(subscription_), _clientImpl(clientImpl_)
    {
    }

    MessageImpl()
        : _message(NULL), _owner(true), _isIgnoreAutoAck(false), _bookmarkSeqNo(0), _subscription(NULL), _clientImpl(NULL)
    {
        // try to create one
        _message = amps_message_create(NULL);
    }

    virtual ~MessageImpl()
    {
        if (_owner && _message) amps_message_destroy(_message);
    }

    MessageImpl* copy() const
    {
        amps_handle copy = amps_message_copy(_message);
        return new MessageImpl(copy, true, _isIgnoreAutoAck, _bookmarkSeqNo,
                               _subscription,_clientImpl);
    }

    void setClientImpl(ClientImpl* clientImpl_)
    {
      _clientImpl = clientImpl_;
    }

    ClientImpl* clientImpl(void) const
    {
      return _clientImpl;
    }

    amps_handle getMessage() const
    {
        return _message;
    }

    void reset()
    {
        Lock<Mutex> l(_lock);
        amps_message_reset(_message);
        _bookmarkSeqNo = 0;
        _subscription = NULL;
        _isIgnoreAutoAck = false;
    }

    void replace(amps_handle message_, bool owner_ = false)
    {
        Lock<Mutex> l(_lock);
        if (_message == message_) return;
        if (_owner && _message)
        {
            amps_message_destroy(_message);
        }
        _owner = owner_;
        _message = message_;
        _bookmarkSeqNo = 0;
        _subscription = NULL;
        _isIgnoreAutoAck = false;
    }

    void disown()
    {
        Lock<Mutex> l(_lock);
        _owner = false;
    }

    static unsigned long newId()
    {
        static ATOMIC_TYPE _id = 0;
        return (unsigned long)(AMPS_FETCH_ADD(&_id,1));
    }

    void setBookmarkSeqNo(size_t val_)
    {
        _bookmarkSeqNo = val_;
    }

    size_t getBookmarkSeqNo(void) const
    {
        return _bookmarkSeqNo;
    }

    void setSubscriptionHandle(amps_subscription_handle subscription_)
    {
        _subscription = subscription_;
    }

    amps_subscription_handle getSubscriptionHandle(void) const
    {
        return _subscription;
    }

    void setIgnoreAutoAck() const
    {
        _isIgnoreAutoAck = true;
    }

    bool getIgnoreAutoAck() const
    {
        return _isIgnoreAutoAck;
    }
};


// This block of macros works with the Doxygen preprocessor to
// create documentation comments for fields defined with the AMPS_FIELD macro.
// A C++ compiler removes comments before expanding macros, so these macros
// must ONLY be defined for Doxygen and not for actual compilation.

#ifdef DOXYGEN_PREPROCESSOR

#define DOX_COMMENTHEAD(s) / ##  ** ## s ## * ## /
#define DOX_GROUPNAME(s) DOX_COMMENTHEAD(@name s Functions) 
#define DOX_OPENGROUP(s) DOX_COMMENTHEAD(@{) DOX_GROUPNAME(s)
#define DOX_CLOSEGROUP() DOX_COMMENTHEAD(@})
#define DOX_MAKEGETCOMMENT(x) DOX_COMMENTHEAD( Retrieves the value of the x header of the Message as a new Field. Notice that not all headers are present on all messages returned by AMPS. See the AMPS %Command Reference for details on which fields will be present in response to a specific command. )
#define DOX_MAKEGETRAWCOMMENT(x) DOX_COMMENTHEAD( Retrieves the value of the x header of self as a Field that references the underlying buffer managed by this Message. Notice that not all headers are present on all messages returned by AMPS. See the AMPS %Command Reference for details on which fields will be present in response to a specific command. )
#define DOX_MAKESETCOMMENT(x) DOX_COMMENTHEAD( Sets the value of the x header for this Message. Not all headers are processed by AMPS for all commands. See the AMPS %Command Reference for which headers are used by AMPS for a specific command. )
#define DOX_MAKEASSIGNCOMMENT(x) DOX_COMMENTHEAD( Sets the value of the x header for this Message. Not all headers are processed by AMPS for all commands. See the AMPS %Command Reference for which headers are used by AMPS for a specific command. )
#define DOX_MAKENEWCOMMENT(x) DOX_COMMENTHEAD(Creates and sets a new sequential value for the x header for this Message. This function is most useful for headers such as CommandId and SubId.)

#else

#define DOX_COMMENTHEAD(s)
#define DOX_GROUPNAME(s)
#define DOX_OPENGROUP(x)
#define DOX_CLOSEGROUP(x)
#define DOX_MAKEGETCOMMENT(x)
#define DOX_MAKEGETRAWCOMMENT(x)
#define DOX_MAKESETCOMMENT(x)
#define DOX_MAKEASSIGNCOMMENT(x)
#define DOX_MAKENEWCOMMENT(x)

#endif

// Macro for defining all of the necessary methods for a field in an AMPS
// message.

#define AMPS_FIELD(x) \
    DOX_OPENGROUP(x) \
    DOX_MAKEGETCOMMENT(x) \
    Field get##x() const { \
        Field returnValue;\
        const char* ptr;\
        size_t sz;\
        amps_message_get_field_value(_body.get().getMessage(),\
                AMPS_##x, &ptr, &sz);\
        returnValue.assign(ptr, sz);\
        return returnValue;\
    }\
    DOX_MAKEGETRAWCOMMENT(x) \
    void getRaw##x(const char** dataptr, size_t* sizeptr) const { \
        amps_message_get_field_value(_body.get().getMessage(), \
                AMPS_##x, dataptr, sizeptr);\
        return;\
    }\
    DOX_MAKESETCOMMENT(x) \
    Message& set##x(const std::string& v) {\
        amps_message_set_field_value(_body.get().getMessage(),\
                AMPS_##x, v.c_str(), v.length());\
        return *this;\
    }\
    DOX_MAKESETCOMMENT(x) \
    Message& set##x(amps_uint64_t v) {\
        char buf[22];\
        AMPS_snprintf_amps_uint64_t(buf,22,v);\
        amps_message_set_field_value_nts(_body.get().getMessage(),\
                AMPS_##x, buf);\
        return *this;\
    }\
    DOX_MAKEASSIGNCOMMENT(x) \
    Message& assign##x(const std::string& v) {\
        amps_message_assign_field_value(_body.get().getMessage(),\
                AMPS_##x, v.c_str(), v.length());\
        return *this;\
    }\
    DOX_MAKEASSIGNCOMMENT(x) \
    Message& assign##x(const char* data, size_t len) {\
        amps_message_assign_field_value(_body.get().getMessage(),\
                AMPS_##x, data, len);\
        return *this;\
    }\
    DOX_MAKESETCOMMENT(x) \
    Message& set##x(const char* str) {\
        amps_message_set_field_value_nts(_body.get().getMessage(),\
                AMPS_##x, str);\
        return *this;\
    }\
    DOX_MAKESETCOMMENT(x) \
    Message& set##x(const char* str,size_t len) {\
        amps_message_set_field_value(_body.get().getMessage(),\
                AMPS_##x, str,len);\
        return *this;\
    }\
    DOX_MAKENEWCOMMENT(x) \
    Message& new##x() {\
        char buf[Message::IdentifierLength+1];\
        buf[Message::IdentifierLength] = 0;\
        AMPS_snprintf(buf, Message::IdentifierLength+1, "%lx" , _body.get().newId());\
        amps_message_set_field_value_nts(_body.get().getMessage(),\
                AMPS_##x, buf);\
        return *this;\
    } \
    DOX_CLOSEGROUP(x)

class Message
{
    RefHandle<MessageImpl> _body;

    Message(MessageImpl* body_) : _body(body_) { ; }

public:
    typedef AMPS::Field Field;

    static const unsigned int IdentifierLength = 16;

    static const size_t BOOKMARK_NONE = AMPS_UNSET_INDEX;

    Message(amps_handle message_, bool owner_ = false)
        : _body(new MessageImpl(message_, owner_))
    {
    }

    Message() : _body(new MessageImpl())
    {
    }

    Message deepCopy(void) const
    {
        return Message(_body.get().copy());
    }

    class Options
    {
        public:
            static const char* None(void) { return AMPS_OPTIONS_NONE; }
            static const char* Live(void) { return AMPS_OPTIONS_LIVE; }
            static const char* OOF(void) { return AMPS_OPTIONS_OOF; }
            static const char* Replace(void) { return AMPS_OPTIONS_REPLACE; }
            static const char* NoEmpties(void) { return AMPS_OPTIONS_NOEMPTIES; }
            static const char* SendKeys(void) { return AMPS_OPTIONS_SENDKEYS; }
            static const char* Timestamp(void) { return AMPS_OPTIONS_TIMESTAMP; }
            static const char* NoSowKey(void) { return AMPS_OPTIONS_NOSOWKEY; }
            static const char* Cancel(void) { return AMPS_OPTIONS_CANCEL; }
            static const char* Resume(void) { return AMPS_OPTIONS_RESUME; }
            static const char* Pause(void) { return AMPS_OPTIONS_PAUSE; }
            static const char* FullyDurable(void) { return AMPS_OPTIONS_FULLY_DURABLE; }
            static const char* Expire(void) { return AMPS_OPTIONS_EXPIRE; }
            static std::string Conflation(const char* conflation_)
            {
                char buf[64];
                AMPS_snprintf(buf, sizeof(buf), "conflation=%s,", conflation_);
                return buf;
            }
            static std::string ConflationKey(const char* conflationKey_)
            {
                std::string option("conflation_key=");
                option.append(conflationKey_).append(",");
                return option;
            }
            static std::string TopN(int topN_)
            {
                char buf[24];
                AMPS_snprintf(buf, sizeof(buf), "top_n=%d,", topN_);
                return buf;
            }
            static std::string MaxBacklog(int maxBacklog_)
            {
                char buf[24];
                AMPS_snprintf(buf, sizeof(buf), "max_backlog=%d,", maxBacklog_);
                return buf;
            }
            static std::string Rate(const char* rate_)
            {
                char buf[64];
                AMPS_snprintf(buf, sizeof(buf), "rate=%s,", rate_);
                return buf;
            }
            static std::string RateMaxGap(const char* rateMaxGap_)
            {
                char buf[64];
                AMPS_snprintf(buf, sizeof(buf), "rate_max_gap=%s,", rateMaxGap_);
                return buf;
            }
            static std::string SkipN(int skipN_)
            {
                char buf[24];
                AMPS_snprintf(buf, sizeof(buf), "skip_n=%d,", skipN_);
                return buf;
            }

            static std::string Projection(std::string projection_)
            {
                return "projection=[" + projection_ + "],";
            }

            template<class Iterator>
            static std::string Projection(Iterator begin_, Iterator end_)
            {
                std::string projection = "projection=[";
                for (Iterator i = begin_; i != end_; ++i)
                {
                    projection += *i;
                    projection += ',';
                }
                projection.insert(projection.length() - 1, "]");
                return projection;
            }

            static std::string Grouping(std::string grouping_)
            {
                return "grouping=[" + grouping_ + "],";
            }

            template<class Iterator>
            static std::string Grouping(Iterator begin_, Iterator end_)
            {
                std::string grouping = "grouping=[";
                for (Iterator i = begin_; i != end_; ++i)
                {
                    grouping += *i;
                    grouping += ',';
                }
                grouping.insert(grouping.length() - 1, "]");
                return grouping;
            }

            static std::string Select(const std::string& select_)
            {
                return "select=[" + select_ + "],";
            }

            template<class Iterator>
            static std::string Select(Iterator begin_, Iterator end_)
            {
                std::string select = "select=[";
                for (Iterator i = begin_; i != end_; ++i)
                {
                    select += *i;
                    select += ',';
                }
                select.insert(select.length() - 1, "]");
                return select;
            }

            static std::string AckConflationInterval(const std::string& interval_)
            {
                return "ack_conflation=" + interval_ + ",";
            }

            static std::string AckConflationInterval(const char* interval_)
            {
                static const std::string start("ack_conflation=");
                return start + interval_ + ",";
            }

            Options(std::string options_ = "")
                : _optionStr(options_)
                , _maxBacklog(0)
                , _topN(0)
            {;}

            int getMaxBacklog(void) const { return _maxBacklog; }
            std::string getConflation(void) const { return _conflation; }
            std::string getConflationKey(void) const { return _conflationKey; }
            int getTopN(void) const { return _topN; }
            std::string getRate(void) const { return _rate; }
            std::string getRateMaxGap(void) const { return _rateMaxGap; }

            void setNone(void) { _optionStr = AMPS_OPTIONS_NONE; }

            void setLive(void) { _optionStr += AMPS_OPTIONS_LIVE; }

            void setOOF(void) { _optionStr += AMPS_OPTIONS_OOF; }

            void setReplace(void) { _optionStr += AMPS_OPTIONS_REPLACE; }

            void setNoEmpties(void) { _optionStr += AMPS_OPTIONS_NOEMPTIES; }

            void setSendKeys(void) { _optionStr += AMPS_OPTIONS_SENDKEYS; }

            void setTimestamp(void) { _optionStr += AMPS_OPTIONS_TIMESTAMP; }

            void setNoSowKey(void) { _optionStr += AMPS_OPTIONS_NOSOWKEY; }

            void setCancel(void) { _optionStr += AMPS_OPTIONS_CANCEL; }

            void setResume(void) { _optionStr += AMPS_OPTIONS_RESUME; }

            void setPause(void) { _optionStr += AMPS_OPTIONS_PAUSE; }

            void setFullyDurable(void) { _optionStr += AMPS_OPTIONS_FULLY_DURABLE; }

            void setMaxBacklog(int maxBacklog_)
            {
                char buf[24];
                AMPS_snprintf(buf, sizeof(buf), "max_backlog=%d,", maxBacklog_);
                _optionStr += buf;
                _maxBacklog = maxBacklog_;
            }

            void setConflation(const char* conflation_)
            {
                char buf[64];
                AMPS_snprintf(buf, sizeof(buf), "conflation=%s,", conflation_);
                _optionStr += buf;
                _conflation = conflation_;
            }

            void setConflationKey(const char* conflationKey_)
            {
                char buf[64];
                AMPS_snprintf(buf, sizeof(buf), "conflation_key=%s,", conflationKey_);
                _optionStr += buf;
                _conflationKey = conflationKey_;
            }

            void setTopN(int topN_)
            {
                char buf[24];
                AMPS_snprintf(buf, sizeof(buf), "top_n=%d,", topN_);
                _optionStr += buf;
                _topN = topN_;
            }

            void setRate(const char* rate_)
            {
                char buf[64];
                AMPS_snprintf(buf, sizeof(buf), "rate=%s,", rate_);
                _optionStr += buf;
                _rate = rate_;
            }

            void setRateMaxGap(const char* rateMaxGap_)
            {
                char buf[64];
                AMPS_snprintf(buf, sizeof(buf), "rate_max_gap=%s,", rateMaxGap_);
                _optionStr += buf;
                _rateMaxGap = rateMaxGap_;
            }

            void setSkipN(int skipN_)
            {
                char buf[24];
                AMPS_snprintf(buf, sizeof(buf), "skip_n=%d,", skipN_);
                _optionStr += buf;
                _skipN = skipN_;
            }

            void setProjection(std::string projection_)
            {
                _projection = "projection=[" + projection_ + "],";
                _optionStr += _projection;
            }


            template<class Iterator>
            void setProjection(Iterator begin_, Iterator end_)
            {
                _projection = "projection=[";
                for (Iterator i = begin_; i != end_; ++i)
                {
                    _projection += *i;
                    _projection += ',';
                }
                _projection.insert(_projection.length() - 1, "]");
                _optionStr += _projection;
            }

            void setGrouping(std::string grouping_)
            {
                _grouping = "grouping=[" + grouping_ + "],";
                _optionStr += _grouping;
            }


            template<class Iterator>
            void setGrouping(Iterator begin_, Iterator end_)
            {
                _grouping = "grouping=[";
                for (Iterator i = begin_; i != end_; ++i)
                {
                    _grouping += *i;
                    _grouping += ',';
                }
                _grouping.insert(_grouping.length() - 1, "]");
                _optionStr += _grouping;
            }

            operator const std::string()
            {
                return _optionStr.substr(0, _optionStr.length()-1);
            }

        private:
            std::string _optionStr;
            int         _maxBacklog;
            std::string _conflation;
            std::string _conflationKey;
            int         _topN;
            std::string _rate;
            std::string _rateMaxGap;
            int         _skipN;
            std::string _projection;
            std::string _grouping;
    };

    struct Command
    {
      typedef enum
      {
          Unknown = 0,
          Publish = 1,
          Subscribe = 2,
          Unsubscribe = 4,
          SOW = 8,
          Heartbeat = 16,
          SOWDelete = 32,
          DeltaPublish = 64,
          Logon = 128,
          SOWAndSubscribe = 256,
          DeltaSubscribe = 512,
          SOWAndDeltaSubscribe = 1024,
          StartTimer = 2048,
          StopTimer = 4096,
          GroupBegin = 8192,
          GroupEnd = 16384,
          OOF = 32768,
          Ack = 65536,
          Flush = 131072
      } Type;
    };
    Command::Type getCommandEnum() const
    {
        const char* data = NULL; size_t len = 0;
        amps_message_get_field_value(_body.get().getMessage(), AMPS_Command, &data, &len);
        switch (len)
        {
        case 1: return Command::Publish;
        case 3:
            switch (data[0])
            {
            case 's': return Command::SOW;
            case 'o': return Command::OOF;
            case 'a': return Command::Ack;
            }
            break;
        case 5:
            switch (data[0])
            {
            case 'l': return Command::Logon;
            case 'f': return Command::Flush;
            }
            break;
        case 7:
            return Command::Publish;
            break;
        case 9:
            switch (data[0])
            {
            case 's': return Command::Subscribe;
            case 'h': return Command::Heartbeat;
            case 'g': return Command::GroupEnd;
            }
            break;
        case 10:
            switch (data[1])
            {
            case 'o': return Command::SOWDelete;
            case 't': return Command::StopTimer;
            }
            break;
        case 11:
            switch (data[0])
            {
            case 'g': return Command::GroupBegin;
            case 'u': return Command::Unsubscribe;
            }
            break;
        case 13:
            return Command::DeltaPublish;
        case 15:
            return Command::DeltaSubscribe;
        case 17:
            return Command::SOWAndSubscribe;
        case 23:
            return Command::SOWAndDeltaSubscribe;
        }
        return Command::Unknown;
    }

    Message& setCommandEnum(Command::Type command_)
    {
        unsigned bits = 0;
        unsigned command = command_;
        while (command > 0) { ++bits; command >>= 1; }
        amps_message_assign_field_value(_body.get().getMessage(), AMPS_Command,
            CommandConstants<0>::Values[bits], CommandConstants<0>::Lengths[bits]);
        return *this;
    }

    AMPS_FIELD(Command)
    AMPS_FIELD(CommandId)
    AMPS_FIELD(ClientName)
    AMPS_FIELD(UserId)
    AMPS_FIELD(Timestamp)

    Field getTransmissionTime() const
    {
        return getTimestamp();
    }

    void getRawTransmissionTime(const char** dataptr, size_t* sizeptr) const
    {
        getRawTimestamp(dataptr, sizeptr);
    }
    
    AMPS_FIELD(Topic)
    AMPS_FIELD(Filter)
    AMPS_FIELD(MessageType)

    struct AckType
    {
      typedef enum
      {
          None = 0, Received = 1, Parsed = 2, Processed = 4, Persisted = 8, Completed = 16, Stats = 32
      } Type;
    };
    AMPS_FIELD(AckType);
    static inline AckType::Type decodeSingleAckType(const char* begin, const char* end)
    {
        switch (end - begin)
        {
        case 5:
            return AckType::Stats;
        case 6:
            return AckType::Parsed;
        case 8:
            return AckType::Received;
        case 9:
            switch (begin[1])
            {
            case 'e': return AckType::Persisted;
            case 'r': return AckType::Processed;
            case 'o': return AckType::Completed;
            default: break;
            }
            break;
        default:
            break;
        }
        return AckType::None;
    }
    unsigned getAckTypeEnum() const
    {
        unsigned result = AckType::None;
        const char* data = NULL; size_t len = 0;
        amps_message_get_field_value(_body.get().getMessage(), AMPS_AckType, &data, &len);
        const char* mark = data;
        for (const char* end = data + len; data != end; ++data)
        {
            if (*data == ',')
            {
                result |= decodeSingleAckType(mark, data);
                mark = data + 1;
            }
        }
        if (mark < data) result |= decodeSingleAckType(mark, data);
        return result;
    }
    Message& setAckTypeEnum(unsigned ackType_)
    {
        if(ackType_ < AckTypeConstants<0>::Entries)
        {
            amps_message_assign_field_value(_body.get().getMessage(), AMPS_AckType,
              AckTypeConstants<0>::Values[ackType_], AckTypeConstants<0>::Lengths[ackType_]);
        }
        return *this;
    }

    AMPS_FIELD(SubscriptionId)
    AMPS_FIELD(Expiration)
    AMPS_FIELD(Heartbeat)
    AMPS_FIELD(TimeoutInterval)
    AMPS_FIELD(LeasePeriod)
    AMPS_FIELD(Status)
    AMPS_FIELD(QueryID)
    AMPS_FIELD(BatchSize)
    AMPS_FIELD(TopNRecordsReturned)
    AMPS_FIELD(OrderBy)
    AMPS_FIELD(SowKeys)
    AMPS_FIELD(CorrelationId)
    AMPS_FIELD(Sequence)
    AMPS_FIELD(Bookmark)
    AMPS_FIELD(RecordsInserted)
    AMPS_FIELD(RecordsUpdated)
    AMPS_FIELD(SowDelete)
    AMPS_FIELD(RecordsReturned)
    AMPS_FIELD(TopicMatches)
    AMPS_FIELD(Matches)
    AMPS_FIELD(MessageLength)
    AMPS_FIELD(SowKey)
    AMPS_FIELD(GroupSequenceNumber)
    AMPS_FIELD(SubscriptionIds)
    AMPS_FIELD(Reason)
    AMPS_FIELD(Password)
    AMPS_FIELD(Version)

    DOX_OPENGROUP(x)
    DOX_MAKEGETCOMMENT(Options) DOX_COMMENTHEAD( Retrieves the value of the Options header of the Message as a new Field.)
    Field getOptions() const {
        Field returnValue;
        const char* ptr;
        size_t sz;
        amps_message_get_field_value(_body.get().getMessage(),
                AMPS_Options, &ptr, &sz);
        if (sz && ptr[sz-1] == ',') --sz;
        returnValue.assign(ptr, sz);
        return returnValue;
    }

    DOX_MAKEGETRAWCOMMENT(x) DOX_COMMENTHEAD( Retrieves the value of the Options header of self as a Field that references the underlying buffer managed by this Message. )
    void getRawOptions(const char** dataptr, size_t* sizeptr) const {
        amps_message_get_field_value(_body.get().getMessage(),
                AMPS_Options, dataptr, sizeptr);
        if (*sizeptr && *dataptr && (*dataptr)[*sizeptr-1] == ',') --*sizeptr;
        return;
    }

    DOX_MAKESETCOMMENT(x) DOX_COMMENTHEAD( Sets the value of the Options header for this Message.)
    Message& setOptions(const std::string& v) {
        size_t sz = v.length();
        if (sz && v[sz-1] == ',') --sz;
        amps_message_set_field_value(_body.get().getMessage(),
                AMPS_Options, v.c_str(), sz);
        return *this;
    }

    DOX_MAKEASSIGNCOMMENT(x) DOX_COMMENTHEAD( Sets the value of the Options header for this Message.)
    Message& assignOptions(const std::string& v) {
        size_t sz = v.length();
        if (sz && v[sz-1] == ',') --sz;
        amps_message_assign_field_value(_body.get().getMessage(),
                AMPS_Options, v.c_str(), sz);
        return *this;
    }

    DOX_MAKEASSIGNCOMMENT(x) DOX_COMMENTHEAD( Sets the value of the Options header for this Message.)
    Message& assignOptions(const char* data, size_t len) {
        if (len && data[len-1] == ',') --len;
        amps_message_assign_field_value(_body.get().getMessage(),
                AMPS_Options, data, len);
        return *this;
    }

    DOX_MAKESETCOMMENT(x) DOX_COMMENTHEAD( Sets the value of the Options header for this Message.)
    Message& setOptions(const char* str) {
        if (str)
        {
            size_t sz = strlen(str);
            if (sz && str[sz-1] == ',') --sz;
            amps_message_set_field_value(_body.get().getMessage(),
                    AMPS_Options, str, sz);
        }
        else
        {
            amps_message_set_field_value(_body.get().getMessage(),
                    AMPS_Options, str, 0);
        }
        return *this;
    }

    DOX_MAKESETCOMMENT(x) DOX_COMMENTHEAD( Sets the value of the Options header for this Message.)
    Message& setOptions(const char* str,size_t len) {
        if (len && str[len-1] == ',') --len;
        amps_message_set_field_value(_body.get().getMessage(),
                AMPS_Options, str, len);
        return *this;
    }
    DOX_CLOSEGROUP(x)

    

    Field getData() const
    {
        Field returnValue;
        char* ptr;
        size_t sz;
        amps_message_get_data(_body.get().getMessage(), &ptr, &sz);
        returnValue.assign(ptr, sz);
        return returnValue;
    }

    void getRawData(const char **data, size_t *sz) const
    {
        amps_message_get_data(_body.get().getMessage(), (char**)data, sz);
    }
    Message& setData(const std::string &v_)
    {
        amps_message_set_data(_body.get().getMessage(), v_.c_str(), v_.length());
        return *this;
    }
    Message& assignData(const std::string &v_)
    {
        amps_message_assign_data(_body.get().getMessage(), v_.c_str(), v_.length());
        return *this;
    }

    Message& setData(const char* data_, size_t length_)
    {
        amps_message_set_data(_body.get().getMessage(), data_, length_);
        return *this;
    }
    Message& assignData(const char* data_,size_t length_)
    {
        amps_message_assign_data(_body.get().getMessage(),data_,length_);
        return *this;
    }

    Message& setData(const char* data_)
    {
        amps_message_set_data_nts(_body.get().getMessage(), data_);
        return *this;
    }
    Message& assignData(const char* data_)
    {
        amps_message_assign_data(_body.get().getMessage(), data_, strlen(data_));
        return *this;
    }
    amps_handle getMessage() const
    {
        return _body.get().getMessage();
    }
    void replace(amps_handle message, bool owner = false)
    {
        _body.get().replace(message, owner);
    }
    void disown()
    {
        _body.get().disown();
    }
    void invalidate()
    {
        _body = NULL;
    }
    bool isValid(void) const
    {
        return _body.isValid();
    }
    Message& reset()
    {
        _body.get().reset();
        return *this;
    }

    void setBookmarkSeqNo(size_t val)
    {
        _body.get().setBookmarkSeqNo(val);
    }

    size_t getBookmarkSeqNo() const
    {
        return _body.get().getBookmarkSeqNo();
    }

    void setSubscriptionHandle(amps_handle val)
    {
        _body.get().setSubscriptionHandle(val);
    }

    amps_handle getSubscriptionHandle() const
    {
        return _body.get().getSubscriptionHandle();
    }

    void ack(const char* options_ = NULL) const;

    void setClientImpl(ClientImpl *pClientImpl)
    {
        _body.get().setClientImpl(pClientImpl);
    }

    void setIgnoreAutoAck() const
    {
        _body.get().setIgnoreAutoAck();
    }

    bool getIgnoreAutoAck() const
    {
        return _body.get().getIgnoreAutoAck();
    }

    template <class T> // static
    void throwFor(const T& /*context_*/, const std::string& ackReason_) const
    {
        switch (ackReason_[0])
        {
            case 'a': // auth failure
                throw AuthenticationException("Logon failed for user \"" +
                                              (std::string)getUserId() + "\"");
                break;
            case 'b':
                switch (ackReason_.length())
                {
                    case 10: // bad filter
                        throw BadFilterException("Filter '" +
                                                 (std::string)getFilter() +
                                                 "' is invalid");
                        break;
                    case 15: // bad regex topic
                        throw BadRegexTopicException("Regular Expression Topic '" +
                                                     (std::string)getTopic() +
                                                     "' is invalid.");
                        break;
                    default:
                        break;
                }
                break;
            case 'i':
                if (ackReason_.length() == 13)
                {
                    switch (ackReason_[8])
                    {
                        case 't': // invalid topic
                            throw InvalidTopicException("Topic '" +
                                                        (std::string)getTopic() +
                                                        "' is invalid.");
                            break;
                        case 's': // invalid subId
                            throw InvalidSubIdException("SubId '" +
                                                        (std::string)getSubscriptionId() +
                                                        "' is invalid.");
                            break;
                        default:
                            break;
                    }
                }
                break;
            case 'n':
                switch (ackReason_[1])
                {
                    case 'a': // name in use
                        throw NameInUseException("Client name '" +
                                                 (std::string)getClientName() +
                                                 "' already exists.");
                        break;
                    case 'o': // not entitled
                        throw NotEntitledException("User \"" +
                                                   (std::string)getUserId() +
                                                   "\" not entitled to topic \"" +
                                                   (std::string)getTopic() +
                                                   "\".");
                        break;
                    default:
                        break;
                }
                break;
            case 's':
                switch (ackReason_.length())
                {
                    case 27: // subscription already exists
                        throw SubscriptionAlreadyExistsException(
                                                    "Subscription for command '" +
                                                    (std::string)getCommandId() +
                                                    "' already exists.");
                        break;
                    case 12: // subid in use
                        throw SubidInUseException("Subscription with id '" +
                                                  (std::string)getSubscriptionId() + 
                                                  "' already exists.");
                        break;
                    default:
                        break;
                }
                break;
            default:
                break;
        }
        throw CommandException("Error from server while processing this command: '" +
                               ackReason_ + "'");
    }
};

inline std::string
operator+(const std::string& lhs, const Message::Field& rhs)
{
    return lhs + std::string(rhs);
}

inline std::basic_ostream<char>&
operator<<(std::basic_ostream<char>& os, const Message::Field& rhs)
{
    os.write(rhs.data(), (std::streamsize)rhs.len());
    return os;
}
inline bool
AMPS::Field::operator<(const AMPS::Field& rhs) const
{
    return std::lexicographical_compare(data(), data()+len(), rhs.data(), rhs.data()+rhs.len());
}

}

#endif