Message.hpp

Go to the documentation of this file.

//
// Copyright (c) 2010-2024 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 "amps/util.hpp"
#include "amps/constants.hpp"
#include "amps/amps_generated.h"
#include "amps/Field.hpp"
#include <stdio.h>
#include <algorithm>
#include <ostream>
#include <string>
#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__ == 4) && (__GNUC_MINOR__) >=5 )
  #define AMPS_USE_LAMBDAS 1
#endif

#if (_MSC_VER >= 1600) || (__GNUC__ > 4) || ( (__GNUC__ == 4) && (__GNUC_MINOR__) >=8 )
  #define AMPS_USE_EMPLACE 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 copy(const MessageImpl& rhs_)
    {
      if (_owner && _message)
      {
        amps_message_destroy(_message);
      }
      _message = amps_message_copy(rhs_._message);
      _owner = true;
      _bookmarkSeqNo = rhs_._bookmarkSeqNo;
      _subscription = rhs_._subscription;
      _isIgnoreAutoAck = rhs_._isIgnoreAutoAck;
      _clientImpl = rhs_._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;
      _clientImpl = NULL;
    }

    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()
    {
#if __cplusplus >= 201100L || _MSC_VER >= 1900
      static std::atomic<uint_fast64_t> _id(0);
      return (unsigned long)++_id;
#else
      static AMPS_ATOMIC_TYPE _id = 0;
      return (unsigned long)(AMPS_FETCH_ADD(&_id, 1));
#endif
    }

    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()
#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, "auto%lu" , (unsigned long)(_body.get().newId()));\
    amps_message_set_field_value_nts(_body.get().getMessage(),\
                                     AMPS_##x, buf);\
    return *this;\
  } \
  DOX_CLOSEGROUP()

