Message.hpp

Go to the documentation of this file.

//
// Copyright (c) 2010-2021 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>
#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_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()
#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%zx" , (size_t)(_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%zx" , (size_t)(_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%zx" , (size_t)(_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;

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

 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
 {
 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':
 switch (ackReason_[1])
 {
 case 'r': // projection clause too large
 throw CommandException("projection clause too large in options '" +
 (std::string)getOptions() +
 "'.");
 break;
 case 'u': // publish filter no match
 throw PublishFilterException("Publish filter '" +
 (std::string)getFilter() +
 "' doesn't match any records.");
 break;
 default:
 break;
 }
 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
{
 return std::lexicographical_compare(data(), data()+len(), rhs.data(), rhs.data()+rhs.len());
}

}

#endif