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

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