AMPS C/C++ Client Class Reference AMPS C/C++ Client Version 5.3.5.4

Message.hpp
Go to the documentation of this file.
1 //
3 // Copyright (c) 2010-2026 60East Technologies Inc., All Rights Reserved.
4 //
5 // This computer software is owned by 60East Technologies Inc. and is
6 // protected by U.S. copyright laws and other laws and by international
7 // treaties. This computer software is furnished by 60East Technologies
8 // Inc. pursuant to a written license agreement and may be used, copied,
9 // transmitted, and stored only in accordance with the terms of such
10 // license agreement and with the inclusion of the above copyright notice.
11 // This computer software or any other copies thereof may not be provided
12 // or otherwise made available to any other person.
13 //
14 // U.S. Government Restricted Rights. This computer software: (a) was
15 // developed at private expense and is in all respects the proprietary
16 // information of 60East Technologies Inc.; (b) was not developed with
17 // government funds; (c) is a trade secret of 60East Technologies Inc.
18 // for all purposes of the Freedom of Information Act; and (d) is a
19 // commercial item and thus, pursuant to Section 12.212 of the Federal
20 // Acquisition Regulations (FAR) and DFAR Supplement Section 227.7202,
21 // Government's use, duplication or disclosure of the computer software
22 // is subject to the restrictions set forth by 60East Technologies Inc..
23 //
25 #ifndef __AMPS_MESSAGE_HPP__
26 #define __AMPS_MESSAGE_HPP__
27 #include "amps/util.hpp"
28 #include "amps/constants.hpp"
29 #include "amps/amps_generated.h"
30 #include "amps/Field.hpp"
31 #include <stdio.h>
32 #include <algorithm>
33 #include <ostream>
34 #include <string>
35 #define AMPS_UNSET_SEQUENCE (amps_uint64_t)-1
36
37 #include <functional>
38 #include <algorithm>
39
43
44 #define AMPS_OPTIONS_NONE ""
45 #define AMPS_OPTIONS_LIVE "live,"
46 #define AMPS_OPTIONS_OOF "oof,"
47 #define AMPS_OPTIONS_REPLACE "replace,"
48 #define AMPS_OPTIONS_NOEMPTIES "no_empties,"
49 #define AMPS_OPTIONS_SENDKEYS "send_keys,"
50 #define AMPS_OPTIONS_TIMESTAMP "timestamp,"
51 #define AMPS_OPTIONS_NOSOWKEY "no_sowkey,"
52 #define AMPS_OPTIONS_CANCEL "cancel,"
53 #define AMPS_OPTIONS_RESUME "resume,"
54 #define AMPS_OPTIONS_PAUSE "pause,"
55 #define AMPS_OPTIONS_FULLY_DURABLE "fully_durable,"
56 #define AMPS_OPTIONS_EXPIRE "expire,"
57 #define AMPS_OPTIONS_TOPN(x) "top_n=##x,"
58 #define AMPS_OPTIONS_MAXBACKLOG(x) "max_backlog=##x,"
59 #define AMPS_OPTIONS_RATE(x) "rate=##x,"
60
61 extern AMPSDLL volatile amps_uint64_t _AMPS_NEXT_ID;
62
63 namespace AMPS
64 {
65 typedef void* amps_subscription_handle;
66
67 class ClientImpl;
68
73 class MessageImpl : public RefBody
74 {
75 private:
76 amps_handle _message;
77 //Mutex _lock;
78 bool _owner;
79 mutable bool _isIgnoreAutoAck;
80 size_t _bookmarkSeqNo;
81 amps_subscription_handle _subscription;
82 ClientImpl* _clientImpl;
83 public:
95 MessageImpl(amps_handle message_, bool owner_ = false,
96 bool ignoreAutoAck_ = false, size_t bookmarkSeqNo_ = 0,
97 amps_subscription_handle subscription_ = NULL,
98 ClientImpl* clientImpl_ = NULL)
99 : _message(message_), _owner(owner_), _isIgnoreAutoAck(ignoreAutoAck_)
100 , _bookmarkSeqNo(bookmarkSeqNo_)
101 , _subscription(subscription_), _clientImpl(clientImpl_)
102 {
103 }
104
108 MessageImpl()
109 : _message(NULL), _owner(true), _isIgnoreAutoAck(false), _bookmarkSeqNo(0), _subscription(NULL), _clientImpl(NULL)
110 {
111 // try to create one
112 _message = amps_message_create(NULL);
113 }
114
115 virtual ~MessageImpl()
116 {
117 if (_owner && _message)
118 {
119 amps_message_destroy(_message);
120 }
121 }
122
123 MessageImpl* copy() const
124 {
125 amps_handle copy = amps_message_copy(_message);
126 return new MessageImpl(copy, true, _isIgnoreAutoAck, _bookmarkSeqNo,
127 _subscription, _clientImpl);
128 }
129
130 void copy(const MessageImpl& rhs_)
131 {
132 if (_owner && _message)
133 {
134 amps_message_destroy(_message);
135 }
136 _message = amps_message_copy(rhs_._message);
137 _owner = true;
138 _bookmarkSeqNo = rhs_._bookmarkSeqNo;
139 _subscription = rhs_._subscription;
140 _isIgnoreAutoAck = rhs_._isIgnoreAutoAck;
141 _clientImpl = rhs_._clientImpl;
142 }
143
144 void setClientImpl(ClientImpl* clientImpl_)
145 {
146 _clientImpl = clientImpl_;
147 }
148
149 ClientImpl* clientImpl(void) const
150 {
151 return _clientImpl;
152 }
153
156 amps_handle getMessage() const
157 {
158 return _message;
159 }
160
161 void reset()
162 {
163 //Lock<Mutex> l(_lock);
164 amps_message_reset(_message);
165 _bookmarkSeqNo = 0;
166 _subscription = NULL;
167 _isIgnoreAutoAck = false;
168 _clientImpl = NULL;
169 }
170
176 void replace(amps_handle message_, bool owner_ = false)
177 {
178 //Lock<Mutex> l(_lock);
179 if (_message == message_)
180 {
181 return;
182 }
183 if (_owner && _message)
184 {
185 amps_message_destroy(_message);
186 }
187 _owner = owner_;
188 _message = message_;
189 _bookmarkSeqNo = 0;
190 _subscription = NULL;
191 _isIgnoreAutoAck = false;
192 }
193
194 void disown()
195 {
196 //Lock<Mutex> l(_lock);
197 _owner = false;
198 }
199
200 static unsigned long newId()
201 {
202 return (unsigned long)(AMPS_FETCH_ADD_VOLATILE(&_AMPS_NEXT_ID, 1));
203 }
204
205 void setBookmarkSeqNo(size_t val_)
206 {
207 _bookmarkSeqNo = val_;
208 }
209
210 size_t getBookmarkSeqNo(void) const
211 {
212 return _bookmarkSeqNo;
213 }
214
215 void setSubscriptionHandle(amps_subscription_handle subscription_)
216 {
217 _subscription = subscription_;
218 }
219
220 amps_subscription_handle getSubscriptionHandle(void) const
221 {
222 return _subscription;
223 }
224
225 void setIgnoreAutoAck() const
226 {
227 _isIgnoreAutoAck = true;
228 }
229
230 bool getIgnoreAutoAck() const
231 {
232 return _isIgnoreAutoAck;
233 }
234 };
235
236
237 // This block of macros works with the Doxygen preprocessor to
238 // create documentation comments for fields defined with the AMPS_FIELD macro.
239 // A C++ compiler removes comments before expanding macros, so these macros
240 // must ONLY be defined for Doxygen and not for actual compilation.
241
242 #ifdef DOXYGEN_PREPROCESSOR
243
244 #define DOX_COMMENTHEAD(s) / ## ** ## s ## * ## /
245 #define DOX_GROUPNAME(s) DOX_COMMENTHEAD(@name s Functions)
246 #define DOX_OPENGROUP(s) DOX_COMMENTHEAD(@{) \
247 DOX_GROUPNAME(s)
248 #define DOX_CLOSEGROUP() DOX_COMMENTHEAD(@})
249 #define DOX_MAKEGETCOMMENT(x) DOX_COMMENTHEAD( Retrieves the value of the x header of the Message as a Field which 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. )
250 #define DOX_MAKEGETRAWCOMMENT(x) DOX_COMMENTHEAD( Modifies the passed in arguments to reference the value of the x header of self in 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. )
251 #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. )
252 #define DOX_MAKEASSIGNCOMMENT(x) DOX_COMMENTHEAD( Assigns the value of the x header for this Message without copying. 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. )
253 #define DOX_MAKEASSIGNOWNCOMMENT(x) DOX_COMMENTHEAD( Assigns the value of the x header for this Message without copying and makes this Message responsible for deleting the value. 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. )
254 #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.)
255
256 #else
257
258 #define DOX_COMMENTHEAD(s)
259 #define DOX_GROUPNAME(s)
260 #define DOX_OPENGROUP(x)
261 #define DOX_CLOSEGROUP()
262 #define DOX_MAKEGETCOMMENT(x)
263 #define DOX_MAKEGETRAWCOMMENT(x)
264 #define DOX_MAKESETCOMMENT(x)
265 #define DOX_MAKEASSIGNCOMMENT(x)
266 #define DOX_MAKEASSIGNOWNCOMMENT(x)
267 #define DOX_MAKENEWCOMMENT(x)
268
269 #endif
270
271 // Macro for defining all of the necessary methods for a field in an AMPS
272 // message.
273
274
275 #define AMPS_FIELD(x) \
276 DOX_OPENGROUP(x) \
277 DOX_MAKEGETCOMMENT(x) \
278 Field get##x() const {\
279 Field returnValue;\
280 const char* ptr;\
281 size_t sz;\
282 amps_message_get_field_value(_body.get().getMessage(),\
283 AMPS_##x, &ptr, &sz);\
284 returnValue.assign(ptr, sz);\
285 return returnValue;\
286 }\
287 DOX_MAKEGETRAWCOMMENT(x) \
288 void getRaw##x(const char** dataptr, size_t* sizeptr) const {\
289 amps_message_get_field_value(_body.get().getMessage(),\
290 AMPS_##x, dataptr, sizeptr);\
291 return;\
292 }\
293 DOX_MAKESETCOMMENT(x) \
294 Message& set##x(const std::string& v) {\
295 amps_message_set_field_value(_body.get().getMessage(),\
296 AMPS_##x, v.c_str(), v.length());\
297 return *this;\
298 }\
299 DOX_MAKESETCOMMENT(x) \
300 Message& set##x(amps_uint64_t v) {\
301 char buf[22];\
302 AMPS_snprintf_amps_uint64_t(buf,22,v);\
303 amps_message_set_field_value_nts(_body.get().getMessage(),\
304 AMPS_##x, buf);\
305 return *this;\
306 }\
307 DOX_MAKEASSIGNCOMMENT(x) \
308 Message& assign##x(const std::string& v) {\
309 amps_message_assign_field_value(_body.get().getMessage(),\
310 AMPS_##x, v.c_str(), v.length());\
311 return *this;\
312 }\
313 DOX_MAKEASSIGNCOMMENT(x) \
314 Message& assign##x(const char* data, size_t len) {\
315 amps_message_assign_field_value(_body.get().getMessage(),\
316 AMPS_##x, data, len);\
317 return *this;\
318 }\
319 DOX_MAKEASSIGNOWNCOMMENT(x) \
320 Message& assignOwnership##x(const Field& f) {\
321 amps_message_assign_field_value_ownership(_body.get().getMessage(),\
322 AMPS_##x, f.data(), f.len());\
323 return *this;\
324 }\
325 DOX_MAKESETCOMMENT(x) \
326 Message& set##x(const char* str) {\
327 amps_message_set_field_value_nts(_body.get().getMessage(),\
328 AMPS_##x, str);\
329 return *this;\
330 }\
331 DOX_MAKESETCOMMENT(x) \
332 Message& set##x(const char* str,size_t len) {\
333 amps_message_set_field_value(_body.get().getMessage(),\
334 AMPS_##x, str,len);\
335 return *this;\
336 }\
337 DOX_MAKENEWCOMMENT(x) \
338 Message& new##x() {\
339 char buf[Message::IdentifierLength+1];\
340 buf[Message::IdentifierLength] = 0;\
341 AMPS_snprintf(buf, Message::IdentifierLength+1, "auto%lu" , (unsigned long)(_body.get().newId()));\
342 amps_message_set_field_value_nts(_body.get().getMessage(),\
343 AMPS_##x, buf);\
344 return *this;\
345 } \
346 DOX_CLOSEGROUP()
347
348 #define AMPS_FIELD_ALIAS(x,y) \
349 DOX_OPENGROUP(y) \
350 DOX_MAKEGETCOMMENT(y) \
351 Field get##y() const {\
352 Field returnValue;\
353 const char* ptr;\
354 size_t sz;\
355 amps_message_get_field_value(_body.get().getMessage(),\
356 AMPS_##y, &ptr, &sz);\
357 returnValue.assign(ptr, sz);\
358 return returnValue;\
359 }\
360 DOX_MAKEGETRAWCOMMENT(y) \
361 void getRaw##y(const char** dataptr, size_t* sizeptr) const {\
362 amps_message_get_field_value(_body.get().getMessage(),\
363 AMPS_##y, dataptr, sizeptr);\
364 return;\
365 }\
366 DOX_MAKESETCOMMENT(y) \
367 Message& set##y(const std::string& v) {\
368 amps_message_set_field_value(_body.get().getMessage(),\
369 AMPS_##y, v.c_str(), v.length());\
370 return *this;\
371 }\
372 DOX_MAKESETCOMMENT(y) \
373 Message& set##y(amps_uint64_t v) {\
374 char buf[22];\
375 AMPS_snprintf_amps_uint64_t(buf,22,v);\
376 amps_message_set_field_value_nts(_body.get().getMessage(),\
377 AMPS_##y, buf);\
378 return *this;\
379 }\
380 DOX_MAKEASSIGNCOMMENT(y) \
381 Message& assign##y(const std::string& v) {\
382 amps_message_assign_field_value(_body.get().getMessage(),\
383 AMPS_##y, v.c_str(), v.length());\
384 return *this;\
385 }\
386 DOX_MAKEASSIGNCOMMENT(y) \
387 Message& assign##y(const char* data, size_t len) {\
388 amps_message_assign_field_value(_body.get().getMessage(),\
389 AMPS_##y, data, len);\
390 return *this;\
391 }\
392 DOX_MAKESETCOMMENT(y) \
393 Message& set##y(const char* str) {\
394 amps_message_set_field_value_nts(_body.get().getMessage(),\
395 AMPS_##y, str);\
396 return *this;\
397 }\
398 DOX_MAKESETCOMMENT(y) \
399 Message& set##y(const char* str,size_t len) {\
400 amps_message_set_field_value(_body.get().getMessage(),\
401 AMPS_##y, str,len);\
402 return *this;\
403 }\
404 DOX_MAKENEWCOMMENT(y) \
405 Message& new##y() {\
406 char buf[Message::IdentifierLength+1];\
407 buf[Message::IdentifierLength] = 0;\
408 AMPS_snprintf(buf, Message::IdentifierLength+1, "auto%lux" , (unsigned long)(_body.get().newId()));\
409 amps_message_set_field_value_nts(_body.get().getMessage(),\
410 AMPS_##y, buf);\
411 return *this;\
412 }\
413 DOX_MAKEGETCOMMENT(y) \
414 Field get##x() const {\
415 Field returnValue;\
416 const char* ptr;\
417 size_t sz;\
418 amps_message_get_field_value(_body.get().getMessage(),\
419 AMPS_##y, &ptr, &sz);\
420 returnValue.assign(ptr, sz);\
421 return returnValue;\
422 }\
423 DOX_MAKEGETRAWCOMMENT(y) \
424 void getRaw##x(const char** dataptr, size_t* sizeptr) const {\
425 amps_message_get_field_value(_body.get().getMessage(),\
426 AMPS_##y, dataptr, sizeptr);\
427 return;\
428 }\
429 DOX_MAKESETCOMMENT(y) \
430 Message& set##x(const std::string& v) {\
431 amps_message_set_field_value(_body.get().getMessage(),\
432 AMPS_##y, v.c_str(), v.length());\
433 return *this;\
434 }\
435 DOX_MAKESETCOMMENT(y) \
436 Message& set##x(amps_uint64_t v) {\
437 char buf[22];\
438 AMPS_snprintf_amps_uint64_t(buf,22,v);\
439 amps_message_set_field_value_nts(_body.get().getMessage(),\
440 AMPS_##y, buf);\
441 return *this;\
442 }\
443 DOX_MAKEASSIGNCOMMENT(y) \
444 Message& assign##x(const std::string& v) {\
445 amps_message_assign_field_value(_body.get().getMessage(),\
446 AMPS_##y, v.c_str(), v.length());\
447 return *this;\
448 }\
449 DOX_MAKEASSIGNCOMMENT(y) \
450 Message& assign##x(const char* data, size_t len) {\
451 amps_message_assign_field_value(_body.get().getMessage(),\
452 AMPS_##y, data, len);\
453 return *this;\
454 }\
455 DOX_MAKESETCOMMENT(y) \
456 Message& set##x(const char* str) {\
457 amps_message_set_field_value_nts(_body.get().getMessage(),\
458 AMPS_##y, str);\
459 return *this;\
460 }\
461 DOX_MAKESETCOMMENT(y) \
462 Message& set##x(const char* str,size_t len) {\
463 amps_message_set_field_value(_body.get().getMessage(),\
464 AMPS_##y, str,len);\
465 return *this;\
466 }\
467 DOX_MAKENEWCOMMENT(y) \
468 Message& new##x() {\
469 char buf[Message::IdentifierLength+1];\
470 buf[Message::IdentifierLength] = 0;\
471 AMPS_snprintf(buf, Message::IdentifierLength+1, "auto%lux" , (unsigned long)(_body.get().newId()));\
472 amps_message_set_field_value_nts(_body.get().getMessage(),\
473 AMPS_##y, buf);\
474 return *this;\
475 } \
476 DOX_CLOSEGROUP()
477
478
519 class Message
520 {
521 RefHandle<MessageImpl> _body;
522
523 Message(MessageImpl* body_) : _body(body_) { ; }
524
525 public:
526 typedef AMPS::Field Field;
527
530 static const unsigned int IdentifierLength = 32;
531
534 static const size_t BOOKMARK_NONE = AMPS_UNSET_INDEX;
535
539 enum CtorFlag { EMPTY };
540
543 Message(CtorFlag) : _body()
544 {
545 }
546
553 Message(amps_handle message_, bool owner_ = false)
554 : _body(new MessageImpl(message_, owner_))
555 {
556 }
557
561 Message() : _body(new MessageImpl())
562 {
563 }
564
567 Message deepCopy(void) const
568 {
569 return Message(_body.get().copy());
570 }
571
574 void deepCopy(const Message& rhs_)
575 {
576 _body.get().copy(rhs_._body.get());
577 }
578
589 class Options
590 {
591 public:
592 static const char* None(void)
593 {
594 return AMPS_OPTIONS_NONE;
595 }
596 static const char* Live(void)
597 {
598 return AMPS_OPTIONS_LIVE;
599 }
600 static const char* OOF(void)
601 {
602 return AMPS_OPTIONS_OOF;
603 }
604 static const char* Replace(void)
605 {
606 return AMPS_OPTIONS_REPLACE;
607 }
608 static const char* NoEmpties(void)
609 {
610 return AMPS_OPTIONS_NOEMPTIES;
611 }
612 static const char* SendKeys(void)
613 {
614 return AMPS_OPTIONS_SENDKEYS;
615 }
616 static const char* Timestamp(void)
617 {
618 return AMPS_OPTIONS_TIMESTAMP;
619 }
620 static const char* NoSowKey(void)
621 {
622 return AMPS_OPTIONS_NOSOWKEY;
623 }
624 static const char* Cancel(void)
625 {
626 return AMPS_OPTIONS_CANCEL;
627 }
628 static const char* Resume(void)
629 {
630 return AMPS_OPTIONS_RESUME;
631 }
632 static const char* Pause(void)
633 {
634 return AMPS_OPTIONS_PAUSE;
635 }
636 static const char* FullyDurable(void)
637 {
638 return AMPS_OPTIONS_FULLY_DURABLE;
639 }
640 static const char* Expire(void)
641 {
642 return AMPS_OPTIONS_EXPIRE;
643 }
644 static std::string Conflation(const char* conflation_)
645 {
646 char buf[64];
647 AMPS_snprintf(buf, sizeof(buf), "conflation=%s,", conflation_);
648 return buf;
649 }
650 static std::string ConflationKey(const char* conflationKey_)
651 {
652 std::string option("conflation_key=");
653 option.append(conflationKey_).append(",");
654 return option;
655 }
656 static std::string TopN(int topN_)
657 {
658 char buf[24];
659 AMPS_snprintf(buf, sizeof(buf), "top_n=%d,", topN_);
660 return buf;
661 }
662 static std::string MaxBacklog(int maxBacklog_)
663 {
664 char buf[24];
665 AMPS_snprintf(buf, sizeof(buf), "max_backlog=%d,", maxBacklog_);
666 return buf;
667 }
668 static std::string Rate(const char* rate_)
669 {
670 char buf[64];
671 AMPS_snprintf(buf, sizeof(buf), "rate=%s,", rate_);
672 return buf;
673 }
674 static std::string RateMaxGap(const char* rateMaxGap_)
675 {
676 char buf[64];
677 AMPS_snprintf(buf, sizeof(buf), "rate_max_gap=%s,", rateMaxGap_);
678 return buf;
679 }
680 static std::string SkipN(int skipN_)
681 {
682 char buf[24];
683 AMPS_snprintf(buf, sizeof(buf), "skip_n=%d,", skipN_);
684 return buf;
685 }
686
687 static std::string Projection(const std::string& projection_)
688 {
689 return "projection=[" + projection_ + "],";
690 }
691
692 template<class Iterator>
693 static std::string Projection(Iterator begin_, Iterator end_)
694 {
695 std::string projection = "projection=[";
696 for (Iterator i = begin_; i != end_; ++i)
697 {
698 projection += *i;
699 projection += ',';
700 }
701 projection.insert(projection.length() - 1, "]");
702 return projection;
703 }
704
705 static std::string Grouping(const std::string& grouping_)
706 {
707 return "grouping=[" + grouping_ + "],";
708 }
709
710 template<class Iterator>
711 static std::string Grouping(Iterator begin_, Iterator end_)
712 {
713 std::string grouping = "grouping=[";
714 for (Iterator i = begin_; i != end_; ++i)
715 {
716 grouping += *i;
717 grouping += ',';
718 }
719 grouping.insert(grouping.length() - 1, "]");
720 return grouping;
721 }
722
723 static std::string Select(const std::string& select_)
724 {
725 return "select=[" + select_ + "],";
726 }
727
728 template<class Iterator>
729 static std::string Select(Iterator begin_, Iterator end_)
730 {
731 std::string select = "select=[";
732 for (Iterator i = begin_; i != end_; ++i)
733 {
734 select += *i;
735 select += ',';
736 }
737 select.insert(select.length() - 1, "]");
738 return select;
739 }
740
741 static std::string AckConflationInterval(const std::string& interval_)
742 {
743 return "ack_conflation=" + interval_ + ",";
744 }
745
746 static std::string AckConflationInterval(const char* interval_)
747 {
748 static const std::string start("ack_conflation="); // -V1096
749 return start + interval_ + ",";
750 }
751
752 static std::string BookmarkNotFound(const char* action_)
753 {
754 static const std::string start("bookmark_not_found="); // -V1096
755 return start + action_ + ",";
756 }
757
758 static std::string BookmarkNotFoundNow()
759 {
760 return BookmarkNotFound("now");
761 }
762
763 static std::string BookmarkNotFoundEpoch()
764 {
765 return BookmarkNotFound("epoch");
766 }
767
768 static std::string BookmarkNotFoundFail()
769 {
770 return BookmarkNotFound("fail");
771 }
772
775 Options(std::string options_ = "")
776 : _optionStr(options_)
777 , _maxBacklog(0)
778 , _topN(0)
779 , _skipN(0)
780 {;}
781
782 int getMaxBacklog(void) const
783 {
784 return _maxBacklog;
785 }
786 std::string getConflation(void) const
787 {
788 return _conflation;
789 }
790 std::string getConflationKey(void) const
791 {
792 return _conflationKey;
793 }
794 int getTopN(void) const
795 {
796 return _topN;
797 }
798 std::string getRate(void) const
799 {
800 return _rate;
801 }
802 std::string getRateMaxGap(void) const
803 {
804 return _rateMaxGap;
805 }
806
810 void setNone(void)
811 {
812 _optionStr.clear();
813 }
814
824 void setLive(void)
825 {
826 _optionStr += AMPS_OPTIONS_LIVE;
827 }
828
833 void setOOF(void)
834 {
835 _optionStr += AMPS_OPTIONS_OOF;
836 }
837
842 void setReplace(void)
843 {
844 _optionStr += AMPS_OPTIONS_REPLACE;
845 }
846
850 void setNoEmpties(void)
851 {
852 _optionStr += AMPS_OPTIONS_NOEMPTIES;
853 }
854
858 void setSendKeys(void)
859 {
860 _optionStr += AMPS_OPTIONS_SENDKEYS;
861 }
862
867 void setTimestamp(void)
868 {
869 _optionStr += AMPS_OPTIONS_TIMESTAMP;
870 }
871
875 void setNoSowKey(void)
876 {
877 _optionStr += AMPS_OPTIONS_NOSOWKEY;
878 }
879
883 void setCancel(void)
884 {
885 _optionStr += AMPS_OPTIONS_CANCEL;
886 }
887
894 void setResume(void)
895 {
896 _optionStr += AMPS_OPTIONS_RESUME;
897 }
898
909 void setPause(void)
910 {
911 _optionStr += AMPS_OPTIONS_PAUSE;
912 }
913
920 void setFullyDurable(void)
921 {
922 _optionStr += AMPS_OPTIONS_FULLY_DURABLE;
923 }
924
935 void setMaxBacklog(int maxBacklog_)
936 {
937 char buf[24];
938 AMPS_snprintf(buf, sizeof(buf), "max_backlog=%d,", maxBacklog_);
939 _optionStr += buf;
940 _maxBacklog = maxBacklog_;
941 }
942
948 void setConflation(const char* conflation_)
949 {
950 char buf[64];
951 AMPS_snprintf(buf, sizeof(buf), "conflation=%s,", conflation_);
952 _optionStr += buf;
953 _conflation = conflation_;
954 }
955
965 void setConflationKey(const char* conflationKey_)
966 {
967 char buf[64];
968 AMPS_snprintf(buf, sizeof(buf), "conflation_key=%s,", conflationKey_);
969 _optionStr += buf;
970 _conflationKey = conflationKey_;
971 }
972
978 void setTopN(int topN_)
979 {
980 char buf[24];
981 AMPS_snprintf(buf, sizeof(buf), "top_n=%d,", topN_);
982 _optionStr += buf;
983 _topN = topN_;
984 }
985
992 void setRate(const char* rate_)
993 {
994 char buf[64];
995 AMPS_snprintf(buf, sizeof(buf), "rate=%s,", rate_);
996 _optionStr += buf;
997 _rate = rate_;
998 }
999
1014 void setRateMaxGap(const char* rateMaxGap_)
1015 {
1016 char buf[64];
1017 AMPS_snprintf(buf, sizeof(buf), "rate_max_gap=%s,", rateMaxGap_);
1018 _optionStr += buf;
1019 _rateMaxGap = rateMaxGap_;
1020 }
1021
1027 void setSkipN(int skipN_)
1028 {
1029 char buf[24];
1030 AMPS_snprintf(buf, sizeof(buf), "skip_n=%d,", skipN_);
1031 _optionStr += buf;
1032 _skipN = skipN_;
1033 }
1034
1039 void setProjection(const std::string& projection_)
1040 {
1041 _projection = "projection=[" + projection_ + "],";
1042 _optionStr += _projection;
1043 }
1044
1045
1051 template<class Iterator>
1052 void setProjection(Iterator begin_, Iterator end_)
1053 {
1054 _projection = "projection=[";
1055 for (Iterator i = begin_; i != end_; ++i)
1056 {
1057 _projection += *i;
1058 _projection += ',';
1059 }
1060 _projection.insert(_projection.length() - 1, "]");
1061 _optionStr += _projection;
1062 }
1063
1068 void setGrouping(const std::string& grouping_)
1069 {
1070 _grouping = "grouping=[" + grouping_ + "],";
1071 _optionStr += _grouping;
1072 }
1073
1074
1080 template<class Iterator>
1081 void setGrouping(Iterator begin_, Iterator end_)
1082 {
1083 _grouping = "grouping=[";
1084 for (Iterator i = begin_; i != end_; ++i)
1085 {
1086 _grouping += *i;
1087 _grouping += ',';
1088 }
1089 _grouping.insert(_grouping.length() - 1, "]");
1090 _optionStr += _grouping;
1091 }
1092
1097 void setBookmarkNotFound(const char* action_)
1098 {
1099 _optionStr += BookmarkNotFound(action_);
1100 }
1101
1105 void setBookmarkNotFoundNow()
1106 {
1107 _optionStr += BookmarkNotFoundNow();
1108 }
1109
1113 void setBookmarkNotFoundEpoch()
1114 {
1115 _optionStr += BookmarkNotFoundEpoch();
1116 }
1117
1121 void setBookmarkNotFoundFail()
1122 {
1123 _optionStr += BookmarkNotFoundFail();
1124 }
1125
1129 operator std::string()
1130 {
1131 return _optionStr.substr(0, _optionStr.length() - 1);
1132 }
1136 size_t getLength() const
1137 {
1138 return (_optionStr.empty() ? 0 : _optionStr.length() - 1);
1139 }
1140
1145 const char* getStr() const
1146 {
1147 return (_optionStr.empty() ? 0 : _optionStr.data());
1148 }
1149
1150 private:
1151 std::string _optionStr;
1152 int _maxBacklog;
1153 std::string _conflation;
1154 std::string _conflationKey;
1155 int _topN;
1156 std::string _rate;
1157 std::string _rateMaxGap;
1158 int _skipN;
1159 std::string _projection;
1160 std::string _grouping;
1161 };
1162
1165 struct AckType
1166 {
1167 typedef enum : unsigned
1168 {
1169 None = 0, Received = 1, Parsed = 2, Processed = 4, Persisted = 8, Completed = 16, Stats = 32
1170 } Type;
1171 };
1172 AMPS_FIELD(AckType)
1175 static inline AckType::Type decodeSingleAckType(const char* begin, const char* end)
1176 {
1177 switch (end - begin)
1178 {
1179 case 5:
1180 return AckType::Stats;
1181 case 6:
1182 return AckType::Parsed;
1183 case 8:
1184 return AckType::Received;
1185 case 9:
1186 switch (begin[1])
1187 {
1188 case 'e': return AckType::Persisted;
1189 case 'r': return AckType::Processed;
1190 case 'o': return AckType::Completed;
1191 default: break;
1192 }
1193 break;
1194 default:
1195 break;
1196 }
1197 return AckType::None;
1198 }
1202 unsigned getAckTypeEnum() const
1203 {
1204 unsigned result = AckType::None;
1205 const char* data = NULL; size_t len = 0;
1206 amps_message_get_field_value(_body.get().getMessage(), AMPS_AckType, &data, &len);
1207 const char* mark = data;
1208 for (const char* end = data + len; data != end; ++data)
1209 {
1210 if (*data == ',')
1211 {
1212 result |= decodeSingleAckType(mark, data);
1213 mark = data + 1;
1214 }
1215 }
1216 if (mark < data)
1217 {
1218 result |= decodeSingleAckType(mark, data);
1219 }
1220 return result;
1221 }
1225 Message& setAckTypeEnum(unsigned ackType_)
1226 {
1227 if (ackType_ < AckTypeConstants<0>::Entries)
1228 {
1229 amps_message_assign_field_value(_body.get().getMessage(), AMPS_AckType,
1230 AckTypeConstants<0>::Values[ackType_], AckTypeConstants<0>::Lengths[ackType_]);
1231 }
1232 return *this;
1233 }
1234
1235 AMPS_FIELD(BatchSize)
1236 AMPS_FIELD(Bookmark)
1237 AMPS_FIELD(Command)
1238
1242 struct Command
1243 {
1244 typedef enum
1245 {
1246 Unknown = 0,
1247 Publish = 1,
1248 Subscribe = 2,
1249 Unsubscribe = 4,
1250 SOW = 8,
1251 Heartbeat = 16,
1252 SOWDelete = 32,
1253 DeltaPublish = 64,
1254 Logon = 128,
1255 SOWAndSubscribe = 256,
1256 DeltaSubscribe = 512,
1257 SOWAndDeltaSubscribe = 1024,
1258 StartTimer = 2048,
1259 StopTimer = 4096,
1260 GroupBegin = 8192,
1261 GroupEnd = 16384,
1262 OOF = 32768,
1263 Ack = 65536,
1264 Flush = 131072,
1265 NoDataCommands = Publish | Unsubscribe | Heartbeat | SOWDelete | DeltaPublish
1266 | Logon | StartTimer | StopTimer | Flush
1267 } Type;
1268 };
1270 Command::Type getCommandEnum() const
1271 {
1272 const char* data = NULL; size_t len = 0;
1273 amps_message_get_field_value(_body.get().getMessage(), AMPS_Command, &data, &len);
1274 switch (len)
1275 {
1276 case 1: return Command::Publish; // -V1037
1277 case 3:
1278 switch (data[0])
1279 {
1280 case 's': return Command::SOW;
1281 case 'o': return Command::OOF;
1282 case 'a': return Command::Ack;
1283 }
1284 break;
1285 case 5:
1286 switch (data[0])
1287 {
1288 case 'l': return Command::Logon;
1289 case 'f': return Command::Flush;
1290 }
1291 break;
1292 case 7:
1293 return Command::Publish; // -V1037
1294 break;
1295 case 9:
1296 switch (data[0])
1297 {
1298 case 's': return Command::Subscribe;
1299 case 'h': return Command::Heartbeat;
1300 case 'g': return Command::GroupEnd;
1301 }
1302 break;
1303 case 10:
1304 switch (data[1])
1305 {
1306 case 'o': return Command::SOWDelete;
1307 case 't': return Command::StopTimer;
1308 }
1309 break;
1310 case 11:
1311 switch (data[0])
1312 {
1313 case 'g': return Command::GroupBegin;
1314 case 'u': return Command::Unsubscribe;
1315 }
1316 break;
1317 case 13:
1318 return Command::DeltaPublish;
1319 case 15:
1320 return Command::DeltaSubscribe;
1321 case 17:
1322 return Command::SOWAndSubscribe;
1323 case 23:
1324 return Command::SOWAndDeltaSubscribe;
1325 }
1326 return Command::Unknown;
1327 }
1328
1330 Message& setCommandEnum(Command::Type command_)
1331 {
1332 unsigned bits = 0;
1333 unsigned command = command_;
1334 while (command > 0)
1335 {
1336 ++bits;
1337 command >>= 1;
1338 }
1339 amps_message_assign_field_value(_body.get().getMessage(), AMPS_Command,
1340 CommandConstants<0>::Values[bits], CommandConstants<0>::Lengths[bits]);
1341 return *this;
1342 }
1343
1344 AMPS_FIELD(CommandId)
1345 AMPS_FIELD(ClientName)
1346 AMPS_FIELD(CorrelationId)
1347 AMPS_FIELD(Expiration)
1348 AMPS_FIELD(Filter)
1349 AMPS_FIELD(GroupSequenceNumber)
1350 AMPS_FIELD(Heartbeat)
1351 AMPS_FIELD(LeasePeriod)
1352 AMPS_FIELD(Matches)
1353 AMPS_FIELD(MessageLength)
1354 AMPS_FIELD(MessageType)
1355
1356 DOX_OPENGROUP(Options)
1357 DOX_MAKEGETCOMMENT(Options)
1358 Field getOptions() const
1359 {
1360 Field returnValue;
1361 const char* ptr;
1362 size_t sz;
1363 amps_message_get_field_value(_body.get().getMessage(),
1364 AMPS_Options, &ptr, &sz);
1365 if (sz && ptr[sz - 1] == ',')
1366 {
1367 --sz;
1368 }
1369 returnValue.assign(ptr, sz);
1370 return returnValue;
1371 }
1372
1373 DOX_MAKEGETRAWCOMMENT(Options)
1374 void getRawOptions(const char** dataptr, size_t* sizeptr) const
1375 {
1376 amps_message_get_field_value(_body.get().getMessage(),
1377 AMPS_Options, dataptr, sizeptr);
1378 if (*sizeptr && *dataptr && (*dataptr)[*sizeptr - 1] == ',')
1379 {
1380 --*sizeptr;
1381 }
1382 return;
1383 }
1384
1385 DOX_MAKESETCOMMENT(Options)
1386 Message& setOptions(const std::string& v)
1387 {
1388 size_t sz = v.length();
1389 if (sz && v[sz - 1] == ',')
1390 {
1391 --sz;
1392 }
1393 amps_message_set_field_value(_body.get().getMessage(),
1394 AMPS_Options, v.c_str(), sz);
1395 return *this;
1396 }
1397
1398 DOX_MAKEASSIGNCOMMENT(Options)
1399 Message& assignOptions(const std::string& v)
1400 {
1401 size_t sz = v.length();
1402 if (sz && v[sz - 1] == ',')
1403 {
1404 --sz;
1405 }
1406 amps_message_assign_field_value(_body.get().getMessage(),
1407 AMPS_Options, v.c_str(), sz);
1408 return *this;
1409 }
1410
1411 DOX_MAKEASSIGNCOMMENT(Options)
1412 Message& assignOptions(const char* data, size_t len)
1413 {
1414 if (len && data[len - 1] == ',')
1415 {
1416 --len;
1417 }
1418 amps_message_assign_field_value(_body.get().getMessage(),
1419 AMPS_Options, data, len);
1420 return *this;
1421 }
1422
1423 DOX_MAKESETCOMMENT(Options)
1424 Message& setOptions(const char* str)
1425 {
1426 if (str)
1427 {
1428 size_t sz = strlen(str);
1429 if (sz && str[sz - 1] == ',')
1430 {
1431 --sz;
1432 }
1433 amps_message_set_field_value(_body.get().getMessage(),
1434 AMPS_Options, str, sz);
1435 }
1436 else
1437 {
1438 amps_message_set_field_value(_body.get().getMessage(),
1439 AMPS_Options, str, 0);
1440 }
1441 return *this;
1442 }
1443
1444 DOX_MAKESETCOMMENT(Options)
1445 Message& setOptions(const char* str, size_t len)
1446 {
1447 if (len && str[len - 1] == ',')
1448 {
1449 --len;
1450 }
1451 amps_message_set_field_value(_body.get().getMessage(),
1452 AMPS_Options, str, len);
1453 return *this;
1454 }
1455 DOX_CLOSEGROUP()
1456
1457 AMPS_FIELD(OrderBy)
1458 AMPS_FIELD(Password)
1459 AMPS_FIELD_ALIAS(QueryId, QueryID)
1460 AMPS_FIELD(Reason)
1461 AMPS_FIELD(RecordsInserted)
1462 AMPS_FIELD(RecordsReturned)
1463 AMPS_FIELD(RecordsUpdated)
1464 AMPS_FIELD(Sequence)
1465 AMPS_FIELD(SowDelete)
1466 AMPS_FIELD(SowKey)
1467 AMPS_FIELD(SowKeys)
1468 AMPS_FIELD(Status)
1469 AMPS_FIELD_ALIAS(SubId, SubscriptionId) // -V524
1470 AMPS_FIELD(SubscriptionIds)
1471 AMPS_FIELD(TimeoutInterval)
1472 AMPS_FIELD(Timestamp)
1473
1477 Field getTransmissionTime() const
1478 {
1479 return getTimestamp();
1480 }
1481
1486 void getRawTransmissionTime(const char** dataptr, size_t* sizeptr) const
1487 {
1488 getRawTimestamp(dataptr, sizeptr);
1489 }
1490
1491 AMPS_FIELD(Topic)
1492 AMPS_FIELD(TopicMatches)
1493 AMPS_FIELD(TopNRecordsReturned)
1494 AMPS_FIELD(Version)
1495 AMPS_FIELD(UserId)
1496
1501
1502 Field getData() const
1503 {
1504 Field returnValue;
1505 char* ptr;
1506 size_t sz;
1507 amps_message_get_data(_body.get().getMessage(), &ptr, &sz);
1508 returnValue.assign(ptr, sz);
1509 return returnValue;
1510 }
1511
1512 void getRawData(const char** data, size_t* sz) const
1513 {
1514 amps_message_get_data(_body.get().getMessage(), (char**)data, sz);
1515 }
1518 Message& setData(const std::string& v_)
1519 {
1520 amps_message_set_data(_body.get().getMessage(), v_.c_str(), v_.length());
1521 return *this;
1522 }
1523 Message& assignData(const std::string& v_)
1524 {
1525 amps_message_assign_data(_body.get().getMessage(), v_.c_str(), v_.length());
1526 return *this;
1527 }
1528
1532 Message& setData(const char* data_, size_t length_)
1533 {
1534 amps_message_set_data(_body.get().getMessage(), data_, length_);
1535 return *this;
1536 }
1537 Message& assignData(const char* data_, size_t length_)
1538 {
1539 amps_message_assign_data(_body.get().getMessage(), data_, length_);
1540 return *this;
1541 }
1542
1545 Message& setData(const char* data_)
1546 {
1547 amps_message_set_data_nts(_body.get().getMessage(), data_);
1548 return *this;
1549 }
1550 Message& assignData(const char* data_)
1551 {
1552 amps_message_assign_data(_body.get().getMessage(), data_, strlen(data_));
1553 return *this;
1554 }
1555 amps_handle getMessage() const
1556 {
1557 return _body.get().getMessage();
1558 }
1559 void replace(amps_handle message, bool owner = false)
1560 {
1561 _body.get().replace(message, owner);
1562 }
1563 void disown()
1564 {
1565 _body.get().disown();
1566 }
1567 void invalidate()
1568 {
1569 _body = NULL;
1570 }
1571 bool isValid(void) const
1572 {
1573 return _body.isValid();
1574 }
1575 Message& reset()
1576 {
1577 _body.get().reset();
1578 return *this;
1579 }
1580
1581 void setBookmarkSeqNo(size_t val)
1582 {
1583 _body.get().setBookmarkSeqNo(val);
1584 }
1585
1586 size_t getBookmarkSeqNo() const
1587 {
1588 return _body.get().getBookmarkSeqNo();
1589 }
1590
1591 void setSubscriptionHandle(amps_handle val)
1592 {
1593 _body.get().setSubscriptionHandle(val);
1594 }
1595
1596 amps_handle getSubscriptionHandle() const
1597 {
1598 return _body.get().getSubscriptionHandle();
1599 }
1600
1601 void ack(const char* options_ = NULL) const;
1602
1603 void setClientImpl(ClientImpl* pClientImpl)
1604 {
1605 _body.get().setClientImpl(pClientImpl);
1606 }
1607
1608 void setIgnoreAutoAck() const
1609 {
1610 _body.get().setIgnoreAutoAck();
1611 }
1612
1613 bool getIgnoreAutoAck() const
1614 {
1615 return _body.get().getIgnoreAutoAck();
1616 }
1617
1618 // static
1619 template <class T>
1620 void throwFor(const T& /*context_*/, const std::string& ackReason_) const
1621 {
1622 switch (ackReason_[0])
1623 {
1624 case 'a': // auth failure
1625 throw AuthenticationException("Logon failed for user \"" +
1626 (std::string)getUserId() + "\"");
1627 break;
1628 case 'b':
1629 switch (ackReason_.length())
1630 {
1631 case 10: // bad filter
1632 throw BadFilterException("bad filter '" +
1633 (std::string)getFilter() +
1634 "'");
1635 break;
1636 case 11: // bad sow key
1637 if (getSowKeys().len())
1638 {
1639 throw BadSowKeyException("bad sow key '" +
1640 (std::string)getSowKeys() +
1641 "'");
1642 }
1643 else
1644 {
1645 throw BadSowKeyException("bad sow key '" +
1646 (std::string)getSowKey() +
1647 "'");
1648 }
1649 break;
1650 case 15: // bad regex topic
1651 throw BadRegexTopicException("bad regex topic '" +
1652 (std::string)getTopic() +
1653 "'.");
1654 break;
1655 default:
1656 break;
1657 }
1658 break;
1659 case 'd':
1660 if (ackReason_.length() == 23) // duplicate logon attempt
1661 {
1662 throw DuplicateLogonException("Client '" +
1663 (std::string)getClientName() +
1664 "' with userid '" +
1665 (std::string)getUserId() +
1666 "' duplicate logon attempt");
1667 }
1668 break;
1669 case 'i':
1670 if (ackReason_.length() >= 9)
1671 {
1672 switch (ackReason_[8])
1673 {
1674 case 'b': // invalid bookmark
1675 throw InvalidBookmarkException("invalid bookmark '" +
1676 (std::string)getBookmark() +
1677 "'.");
1678 break;
1679 case 'm': // invalid message type
1680 throw CommandException(std::string("invalid message type '") +
1681 (std::string)getMessageType() +
1682 "'.");
1683 break;
1684 case 'o':
1685 if (ackReason_[9] == 'p') // invalid options
1686 {
1687 throw InvalidOptionsException("invalid options '" +
1688 (std::string)getOptions() +
1689 "'.");
1690 }
1691 else if (ackReason_[9] == 'r') // invalid order by
1692 {
1693 throw InvalidOrderByException("invalid order by '" +
1694 (std::string)getOrderBy() +
1695 "'.");
1696 }
1697 break;
1698 case 's': // invalid subId
1699 throw InvalidSubIdException("invalid subid '" +
1700 (std::string)getSubscriptionId() +
1701 "'.");
1702 break;
1703 case 't':
1704 if (ackReason_.length() == 13) // invalid topic
1705 {
1706 throw InvalidTopicException("invalid topic '" +
1707 (std::string)getTopic() +
1708 "'.");
1709 }
1710 else if (ackReason_.length() == 23) // invalid topic or filter
1711 {
1712 throw InvalidTopicException("invalid topic or filter. Topic '" +
1713 (std::string)getTopic() +
1714 "' Filter '" +
1715 (std::string)getFilter() +
1716 "'.");
1717 }
1718 break;
1719 default:
1720 break;
1721 }
1722 }
1723 break;
1724 case 'l':
1725 if (ackReason_.length() == 14) // logon required
1726 {
1727 throw LogonRequiredException("logon required before command");
1728 }
1729 break;
1730 case 'n':
1731 switch (ackReason_[4])
1732 {
1733 case ' ': // name in use
1734 throw NameInUseException("name in use '" +
1735 (std::string)getClientName() +
1736 "'.");
1737 break;
1738 case 'e': // not entitled
1739 throw NotEntitledException("User \"" +
1740 (std::string)getUserId() +
1741 "\" not entitled to topic \"" +
1742 (std::string)getTopic() +
1743 "\".");
1744 break;
1745 case 'i': // no filter or bookmark
1746 throw MissingFieldsException("command sent with no filter or bookmark.");
1747 break;
1748 case 'l': // no client name
1749 throw MissingFieldsException("command sent with no client name.");
1750 break;
1751 case 'o': // no topic or filter
1752 throw MissingFieldsException("command sent with no topic or filter.");
1753 break;
1754 case 's': // not supported
1755 throw CommandException("operation on topic '" +
1756 (std::string)getTopic() +
1757 "' with options '" +
1758 (std::string)getOptions() +
1759 "' not supported.");
1760 break;
1761 default:
1762 break;
1763 }
1764 break;
1765 case 'o':
1766 switch (ackReason_.length())
1767 {
1768 case 16: // orderby required
1769 throw MissingFieldsException("orderby required");
1770 break;
1771 case 17: // orderby too large
1772 throw CommandException("orderby too large '" +
1773 (std::string)getOrderBy() +
1774 "'.");
1775 break;
1776 }
1777 break;
1778 case 'p':
1779 throw CommandException("projection clause too large in options '" +
1780 (std::string)getOptions() +
1781 "'.");
1782 break;
1783 case 'r':
1784 switch (ackReason_[2])
1785 {
1786 case 'g': // regex topic not supported
1787 throw BadRegexTopicException("'regex topic not supported '" +
1788 (std::string)getTopic() +
1789 "'.");
1790 break;
1791 default:
1792 break;
1793 }
1794 break;
1795 case 's':
1796 switch (ackReason_[5])
1797 {
1798 case ' ': // subid in use
1799 throw SubidInUseException("subid in use '" +
1800 (std::string)getSubscriptionId() +
1801 "'.");
1802 break;
1803 case 'e': // sow_delete command only supports one of: filter, sow_keys, bookmark, or data
1804 throw CommandException("sow_delete command only supports one of: filter '" +
1805 (std::string)getFilter() +
1806 "', sow_keys '" +
1807 (std::string)getSowKeys() +
1808 "', bookmark '" +
1809 (std::string)getBookmark() +
1810 "', or data '" +
1811 (std::string)getData() +
1812 "'.");
1813 break;
1814 case 't': // sow store failed
1815 throw PublishException("sow store failed.");
1816 break;
1817 default:
1818 break;
1819 }
1820 break;
1821 case 't':
1822 switch (ackReason_[2])
1823 {
1824 case ' ': // tx store failure
1825 throw PublishException("tx store failure.");
1826 break;
1827 case 'n': // txn replay failed
1828 throw ReplayFailedException("txn replay failed for '"
1829 + (std::string)getSubId()
1830 + "'.");
1831 break;
1832 }
1833 break;
1834 default:
1835 break;
1836 }
1837 throw CommandException("Error from server while processing this command: '" +
1838 ackReason_ + "'");
1839 }
1840 };
1841
1842 inline std::string
1843 operator+(const std::string& lhs, const Message::Field& rhs)
1844 {
1845 return lhs + std::string(rhs);
1846 }
1847
1848 inline std::basic_ostream<char>&
1849 operator<<(std::basic_ostream<char>& os, const Message::Field& rhs)
1850 {
1851 os.write(rhs.data(), (std::streamsize)rhs.len());
1852 return os;
1853 }
1854 inline bool
1855 AMPS::Field::operator<(const AMPS::Field& rhs) const
1856 {
1857 if (!data())
1858 {
1859 return rhs.data() != NULL; // -V547
1860 }
1861 if (!rhs.data()) // -V547
1862 {
1863 return false;
1864 }
1865 return std::lexicographical_compare(data(), data() + len(), rhs.data(), rhs.data() + rhs.len());
1866 }
1867
1868 }
1869
1870 #endif
Field.hpp
Defines the AMPS::Field class, which represents the value of a field in a message.
amps_message_create
AMPSDLL amps_handle amps_message_create(amps_handle client)
Functions for creation and manipulation of AMPS messages.
amps_message_set_data
AMPSDLL void amps_message_set_data(amps_handle message, const amps_char *value, size_t length)
Sets the data component of an AMPS message.
amps_message_get_field_value
AMPSDLL 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.
amps_message_assign_field_value
AMPSDLL 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.
amps_message_set_data_nts
AMPSDLL void amps_message_set_data_nts(amps_handle message, const amps_char *value)
Sets the data component of an AMPS message.
amps_message_reset
AMPSDLL void amps_message_reset(amps_handle message)
Clears all fields and data in a message.
amps_message_destroy
AMPSDLL void amps_message_destroy(amps_handle message)
Destroys and frees the memory associated with an AMPS message object.
amps_message_assign_data
AMPSDLL 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_message_copy
AMPSDLL 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...
amps_message_set_field_value
AMPSDLL 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.
amps_handle
void * amps_handle
Opaque handle type used to refer to objects in the AMPS api.
Definition: amps.h:211
amps_message_get_data
AMPSDLL 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::Field
Field represents the value of a single field in a Message.
Definition: Field.hpp:88
AMPS::Field::len
size_t len() const
Returns the length of the data underlying this field.
Definition: Field.hpp:286
AMPS::Field::data
const char * data() const
Returns the (non-null-terminated) data underlying this field.
Definition: Field.hpp:279
AMPS::MessageImpl
Implementation class for a Message.
Definition: Message.hpp:74
AMPS::MessageImpl::replace
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:176
AMPS::MessageImpl::MessageImpl
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:95
AMPS::MessageImpl::getMessage
amps_handle getMessage() const
Returns the underling AMPS message object from the C layer.
Definition: Message.hpp:156
AMPS::MessageImpl::MessageImpl
MessageImpl()
Constructs a MessageImpl with a new, empty AMPS message.
Definition: Message.hpp:108
AMPS::Message::Options
Class for constructing the options string to pass to AMPS in a Message.
Definition: Message.hpp:590
AMPS::Message::Options::setBookmarkNotFoundEpoch
void setBookmarkNotFoundEpoch()
Set the option for the action to take if the requested bookmark to start the subscription is not foun...
Definition: Message.hpp:1113
AMPS::Message::Options::getLength
size_t getLength() const
Return the length of this Options object as a string.
Definition: Message.hpp:1136
AMPS::Message::Options::setConflationKey
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:965
AMPS::Message::Options::setReplace
void setReplace(void)
Set the option to replace a current subscription with this one.
Definition: Message.hpp:842
AMPS::Message::Options::setNone
void setNone(void)
Clear any previously set options and set the options to an empty string (AMPS_OPTIONS_NONE).
Definition: Message.hpp:810
AMPS::Message::Options::setResume
void setResume(void)
Set the option to resume a subscription.
Definition: Message.hpp:894
AMPS::Message::Options::setProjection
void setProjection(const std::string &projection_)
Set the option for projecting the results of an aggregated query or subscription.
Definition: Message.hpp:1039
AMPS::Message::Options::setGrouping
void setGrouping(const std::string &grouping_)
Set the option for grouping the results of an aggregated query or subscription.
Definition: Message.hpp:1068
AMPS::Message::Options::setNoEmpties
void setNoEmpties(void)
Set the option to not send empty messages on a delta subscription.
Definition: Message.hpp:850
AMPS::Message::Options::setNoSowKey
void setNoSowKey(void)
Set the option to not set the SowKey header on messages.
Definition: Message.hpp:875
AMPS::Message::Options::setFullyDurable
void setFullyDurable(void)
Set the option to only provide messages that have been persisted to all replication destinations that...
Definition: Message.hpp:920
AMPS::Message::Options::setConflation
void setConflation(const char *conflation_)
Set the options for conflation on a subscription.
Definition: Message.hpp:948
AMPS::Message::Options::setBookmarkNotFoundFail
void setBookmarkNotFoundFail()
Set the option for the action to take if the requested bookmark to start the subscription is not foun...
Definition: Message.hpp:1121
AMPS::Message::Options::setTopN
void setTopN(int topN_)
Set the top N option, which specifies the maximum number of messages to return for this command.
Definition: Message.hpp:978
AMPS::Message::Options::setSendKeys
void setSendKeys(void)
Set the option to send key fields with a delta subscription.
Definition: Message.hpp:858
AMPS::Message::Options::setTimestamp
void setTimestamp(void)
Set the option to send a timestamp that the message was processed on a subscription or query.
Definition: Message.hpp:867
AMPS::Message::Options::setProjection
void setProjection(Iterator begin_, Iterator end_)
Set the option for projecting the results of an aggregated query or subscription.
Definition: Message.hpp:1052
AMPS::Message::Options::setMaxBacklog
void setMaxBacklog(int maxBacklog_)
Set the option for maximum backlog this subscription is willing to accept.
Definition: Message.hpp:935
AMPS::Message::Options::setBookmarkNotFound
void setBookmarkNotFound(const char *action_)
Set the option for the action to take if the requested bookmark to start the subscription is ot found...
Definition: Message.hpp:1097
AMPS::Message::Options::setSkipN
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:1027
AMPS::Message::Options::getStr
const char * getStr() const
Return this Options object as a non-NULL-terminated string.
Definition: Message.hpp:1145
AMPS::Message::Options::setGrouping
void setGrouping(Iterator begin_, Iterator end_)
Set the option for grouping the results of an aggregated query or subscription.
Definition: Message.hpp:1081
AMPS::Message::Options::setRate
void setRate(const char *rate_)
Set the option for the maximum rate at which messages are provided to the subscription.
Definition: Message.hpp:992
AMPS::Message::Options::setOOF
void setOOF(void)
Set the option to receive out of focus (OOF) messages on a subscription, where applicable.
Definition: Message.hpp:833
AMPS::Message::Options::setPause
void setPause(void)
Set the option to pause a bookmark subscription.
Definition: Message.hpp:909
AMPS::Message::Options::Options
Options(std::string options_="")
ctor - default to None
Definition: Message.hpp:775
AMPS::Message::Options::setBookmarkNotFoundNow
void setBookmarkNotFoundNow()
Set the option for the action to take if the requested bookmark to start the subscription is not foun...
Definition: Message.hpp:1105
AMPS::Message::Options::setCancel
void setCancel(void)
Set the cancel option, used on a sow_delete command to return a message to the queue.
Definition: Message.hpp:883
AMPS::Message::Options::setRateMaxGap
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:1014
AMPS::Message::Options::setLive
void setLive(void)
Set the live option for a bookmark subscription, which requests that the subscription receives messag...
Definition: Message.hpp:824
AMPS::Message
Message encapsulates a single message sent to or received from an AMPS server, and provides methods f...
Definition: Message.hpp:520
AMPS::Message::getUserId
Field getUserId() const
Retrieves the value of the UserId header of the Message as a Field which references the underlying bu...
Definition: Message.hpp:1495
AMPS::Message::getRawTimestamp
void getRawTimestamp(const char **dataptr, size_t *sizeptr) const
Modifies the passed in arguments to reference the value of the Timestamp header of self in the underl...
Definition: Message.hpp:1472
AMPS::Message::getBookmark
Field getBookmark() const
Retrieves the value of the Bookmark header of the Message as a Field which references the underlying ...
Definition: Message.hpp:1236
AMPS::Message::deepCopy
void deepCopy(const Message &rhs_)
Makes self a deep copy of rhs_.
Definition: Message.hpp:574
AMPS::Message::IdentifierLength
static const unsigned int IdentifierLength
The length of identifiers used for unique identification of commands and subscriptions.
Definition: Message.hpp:530
AMPS::Message::getOrderBy
Field getOrderBy() const
Retrieves the value of the OrderBy header of the Message as a Field which references the underlying b...
Definition: Message.hpp:1457
AMPS::Message::Message
Message(amps_handle message_, bool owner_=false)
Constructs a new Message to wrap message.
Definition: Message.hpp:553
AMPS::Message::getData
Field getData() const
Returns the data from this message.
Definition: Message.hpp:1502
AMPS::Message::deepCopy
Message deepCopy(void) const
Returns a deep copy of self.
Definition: Message.hpp:567
AMPS::Message::CtorFlag
CtorFlag
A flag to indicate not to create a body.
Definition: Message.hpp:539
AMPS::Message::BOOKMARK_NONE
static const size_t BOOKMARK_NONE
An indicator of no bookmark value.
Definition: Message.hpp:534
AMPS::Message::getRawTransmissionTime
void getRawTransmissionTime(const char **dataptr, size_t *sizeptr) const
Definition: Message.hpp:1486
AMPS::Message::assignOptions
Message & assignOptions(const std::string &v)
Assigns the value of the Options header for this Message without copying.
Definition: Message.hpp:1399
AMPS::Message::getMessageType
Field getMessageType() const
Retrieves the value of the MessageType header of the Message as a Field which references the underlyi...
Definition: Message.hpp:1354
AMPS::Message::Message
Message(CtorFlag)
Constructs a new empty, invalid Message.
Definition: Message.hpp:543
AMPS::Message::getSowKeys
Field getSowKeys() const
Retrieves the value of the SowKeys header of the Message as a Field which references the underlying b...
Definition: Message.hpp:1467
AMPS::Message::getTimestamp
Field getTimestamp() const
Retrieves the value of the Timestamp header of the Message as a Field which references the underlying...
Definition: Message.hpp:1472
AMPS::Message::getSubscriptionId
Field getSubscriptionId() const
Retrieves the value of the SubscriptionId header of the Message as a Field which references the under...
Definition: Message.hpp:1469
AMPS::Message::getFilter
Field getFilter() const
Retrieves the value of the Filter header of the Message as a Field which references the underlying bu...
Definition: Message.hpp:1348
AMPS::Message::getRawOptions
void getRawOptions(const char **dataptr, size_t *sizeptr) const
Modifies the passed in arguments to reference the value of the Options header of self in the underlyi...
Definition: Message.hpp:1374
AMPS::Message::decodeSingleAckType
static AckType::Type decodeSingleAckType(const char *begin, const char *end)
Decodes a single ack string.
Definition: Message.hpp:1175
AMPS::Message::setAckTypeEnum
Message & setAckTypeEnum(unsigned ackType_)
Encode self's "ack type" field from a bitmask of values from AckType.
Definition: Message.hpp:1225
AMPS::Message::getAckTypeEnum
unsigned getAckTypeEnum() const
Decode self's "ack type" field and return the corresponding bitmask of values from AckType.
Definition: Message.hpp:1202
AMPS::Message::setOptions
Message & setOptions(const std::string &v)
Sets the value of the Options header for this Message.
Definition: Message.hpp:1386
AMPS::Message::getCommandEnum
Command::Type getCommandEnum() const
Decode self's "command" field and return one of the values from Command.
Definition: Message.hpp:1270
AMPS::Message::getSowKey
Field getSowKey() const
Retrieves the value of the SowKey header of the Message as a Field which references the underlying bu...
Definition: Message.hpp:1466
AMPS::Message::Message
Message()
Construct a new, empty Message.
Definition: Message.hpp:561
AMPS::Message::setCommandEnum
Message & setCommandEnum(Command::Type command_)
Set self's "command" field from one of the values in Command.
Definition: Message.hpp:1330
AMPS::Message::getClientName
Field getClientName() const
Retrieves the value of the ClientName header of the Message as a Field which references the underlyin...
Definition: Message.hpp:1345
AMPS::Message::getTransmissionTime
Field getTransmissionTime() const
Definition: Message.hpp:1477
AMPS::Message::getSubId
Field getSubId() const
Retrieves the value of the SubscriptionId header of the Message as a Field which references the under...
Definition: Message.hpp:1469
AMPS::Message::getTopic
Field getTopic() const
Retrieves the value of the Topic header of the Message as a Field which references the underlying buf...
Definition: Message.hpp:1491
AMPS::Message::setData
Message & setData(const char *data_, size_t length_)
Sets the data portion of self from a char array.
Definition: Message.hpp:1532
AMPS::Message::setData
Message & setData(const char *data_)
Sets the data portion of self from a null-terminated string.
Definition: Message.hpp:1545
AMPS::Message::getOptions
Field getOptions() const
Retrieves the value of the Options header of the Message as a Field which references the underlying b...
Definition: Message.hpp:1358
AMPS::Message::setData
Message & setData(const std::string &v_)
Sets the data portion of self.
Definition: Message.hpp:1518
AMPS::Reason
Class to hold string versions of failure reasons.
Definition: ampsplusplus.hpp:152
AMPS::Message::AckType
Valid values for the setAckTypeEnum() and getAckTypeEnum() methods.
Definition: Message.hpp:1166
AMPS::Message::Command
Valid values for setCommandEnum() and getCommandEnum().
Definition: Message.hpp:1243