25 #ifndef __AMPS_MESSAGE_HPP__ 26 #define __AMPS_MESSAGE_HPP__ 28 #include "constants.hpp" 29 #include "amps_generated.h" 35 #define AMPS_UNSET_SEQUENCE (amps_uint64_t)-1 38 #if defined(__GXX_EXPERIMENTAL_CXX0X__) || (_MSC_VER >= 1600) 39 #define AMPS_USE_FUNCTIONAL 1 42 #if (_MSC_VER >= 1600) || ( (__GNUC__ >= 4) && (__GNUC_MINOR__) >=5 ) 43 #define AMPS_USE_LAMBDAS 1 46 #ifdef AMPS_USE_FUNCTIONAL 56 #define AMPS_OPTIONS_NONE "" 57 #define AMPS_OPTIONS_LIVE "live," 58 #define AMPS_OPTIONS_OOF "oof," 59 #define AMPS_OPTIONS_REPLACE "replace," 60 #define AMPS_OPTIONS_NOEMPTIES "no_empties," 61 #define AMPS_OPTIONS_SENDKEYS "send_keys," 62 #define AMPS_OPTIONS_TIMESTAMP "timestamp," 63 #define AMPS_OPTIONS_NOSOWKEY "no_sowkey," 64 #define AMPS_OPTIONS_CANCEL "cancel," 65 #define AMPS_OPTIONS_RESUME "resume," 66 #define AMPS_OPTIONS_PAUSE "pause," 67 #define AMPS_OPTIONS_FULLY_DURABLE "fully_durable," 68 #define AMPS_OPTIONS_EXPIRE "expire," 69 #define AMPS_OPTIONS_TOPN(x) "top_n=##x," 70 #define AMPS_OPTIONS_MAXBACKLOG(x) "max_backlog=##x," 71 #define AMPS_OPTIONS_RATE(x) "rate=##x," 75 typedef void* amps_subscription_handle;
89 mutable bool _isIgnoreAutoAck;
90 size_t _bookmarkSeqNo;
91 amps_subscription_handle _subscription;
92 ClientImpl* _clientImpl;
106 bool ignoreAutoAck_ =
false,
size_t bookmarkSeqNo_ = 0,
107 amps_subscription_handle subscription_ = NULL,
108 ClientImpl* clientImpl_=NULL)
109 : _message(message_), _owner(owner_), _isIgnoreAutoAck(ignoreAutoAck_)
110 , _bookmarkSeqNo(bookmarkSeqNo_)
111 , _subscription(subscription_), _clientImpl(clientImpl_)
119 : _message(NULL), _owner(true), _isIgnoreAutoAck(false), _bookmarkSeqNo(0), _subscription(NULL), _clientImpl(NULL)
133 return new MessageImpl(copy,
true, _isIgnoreAutoAck, _bookmarkSeqNo,
134 _subscription,_clientImpl);
137 void setClientImpl(ClientImpl* clientImpl_)
139 _clientImpl = clientImpl_;
142 ClientImpl* clientImpl(
void)
const 156 Lock<Mutex> l(_lock);
159 _subscription = NULL;
160 _isIgnoreAutoAck =
false;
170 Lock<Mutex> l(_lock);
171 if (_message == message_)
return;
172 if (_owner && _message)
179 _subscription = NULL;
180 _isIgnoreAutoAck =
false;
185 Lock<Mutex> l(_lock);
189 static unsigned long newId()
191 static ATOMIC_TYPE _id = 0;
192 return (
unsigned long)(AMPS_FETCH_ADD(&_id,1));
195 void setBookmarkSeqNo(
size_t val_)
197 _bookmarkSeqNo = val_;
200 size_t getBookmarkSeqNo(
void)
const 202 return _bookmarkSeqNo;
205 void setSubscriptionHandle(amps_subscription_handle subscription_)
207 _subscription = subscription_;
210 amps_subscription_handle getSubscriptionHandle(
void)
const 212 return _subscription;
215 void setIgnoreAutoAck()
const 217 _isIgnoreAutoAck =
true;
220 bool getIgnoreAutoAck()
const 222 return _isIgnoreAutoAck;
232 #ifdef DOXYGEN_PREPROCESSOR 234 #define DOX_COMMENTHEAD(s) / ## ** ## s ## * ## / 235 #define DOX_GROUPNAME(s) DOX_COMMENTHEAD(@name s Functions) 236 #define DOX_OPENGROUP(s) DOX_COMMENTHEAD(@{) DOX_GROUPNAME(s) 237 #define DOX_CLOSEGROUP() DOX_COMMENTHEAD(@}) 238 #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. ) 239 #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. ) 240 #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. ) 241 #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. ) 242 #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.) 244 #define DOX_ALIAS_MAKEGETCOMMENT(x,y) DOX_COMMENTHEAD( Retrieves the value of the y [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. ) 245 #define DOX_ALIAS_MAKEGETRAWCOMMENT(x,y) DOX_COMMENTHEAD( Retrieves the value of the y [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. ) 246 #define DOX_ALIAS_MAKESETCOMMENT(x,y) DOX_COMMENTHEAD( Sets the value of the y [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. ) 247 #define DOX_ALIAS_MAKEASSIGNCOMMENT(x,y) DOX_COMMENTHEAD( Sets the value of the y [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. ) 248 #define DOX_ALIAS_MAKENEWCOMMENT(x,y) DOX_COMMENTHEAD(Creates and sets a new sequential value for the y [x] header for this Message. This function is most useful for headers such as CommandId and SubId.) 252 #define DOX_COMMENTHEAD(s) 253 #define DOX_GROUPNAME(s) 254 #define DOX_OPENGROUP(x) 255 #define DOX_CLOSEGROUP(x) 256 #define DOX_MAKEGETCOMMENT(x) 257 #define DOX_MAKEGETRAWCOMMENT(x) 258 #define DOX_MAKESETCOMMENT(x) 259 #define DOX_MAKEASSIGNCOMMENT(x) 260 #define DOX_MAKENEWCOMMENT(x) 261 #define DOX_ALIAS_MAKEGETCOMMENT(x,y) 262 #define DOX_ALIAS_MAKEGETRAWCOMMENT(x,y) 263 #define DOX_ALIAS_MAKESETCOMMENT(x,y) 264 #define DOX_ALIAS_MAKEASSIGNCOMMENT(x,y) 265 #define DOX_ALIAS_MAKENEWCOMMENT(x,y) 273 #define AMPS_FIELD(x) \ 275 DOX_MAKEGETCOMMENT(x) \ 276 Field get##x() const { \ 280 amps_message_get_field_value(_body.get().getMessage(),\ 281 AMPS_##x, &ptr, &sz);\ 282 returnValue.assign(ptr, sz);\ 285 DOX_MAKEGETRAWCOMMENT(x) \ 286 void getRaw##x(const char** dataptr, size_t* sizeptr) const { \ 287 amps_message_get_field_value(_body.get().getMessage(), \ 288 AMPS_##x, dataptr, sizeptr);\ 291 DOX_MAKESETCOMMENT(x) \ 292 Message& set##x(const std::string& v) {\ 293 amps_message_set_field_value(_body.get().getMessage(),\ 294 AMPS_##x, v.c_str(), v.length());\ 297 DOX_MAKESETCOMMENT(x) \ 298 Message& set##x(amps_uint64_t v) {\ 300 AMPS_snprintf_amps_uint64_t(buf,22,v);\ 301 amps_message_set_field_value_nts(_body.get().getMessage(),\ 305 DOX_MAKEASSIGNCOMMENT(x) \ 306 Message& assign##x(const std::string& v) {\ 307 amps_message_assign_field_value(_body.get().getMessage(),\ 308 AMPS_##x, v.c_str(), v.length());\ 311 DOX_MAKEASSIGNCOMMENT(x) \ 312 Message& assign##x(const char* data, size_t len) {\ 313 amps_message_assign_field_value(_body.get().getMessage(),\ 314 AMPS_##x, data, len);\ 317 DOX_MAKESETCOMMENT(x) \ 318 Message& set##x(const char* str) {\ 319 amps_message_set_field_value_nts(_body.get().getMessage(),\ 323 DOX_MAKESETCOMMENT(x) \ 324 Message& set##x(const char* str,size_t len) {\ 325 amps_message_set_field_value(_body.get().getMessage(),\ 329 DOX_MAKENEWCOMMENT(x) \ 331 char buf[Message::IdentifierLength+1];\ 332 buf[Message::IdentifierLength] = 0;\ 333 AMPS_snprintf(buf, Message::IdentifierLength+1, "auto%lx" , (size_t)(_body.get().newId()));\ 334 amps_message_set_field_value_nts(_body.get().getMessage(),\ 340 #define AMPS_FIELD_ALIAS(x,y) \ 342 DOX_ALIAS_MAKEGETCOMMENT(x,y) \ 343 Field get##x() const { \ 347 amps_message_get_field_value(_body.get().getMessage(),\ 348 AMPS_##y, &ptr, &sz);\ 349 returnValue.assign(ptr, sz);\ 352 DOX_ALIAS_MAKEGETRAWCOMMENT(x,y) \ 353 void getRaw##x(const char** dataptr, size_t* sizeptr) const { \ 354 amps_message_get_field_value(_body.get().getMessage(), \ 355 AMPS_##y, dataptr, sizeptr);\ 358 DOX_ALIAS_MAKESETCOMMENT(x,y) \ 359 Message& set##x(const std::string& v) {\ 360 amps_message_set_field_value(_body.get().getMessage(),\ 361 AMPS_##y, v.c_str(), v.length());\ 364 DOX_ALIAS_MAKESETCOMMENT(x,y) \ 365 Message& set##x(amps_uint64_t v) {\ 367 AMPS_snprintf_amps_uint64_t(buf,22,v);\ 368 amps_message_set_field_value_nts(_body.get().getMessage(),\ 372 DOX_ALIAS_MAKEASSIGNCOMMENT(x,y) \ 373 Message& assign##x(const std::string& v) {\ 374 amps_message_assign_field_value(_body.get().getMessage(),\ 375 AMPS_##y, v.c_str(), v.length());\ 378 DOX_ALIAS_MAKEASSIGNCOMMENT(x,y) \ 379 Message& assign##x(const char* data, size_t len) {\ 380 amps_message_assign_field_value(_body.get().getMessage(),\ 381 AMPS_##y, data, len);\ 384 DOX_ALIAS_MAKESETCOMMENT(x,y) \ 385 Message& set##x(const char* str) {\ 386 amps_message_set_field_value_nts(_body.get().getMessage(),\ 390 DOX_ALIAS_MAKESETCOMMENT(x,y) \ 391 Message& set##x(const char* str,size_t len) {\ 392 amps_message_set_field_value(_body.get().getMessage(),\ 396 DOX_ALIAS_MAKENEWCOMMENT(x,y) \ 398 char buf[Message::IdentifierLength+1];\ 399 buf[Message::IdentifierLength] = 0;\ 400 AMPS_snprintf(buf, Message::IdentifierLength+1, "%lx" , _body.get().newId());\ 401 amps_message_set_field_value_nts(_body.get().getMessage(),\ 449 RefHandle<MessageImpl> _body;
458 static const unsigned int IdentifierLength = 32;
462 static const size_t BOOKMARK_NONE = AMPS_UNSET_INDEX;
486 return Message(_body.get().copy());
502 static const char* None(
void) {
return AMPS_OPTIONS_NONE; }
503 static const char* Live(
void) {
return AMPS_OPTIONS_LIVE; }
504 static const char* OOF(
void) {
return AMPS_OPTIONS_OOF; }
505 static const char* Replace(
void) {
return AMPS_OPTIONS_REPLACE; }
506 static const char* NoEmpties(
void) {
return AMPS_OPTIONS_NOEMPTIES; }
507 static const char* SendKeys(
void) {
return AMPS_OPTIONS_SENDKEYS; }
508 static const char* Timestamp(
void) {
return AMPS_OPTIONS_TIMESTAMP; }
509 static const char* NoSowKey(
void) {
return AMPS_OPTIONS_NOSOWKEY; }
510 static const char* Cancel(
void) {
return AMPS_OPTIONS_CANCEL; }
511 static const char* Resume(
void) {
return AMPS_OPTIONS_RESUME; }
512 static const char* Pause(
void) {
return AMPS_OPTIONS_PAUSE; }
513 static const char* FullyDurable(
void) {
return AMPS_OPTIONS_FULLY_DURABLE; }
514 static const char* Expire(
void) {
return AMPS_OPTIONS_EXPIRE; }
515 static std::string Conflation(
const char* conflation_)
518 AMPS_snprintf(buf,
sizeof(buf),
"conflation=%s,", conflation_);
521 static std::string ConflationKey(
const char* conflationKey_)
523 std::string option(
"conflation_key=");
524 option.append(conflationKey_).append(
",");
527 static std::string TopN(
int topN_)
530 AMPS_snprintf(buf,
sizeof(buf),
"top_n=%d,", topN_);
533 static std::string MaxBacklog(
int maxBacklog_)
536 AMPS_snprintf(buf,
sizeof(buf),
"max_backlog=%d,", maxBacklog_);
539 static std::string Rate(
const char* rate_)
542 AMPS_snprintf(buf,
sizeof(buf),
"rate=%s,", rate_);
545 static std::string RateMaxGap(
const char* rateMaxGap_)
548 AMPS_snprintf(buf,
sizeof(buf),
"rate_max_gap=%s,", rateMaxGap_);
551 static std::string SkipN(
int skipN_)
554 AMPS_snprintf(buf,
sizeof(buf),
"skip_n=%d,", skipN_);
558 static std::string Projection(
const std::string& projection_)
560 return "projection=[" + projection_ +
"],";
563 template<
class Iterator>
564 static std::string Projection(Iterator begin_, Iterator end_)
566 std::string projection =
"projection=[";
567 for (Iterator i = begin_; i != end_; ++i)
572 projection.insert(projection.length() - 1,
"]");
576 static std::string Grouping(
const std::string& grouping_)
578 return "grouping=[" + grouping_ +
"],";
581 template<
class Iterator>
582 static std::string Grouping(Iterator begin_, Iterator end_)
584 std::string grouping =
"grouping=[";
585 for (Iterator i = begin_; i != end_; ++i)
590 grouping.insert(grouping.length() - 1,
"]");
594 static std::string Select(
const std::string& select_)
596 return "select=[" + select_ +
"],";
599 template<
class Iterator>
600 static std::string Select(Iterator begin_, Iterator end_)
602 std::string select =
"select=[";
603 for (Iterator i = begin_; i != end_; ++i)
608 select.insert(select.length() - 1,
"]");
612 static std::string AckConflationInterval(
const std::string& interval_)
614 return "ack_conflation=" + interval_ +
",";
617 static std::string AckConflationInterval(
const char* interval_)
619 static const std::string start(
"ack_conflation=");
620 return start + interval_ +
",";
626 : _optionStr(options_)
632 int getMaxBacklog(
void)
const {
return _maxBacklog; }
633 std::string getConflation(
void)
const {
return _conflation; }
634 std::string getConflationKey(
void)
const {
return _conflationKey; }
635 int getTopN(
void)
const {
return _topN; }
636 std::string getRate(
void)
const {
return _rate; }
637 std::string getRateMaxGap(
void)
const {
return _rateMaxGap; }
653 void setLive(
void) { _optionStr += AMPS_OPTIONS_LIVE; }
659 void setOOF(
void) { _optionStr += AMPS_OPTIONS_OOF; }
665 void setReplace(
void) { _optionStr += AMPS_OPTIONS_REPLACE; }
691 void setCancel(
void) { _optionStr += AMPS_OPTIONS_CANCEL; }
699 void setResume(
void) { _optionStr += AMPS_OPTIONS_RESUME; }
711 void setPause(
void) { _optionStr += AMPS_OPTIONS_PAUSE; }
734 AMPS_snprintf(buf,
sizeof(buf),
"max_backlog=%d,", maxBacklog_);
736 _maxBacklog = maxBacklog_;
747 AMPS_snprintf(buf,
sizeof(buf),
"conflation=%s,", conflation_);
749 _conflation = conflation_;
764 AMPS_snprintf(buf,
sizeof(buf),
"conflation_key=%s,", conflationKey_);
766 _conflationKey = conflationKey_;
777 AMPS_snprintf(buf,
sizeof(buf),
"top_n=%d,", topN_);
791 AMPS_snprintf(buf,
sizeof(buf),
"rate=%s,", rate_);
813 AMPS_snprintf(buf,
sizeof(buf),
"rate_max_gap=%s,", rateMaxGap_);
815 _rateMaxGap = rateMaxGap_;
826 AMPS_snprintf(buf,
sizeof(buf),
"skip_n=%d,", skipN_);
837 _projection =
"projection=[" + projection_ +
"],";
838 _optionStr += _projection;
847 template<
class Iterator>
850 _projection =
"projection=[";
851 for (Iterator i = begin_; i != end_; ++i)
856 _projection.insert(_projection.length() - 1,
"]");
857 _optionStr += _projection;
866 _grouping =
"grouping=[" + grouping_ +
"],";
867 _optionStr += _grouping;
876 template<
class Iterator>
879 _grouping =
"grouping=[";
880 for (Iterator i = begin_; i != end_; ++i)
885 _grouping.insert(_grouping.length() - 1,
"]");
886 _optionStr += _grouping;
892 operator const std::string()
894 return _optionStr.substr(0, _optionStr.length()-1);
898 std::string _optionStr;
900 std::string _conflation;
901 std::string _conflationKey;
904 std::string _rateMaxGap;
906 std::string _projection;
907 std::string _grouping;
926 SOWAndSubscribe = 256,
927 DeltaSubscribe = 512,
928 SOWAndDeltaSubscribe = 1024,
936 NoDataCommands = Publish|Unsubscribe|Heartbeat|SOWDelete|DeltaPublish
937 |Logon|StartTimer|StopTimer|Flush
943 const char* data = NULL;
size_t len = 0;
947 case 1:
return Command::Publish;
951 case 's':
return Command::SOW;
952 case 'o':
return Command::OOF;
953 case 'a':
return Command::Ack;
959 case 'l':
return Command::Logon;
960 case 'f':
return Command::Flush;
964 return Command::Publish;
969 case 's':
return Command::Subscribe;
970 case 'h':
return Command::Heartbeat;
971 case 'g':
return Command::GroupEnd;
977 case 'o':
return Command::SOWDelete;
978 case 't':
return Command::StopTimer;
984 case 'g':
return Command::GroupBegin;
985 case 'u':
return Command::Unsubscribe;
989 return Command::DeltaPublish;
991 return Command::DeltaSubscribe;
993 return Command::SOWAndSubscribe;
995 return Command::SOWAndDeltaSubscribe;
997 return Command::Unknown;
1004 unsigned command = command_;
1005 while (command > 0) { ++bits; command >>= 1; }
1007 CommandConstants<0>::Values[bits], CommandConstants<0>::Lengths[bits]);
1012 AMPS_FIELD(CommandId)
1013 AMPS_FIELD(ClientName)
1015 AMPS_FIELD(Timestamp)
1020 Field getTransmissionTime()
const 1022 return getTimestamp();
1031 getRawTimestamp(dataptr, sizeptr);
1036 AMPS_FIELD(MessageType)
1044 None = 0, Received = 1, Parsed = 2, Processed = 4, Persisted = 8, Completed = 16, Stats = 32
1052 switch (end - begin)
1055 return AckType::Stats;
1057 return AckType::Parsed;
1059 return AckType::Received;
1063 case 'e':
return AckType::Persisted;
1064 case 'r':
return AckType::Processed;
1065 case 'o':
return AckType::Completed;
1072 return AckType::None;
1079 unsigned result = AckType::None;
1080 const char* data = NULL;
size_t len = 0;
1082 const char* mark = data;
1083 for (
const char* end = data + len; data != end; ++data)
1087 result |= decodeSingleAckType(mark, data);
1091 if (mark < data) result |= decodeSingleAckType(mark, data);
1099 if(ackType_ < AckTypeConstants<0>::Entries)
1102 AckTypeConstants<0>::Values[ackType_], AckTypeConstants<0>::Lengths[ackType_]);
1107 AMPS_FIELD(SubscriptionId)
1108 AMPS_FIELD_ALIAS(SubId, SubscriptionId)
1109 AMPS_FIELD(Expiration)
1110 AMPS_FIELD(Heartbeat)
1111 AMPS_FIELD(TimeoutInterval)
1112 AMPS_FIELD(LeasePeriod)
1115 AMPS_FIELD_ALIAS(QueryId, QueryID)
1116 AMPS_FIELD(BatchSize)
1117 AMPS_FIELD(TopNRecordsReturned)
1120 AMPS_FIELD(CorrelationId)
1121 AMPS_FIELD(Sequence)
1122 AMPS_FIELD(Bookmark)
1123 AMPS_FIELD(RecordsInserted)
1124 AMPS_FIELD(RecordsUpdated)
1125 AMPS_FIELD(SowDelete)
1126 AMPS_FIELD(RecordsReturned)
1127 AMPS_FIELD(TopicMatches)
1129 AMPS_FIELD(MessageLength)
1131 AMPS_FIELD(GroupSequenceNumber)
1132 AMPS_FIELD(SubscriptionIds)
1134 AMPS_FIELD(Password)
1138 DOX_MAKEGETCOMMENT(
Options) DOX_COMMENTHEAD( Retrieves the value of the
Options header of the
Message as a new Field.)
1139 Field getOptions()
const {
1144 AMPS_Options, &ptr, &sz);
1145 if (sz && ptr[sz-1] ==
',') --sz;
1146 returnValue.assign(ptr, sz);
1150 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. )
1151 void getRawOptions(
const char** dataptr,
size_t* sizeptr)
const {
1153 AMPS_Options, dataptr, sizeptr);
1154 if (*sizeptr && *dataptr && (*dataptr)[*sizeptr-1] ==
',') --*sizeptr;
1158 DOX_MAKESETCOMMENT(x) DOX_COMMENTHEAD( Sets the value of the
Options header
for this Message.)
1159 Message& setOptions(
const std::string& v) {
1160 size_t sz = v.length();
1161 if (sz && v[sz-1] ==
',') --sz;
1163 AMPS_Options, v.c_str(), sz);
1167 DOX_MAKEASSIGNCOMMENT(x) DOX_COMMENTHEAD( Sets the value of the
Options header
for this Message.)
1168 Message& assignOptions(
const std::string& v) {
1169 size_t sz = v.length();
1170 if (sz && v[sz-1] ==
',') --sz;
1172 AMPS_Options, v.c_str(), sz);
1176 DOX_MAKEASSIGNCOMMENT(x) DOX_COMMENTHEAD( Sets the value of the
Options header
for this Message.)
1177 Message& assignOptions(
const char* data,
size_t len) {
1178 if (len && data[len-1] ==
',') --len;
1180 AMPS_Options, data, len);
1184 DOX_MAKESETCOMMENT(x) DOX_COMMENTHEAD( Sets the value of the
Options header
for this Message.)
1185 Message& setOptions(
const char* str) {
1188 size_t sz = strlen(str);
1189 if (sz && str[sz-1] ==
',') --sz;
1191 AMPS_Options, str, sz);
1196 AMPS_Options, str, 0);
1201 DOX_MAKESETCOMMENT(x) DOX_COMMENTHEAD( Sets the value of the
Options header
for this Message.)
1202 Message& setOptions(
const char* str,
size_t len) {
1203 if (len && str[len-1] ==
',') --len;
1205 AMPS_Options, str, len);
1215 Field getData()
const 1221 returnValue.assign(ptr, sz);
1225 void getRawData(
const char **data,
size_t *sz)
const 1236 Message& assignData(
const std::string &v_)
1250 Message& assignData(
const char* data_,
size_t length_)
1263 Message& assignData(
const char* data_)
1270 return _body.get().getMessage();
1274 _body.get().replace(message, owner);
1278 _body.get().disown();
1284 bool isValid(
void)
const 1286 return _body.isValid();
1290 _body.get().reset();
1294 void setBookmarkSeqNo(
size_t val)
1296 _body.get().setBookmarkSeqNo(val);
1299 size_t getBookmarkSeqNo()
const 1301 return _body.get().getBookmarkSeqNo();
1306 _body.get().setSubscriptionHandle(val);
1311 return _body.get().getSubscriptionHandle();
1314 void ack(
const char* options_ = NULL)
const;
1316 void setClientImpl(ClientImpl *pClientImpl)
1318 _body.get().setClientImpl(pClientImpl);
1321 void setIgnoreAutoAck()
const 1323 _body.get().setIgnoreAutoAck();
1326 bool getIgnoreAutoAck()
const 1328 return _body.get().getIgnoreAutoAck();
1332 void throwFor(
const T& ,
const std::string& ackReason_)
const 1334 switch (ackReason_[0])
1337 throw AuthenticationException(
"Logon failed for user \"" +
1338 (std::string)getUserId() +
"\"");
1341 switch (ackReason_.length())
1344 throw BadFilterException(
"bad filter '" +
1345 (std::string)getFilter() +
1349 if (getSowKeys().len())
1351 throw BadSowKeyException(
"bad sow key '" +
1352 (std::string)getSowKeys() +
1357 throw BadSowKeyException(
"bad sow key '" +
1358 (std::string)getSowKey() +
1363 throw BadRegexTopicException(
"bad regex topic '" +
1364 (std::string)getTopic() +
1372 if (ackReason_.length() == 23)
1374 throw DuplicateLogonException(
"Client '" +
1375 (std::string)getClientName() +
1377 (std::string)getUserId() +
1378 "' duplicate logon attempt");
1382 if (ackReason_.length() >= 9)
1384 switch (ackReason_[8])
1387 throw InvalidBookmarkException(
"invalid bookmark '" +
1388 (std::string)getBookmark() +
1392 throw CommandException(std::string(
"invalid message type '") +
1393 (std::string)getMessageType() +
1397 if (ackReason_[9] ==
'p')
1399 throw InvalidOptionsException(
"invalid options '" +
1400 (std::string)getOptions() +
1403 else if (ackReason_[9] ==
'r')
1405 throw InvalidOrderByException(
"invalid order by '" +
1406 (std::string)getOrderBy() +
1411 throw InvalidSubIdException(
"invalid subid '" +
1412 (std::string)getSubscriptionId() +
1416 if (ackReason_.length() == 13)
1418 throw InvalidTopicException(
"invalid topic '" +
1419 (std::string)getTopic() +
1422 else if (ackReason_.length() == 23)
1424 throw InvalidTopicException(
"invalid topic or filter. Topic '" +
1425 (std::string)getTopic() +
1427 (std::string)getFilter() +
1437 if (ackReason_.length() == 14)
1439 throw LogonRequiredException(
"logon required before command");
1443 switch (ackReason_[5])
1446 throw NameInUseException(
"name in use '" +
1447 (std::string)getClientName() +
1451 throw NotEntitledException(
"User \"" +
1452 (std::string)getUserId() +
1453 "\" not entitled to topic \"" +
1454 (std::string)getTopic() +
1458 throw MissingFieldsException(
"command sent with no filter or bookmark.");
1461 throw MissingFieldsException(
"command sent with no client name.");
1464 throw MissingFieldsException(
"command sent with no topic or filter.");
1467 throw CommandException(
"operation on topic '" +
1468 (std::string)getTopic() +
1469 "' with options '" +
1470 (std::string)getOptions() +
1471 "' not supported.");
1478 switch (ackReason_.length())
1481 throw MissingFieldsException(
"orderby required");
1484 throw CommandException(
"orderby too large '" +
1485 (std::string)getOrderBy() +
1491 switch (ackReason_[1])
1494 throw CommandException(
"projection clause too large in options '" +
1495 (std::string)getOptions() +
1499 throw PublishFilterException(
"Publish filter '" +
1500 (std::string)getFilter() +
1501 "' doesn't match any records.");
1508 switch (ackReason_[2])
1511 throw BadRegexTopicException(
"'regex topic not supported '" +
1512 (std::string)getTopic() +
1520 switch (ackReason_[5])
1523 throw SubidInUseException(
"subid in use '" +
1524 (std::string)getSubscriptionId() +
1528 throw CommandException(
"sow_delete command only supports one of: filter '" +
1529 (std::string)getFilter() +
1531 (std::string)getSowKeys() +
1533 (std::string)getBookmark() +
1535 (std::string)getData() +
1539 throw PublishException(
"sow store failed.");
1546 switch (ackReason_[2])
1549 throw PublishException(
"tx store failure.");
1552 throw CommandException(
"txn replay failed for '" +
1553 (std::string)getSubId() +
1561 throw CommandException(
"Error from server while processing this command: '" +
1569 return lhs + std::string(rhs);
1572 inline std::basic_ostream<char>&
1573 operator<<(std::basic_ostream<char>& os,
const Message::Field& rhs)
1575 os.write(rhs.
data(), (std::streamsize)rhs.
len());
1579 AMPS::Field::operator<(
const AMPS::Field& rhs)
const 1581 return std::lexicographical_compare(data(), data()+len(), rhs.
data(), rhs.
data()+rhs.
len());
void setFullyDurable(void)
Set the option to only provide messages that have been persisted to all replication destinations that...
Definition: Message.hpp:719
Class to hold string versions of failure reasons.
Definition: ampsplusplus.hpp:179
Message & setData(const std::string &v_)
Sets the data portion of self.
Definition: Message.hpp:1231
void amps_message_reset(amps_handle message)
Clears all fields and data in a message.
void setRateMaxGap(const char *rateMaxGap_)
Set the option for the maximum amount of time that a bookmark replay with a specified rate will allow...
Definition: Message.hpp:810
Command::Type getCommandEnum() const
Decode self's "command" field and return one of the values from Command.
Definition: Message.hpp:941
void amps_message_assign_data(amps_handle message, const amps_char *value, size_t length)
Assigns the data component of an AMPS message, without copying the value.
amps_handle amps_message_copy(amps_handle message)
Creates and returns a handle to a new AMPS message object that is a deep copy of the message passed i...
void setNoEmpties(void)
Set the option to not send empty messages on a delta subscription.
Definition: Message.hpp:670
Message deepCopy(void) const
Returns a deep copy of self.
Definition: Message.hpp:484
void setOOF(void)
Set the option to receive out of focus (OOF) messages on a subscription, where applicable.
Definition: Message.hpp:659
Message encapsulates a single message sent to or received from an AMPS server, and provides methods f...
Definition: Message.hpp:447
Message & setAckTypeEnum(unsigned ackType_)
Encode self's "ack type" field from a bitmask of values from AckType.
Definition: Message.hpp:1097
void setNoSowKey(void)
Set the option to not set the SowKey header on messages.
Definition: Message.hpp:686
MessageImpl(amps_handle message_, bool owner_=false, bool ignoreAutoAck_=false, size_t bookmarkSeqNo_=0, amps_subscription_handle subscription_=NULL, ClientImpl *clientImpl_=NULL)
Constructs a messageImpl from an existing AMPS message.
Definition: Message.hpp:105
Message & setCommandEnum(Command::Type command_)
Set self's "command" field from one of the values in Command.
Definition: Message.hpp:1001
void setSendKeys(void)
Set the option to send key fields with a delta subscription.
Definition: Message.hpp:675
void * amps_handle
Opaque handle type used to refer to objects in the AMPS api.
Definition: amps.h:179
Class for constructing the options string to pass to AMPS in a Message.
Definition: Message.hpp:499
const char * data() const
Returns the (non-null-terminated) data underlying this field.
Definition: Field.hpp:251
void setPause(void)
Set the option to pause a bookmark subscription.
Definition: Message.hpp:711
DOX_CLOSEGROUP(x) Field getData() const
Returns the data from this message.
Definition: Message.hpp:1208
void setReplace(void)
Set the option to replace a current subscription with this one.
Definition: Message.hpp:665
Valid values for setCommandEnum() and getCommandEnum().
Definition: Message.hpp:913
Message & setData(const char *data_)
Sets the data portion of self from a null-terminated string.
Definition: Message.hpp:1258
void setGrouping(Iterator begin_, Iterator end_)
Set the option for grouping the results of an aggregated query or subscription.
Definition: Message.hpp:877
void setTimestamp(void)
Set the option to send a timestamp that the message was processed on a subscription or query...
Definition: Message.hpp:681
amps_handle getMessage() const
Returns the underling AMPS message object from the C layer.
Definition: Message.hpp:149
Defines the AMPS::Field class, which represents the value of a field in a message.
size_t len() const
Returns the length of the data underlying this field.
Definition: Field.hpp:258
void setResume(void)
Set the option to resume a subscription.
Definition: Message.hpp:699
static AckType::Type decodeSingleAckType(const char *begin, const char *end)
Decodes a single ack string.
Definition: Message.hpp:1050
void setConflation(const char *conflation_)
Set the options for conflation on a subscription.
Definition: Message.hpp:744
void replace(amps_handle message_, bool owner_=false)
Causes self to refer to a new AMPS message, freeing any current message owned by self along the way...
Definition: Message.hpp:168
Options(std::string options_="")
ctor - default to None
Definition: Message.hpp:625
void amps_message_get_data(amps_handle message, amps_char **value_ptr, size_t *length_ptr)
Gets the data component of an AMPS message.
amps_handle amps_message_create(amps_handle client)
Functions for creation and manipulation of AMPS messages.
void setRate(const char *rate_)
Set the option for the maximum rate at which messages are provided to the subscription.
Definition: Message.hpp:788
Valid values for the setAckTypeEnum() and getAckTypeEnum() methods.
Definition: Message.hpp:1040
void amps_message_assign_field_value(amps_handle message, FieldId field, const amps_char *value, size_t length)
Assigns the value of a header field in an AMPS message, without copying the value.
void setGrouping(const std::string &grouping_)
Set the option for grouping the results of an aggregated query or subscription.
Definition: Message.hpp:864
void amps_message_set_field_value(amps_handle message, FieldId field, const amps_char *value, size_t length)
Sets the value of a header field in an AMPS message.
void setTopN(int topN_)
Set the top N option, which specifies the maximum number of messages to return for this command...
Definition: Message.hpp:774
void amps_message_set_data_nts(amps_handle message, const amps_char *value)
Sets the data component of an AMPS message.
void setMaxBacklog(int maxBacklog_)
Set the option for maximum backlog this subscription is willing to accept.
Definition: Message.hpp:731
void getRawTransmissionTime(const char **dataptr, size_t *sizeptr) const
Definition: Message.hpp:1029
void amps_message_get_field_value(amps_handle message, FieldId field, const amps_char **value_ptr, size_t *length_ptr)
Retrieves the value of a header field in an AMPS message.
MessageImpl()
Constructs a MessageImpl with a new, empty AMPS message.
Definition: Message.hpp:118
Field represents the value of a single field in a Message.
Definition: Field.hpp:84
void setConflationKey(const char *conflationKey_)
Set the options for the conflation key, the identifiers for the field or fields used by AMPS to deter...
Definition: Message.hpp:761
void setProjection(const std::string &projection_)
Set the option for projecting the results of an aggregated query or subscription. ...
Definition: Message.hpp:835
void setSkipN(int skipN_)
Set the option for skip N, the number of messages in the result set to skip before returning messages...
Definition: Message.hpp:823
void amps_message_set_data(amps_handle message, const amps_char *value, size_t length)
Sets the data component of an AMPS message.
Message(amps_handle message_, bool owner_=false)
Constructs a new Message to wrap message.
Definition: Message.hpp:470
void setNone(void)
Clear any previously set options and set the options to an empty string (AMPS_OPTIONS_NONE).
Definition: Message.hpp:642
Implementation class for a Message.
Definition: Message.hpp:83
void setProjection(Iterator begin_, Iterator end_)
Set the option for projecting the results of an aggregated query or subscription. ...
Definition: Message.hpp:848
unsigned getAckTypeEnum() const
Decode self's "ack type" field and return the corresponding bitmask of values from AckType...
Definition: Message.hpp:1077
Message & setData(const char *data_, size_t length_)
Sets the data portion of self from a char array.
Definition: Message.hpp:1245
Definition: ampsplusplus.hpp:103
void setCancel(void)
Set the cancel option, used on a sow_delete command to return a message to the queue.
Definition: Message.hpp:691
Message()
Construct a new, empty Message.
Definition: Message.hpp:478
void setLive(void)
Set the live option for a bookmark subscription, which requests that the subscription receives messag...
Definition: Message.hpp:653
void amps_message_destroy(amps_handle message)
Destroys and frees the memory associated with an AMPS message object.