#define AMPS_FIELD_ALIAS(x,y) \
  DOX_OPENGROUP(y) \
  DOX_MAKEGETCOMMENT(y) \
  Field get##y() const {\
    Field returnValue;\
    const char* ptr;\
    size_t sz;\
    amps_message_get_field_value(_body.get().getMessage(),\
                                 AMPS_##y, &ptr, &sz);\
    returnValue.assign(ptr, sz);\
    return returnValue;\
  }\
  DOX_MAKEGETRAWCOMMENT(y) \
  void getRaw##y(const char** dataptr, size_t* sizeptr) const {\
    amps_message_get_field_value(_body.get().getMessage(),\
                                 AMPS_##y, dataptr, sizeptr);\
    return;\
  }\
  DOX_MAKESETCOMMENT(y) \
  Message& set##y(const std::string& v) {\
    amps_message_set_field_value(_body.get().getMessage(),\
                                 AMPS_##y, v.c_str(), v.length());\
    return *this;\
  }\
  DOX_MAKESETCOMMENT(y) \
  Message& set##y(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_##y, buf);\
    return *this;\
  }\
  DOX_MAKEASSIGNCOMMENT(y) \
  Message& assign##y(const std::string& v) {\
    amps_message_assign_field_value(_body.get().getMessage(),\
                                    AMPS_##y, v.c_str(), v.length());\
    return *this;\
  }\
  DOX_MAKEASSIGNCOMMENT(y) \
  Message& assign##y(const char* data, size_t len) {\
    amps_message_assign_field_value(_body.get().getMessage(),\
                                    AMPS_##y, data, len);\
    return *this;\
  }\
  DOX_MAKESETCOMMENT(y) \
  Message& set##y(const char* str) {\
    amps_message_set_field_value_nts(_body.get().getMessage(),\
                                     AMPS_##y, str);\
    return *this;\
  }\
  DOX_MAKESETCOMMENT(y) \
  Message& set##y(const char* str,size_t len) {\
    amps_message_set_field_value(_body.get().getMessage(),\
                                 AMPS_##y, str,len);\
    return *this;\
  }\
  DOX_MAKENEWCOMMENT(y) \
  Message& new##y() {\
    char buf[Message::IdentifierLength+1];\
    buf[Message::IdentifierLength] = 0;\
    AMPS_snprintf(buf, Message::IdentifierLength+1, "auto%lux" , (unsigned long)(_body.get().newId()));\
    amps_message_set_field_value_nts(_body.get().getMessage(),\
                                     AMPS_##y, buf);\
    return *this;\
  }\
  DOX_MAKEGETCOMMENT(y) \
  Field get##x() const {\
    Field returnValue;\
    const char* ptr;\
    size_t sz;\
    amps_message_get_field_value(_body.get().getMessage(),\
                                 AMPS_##y, &ptr, &sz);\
    returnValue.assign(ptr, sz);\
    return returnValue;\
  }\
  DOX_MAKEGETRAWCOMMENT(y) \
  void getRaw##x(const char** dataptr, size_t* sizeptr) const {\
    amps_message_get_field_value(_body.get().getMessage(),\
                                 AMPS_##y, dataptr, sizeptr);\
    return;\
  }\
  DOX_MAKESETCOMMENT(y) \
  Message& set##x(const std::string& v) {\
    amps_message_set_field_value(_body.get().getMessage(),\
                                 AMPS_##y, v.c_str(), v.length());\
    return *this;\
  }\
  DOX_MAKESETCOMMENT(y) \
  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_##y, buf);\
    return *this;\
  }\
  DOX_MAKEASSIGNCOMMENT(y) \
  Message& assign##x(const std::string& v) {\
    amps_message_assign_field_value(_body.get().getMessage(),\
                                    AMPS_##y, v.c_str(), v.length());\
    return *this;\
  }\
  DOX_MAKEASSIGNCOMMENT(y) \
  Message& assign##x(const char* data, size_t len) {\
    amps_message_assign_field_value(_body.get().getMessage(),\
                                    AMPS_##y, data, len);\
    return *this;\
  }\
  DOX_MAKESETCOMMENT(y) \
  Message& set##x(const char* str) {\
    amps_message_set_field_value_nts(_body.get().getMessage(),\
                                     AMPS_##y, str);\
    return *this;\
  }\
  DOX_MAKESETCOMMENT(y) \
  Message& set##x(const char* str,size_t len) {\
    amps_message_set_field_value(_body.get().getMessage(),\
                                 AMPS_##y, str,len);\
    return *this;\
  }\
  DOX_MAKENEWCOMMENT(y) \
  Message& new##x() {\
    char buf[Message::IdentifierLength+1];\
    buf[Message::IdentifierLength] = 0;\
    AMPS_snprintf(buf, Message::IdentifierLength+1, "auto%lux" , (unsigned long)(_body.get().newId()));\
    amps_message_set_field_value_nts(_body.get().getMessage(),\
                                     AMPS_##y, buf);\
    return *this;\
  } \
  DOX_CLOSEGROUP()


  class Message
  {
    RefHandle<MessageImpl> _body;

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

  public:
    typedef AMPS::Field Field;

    static const unsigned int IdentifierLength = 32;

    static const size_t BOOKMARK_NONE = AMPS_UNSET_INDEX;

    enum CtorFlag { EMPTY };

    Message(CtorFlag) : _body()
    {
    }

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

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

    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(const 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(const 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)
        , _skipN(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.clear();
      }

      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(const 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(const 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);
      }
      size_t getLength() const
      {
        return (_optionStr.empty() ? 0 : _optionStr.length() - 1);
      }

      const char* getStr() const
      {
        return (_optionStr.empty() ? 0 : _optionStr.data());
      }

    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 AckType
    {
      typedef enum : unsigned
      {
        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(BatchSize)
    AMPS_FIELD(Bookmark)
    AMPS_FIELD(Command)

    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,
        NoDataCommands = Publish | Unsubscribe | Heartbeat | SOWDelete | DeltaPublish
                         | Logon | StartTimer | StopTimer | Flush
      } 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; // -V1037
      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; // -V1037
        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(CommandId)
    AMPS_FIELD(ClientName)
    AMPS_FIELD(CorrelationId)
    AMPS_FIELD(Expiration)
    AMPS_FIELD(Filter)
    AMPS_FIELD(GroupSequenceNumber)
    AMPS_FIELD(Heartbeat)
    AMPS_FIELD(LeasePeriod)
    AMPS_FIELD(Matches)
    AMPS_FIELD(MessageLength)
    AMPS_FIELD(MessageType)

    DOX_OPENGROUP(Options)
    DOX_MAKEGETCOMMENT(Options)
    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(Options)
    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(Options)
    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(Options)
    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(Options)
    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(Options)
    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(Options)
    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()

    AMPS_FIELD(OrderBy)
    AMPS_FIELD(Password)
    AMPS_FIELD_ALIAS(QueryId, QueryID)
    AMPS_FIELD(Reason)
    AMPS_FIELD(RecordsInserted)
    AMPS_FIELD(RecordsReturned)
    AMPS_FIELD(RecordsUpdated)
    AMPS_FIELD(Sequence)
    AMPS_FIELD(SowDelete)
    AMPS_FIELD(SowKey)
    AMPS_FIELD(SowKeys)
    AMPS_FIELD(Status)
    AMPS_FIELD_ALIAS(SubId, SubscriptionId) // -V524
    AMPS_FIELD(SubscriptionIds)
    AMPS_FIELD(TimeoutInterval)
    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(TopicMatches)
    AMPS_FIELD(TopNRecordsReturned)
    AMPS_FIELD(Version)
    AMPS_FIELD(UserId)


    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("bad filter '" +
                                   (std::string)getFilter() +
                                   "'");
          break;
        case 11: // bad sow key
          if (getSowKeys().len())
          {
            throw BadSowKeyException("bad sow key '" +
                                     (std::string)getSowKeys() +
                                     "'");
          }
          else
          {
            throw BadSowKeyException("bad sow key '" +
                                     (std::string)getSowKey() +
                                     "'");
          }
          break;
        case 15: // bad regex topic
          throw BadRegexTopicException("bad regex topic '" +
                                       (std::string)getTopic() +
                                       "'.");
          break;
        default:
          break;
        }
        break;
      case 'd':
        if (ackReason_.length() == 23) // duplicate logon attempt
        {
          throw DuplicateLogonException("Client '" +
                                        (std::string)getClientName() +
                                        "' with userid '" +
                                        (std::string)getUserId() +
                                        "' duplicate logon attempt");
        }
        break;
      case 'i':
        if (ackReason_.length() >= 9)
        {
          switch (ackReason_[8])
          {
          case 'b': // invalid bookmark
            throw InvalidBookmarkException("invalid bookmark '" +
                                           (std::string)getBookmark() +
                                           "'.");
            break;
          case 'm': // invalid message type
            throw CommandException(std::string("invalid message type '") +
                                   (std::string)getMessageType() +
                                   "'.");
            break;
          case 'o':
            if (ackReason_[9] == 'p') // invalid options
            {
              throw InvalidOptionsException("invalid options '" +
                                            (std::string)getOptions() +
                                            "'.");
            }
            else if (ackReason_[9] == 'r') // invalid order by
            {
              throw InvalidOrderByException("invalid order by '" +
                                            (std::string)getOrderBy() +
                                            "'.");
            }
            break;
          case 's': // invalid subId
            throw InvalidSubIdException("invalid subid '" +
                                        (std::string)getSubscriptionId() +
                                        "'.");
            break;
          case 't':
            if (ackReason_.length() == 13) // invalid topic
            {
              throw InvalidTopicException("invalid topic '" +
                                          (std::string)getTopic() +
                                          "'.");
            }
            else if (ackReason_.length() == 23) // invalid topic or filter
            {
              throw InvalidTopicException("invalid topic or filter. Topic '" +
                                          (std::string)getTopic() +
                                          "' Filter '" +
                                          (std::string)getFilter() +
                                          "'.");
            }
            break;
          default:
            break;
          }
        }
        break;
      case 'l':
        if (ackReason_.length() == 14) // logon required
        {
          throw LogonRequiredException("logon required before command");
        }
        break;
      case 'n':
        switch (ackReason_[4])
        {
        case ' ': // name in use
          throw NameInUseException("name in use '" +
                                   (std::string)getClientName() +
                                   "'.");
          break;
        case 'e': // not entitled
          throw NotEntitledException("User \"" +
                                     (std::string)getUserId() +
                                     "\" not entitled to topic \"" +
                                     (std::string)getTopic() +
                                     "\".");
          break;
        case 'i': // no filter or bookmark
          throw MissingFieldsException("command sent with no filter or bookmark.");
          break;
        case 'l': // no client name
          throw MissingFieldsException("command sent with no client name.");
          break;
        case 'o': // no topic or filter
          throw MissingFieldsException("command sent with no topic or filter.");
          break;
        case 's': // not supported
          throw CommandException("operation on topic '" +
                                 (std::string)getTopic() +
                                 "' with options '" +
                                 (std::string)getOptions() +
                                 "' not supported.");
          break;
        default:
          break;
        }
        break;
      case 'o':
        switch (ackReason_.length())
        {
        case 16: // orderby required
          throw MissingFieldsException("orderby required");
          break;
        case 17: // orderby too large
          throw CommandException("orderby too large '" +
                                 (std::string)getOrderBy() +
                                 "'.");
          break;
        }
        break;
      case 'p':
        throw CommandException("projection clause too large in options '" +
                               (std::string)getOptions() +
                               "'.");
        break;
      case 'r':
        switch (ackReason_[2])
        {
        case 'g': // regex topic not supported
          throw BadRegexTopicException("'regex topic not supported '" +
                                       (std::string)getTopic() +
                                       "'.");
          break;
        default:
          break;
        }
        break;
      case 's':
        switch (ackReason_[5])
        {
        case ' ': // subid in use
          throw SubidInUseException("subid in use '" +
                                    (std::string)getSubscriptionId() +
                                    "'.");
          break;
        case 'e': // sow_delete command only supports one of: filter, sow_keys, bookmark, or data
          throw CommandException("sow_delete command only supports one of: filter '" +
                                 (std::string)getFilter() +
                                 "', sow_keys '" +
                                 (std::string)getSowKeys() +
                                 "', bookmark '" +
                                 (std::string)getBookmark() +
                                 "', or data '" +
                                 (std::string)getData() +
                                 "'.");
          break;
        case 't': // sow store failed
          throw PublishException("sow store failed.");
          break;
        default:
          break;
        }
        break;
      case 't':
        switch (ackReason_[2])
        {
        case ' ': // tx store failure
          throw PublishException("tx store failure.");
          break;
        case 'n': // txn replay failed
          throw CommandException("txn replay failed for '" +
                                 (std::string)getSubId() +
                                 "'.");
          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
  {
    if (!data())
    {
      return rhs.data() != NULL;
    }
    if (!rhs.data())
    {
      return false;
    }
    return std::lexicographical_compare(data(), data() + len(), rhs.data(), rhs.data() + rhs.len());
  }

}

#endif