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

ampsplusplus.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 _AMPSPLUSPLUS_H_
26 #define _AMPSPLUSPLUS_H_
27 #include "amps/amps.h"
28 #include "amps/ampsver.h"
29 #include <string>
30 #include <map>
31 #include <sstream>
32 #include <iostream>
33 #include <memory>
34 #include <stdexcept>
35 #include <limits.h>
36 #include <list>
37 #include <memory>
38 #include <regex>
39 #include <set>
40 #include <deque>
41 #include <vector>
42 #include <assert.h>
43 #ifndef _WIN32
44 #include <inttypes.h>
45 #endif
46 #if defined(sun)
47 #include <sys/atomic.h>
48 #endif
49 #include "amps/BookmarkStore.hpp"
50 #include "amps/MessageRouter.hpp"
51 #include "amps/util.hpp"
52 #include "amps/ampscrc.hpp"
53 #if __cplusplus >= 201100L || _MSC_VER >= 1900
54 #include <atomic>
55 #endif
56
57 #ifndef AMPS_TESTING_SLOW_MESSAGE_STREAM
58 #define AMPS_TESTING_SLOW_MESSAGE_STREAM
59 #endif
60
65
66
73
84
85 // For StoreBuffer implementations
86 #define AMPS_MEMORYBUFFER_DEFAULT_BUFFERS 10
87 #define AMPS_MEMORYBUFFER_DEFAULT_LENGTH 40960
88 #define AMPS_SUBSCRIPTION_MANAGER_DEFAULT_TIMEOUT 0
89 #define AMPS_HACLIENT_TIMEOUT_DEFAULT 10000
90 #define AMPS_HACLIENT_RECONNECT_DEFAULT 200
91 #define AMPS_DEFAULT_COMMAND_TIMEOUT 5000
92 #define AMPS_DEFAULT_TOP_N -1
93 #define AMPS_DEFAULT_BATCH_SIZE 10
94 #define AMPS_NUMBER_BUFFER_LEN 20
95 #define AMPS_DEFAULT_QUEUE_ACK_TIMEOUT 1000
96
97 #if defined(_M_X64) || defined(__x86_64) || defined(_WIN64)
98 #define AMPS_X64 1
99 #endif
100
101 static thread_local AMPS::Message threadLocalMessage;
102
103 namespace AMPS
104 {
105
106 typedef std::map<std::string, std::string> ConnectionInfo;
107
108 template<class Type>
109 inline std::string asString(Type x_)
110 {
111 std::ostringstream os;
112 os << x_;
113 return os.str();
114 }
115
116 inline
117 size_t convertToCharArray(char* buf_, amps_uint64_t seqNo_)
118 {
119 size_t pos = AMPS_NUMBER_BUFFER_LEN;
120 for (int i = 0; i < AMPS_NUMBER_BUFFER_LEN; ++i)
121 {
122 if (seqNo_ > 0)
123 {
124 buf_[--pos] = (char)(seqNo_ % 10 + '0');
125 seqNo_ /= 10;
126 }
127 }
128 return pos;
129 }
130
131 #ifdef _WIN32
132 inline
133 size_t convertToCharArray(char* buf_, unsigned long seqNo_)
134 {
135 size_t pos = AMPS_NUMBER_BUFFER_LEN;
136 for (int i = 0; i < AMPS_NUMBER_BUFFER_LEN; ++i)
137 {
138 if (seqNo_ > 0)
139 {
140 buf_[--pos] = (char)(seqNo_ % 10 + '0');
141 seqNo_ /= 10;
142 }
143 }
144 return pos;
145 }
146 #endif
147
151 class Reason
152 {
153 public:
154 static const char* duplicate()
155 {
156 return "duplicate";
157 }
158 static const char* badFilter()
159 {
160 return "bad filter";
161 }
162 static const char* badRegexTopic()
163 {
164 return "bad regex topic";
165 }
166 static const char* subscriptionAlreadyExists()
167 {
168 return "subscription already exists";
169 }
170 static const char* nameInUse()
171 {
172 return "name in use";
173 }
174 static const char* authFailure()
175 {
176 return "auth failure";
177 }
178 static const char* notEntitled()
179 {
180 return "not entitled";
181 }
182 static const char* authDisabled()
183 {
184 return "authentication disabled";
185 }
186 static const char* subidInUse()
187 {
188 return "subid in use";
189 }
190 static const char* noTopic()
191 {
192 return "no topic";
193 }
194 };
195
204 class ExceptionListener
205 {
206 public:
207 virtual ~ExceptionListener() {;}
208 virtual void exceptionThrown(const std::exception&) const {;}
209 };
210
211 typedef ExceptionListener DefaultExceptionListener;
212
213
214 #define AMPS_CALL_EXCEPTION_WRAPPER(x) \
215 try\
216 {\
217 x;\
218 }\
219 catch (std::exception& stdEx_)\
220 {\
221 try\
222 {\
223 _exceptionListener->exceptionThrown(stdEx_);\
224 }\
225 catch(...)\
226 {\
227 ;\
228 }\
229 }
230 /*
231 * Note : we don't attempt to trap non std::exception exceptions
232 * here because doing so interferes with pthread_exit on some OSes.
233 catch (...)\
234 {\
235 try\
236 {\
237 _exceptionListener->exceptionThrown(AMPS::AMPSException(\
238 "An unhandled exception of unknown type was thrown by "\
239 "the registered handler.", AMPS_E_USAGE));\
240 }\
241 catch(...)\
242 {\
243 ;\
244 }\
245 }*/
246 #ifdef _WIN32
247 #define AMPS_CALL_EXCEPTION_WRAPPER_2(me,x) \
248 try\
249 {\
250 while(me->_connected)\
251 {\
252 try\
253 {\
254 x;\
255 break;\
256 }\
257 catch(MessageStreamFullException&)\
258 {\
259 try\
260 {\
261 me->checkAndSendHeartbeat(false);\
262 }\
263 catch (std::exception& stdEx_)\
264 {\
265 try\
266 {\
267 me->_exceptionListener->exceptionThrown(stdEx_);\
268 }\
269 catch(...)\
270 {\
271 ;\
272 }\
273 break;\
274 }\
275 }\
276 }\
277 }\
278 catch (std::exception& stdEx_)\
279 {\
280 try\
281 {\
282 me->_exceptionListener->exceptionThrown(stdEx_);\
283 }\
284 catch(...)\
285 {\
286 ;\
287 }\
288 }
289 /*
290 * Note : we don't attempt to trap non std::exception exceptions
291 * here because doing so interferes with pthread_exit on some OSes.
292 catch (...)\
293 {\
294 try\
295 {\
296 me->_exceptionListener->exceptionThrown(AMPS::AMPSException(\
297 "An unhandled exception of unknown type was thrown by "\
298 "the registered handler.", AMPS_E_USAGE));\
299 }\
300 catch(...)\
301 {\
302 ;\
303 }\
304 }*/
305
306 #define AMPS_CALL_EXCEPTION_WRAPPER_STREAM_FULL_2(me, x)\
307 while(me->_connected)\
308 {\
309 try\
310 {\
311 x;\
312 break;\
313 }\
314 catch(MessageStreamFullException&)\
315 {\
316 try\
317 {\
318 me->checkAndSendHeartbeat(false);\
319 }\
320 catch (std::exception& stdEx_)\
321 {\
322 try\
323 {\
324 me->_exceptionListener->exceptionThrown(stdEx_);\
325 }\
326 catch(...)\
327 {\
328 ;\
329 }\
330 break;\
331 }\
332 }\
333 }
334 #else
335 #define AMPS_CALL_EXCEPTION_WRAPPER_2(me,x) \
336 try\
337 {\
338 while(me->_connected)\
339 {\
340 try\
341 {\
342 x;\
343 break;\
344 }\
345 catch(MessageStreamFullException& msfEx_)\
346 {\
347 try\
348 {\
349 me->checkAndSendHeartbeat(false);\
350 }\
351 catch (std::exception& stdEx_)\
352 {\
353 try\
354 {\
355 me->_exceptionListener->exceptionThrown(stdEx_);\
356 }\
357 catch(...)\
358 {\
359 ;\
360 }\
361 break;\
362 }\
363 }\
364 }\
365 }\
366 catch (std::exception& stdEx_)\
367 {\
368 try\
369 {\
370 me->_exceptionListener->exceptionThrown(stdEx_);\
371 }\
372 catch(...)\
373 {\
374 ;\
375 }\
376 }
377 /*
378 * Note : we don't attempt to trap non std::exception exceptions
379 * here because doing so interferes with pthread_exit on some OSes.
380 catch (...)\
381 {\
382 try\
383 {\
384 me->_exceptionListener->exceptionThrown(AMPS::AMPSException(\
385 "An unhandled exception of unknown type was thrown by "\
386 "the registered handler.", AMPS_E_USAGE));\
387 }\
388 catch(...)\
389 {\
390 ;\
391 }\
392 }*/
393
394 #define AMPS_CALL_EXCEPTION_WRAPPER_STREAM_FULL_2(me, x)\
395 while(me->_connected)\
396 {\
397 try\
398 {\
399 x;\
400 break;\
401 }\
402 catch(MessageStreamFullException& msfEx_)\
403 {\
404 try\
405 {\
406 me->checkAndSendHeartbeat(false);\
407 }\
408 catch (std::exception& stdEx_)\
409 {\
410 try\
411 {\
412 me->_exceptionListener->exceptionThrown(stdEx_);\
413 }\
414 catch(...)\
415 {\
416 ;\
417 }\
418 break;\
419 }\
420 }\
421 }
422 #endif
423
424 #define AMPS_UNHANDLED_EXCEPTION(ex) \
425 try\
426 {\
427 _exceptionListener->exceptionThrown(ex);\
428 }\
429 catch(...)\
430 {\
431 ;\
432 }
433
434 #define AMPS_UNHANDLED_EXCEPTION_2(me,ex) \
435 try\
436 {\
437 me->_exceptionListener->exceptionThrown(ex);\
438 }\
439 catch(...)\
440 {\
441 ;\
442 }
443
444
445 class Client;
446
471
472 class Command
473 {
474 Message _message;
475 unsigned _timeout;
476 unsigned _batchSize;
477 unsigned _flags;
478 static const unsigned Subscribe = 1;
479 static const unsigned SOW = 2;
480 static const unsigned NeedsSequenceNumber = 4;
481 static const unsigned ProcessedAck = 8;
482 static const unsigned StatsAck = 16;
483 void init(Message::Command::Type command_)
484 {
485 _timeout = 0;
486 _batchSize = 0;
487 _flags = 0;
488 _message.reset();
489 _message.setCommandEnum(command_);
490 _setIds();
491 }
492 void init(const std::string& command_)
493 {
494 _timeout = 0;
495 _batchSize = 0;
496 _flags = 0;
497 _message.reset();
498 _message.setCommand(command_);
499 _setIds();
500 }
501 void init(const char* command_, size_t commandLen_)
502 {
503 _timeout = 0;
504 _batchSize = 0;
505 _flags = 0;
506 _message.reset();
507 _message.setCommand(command_, commandLen_);
508 _setIds();
509 }
510 void _setIds(void)
511 {
512 Message::Command::Type command = _message.getCommandEnum();
513 if (!(command & Message::Command::NoDataCommands))
514 {
515 _message.newCommandId();
516 if (command == Message::Command::Subscribe ||
517 command == Message::Command::SOWAndSubscribe ||
518 command == Message::Command::DeltaSubscribe ||
519 command == Message::Command::SOWAndDeltaSubscribe)
520 {
521 _message.setSubscriptionId(_message.getCommandId());
522 _flags |= Subscribe;
523 }
524 if (command == Message::Command::SOW
525 || command == Message::Command::SOWAndSubscribe
526 || command == Message::Command::SOWAndDeltaSubscribe)
527 {
528 _message.setQueryID(_message.getCommandId());
529 if (_batchSize == 0)
530 {
531 setBatchSize(AMPS_DEFAULT_BATCH_SIZE);
532 }
533 if (command == Message::Command::SOW)
534 {
535 _flags |= SOW;
536 }
537 }
538 _flags |= ProcessedAck;
539 }
540 else if (command == Message::Command::SOWDelete)
541 {
542 _message.newCommandId();
543 _flags |= ProcessedAck;
544 _flags |= NeedsSequenceNumber;
545 }
546 else if (command == Message::Command::Publish
547 || command == Message::Command::DeltaPublish)
548 {
549 _flags |= NeedsSequenceNumber;
550 }
551 else if (command == Message::Command::StopTimer)
552 {
553 _message.newCommandId();
554 }
555 }
556 public:
560 Command(const std::string& command_)
561 {
562 init(command_);
563 }
568 Command(const char* command_, size_t commandLen_)
569 {
570 init(command_, commandLen_);
571 }
575 Command(Message::Command::Type command_)
576 {
577 init(command_);
578 }
579
583 Command& reset(const std::string& command_)
584 {
585 init(command_);
586 return *this;
587 }
592 Command& reset(const char* command_, size_t commandLen_)
593 {
594 init(command_, commandLen_);
595 return *this;
596 }
600 Command& reset(Message::Command::Type command_)
601 {
602 init(command_);
603 return *this;
604 }
612 Command& setSowKey(const std::string& sowKey_)
613 {
614 _message.setSowKey(sowKey_);
615 return *this;
616 }
625 Command& setSowKey(const char* sowKey_, size_t sowKeyLen_)
626 {
627 _message.setSowKey(sowKey_, sowKeyLen_);
628 return *this;
629 }
642 Command& setSowKeys(const std::string& sowKeys_)
643 {
644 _message.setSowKeys(sowKeys_);
645 return *this;
646 }
660 Command& setSowKeys(const char* sowKeys_, size_t sowKeysLen_)
661 {
662 _message.setSowKeys(sowKeys_, sowKeysLen_);
663 return *this;
664 }
666 Command& setCommandId(const std::string& cmdId_)
667 {
668 _message.setCommandId(cmdId_);
669 return *this;
670 }
673 Command& setCommandId(const char* cmdId_, size_t cmdIdLen_)
674 {
675 _message.setCommandId(cmdId_, cmdIdLen_);
676 return *this;
677 }
679 Command& setTopic(const std::string& topic_)
680 {
681 _message.setTopic(topic_);
682 return *this;
683 }
686 Command& setTopic(const char* topic_, size_t topicLen_)
687 {
688 _message.setTopic(topic_, topicLen_);
689 return *this;
690 }
692 Command& setFilter(const std::string& filter_)
693 {
694 _message.setFilter(filter_);
695 return *this;
696 }
699 Command& setFilter(const char* filter_, size_t filterLen_)
700 {
701 _message.setFilter(filter_, filterLen_);
702 return *this;
703 }
705 Command& setOrderBy(const std::string& orderBy_)
706 {
707 _message.setOrderBy(orderBy_);
708 return *this;
709 }
712 Command& setOrderBy(const char* orderBy_, size_t orderByLen_)
713 {
714 _message.setOrderBy(orderBy_, orderByLen_);
715 return *this;
716 }
718 Command& setSubId(const std::string& subId_)
719 {
720 _message.setSubscriptionId(subId_);
721 return *this;
722 }
725 Command& setSubId(const char* subId_, size_t subIdLen_)
726 {
727 _message.setSubscriptionId(subId_, subIdLen_);
728 return *this;
729 }
731 Command& setQueryId(const std::string& queryId_)
732 {
733 _message.setQueryId(queryId_);
734 return *this;
735 }
738 Command& setQueryId(const char* queryId_, size_t queryIdLen_)
739 {
740 _message.setQueryId(queryId_, queryIdLen_);
741 return *this;
742 }
748 Command& setBookmark(const std::string& bookmark_)
749 {
750 _message.setBookmark(bookmark_);
751 return *this;
752 }
759 Command& setBookmark(const char* bookmark_, size_t bookmarkLen_)
760 {
761 _message.setBookmark(bookmark_, bookmarkLen_);
762 return *this;
763 }
770 Command& setCorrelationId(const std::string& correlationId_)
771 {
772 _message.setCorrelationId(correlationId_);
773 return *this;
774 }
782 Command& setCorrelationId(const char* correlationId_, size_t correlationIdLen_)
783 {
784 _message.setCorrelationId(correlationId_, correlationIdLen_);
785 return *this;
786 }
789 Command& setOptions(const std::string& options_)
790 {
791 _message.setOptions(options_);
792 return *this;
793 }
797 Command& setOptions(const char* options_, size_t optionsLen_)
798 {
799 _message.setOptions(options_, optionsLen_);
800 return *this;
801 }
803 Command& setSequence(const std::string& seq_)
804 {
805 _message.setSequence(seq_);
806 return *this;
807 }
810 Command& setSequence(const char* seq_, size_t seqLen_)
811 {
812 _message.setSequence(seq_, seqLen_);
813 return *this;
814 }
816 Command& setSequence(const amps_uint64_t seq_)
817 {
818 std::ostringstream os;
819 os << seq_;
820 _message.setSequence(os.str());
821 return *this;
822 }
823 amps_uint64_t getSequence() const
824 {
825 return amps_message_get_field_uint64(_message.getMessage(), AMPS_Sequence);
826 }
829 Command& setData(const std::string& data_)
830 {
831 _message.setData(data_);
832 return *this;
833 }
837 Command& setData(const char* data_, size_t dataLen_)
838 {
839 _message.setData(data_, dataLen_);
840 return *this;
841 }
851 Command& setTimeout(unsigned timeout_)
852 {
853 _timeout = timeout_;
854 return *this;
855 }
857 Command& setTopN(unsigned topN_)
858 {
859 if (topN_ != (unsigned)AMPS_DEFAULT_TOP_N)
860 {
861 _message.setTopNRecordsReturned(topN_);
862 }
863 else
864 {
865 _message.setTopNRecordsReturned(nullptr, 0);
866 }
867 return *this;
868 }
873 Command& setBatchSize(unsigned batchSize_)
874 {
875 _message.setBatchSize(batchSize_);
876 _batchSize = batchSize_;
877 return *this;
878 }
889 Command& setExpiration(unsigned expiration_)
890 {
891 _message.setExpiration(expiration_);
892 return *this;
893 }
895 Command& addAckType(const std::string& ackType_)
896 {
897 _message.setAckType(_message.getAckType() + "," + ackType_);
898 if (ackType_ == "processed")
899 {
900 _flags |= ProcessedAck;
901 }
902 else if (ackType_ == "stats")
903 {
904 _flags |= StatsAck;
905 }
906 return *this;
907 }
909 Command& setAckType(const std::string& ackType_)
910 {
911 _message.setAckType(ackType_);
912 if (ackType_.find("processed") != std::string::npos)
913 {
914 _flags |= ProcessedAck;
915 }
916 else
917 {
918 _flags &= ~ProcessedAck;
919 }
920 if (ackType_.find("stats") != std::string::npos)
921 {
922 _flags |= StatsAck;
923 }
924 else
925 {
926 _flags &= ~StatsAck;
927 }
928 return *this;
929 }
931 Command& setAckType(unsigned ackType_)
932 {
933 _message.setAckTypeEnum(ackType_);
934 if (ackType_ & Message::AckType::Processed)
935 {
936 _flags |= ProcessedAck;
937 }
938 else
939 {
940 _flags &= ~ProcessedAck;
941 }
942 if (ackType_ & Message::AckType::Stats)
943 {
944 _flags |= StatsAck;
945 }
946 else
947 {
948 _flags &= ~StatsAck;
949 }
950 return *this;
951 }
953 std::string getAckType() const
954 {
955 return (std::string)(_message.getAckType());
956 }
958 unsigned getAckTypeEnum() const
959 {
960 return _message.getAckTypeEnum();
961 }
962
963 Message& getMessage(void)
964 {
965 return _message;
966 }
967 unsigned getTimeout(void) const
968 {
969 return _timeout;
970 }
971 unsigned getBatchSize(void) const
972 {
973 return _batchSize;
974 }
975 bool isSubscribe(void) const
976 {
977 return _flags & Subscribe;
978 }
979 bool isSow(void) const
980 {
981 return (_flags & SOW) != 0;
982 }
983 bool hasProcessedAck(void) const
984 {
985 return (_flags & ProcessedAck) != 0;
986 }
987 bool hasStatsAck(void) const
988 {
989 return (_flags & StatsAck) != 0;
990 }
991 bool needsSequenceNumber(void) const
992 {
993 return (_flags & NeedsSequenceNumber) != 0;
994 }
995 };
996
999 typedef void(*DisconnectHandlerFunc)(Client&, void* userData);
1000
1001 class Message;
1002 typedef Handler<DisconnectHandlerFunc, Client&> DisconnectHandler;
1003
1006 class Authenticator
1007 {
1008 public:
1009 virtual ~Authenticator() {;}
1010
1016 virtual std::string authenticate(const std::string& userName_, const std::string& password_) = 0;
1024 virtual std::string retry(const std::string& userName_, const std::string& password_) = 0;
1031 virtual void completed(const std::string& userName_, const std::string& password_, const std::string& reason_) = 0;
1032 };
1033
1037 class DefaultAuthenticator : public Authenticator
1038 {
1039 public:
1040 virtual ~DefaultAuthenticator() {;}
1043 std::string authenticate(const std::string& /*userName_*/, const std::string& password_)
1044 {
1045 return password_;
1046 }
1047
1050 std::string retry(const std::string& /*userName_*/, const std::string& /*password_*/)
1051 {
1052 throw AuthenticationException("retry not implemented by DefaultAuthenticator.");
1053 }
1054
1055 void completed(const std::string& /*userName_*/, const std::string& /* password_ */, const std::string& /* reason */) {;}
1056
1060 static Authenticator& instance()
1061 {
1062 static DefaultAuthenticator d; // -V1096
1063 return d;
1064 }
1065 };
1066
1069 class StoreReplayer
1070 {
1071 public:
1072
1076 virtual void execute(Message& message_) = 0;
1077
1078 virtual ~StoreReplayer() {;}
1079 };
1080
1081 class Store;
1082
1091 typedef bool (*PublishStoreResizeHandler)(Store store_,
1092 size_t size_,
1093 void* userData_);
1094
1097 class StoreImpl : public RefBody
1098 {
1099 public:
1105 StoreImpl(bool errorOnPublishGap_ = false)
1106 : _resizeHandler(NULL)
1107 , _resizeHandlerData(NULL)
1108 , _errorOnPublishGap(errorOnPublishGap_)
1109 {;}
1110
1115 virtual amps_uint64_t store(const Message& message_) = 0;
1116
1123 virtual void discardUpTo(amps_uint64_t index_) = 0;
1124
1129 virtual void replay(StoreReplayer& replayer_) = 0;
1130
1138 virtual bool replaySingle(StoreReplayer& replayer_, amps_uint64_t index_) = 0;
1139
1144 virtual size_t unpersistedCount() const = 0;
1145
1146 virtual ~StoreImpl() {;}
1147
1156 virtual void flush(long timeout_) = 0;
1157
1160 static inline size_t getUnsetPosition()
1161 {
1162 return AMPS_UNSET_INDEX;
1163 }
1164
1167 static inline amps_uint64_t getUnsetSequence()
1168 {
1169 return AMPS_UNSET_SEQUENCE;
1170 }
1171
1175 virtual amps_uint64_t getLowestUnpersisted() const = 0;
1176
1180 virtual amps_uint64_t getLastPersisted() = 0;
1181
1191 inline virtual void setResizeHandler(PublishStoreResizeHandler handler_,
1192 void* userData_)
1193 {
1194 _resizeHandler = handler_;
1195 _resizeHandlerData = userData_;
1196 }
1197
1198 inline virtual PublishStoreResizeHandler getResizeHandler() const
1199 {
1200 return _resizeHandler;
1201 }
1202
1203 bool callResizeHandler(size_t newSize_);
1204
1205 inline virtual void setErrorOnPublishGap(bool errorOnPublishGap_)
1206 {
1207 _errorOnPublishGap = errorOnPublishGap_;
1208 }
1209
1210 inline virtual bool getErrorOnPublishGap() const
1211 {
1212 return _errorOnPublishGap;
1213 }
1214
1215 private:
1216 PublishStoreResizeHandler _resizeHandler;
1217 void* _resizeHandlerData;
1218 bool _errorOnPublishGap;
1219 };
1220
1223 class Store
1224 {
1225 RefHandle<StoreImpl> _body;
1226 public:
1227 Store() = default;
1228 Store(StoreImpl* body_) : _body(body_) {;}
1229 Store(const Store& rhs) : _body(rhs._body) {;}
1230 Store& operator=(const Store& rhs)
1231 {
1232 _body = rhs._body;
1233 return *this;
1234 }
1235
1239 amps_uint64_t store(const Message& message_)
1240 {
1241 return _body.get().store(message_);
1242 }
1243
1250 void discardUpTo(amps_uint64_t index_)
1251 {
1252 _body.get().discardUpTo(index_);
1253 }
1254
1259 void replay(StoreReplayer& replayer_)
1260 {
1261 _body.get().replay(replayer_);
1262 }
1263
1271 bool replaySingle(StoreReplayer& replayer_, amps_uint64_t index_)
1272 {
1273 return _body.get().replaySingle(replayer_, index_);
1274 }
1275
1280 size_t unpersistedCount() const
1281 {
1282 return _body.get().unpersistedCount();
1283 }
1284
1288 bool isValid() const
1289 {
1290 return _body.isValid();
1291 }
1292
1301 void flush(long timeout_ = 0)
1302 {
1303 return _body.get().flush(timeout_);
1304 }
1305
1309 amps_uint64_t getLowestUnpersisted()
1310 {
1311 return _body.get().getLowestUnpersisted();
1312 }
1313
1317 amps_uint64_t getLastPersisted()
1318 {
1319 return _body.get().getLastPersisted();
1320 }
1321
1331 void setResizeHandler(PublishStoreResizeHandler handler_,
1332 void* userData_)
1333 {
1334 _body.get().setResizeHandler(handler_, userData_);
1335 }
1336
1337 PublishStoreResizeHandler getResizeHandler() const
1338 {
1339 return _body.get().getResizeHandler();
1340 }
1341
1346 inline void setErrorOnPublishGap(bool errorOnPublishGap_)
1347 {
1348 _body.get().setErrorOnPublishGap(errorOnPublishGap_);
1349 }
1350
1355 inline bool getErrorOnPublishGap() const
1356 {
1357 return _body.get().getErrorOnPublishGap();
1358 }
1359
1363 StoreImpl* get()
1364 {
1365 if (_body.isValid())
1366 {
1367 return &_body.get();
1368 }
1369 else
1370 {
1371 return NULL;
1372 }
1373 }
1374
1375 };
1376
1381 class FailedWriteHandler
1382 {
1383 public:
1384 virtual ~FailedWriteHandler() {;}
1391 virtual void failedWrite(const Message& message_,
1392 const char* reason_, size_t reasonLength_) = 0;
1393 };
1394
1395
1396 inline bool StoreImpl::callResizeHandler(size_t newSize_)
1397 {
1398 if (_resizeHandler)
1399 {
1400 return _resizeHandler(Store(this), newSize_, _resizeHandlerData);
1401 }
1402 return true;
1403 }
1404
1411 inline bool DangerousFlushPublishStoreResizeHandler(Store store_, size_t /*size_*/,
1412 void* data_)
1413 {
1414 long* timeoutp = (long*)data_;
1415 size_t count = store_.unpersistedCount();
1416 if (count == 0)
1417 {
1418 return false;
1419 }
1420 try
1421 {
1422 store_.flush(*timeoutp);
1423 }
1424 #ifdef _WIN32
1425 catch (const TimedOutException&)
1426 #else
1427 catch (const TimedOutException& e)
1428 #endif
1429 {
1430 return true;
1431 }
1432 return (count == store_.unpersistedCount());
1433 }
1434
1439 class FailedResubscribeHandler
1440 {
1441 public:
1453 virtual bool failure(const Message& message_, const MessageHandler& handler_,
1454 unsigned requestedAckTypes_,
1455 const AMPSException& exception_) = 0;
1456 };
1457
1461 class SubscriptionManager
1462 {
1463 public:
1464 virtual ~SubscriptionManager() {;}
1474 virtual AMPS_ATOMIC_BASE_TYPE subscribe(MessageHandler messageHandler_, const Message& message_,
1475 unsigned requestedAckTypes_) = 0;
1479 virtual AMPS_ATOMIC_BASE_TYPE getGenerationCount() const
1480 {
1481 return 0;
1482 }
1486 virtual void unsubscribe(const Message::Field& subId_) = 0;
1489 virtual void clear() = 0;
1493 virtual void resubscribe(Client& client_) = 0;
1498 virtual void setFailedResubscribeHandler(std::shared_ptr<FailedResubscribeHandler> handler_)
1499 {
1500 _failedResubscribeHandler = handler_;
1501 }
1502 protected:
1503 std::shared_ptr<FailedResubscribeHandler> _failedResubscribeHandler;
1504 };
1505
1509
1510 class ConnectionStateListener
1511 {
1512 public:
1514 typedef enum { Disconnected = 0,
1515 Shutdown = 1,
1516 Connected = 2,
1517 LoggedOn = 4,
1518 PublishReplayed = 8,
1519 HeartbeatInitiated = 16,
1520 Resubscribed = 32,
1521 UNKNOWN = 16384
1522 } State;
1523
1533 virtual void connectionStateChanged(State newState_) = 0;
1534 virtual ~ConnectionStateListener() {;}
1535 };
1536
1537
1538 class MessageStreamImpl;
1539 class MessageStream;
1540
1541 typedef void(*DeferredExecutionFunc)(void*);
1542
1543 class ClientImpl : public RefBody // -V553
1544 {
1545 // Class to wrap turning of Nagle for things like flush and logon
1546 class NoDelay
1547 {
1548 private:
1549 AMPS_SOCKET _socket;
1550 int _noDelay;
1551 char* _valuePtr;
1552 #ifdef _WIN32
1553 int _valueLen;
1554 #else
1555 socklen_t _valueLen;
1556 #endif
1557 public:
1558 NoDelay(amps_handle client_)
1559 : _socket(AMPS_INVALID_SOCKET), _noDelay(0), _valueLen(sizeof(int))
1560 {
1561 _valuePtr = (char*)&_noDelay;
1562 _socket = amps_client_get_socket(client_);
1563 if (_socket != AMPS_INVALID_SOCKET)
1564 {
1565 getsockopt(_socket, IPPROTO_TCP, TCP_NODELAY, _valuePtr, &_valueLen);
1566 if (!_noDelay)
1567 {
1568 _noDelay = 1;
1569 setsockopt(_socket, IPPROTO_TCP, TCP_NODELAY, _valuePtr, _valueLen);
1570 }
1571 else
1572 {
1573 _socket = AMPS_INVALID_SOCKET;
1574 }
1575 }
1576 }
1577
1578 ~NoDelay()
1579 {
1580 if (_socket != AMPS_INVALID_SOCKET)
1581 {
1582 _noDelay = 0;
1583 setsockopt(_socket, IPPROTO_TCP, TCP_NODELAY, _valuePtr, _valueLen);
1584 }
1585 }
1586 };
1587
1588 friend class Client;
1589 protected:
1590 amps_handle _client;
1591 DisconnectHandler _disconnectHandler;
1592 enum GlobalCommandTypeHandlers : size_t
1593 {
1594 Publish = 0,
1595 SOW = 1,
1596 GroupBegin = 2,
1597 GroupEnd = 3,
1598 Heartbeat = 4,
1599 OOF = 5,
1600 Ack = 6,
1601 LastChance = 7,
1602 DuplicateMessage = 8,
1603 COUNT = 9
1604 };
1605 std::vector<MessageHandler> _globalCommandTypeHandlers;
1606 Message _beatMessage, _readMessage;
1607 MessageRouter _routes;
1608 MessageRouter::RouteCache _routeCache;
1609 mutable Mutex _lock;
1610 std::string _name, _nameHash, _connectionName, _lastUri, _logonCorrelationData, _preflightMessage;
1611 std::vector<std::string> _httpPreflightHeaders;
1612 amps_uint64_t _nameHashValue;
1613 BookmarkStore _bookmarkStore;
1614 Store _publishStore;
1615 bool _isRetryOnDisconnect;
1616 amps_unique_ptr<FailedWriteHandler> _failedWriteHandler;
1617 #if __cplusplus >= 201100L || _MSC_VER >= 1900
1618 std::atomic<amps_uint64_t> _lastSentHaSequenceNumber;
1619 #else
1620 volatile amps_uint64_t _lastSentHaSequenceNumber;
1621 #endif
1622 AMPS_ATOMIC_TYPE_8 _logonInProgress;
1623 AMPS_ATOMIC_TYPE _subscriptionManagerGenerationCount;
1624 VersionInfo _serverVersion;
1625 Timer _heartbeatTimer;
1626 amps_unique_ptr<MessageStream> _pEmptyMessageStream;
1627
1628 // queue data
1629 int _queueAckTimeout;
1630 bool _isAutoAckEnabled;
1631 unsigned _ackBatchSize;
1632 unsigned _queuedAckCount;
1633 unsigned _defaultMaxDepth;
1634 struct QueueBookmarks
1635 {
1636 QueueBookmarks(const std::string& topic_)
1637 : _topic(topic_)
1638 , _oldestTime(0)
1639 , _bookmarkCount(0)
1640 {;}
1641 std::string _topic;
1642 std::string _data;
1643 amps_uint64_t _oldestTime;
1644 unsigned _bookmarkCount;
1645 };
1646 typedef amps_uint64_t topic_hash;
1647 typedef std::map<topic_hash, QueueBookmarks> TopicHashMap;
1648 TopicHashMap _topicHashMap;
1649
1650 class ClientStoreReplayer : public StoreReplayer
1651 {
1652 ClientImpl* _client;
1653 public:
1654 unsigned _version;
1655 amps_result _res;
1656
1657 ClientStoreReplayer()
1658 : _client(NULL), _version(0), _res(AMPS_E_OK)
1659 {}
1660
1661 ClientStoreReplayer(ClientImpl* client_)
1662 : _client(client_), _version(0), _res(AMPS_E_OK)
1663 {}
1664
1665 void setClient(ClientImpl* client_)
1666 {
1667 _client = client_;
1668 }
1669
1670 void execute(Message& message_)
1671 {
1672 if (!_client)
1673 {
1674 throw CommandException("Can't replay without a client.");
1675 }
1676 amps_uint64_t index = amps_message_get_field_uint64(message_.getMessage(),
1677 AMPS_Sequence);
1678 if (index > _client->_lastSentHaSequenceNumber)
1679 {
1680 _client->_lastSentHaSequenceNumber = index;
1681 }
1682
1683 _res = AMPS_E_OK;
1684 // Don't replay a queue cancel message after a reconnect.
1685 // Currently, the only messages that will have anything in options
1686 // are cancel messages.
1687 if (!message_.getCommand().empty() &&
1688 (!_client->_logonInProgress ||
1689 message_.getOptions().len() < 6))
1690 {
1691 _res = amps_client_send_batch(_client->_client,
1692 message_.getMessage(),
1693 &_version,
1694 1);
1695 if (_res != AMPS_E_OK)
1696 {
1697 throw DisconnectedException("AMPS Server disconnected during replay");
1698 }
1699 }
1700 }
1701
1702 };
1703 ClientStoreReplayer _replayer;
1704
1705 class FailedWriteStoreReplayer : public StoreReplayer
1706 {
1707 ClientImpl* _parent;
1708 const char* _reason;
1709 size_t _reasonLength;
1710 size_t _replayCount;
1711 public:
1712 FailedWriteStoreReplayer(ClientImpl* parent, const char* reason_, size_t reasonLength_)
1713 : _parent(parent),
1714 _reason(reason_),
1715 _reasonLength(reasonLength_),
1716 _replayCount(0)
1717 {;}
1718 void execute(Message& message_)
1719 {
1720 if (_parent->_failedWriteHandler)
1721 {
1722 ++_replayCount;
1723 _parent->_failedWriteHandler->failedWrite(message_,
1724 _reason, _reasonLength);
1725 }
1726 }
1727 size_t replayCount(void) const
1728 {
1729 return _replayCount;
1730 }
1731 };
1732
1733 struct AckResponseImpl : public RefBody
1734 {
1735 std::string username, password, reason, status, bookmark, options;
1736 amps_uint64_t sequenceNo;
1737 amps_uint64_t nameHashValue;
1738 VersionInfo serverVersion;
1739 #if __cplusplus >= 201100L || _MSC_VER >= 1900
1740 std::atomic<bool> responded;
1741 std::atomic<bool> abandoned;
1742 #else
1743 volatile bool responded;
1744 volatile bool abandoned;
1745 #endif
1746 unsigned connectionVersion;
1747 AckResponseImpl() :
1748 RefBody(),
1749 username(), password(), reason(), status(), bookmark(), options(),
1750 sequenceNo((amps_uint64_t)0),
1751 nameHashValue((amps_uint64_t)0),
1752 serverVersion(),
1753 responded(false),
1754 abandoned(false),
1755 connectionVersion(UINT_MAX) // Don't abandon if unsent AC-1329
1756 {
1757 }
1758 };
1759
1760 class AckResponse
1761 {
1762 RefHandle<AckResponseImpl> _body;
1763 public:
1764 AckResponse() : _body(NULL) {;}
1765 AckResponse(const AckResponse& rhs) : _body(rhs._body) {;}
1766 static AckResponse create()
1767 {
1768 AckResponse r;
1769 r._body = new AckResponseImpl();
1770 return r;
1771 }
1772
1773 const std::string& username()
1774 {
1775 return _body.get().username;
1776 }
1777 void setUsername(const char* data_, size_t len_)
1778 {
1779 if (data_)
1780 {
1781 _body.get().username.assign(data_, len_);
1782 }
1783 else
1784 {
1785 _body.get().username.clear();
1786 }
1787 }
1788 const std::string& password()
1789 {
1790 return _body.get().password;
1791 }
1792 void setPassword(const char* data_, size_t len_)
1793 {
1794 if (data_)
1795 {
1796 _body.get().password.assign(data_, len_);
1797 }
1798 else
1799 {
1800 _body.get().password.clear();
1801 }
1802 }
1803 const std::string& reason()
1804 {
1805 return _body.get().reason;
1806 }
1807 void setReason(const char* data_, size_t len_)
1808 {
1809 if (data_)
1810 {
1811 _body.get().reason.assign(data_, len_);
1812 }
1813 else
1814 {
1815 _body.get().reason.clear();
1816 }
1817 }
1818 const std::string& status()
1819 {
1820 return _body.get().status;
1821 }
1822 void setStatus(const char* data_, size_t len_)
1823 {
1824 if (data_)
1825 {
1826 _body.get().status.assign(data_, len_);
1827 }
1828 else
1829 {
1830 _body.get().status.clear();
1831 }
1832 }
1833 const std::string& bookmark()
1834 {
1835 return _body.get().bookmark;
1836 }
1837 void setBookmark(const Field& bookmark_)
1838 {
1839 AckResponseImpl& body = _body.get();
1840 if (!bookmark_.empty())
1841 {
1842 body.bookmark.assign(bookmark_.data(), bookmark_.len());
1843 Field::parseBookmark(bookmark_, body.nameHashValue,
1844 body.sequenceNo);
1845 }
1846 else
1847 {
1848 body.bookmark.clear();
1849 body.sequenceNo = (amps_uint64_t)0;
1850 body.nameHashValue = (amps_uint64_t)0;
1851 }
1852 }
1853 amps_uint64_t sequenceNo() const
1854 {
1855 return _body.get().sequenceNo;
1856 }
1857 amps_uint64_t nameHashValue() const
1858 {
1859 return _body.get().nameHashValue;
1860 }
1861 void setSequenceNo(const char* data_, size_t len_)
1862 {
1863 amps_uint64_t result = (amps_uint64_t)0;
1864 if (data_)
1865 {
1866 for (size_t i = 0; i < len_; ++i)
1867 {
1868 result *= (amps_uint64_t)10;
1869 result += (amps_uint64_t)(data_[i] - '0');
1870 }
1871 }
1872 _body.get().sequenceNo = result;
1873 }
1874 VersionInfo serverVersion() const
1875 {
1876 return _body.get().serverVersion;
1877 }
1878 void setServerVersion(const char* data_, size_t len_)
1879 {
1880 if (data_)
1881 {
1882 _body.get().serverVersion.setVersion(std::string(data_, len_));
1883 }
1884 }
1885 bool responded()
1886 {
1887 return _body.get().responded;
1888 }
1889 void setResponded()
1890 {
1891 _body.get().responded = true;
1892 }
1893 bool abandoned()
1894 {
1895 return _body.get().abandoned;
1896 }
1897 void setAbandoned()
1898 {
1899 if (_body.isValid())
1900 {
1901 _body.get().abandoned = true;
1902 }
1903 }
1904
1905 void setConnectionVersion(unsigned connectionVersion)
1906 {
1907 _body.get().connectionVersion = connectionVersion;
1908 }
1909
1910 unsigned getConnectionVersion()
1911 {
1912 return _body.get().connectionVersion;
1913 }
1914
1915 void setOptions(const char* data_, size_t len_)
1916 {
1917 if (data_)
1918 {
1919 _body.get().options.assign(data_, len_);
1920 }
1921 else
1922 {
1923 _body.get().options.clear();
1924 }
1925 }
1926
1927 const std::string& options()
1928 {
1929 return _body.get().options;
1930 }
1931
1932 AckResponse& operator=(const AckResponse& rhs)
1933 {
1934 _body = rhs._body;
1935 return *this;
1936 }
1937 };
1938
1939
1940 typedef std::map<std::string, AckResponse> AckMap;
1941 AckMap _ackMap;
1942 Mutex _ackMapLock;
1943 DefaultExceptionListener _defaultExceptionListener;
1944 protected:
1945
1946 struct DeferredExecutionRequest
1947 {
1948 DeferredExecutionRequest(DeferredExecutionFunc func_,
1949 void* userData_)
1950 : _func(func_),
1951 _userData(userData_)
1952 {;}
1953
1954 DeferredExecutionFunc _func;
1955 void* _userData;
1956 };
1957 const ExceptionListener* _exceptionListener;
1958 std::shared_ptr<const ExceptionListener> _pExceptionListener;
1959 amps_unique_ptr<SubscriptionManager> _subscriptionManager;
1960 volatile bool _connected;
1961 std::string _username;
1962 typedef std::set<ConnectionStateListener*> ConnectionStateListeners;
1963 ConnectionStateListeners _connectionStateListeners;
1964 typedef std::vector<DeferredExecutionRequest> DeferredExecutionList;
1965 Mutex _deferredExecutionLock;
1966 DeferredExecutionList _deferredExecutionList;
1967 unsigned _heartbeatInterval;
1968 unsigned _readTimeout;
1969
1970 void broadcastConnectionStateChanged(ConnectionStateListener::State newState_)
1971 {
1972 // If we disconnected before we got to notification, don't notify.
1973 // This should only be able to happen for Resubscribed, since the lock
1974 // is released to let the subscription manager run resubscribe so a
1975 // disconnect could be called before the change is broadcast.
1976 if (!_connected && newState_ > ConnectionStateListener::Connected)
1977 {
1978 return;
1979 }
1980 for (ConnectionStateListeners::iterator it = _connectionStateListeners.begin(); it != _connectionStateListeners.end(); ++it)
1981 {
1982 AMPS_CALL_EXCEPTION_WRAPPER(
1983 (*it)->connectionStateChanged(newState_));
1984 }
1985 }
1986 void captureSubscriptionManagerGenerationCount()
1987 {
1988 if (_subscriptionManager)
1989 {
1990 _subscriptionManagerGenerationCount = _subscriptionManager->getGenerationCount();
1991 }
1992 }
1993
1994 unsigned processedAck(Message& message);
1995 unsigned persistedAck(Message& meesage);
1996 void lastChance(Message& message);
1997 void checkAndSendHeartbeat(bool force = false);
1998 virtual ConnectionInfo getConnectionInfo() const;
1999 static amps_result
2000 ClientImplMessageHandler(amps_handle message, void* userData);
2001 static void
2002 ClientImplPreDisconnectHandler(amps_handle client, unsigned failedConnectionVersion, void* userData);
2003 static amps_result
2004 ClientImplDisconnectHandler(amps_handle client, void* userData);
2005 static const char*
2006 ClientImplGetHttpPreflightMessage(void* userData);
2007
2008 void unsubscribeInternal(const std::string& id)
2009 {
2010 if (id.empty())
2011 {
2012 return;
2013 }
2014 // remove the handler first to avoid any more message delivery
2015 Message::Field subId;
2016 subId.assign(id.data(), id.length());
2017 _routes.removeRoute(subId);
2018 // Lock is already acquired
2019 if (_subscriptionManager)
2020 {
2021 // Have to unlock before calling into sub manager to avoid deadlock
2022 Unlock<Mutex> unlock(_lock);
2023 _subscriptionManager->unsubscribe(subId);
2024 }
2025 threadLocalMessage.reset();
2026 threadLocalMessage.setCommandEnum(Message::Command::Unsubscribe);
2027 threadLocalMessage.newCommandId();
2028 threadLocalMessage.setSubscriptionId(id);
2029 _sendWithoutRetry(threadLocalMessage);
2030 deferredExecution(&amps_noOpFn, NULL);
2031 }
2032
2033 AckResponse syncAckProcessing(long timeout_, Message& message_,
2034 amps_uint64_t haSeq = (amps_uint64_t)0,
2035 AMPS_ATOMIC_BASE_TYPE subscribeGeneration_ = 0)
2036 {
2037 bool isLogon = message_.getCommand() == "logon";
2038 if (_logonInProgress && !isLogon)
2039 {
2040 while (_logonInProgress)
2041 {
2042 if (!_lock.wait(200))
2043 {
2044 amps_invoke_waiting_function();
2045 if (!_isRetryOnDisconnect)
2046 {
2047 // retrySend is disabled so throw the error
2048 AMPSException::throwFor(_client, AMPS_E_RETRY);
2049 }
2050 }
2051 }
2052 }
2053 // inv: we already have _lock locked up.
2054 AckResponse ack = AckResponse::create();
2055 if (1)
2056 {
2057 Lock<Mutex> guard(_ackMapLock);
2058 _ackMap[message_.getCommandId()] = ack;
2059 }
2060 if (subscribeGeneration_ == 0)
2061 {
2062 ack.setConnectionVersion((unsigned)_sendSequenced(message_, haSeq, 0));
2063 }
2064 else
2065 {
2066 ack.setConnectionVersion((unsigned)_send(message_, subscribeGeneration_));
2067 }
2068 if (ack.getConnectionVersion() == 0)
2069 {
2070 // Send failed
2071 throw DisconnectedException("Connection closed while waiting for response.");
2072 }
2073 else if (ack.getConnectionVersion() == UINT_MAX)
2074 {
2075 ack.setResponded();
2076 Lock<Mutex> guard(_ackMapLock);
2077 AckMap::iterator i = _ackMap.find(message_.getCommandId());
2078 if (i != _ackMap.end())
2079 {
2080 _ackMap.erase(i);
2081 }
2082 return ack;
2083 }
2084 bool timedOut = false;
2085 AMPS_START_TIMER(timeout_)
2086 while (!timedOut && !ack.responded() && !ack.abandoned())
2087 {
2088 if (timeout_)
2089 {
2090 timedOut = !_lock.wait(timeout_);
2091 // May have woken up early, check real time
2092 if (timedOut)
2093 {
2094 AMPS_RESET_TIMER(timedOut, timeout_);
2095 }
2096 }
2097 else
2098 {
2099 // Using a timeout version to ensure python can interrupt
2100 _lock.wait(1000);
2101 Unlock<Mutex> unlck(_lock);
2102 amps_invoke_waiting_function();
2103 }
2104 }
2105 if (ack.responded())
2106 {
2107 if (ack.status() != "failure")
2108 {
2109 if (message_.getCommand() == "logon")
2110 {
2111 amps_uint64_t ackSequence = ack.sequenceNo();
2112 if (_lastSentHaSequenceNumber < ackSequence)
2113 {
2114 _lastSentHaSequenceNumber = ackSequence;
2115 }
2116 if (_publishStore.isValid())
2117 {
2118 // If this throws, logon will fail and eitehr be
2119 // handled in HAClient/ServerChooser or by the caller
2120 // of logon.
2121 _publishStore.discardUpTo(ackSequence);
2122 if (_lastSentHaSequenceNumber < _publishStore.getLastPersisted())
2123 {
2124 _lastSentHaSequenceNumber = _publishStore.getLastPersisted();
2125 }
2126 }
2127 _nameHash = ack.bookmark().substr(0, ack.bookmark().find('|'));
2128 _nameHashValue = ack.nameHashValue();
2129 const std::string& ackOptions = ack.options();
2130 size_t start = ackOptions.find("connection_name=");
2131 if (start != std::string::npos)
2132 {
2133 start += 16;
2134 size_t end = ackOptions.find(',', start);
2135 if (end != std::string::npos)
2136 {
2137 end -= start;
2138 }
2139 _connectionName = ackOptions.substr(start, end);
2140 }
2141 _serverVersion = ack.serverVersion();
2142 if (_bookmarkStore.isValid())
2143 {
2144 _bookmarkStore.setServerVersion(_serverVersion);
2145 }
2146 }
2147 if (_ackBatchSize)
2148 {
2149 const std::string& options = ack.options();
2150 size_t index = options.find_first_of("max_backlog=");
2151 if (index != std::string::npos)
2152 {
2153 unsigned data = 0;
2154 const char* c = options.c_str() + index + 12;
2155 while (*c && *c != ',')
2156 {
2157 data = (data * 10) + (unsigned)(*c++ -48);
2158 }
2159 if (_ackBatchSize > data)
2160 {
2161 _ackBatchSize = data;
2162 }
2163 }
2164 }
2165 return ack;
2166 }
2167 const size_t NotEntitled = 12;
2168 std::string ackReason = ack.reason();
2169 if (ackReason.length() == 0)
2170 {
2171 return ack; // none
2172 }
2173 if (ackReason.length() == NotEntitled &&
2174 ackReason[0] == 'n' &&
2175 message_.getUserId().len() == 0)
2176 {
2177 message_.assignUserId(_username);
2178 }
2179 message_.throwFor(_client, ackReason);
2180 }
2181 else // !ack.responded()
2182 {
2183 if (!ack.abandoned())
2184 {
2185 throw TimedOutException("timed out waiting for operation.");
2186 }
2187 else
2188 {
2189 throw DisconnectedException("Connection closed while waiting for response.");
2190 }
2191 }
2192 return ack;
2193 }
2194
2195 void _cleanup(void)
2196 {
2197 if (!_client)
2198 {
2199 return;
2200 }
2201 amps_client_set_predisconnect_handler(_client, NULL, 0L);
2202 amps_client_set_disconnect_handler(_client, NULL, 0L);
2203 AMPS_CALL_EXCEPTION_WRAPPER(ClientImpl::disconnect());
2204 _pEmptyMessageStream.reset(NULL);
2205 amps_client_destroy(_client);
2206 _client = NULL;
2207 }
2208
2209 public:
2210
2211 ClientImpl(const std::string& clientName)
2212 : _client(NULL), _name(clientName)
2213 , _isRetryOnDisconnect(true)
2214 , _lastSentHaSequenceNumber((amps_uint64_t)0), _logonInProgress(0)
2215 , _subscriptionManagerGenerationCount(0), _serverVersion()
2216 , _queueAckTimeout(AMPS_DEFAULT_QUEUE_ACK_TIMEOUT)
2217 , _isAutoAckEnabled(false)
2218 , _ackBatchSize(0)
2219 , _queuedAckCount(0)
2220 , _defaultMaxDepth(0)
2221 , _connected(false)
2222 , _heartbeatInterval(0)
2223 , _readTimeout(0)
2224 {
2225 _replayer.setClient(this);
2226 _client = amps_client_create(clientName.c_str());
2227 amps_client_set_message_handler(_client,
2228 (amps_handler)ClientImpl::ClientImplMessageHandler,
2229 this);
2230 amps_client_set_predisconnect_handler(_client,
2231 (amps_predisconnect_handler)ClientImpl::ClientImplPreDisconnectHandler,
2232 this);
2233 amps_client_set_disconnect_handler(_client,
2234 (amps_handler)ClientImpl::ClientImplDisconnectHandler,
2235 this);
2236 amps_client_set_http_preflight_callback(_client,
2237 ClientImpl::ClientImplGetHttpPreflightMessage,
2238 this);
2239 _exceptionListener = &_defaultExceptionListener;
2240 for (size_t i = 0; i < GlobalCommandTypeHandlers::COUNT; ++i)
2241 {
2242 _globalCommandTypeHandlers.emplace_back(MessageHandler());
2243 }
2244 }
2245
2246 virtual ~ClientImpl()
2247 {
2248 _cleanup();
2249 }
2250
2251 const std::string& getName() const
2252 {
2253 return _name;
2254 }
2255
2256 const std::string& getNameHash() const
2257 {
2258 return _nameHash;
2259 }
2260
2261 const amps_uint64_t getNameHashValue() const
2262 {
2263 return _nameHashValue;
2264 }
2265
2266 const std::string& getConnectionName() const
2267 {
2268 return _connectionName;
2269 }
2270
2271 void setName(const std::string& name)
2272 {
2273 // This operation will fail if the client's
2274 // name is already set.
2275 amps_result result = amps_client_set_name(_client, name.c_str());
2276 if (result != AMPS_E_OK)
2277 {
2278 AMPSException::throwFor(_client, result);
2279 }
2280 _name = name;
2281 }
2282
2283 const std::string& getLogonCorrelationData() const
2284 {
2285 return _logonCorrelationData;
2286 }
2287
2288 void setLogonCorrelationData(const std::string& logonCorrelationData_)
2289 {
2290 _logonCorrelationData = logonCorrelationData_;
2291 }
2292
2293 size_t getServerVersion() const
2294 {
2295 return _serverVersion.getOldStyleVersion();
2296 }
2297
2298 VersionInfo getServerVersionInfo() const
2299 {
2300 return _serverVersion;
2301 }
2302
2303 const std::string& getURI() const
2304 {
2305 return _lastUri;
2306 }
2307
2308 virtual void connect(const std::string& uri)
2309 {
2310 Lock<Mutex> l(_lock);
2311 _connect(uri);
2312 }
2313
2314 virtual void _connect(const std::string& uri)
2315 {
2316 _lastUri = uri;
2317 amps_result result = amps_client_connect(_client, uri.c_str());
2318 if (result != AMPS_E_OK)
2319 {
2320 AMPSException::throwFor(_client, result);
2321 }
2322 threadLocalMessage.reset();
2323 _beatMessage.setCommandEnum(Message::Command::Heartbeat);
2324 _beatMessage.setOptions("beat");
2325 _readMessage.setClientImpl(this);
2326 if (_queueAckTimeout)
2327 {
2328 result = amps_client_set_idle_time(_client, _queueAckTimeout);
2329 if (result != AMPS_E_OK)
2330 {
2331 AMPSException::throwFor(_client, result);
2332 }
2333 }
2334 _connected = true;
2335 broadcastConnectionStateChanged(ConnectionStateListener::Connected);
2336 }
2337
2338 void addHttpPreflightHeader(const std::string& header_)
2339 {
2340 _httpPreflightHeaders.emplace_back(header_);
2341 }
2342
2343 void addHttpPreflightHeader(const std::string& key_, const std::string& value_)
2344 {
2345 _httpPreflightHeaders.emplace_back(key_ + std::string(": ") + value_);
2346 }
2347
2348 void clearHttpPreflightHeaders()
2349 {
2350 _httpPreflightHeaders.clear();
2351 }
2352
2353 template<class T>
2354 void setHttpPreflightHeaders(const T& headers_)
2355 {
2356 _httpPreflightHeaders.clear();
2357 for (typename T::const_iterator i = headers_.begin(); i != headers_.end(); ++i)
2358 {
2359 _httpPreflightHeaders.push_back(*i);
2360 }
2361 }
2362
2363 void setDisconnected()
2364 {
2365 {
2366 Lock<Mutex> l(_lock);
2367 if (_connected)
2368 {
2369 AMPS_CALL_EXCEPTION_WRAPPER(broadcastConnectionStateChanged(ConnectionStateListener::Disconnected));
2370 }
2371 _connected = false;
2372 _heartbeatTimer.setTimeout(0.0);
2373 // AC-1189 AC-1329 AC-1337 We need acks cleared while lock is held,
2374 // but not for unsent commands.
2375 clearAcks(UINT_MAX-1);
2376 }
2377 amps_client_disconnect(_client);
2378 _routes.clear();
2379 }
2380
2381 virtual void disconnect()
2382 {
2383 AMPS_CALL_EXCEPTION_WRAPPER(flushAcks());
2384 setDisconnected();
2385 // Abandon all acks, sent and unsent
2386 clearAcks(UINT_MAX);
2387 AMPS_CALL_EXCEPTION_WRAPPER(processDeferredExecutions());
2388 Lock<Mutex> l(_lock);
2389 broadcastConnectionStateChanged(ConnectionStateListener::Shutdown);
2390 }
2391
2392 void clearAcks(unsigned failedVersion)
2393 {
2394 // Have to lock to prevent race conditions
2395 Lock<Mutex> guard(_ackMapLock);
2396 {
2397 // Go ahead and signal any waiters if they are around...
2398 std::vector<std::string> worklist;
2399 for (AckMap::iterator i = _ackMap.begin(), e = _ackMap.end(); i != e; ++i)
2400 {
2401 if (i->second.getConnectionVersion() <= failedVersion)
2402 {
2403 i->second.setAbandoned();
2404 worklist.push_back(i->first);
2405 }
2406 }
2407
2408 for (std::vector<std::string>::iterator j = worklist.begin(), e = worklist.end(); j != e; ++j)
2409 {
2410 _ackMap.erase(*j);
2411 }
2412 }
2413
2414 _lock.signalAll();
2415 }
2416
2417 unsigned send(const Message& message)
2418 {
2419 Lock<Mutex> l(_lock);
2420 return _send(message);
2421 }
2422
2423 void sendWithoutRetry(const Message& message_)
2424 {
2425 Lock<Mutex> l(_lock);
2426 // If we got here while logon was in progress, then we tried to send
2427 // while we were disconnected so throw DisconnectedException
2428 if (_logonInProgress)
2429 {
2430 throw DisconnectedException("The client has been disconnected.");
2431 }
2432 _sendWithoutRetry(message_);
2433 }
2434
2435 void _sendWithoutRetry(const Message& message_)
2436 {
2437 amps_result result = amps_client_send(_client, message_.getMessage());
2438 if (result != AMPS_E_OK)
2439 {
2440 AMPSException::throwFor(_client, result);
2441 }
2442 }
2443
2444 // This version of send is used primarily by publish messages and anything
2445 // in a Store.
2446 unsigned _sendSequenced(const Message& message, amps_uint64_t haSeq, int isBatch_)
2447 {
2448 // Lock is already acquired
2449 amps_result result = AMPS_E_RETRY;
2450
2451 // Create a local reference to this message, as we'll need to hold on
2452 // to a reference to it in case reconnect occurs.
2453 Message localMessage = message;
2454 unsigned version = 0;
2455 bool isNotLogon = (haSeq) || (localMessage.getCommand().data()[0] != 'l');
2456
2457 while (result == AMPS_E_RETRY)
2458 {
2459 if (_logonInProgress && isNotLogon)
2460 {
2461 if (!_isRetryOnDisconnect)
2462 {
2463 if (haSeq)
2464 {
2465 // Prefer to return UINT_MAX to 0 so syncAckProcessing won't throw
2466 return UINT_MAX;
2467 }
2468 else
2469 {
2470 // retrySend is disabled so throw the error
2471 AMPSException::throwFor(_client, result);
2472 }
2473 }
2474 if (localMessage.getMessage() == message.getMessage())
2475 {
2476 localMessage = message.deepCopy();
2477 }
2478 while (_logonInProgress)
2479 {
2480 if (!_lock.wait(1000))
2481 {
2482 amps_invoke_waiting_function();
2483 }
2484 }
2485 }
2486 if (haSeq)
2487 {
2488 // Path for PublishStore messages
2489 if (haSeq <= _lastSentHaSequenceNumber)
2490 {
2491 // Prefer to return UINT_MAX to 0 so syncAckProcessing won't throw
2492 return UINT_MAX;
2493 }
2494 // It's possible to get here out of order, but this way we'll
2495 // always send in order.
2496 while (haSeq > _lastSentHaSequenceNumber + 1)
2497 {
2498 try
2499 {
2500 // Replayer updates _lastSentHaSsequenceNumber
2501 if (!_publishStore.replaySingle(_replayer,
2502 _lastSentHaSequenceNumber + 1))
2503 {
2504 continue;
2505 }
2506 result = AMPS_E_OK;
2507 version = _replayer._version;
2508 }
2509 #ifdef _WIN32
2510 catch (const DisconnectedException&)
2511 #else
2512 catch (const DisconnectedException& e)
2513 #endif
2514 {
2515 if (!_isRetryOnDisconnect)
2516 {
2517 // Prefer to return UINT_MAX to 0 so syncAckProcessing won't throw
2518 return UINT_MAX;
2519 }
2520 result = _replayer._res;
2521 break;
2522 }
2523 }
2524 ++_lastSentHaSequenceNumber;
2525 }
2526 result = amps_client_send_batch(_client,
2527 localMessage.getMessage(),
2528 &version,
2529 isBatch_);
2530 if (result != AMPS_E_OK)
2531 {
2532 if (!haSeq
2533 && localMessage.getMessage() == message.getMessage())
2534 {
2535 localMessage = message.deepCopy();
2536 }
2537 if (_isRetryOnDisconnect)
2538 {
2539 Unlock<Mutex> u(_lock);
2540 result = amps_client_attempt_reconnect(_client, version);
2541 // If this is in the publish store, it was
2542 // stored first and will have already been replayed by the
2543 // store after reconnect, so just return.
2544 if (haSeq
2545 && result == AMPS_E_RETRY)
2546 {
2547 // Prefer to return UINT_MAX to 0 so syncAckProcessing won't throw
2548 return UINT_MAX;
2549 }
2550 }
2551 else
2552 {
2553 // we don't throw if the message is in the publish store
2554 if (haSeq)
2555 {
2556 // Prefer to return UINT_MAX to 0 so syncAckProcessing won't throw
2557 return UINT_MAX;
2558 }
2559 // retrySend is disabled so throw the error
2560 // from the send as an exception, do not retry.
2561 AMPSException::throwFor(_client, result);
2562 }
2563 }
2564 if (result == AMPS_E_RETRY)
2565 {
2566 amps_invoke_waiting_function();
2567 }
2568 }
2569
2570 if (result != AMPS_E_OK)
2571 {
2572 // We should never get here for messages in the publish store
2573 AMPSException::throwFor(_client, result);
2574 }
2575 return version;
2576 }
2577
2578 // This version of send is used primarily by messages requesting acks
2579 unsigned _send(const Message& message, AMPS_ATOMIC_BASE_TYPE subscribeGeneration_ = 0)
2580 {
2581 // Lock is already acquired
2582 amps_result result = AMPS_E_RETRY;
2583
2584 // Create a local reference to this message, as we'll need to hold on
2585 // to a reference to it in case reconnect occurs.
2586 Message localMessage = message;
2587 unsigned version = 0;
2588 bool isHASubscribe = subscribeGeneration_ > 0;
2589
2590 while (result == AMPS_E_RETRY)
2591 {
2592 if (_logonInProgress && localMessage.getCommand().data()[0] != 'l')
2593 {
2594 if (!isHASubscribe
2595 && localMessage.getMessage() == message.getMessage())
2596 {
2597 localMessage = message.deepCopy();
2598 }
2599 while (_logonInProgress)
2600 {
2601 if (!_lock.wait(1000))
2602 {
2603 amps_invoke_waiting_function();
2604 if (!_isRetryOnDisconnect)
2605 {
2606 // retrySend is disabled so throw the error
2607 AMPSException::throwFor(_client, result);
2608 }
2609 }
2610 }
2611 }
2612 if (isHASubscribe
2613 && subscribeGeneration_ <= (AMPS_ATOMIC_BASE_TYPE)_subscriptionManagerGenerationCount)
2614 {
2615 // We return UINT_MAX because this is in the sub manager and will be sent
2616 // If we return 0, that will throw DisconnectedException and call
2617 // unsubscribe to remove it from sub manager and message routes.
2618 return UINT_MAX;
2619 }
2620 result = amps_client_send_batch(_client,
2621 localMessage.getMessage(),
2622 &version,
2623 0);
2624 if (result != AMPS_E_OK)
2625 {
2626 if (!isHASubscribe
2627 && localMessage.getMessage() == message.getMessage())
2628 {
2629 localMessage = message.deepCopy();
2630 }
2631 if (_isRetryOnDisconnect)
2632 {
2633 Unlock<Mutex> u(_lock);
2634 result = amps_client_attempt_reconnect(_client, version);
2635 // If this is an HA subscribe command, it was
2636 // stored first and will have already been replayed by the
2637 // store or sub manager after reconnect, so just return.
2638 if (isHASubscribe &&
2639 result == AMPS_E_RETRY)
2640 {
2641 // We return UINT_MAX because this is in the sub manager and will be sent
2642 // If we return 0, that will throw DisconnectedException and call
2643 // unsubscribe to remove it from sub manager and message routes.
2644 return UINT_MAX;
2645 }
2646 }
2647 else
2648 {
2649 // retrySend is disabled so throw the error
2650 // from the send as an exception, do not retry.
2651 AMPSException::throwFor(_client, result);
2652 }
2653 }
2654 if (result == AMPS_E_RETRY)
2655 {
2656 amps_invoke_waiting_function();
2657 }
2658 }
2659
2660 if (result != AMPS_E_OK)
2661 {
2662 AMPSException::throwFor(_client, result);
2663 }
2664 return version;
2665 }
2666
2667 void addMessageHandler(const Field& commandId_,
2668 const AMPS::MessageHandler& messageHandler_,
2669 unsigned requestedAcks_, Message::Command::Type commandType_)
2670 {
2671 Lock<Mutex> lock(_lock);
2672 _routes.addRoute(commandId_, messageHandler_, requestedAcks_,
2673 0, commandType_);
2674 }
2675
2676 bool removeMessageHandler(const Field& commandId_)
2677 {
2678 Lock<Mutex> lock(_lock);
2679 return _routes.removeRoute(commandId_);
2680 }
2681
2682 std::string send(const MessageHandler& messageHandler_,
2683 Message& message_,
2684 int timeout_ = 0,
2685 AMPS_ATOMIC_BASE_TYPE subscribeGeneration_ = 0)
2686 {
2687 Field id = message_.getCommandId();
2688 Field subId = message_.getSubscriptionId();
2689 Field qid = message_.getQueryId();
2690 bool isSubscribeOnly = false;
2691 bool replace = false;
2692 unsigned requestedAcks = message_.getAckTypeEnum();
2693 unsigned systemAddedAcks = Message::AckType::None;
2694 Message::Command::Type commandType = message_.getCommandEnum();
2695
2696 switch (commandType)
2697 {
2698 case Message::Command::Subscribe:
2699 case Message::Command::DeltaSubscribe:
2700 replace = message_.getOptions().operator std::string().find(AMPS_OPTIONS_REPLACE, 0, strlen(AMPS_OPTIONS_REPLACE) - 1) != std::string::npos;
2701 isSubscribeOnly = true;
2702 // fall through
2703 case Message::Command::SOWAndSubscribe:
2704 case Message::Command::SOWAndDeltaSubscribe:
2705 if (id.empty())
2706 {
2707 id = message_.newCommandId().getCommandId();
2708 }
2709 else
2710 {
2711 while (!replace && id != subId && _routes.hasRoute(id))
2712 {
2713 id = message_.newCommandId().getCommandId();
2714 }
2715 }
2716 if (subId.empty())
2717 {
2718 message_.setSubscriptionId(id);
2719 subId = id;
2720 }
2721 if (!message_.getBookmark().empty() && _bookmarkStore.isValid())
2722 {
2723 systemAddedAcks |= Message::AckType::Persisted;
2724 }
2725 // fall through
2726 case Message::Command::SOW:
2727 if (id.empty())
2728 {
2729 id = message_.newCommandId().getCommandId();
2730 }
2731 else
2732 {
2733 while (!replace && id != subId && _routes.hasRoute(id))
2734 {
2735 message_.newCommandId();
2736 if (qid == id)
2737 {
2738 qid = message_.getCommandId();
2739 message_.setQueryId(qid);
2740 }
2741 id = message_.getCommandId();
2742 }
2743 }
2744 if (!isSubscribeOnly)
2745 {
2746 if (qid.empty())
2747 {
2748 message_.setQueryID(id);
2749 qid = id;
2750 }
2751 else
2752 {
2753 while (!replace && qid != subId && qid != id
2754 && _routes.hasRoute(qid))
2755 {
2756 qid = message_.newQueryId().getQueryId();
2757 }
2758 }
2759 }
2760 systemAddedAcks |= Message::AckType::Processed;
2761 message_.setAckTypeEnum(requestedAcks | systemAddedAcks);
2762 {
2763 int routesAdded = 0;
2764 Lock<Mutex> l(_lock);
2765 // We can't check this until we have the lock
2766 if (subscribeGeneration_ > 0
2767 && subscribeGeneration_ <= _subscriptionManagerGenerationCount)
2768 {
2769 message_.setAckTypeEnum(requestedAcks);
2770 return std::string();
2771 }
2772 if (!subId.empty() && messageHandler_.isValid())
2773 {
2774 if (!_routes.hasRoute(subId))
2775 {
2776 ++routesAdded;
2777 }
2778 // This can replace a non-subscribe with a matching id
2779 // with a subscription but not another subscription.
2780 _routes.addRoute(subId, messageHandler_, requestedAcks,
2781 systemAddedAcks, commandType);
2782 }
2783 if (!isSubscribeOnly && !qid.empty()
2784 && messageHandler_.isValid() && qid != subId)
2785 {
2786 if (routesAdded == 0)
2787 {
2788 _routes.addRoute(qid, messageHandler_,
2789 requestedAcks, systemAddedAcks, commandType);
2790 }
2791 else
2792 {
2793 void* data = NULL;
2794 {
2795 Unlock<Mutex> u(_lock);
2796 data = amps_invoke_copy_route_function(
2797 messageHandler_.userData());
2798 }
2799 if (!data)
2800 {
2801 _routes.addRoute(qid, messageHandler_, requestedAcks,
2802 systemAddedAcks, commandType);
2803 }
2804 else
2805 {
2806 _routes.addRoute(qid,
2807 MessageHandler(messageHandler_.function(),
2808 data),
2809 requestedAcks, systemAddedAcks, commandType);
2810 }
2811 }
2812 ++routesAdded;
2813 }
2814 if (!id.empty() && messageHandler_.isValid()
2815 && requestedAcks & ~Message::AckType::Persisted
2816 && id != subId && id != qid)
2817 {
2818 if (routesAdded == 0)
2819 {
2820 _routes.addRoute(id, messageHandler_, requestedAcks,
2821 systemAddedAcks, commandType);
2822 }
2823 else
2824 {
2825 void* data = NULL;
2826 {
2827 Unlock<Mutex> u(_lock);
2828 data = amps_invoke_copy_route_function(
2829 messageHandler_.userData());
2830 }
2831 if (!data)
2832 {
2833 _routes.addRoute(id, messageHandler_, requestedAcks,
2834 systemAddedAcks, commandType);
2835 }
2836 else
2837 {
2838 _routes.addRoute(id,
2839 MessageHandler(messageHandler_.function(),
2840 data),
2841 requestedAcks,
2842 systemAddedAcks, commandType);
2843 }
2844 }
2845 ++routesAdded;
2846 }
2847 try
2848 {
2849 AckResponse resp = syncAckProcessing(timeout_, message_, 0, subscribeGeneration_);
2850 message_.setAckTypeEnum(requestedAcks);
2851 if (resp.getConnectionVersion() == UINT_MAX)
2852 {
2853 // Special case where message wasn't sent
2854 // Don't remove subId/qid routes as next try may have already
2855 // registered since an unlock was possible, but we can
2856 // remove cid if unique, as it could be used for acks.
2857 if (id != subId
2858 && id != qid)
2859 {
2860 _routes.removeRoute(id);
2861 }
2862 }
2863 }
2864 catch (...)
2865 {
2866 _routes.removeRoute(message_.getQueryID());
2867 _routes.removeRoute(message_.getSubscriptionId());
2868 _routes.removeRoute(id);
2869 message_.setAckTypeEnum(requestedAcks);
2870 throw;
2871 }
2872 }
2873 break;
2874 // These are valid commands that are used as-is
2875 case Message::Command::Unsubscribe:
2876 case Message::Command::Heartbeat:
2877 case Message::Command::Logon:
2878 case Message::Command::StartTimer:
2879 case Message::Command::StopTimer:
2880 case Message::Command::SOWDelete:
2881 {
2882 Lock<Mutex> l(_lock);
2883 // if an ack is requested, it'll need a command ID.
2884 if (message_.getAckTypeEnum() != Message::AckType::None)
2885 {
2886 if (id.empty())
2887 {
2888 message_.newCommandId();
2889 id = message_.getCommandId();
2890 }
2891 if (messageHandler_.isValid())
2892 {
2893 _routes.addRoute(id, messageHandler_, requestedAcks,
2894 Message::AckType::None, commandType);
2895 }
2896 }
2897 _send(message_);
2898 }
2899 break;
2900 case Message::Command::DeltaPublish:
2901 case Message::Command::Publish:
2902 {
2903 bool useSync = message_.getFilter().len() > 0;
2904 Lock<Mutex> l(_lock);
2905 // if an ack is requested, it'll need a command ID.
2906 unsigned ackType = message_.getAckTypeEnum();
2907 if (ackType != Message::AckType::None
2908 || useSync)
2909 {
2910 if (id.empty())
2911 {
2912 message_.newCommandId();
2913 id = message_.getCommandId();
2914 }
2915 if (messageHandler_.isValid())
2916 {
2917 _routes.addRoute(id, messageHandler_, requestedAcks,
2918 Message::AckType::None, commandType);
2919 }
2920 }
2921 if (useSync)
2922 {
2923 message_.setAckTypeEnum(ackType | Message::AckType::Processed);
2924 syncAckProcessing(timeout_, message_, 0, 0);
2925 }
2926 else
2927 {
2928 _sendSequenced(message_, 0, 1);
2929 }
2930 }
2931 break;
2932 // These are things that shouldn't be sent (not meaningful)
2933 case Message::Command::GroupBegin:
2934 case Message::Command::GroupEnd:
2935 case Message::Command::OOF:
2936 case Message::Command::Ack:
2937 case Message::Command::Unknown:
2938 default:
2939 throw CommandException("Command type " + message_.getCommand() + " can not be sent directly to AMPS");
2940 }
2941 message_.setAckTypeEnum(requestedAcks);
2942 return id;
2943 }
2944
2945 void setDisconnectHandler(const DisconnectHandler& disconnectHandler)
2946 {
2947 Lock<Mutex> l(_lock);
2948 _disconnectHandler = disconnectHandler;
2949 }
2950
2951 void setGlobalCommandTypeMessageHandler(const std::string& command_, const MessageHandler& handler_)
2952 {
2953 switch (command_[0])
2954 {
2955 #if 0 // Not currently implemented to avoid an extra branch in delivery
2956 case 'p':
2957 _globalCommandTypeHandlers[GlobalCommandTypeHandlers::Publish] = handler_;
2958 break;
2959 case 's':
2960 _globalCommandTypeHandlers[GlobalCommandTypeHandlers::SOW] = handler_;
2961 break;
2962 #endif
2963 case 'h':
2964 _globalCommandTypeHandlers[GlobalCommandTypeHandlers::Heartbeat] = handler_;
2965 break;
2966 #if 0 // Not currently implemented to avoid an extra branch in delivery
2967 case 'g':
2968 if (command_[6] == 'b')
2969 {
2970 _globalCommandTypeHandlers[GlobalCommandTypeHandlers::GroupBegin] = handler_;
2971 }
2972 else if (command_[6] == 'e')
2973 {
2974 _globalCommandTypeHandlers[GlobalCommandTypeHandlers::GroupEnd] = handler_;
2975 }
2976 else
2977 {
2978 std::ostringstream os;
2979 os << "Invalid command '" << command_ << "' passed to setGlobalCommandTypeHandler";
2980 throw CommandException(os.str());
2981 }
2982 break;
2983 case 'o':
2984 _globalCommandTypeHandlers[GlobalCommandTypeHandlers::OOF] = handler_;
2985 break;
2986 #endif
2987 case 'a':
2988 _globalCommandTypeHandlers[GlobalCommandTypeHandlers::Ack] = handler_;
2989 break;
2990 case 'l':
2991 case 'L':
2992 _globalCommandTypeHandlers[GlobalCommandTypeHandlers::LastChance] = handler_;
2993 break;
2994 case 'd':
2995 case 'D':
2996 _globalCommandTypeHandlers[GlobalCommandTypeHandlers::DuplicateMessage] = handler_;
2997 break;
2998 default:
2999 std::ostringstream os;
3000 os << "Invalid command '" << command_ << "' passed to setGlobalCommandTypeHandler";
3001 throw CommandException(os.str());
3002 break;
3003 }
3004 }
3005
3006 void setGlobalCommandTypeMessageHandler(const Message::Command::Type command_, const MessageHandler& handler_)
3007 {
3008 switch (command_)
3009 {
3010 #if 0 // Not currently implemented to avoid an extra branch in delivery
3011 case Message::Command::Publish:
3012 _globalCommandTypeHandlers[GlobalCommandTypeHandlers::Publish] = handler_;
3013 break;
3014 case Message::Command::SOW:
3015 _globalCommandTypeHandlers[GlobalCommandTypeHandlers::SOW] = handler_;
3016 break;
3017 #endif
3018 case Message::Command::Heartbeat:
3019 _globalCommandTypeHandlers[GlobalCommandTypeHandlers::Heartbeat] = handler_;
3020 break;
3021 #if 0 // Not currently implemented to avoid an extra branch in delivery
3022 case Message::Command::GroupBegin:
3023 _globalCommandTypeHandlers[GlobalCommandTypeHandlers::GroupBegin] = handler_;
3024 break;
3025 case Message::Command::GroupEnd:
3026 _globalCommandTypeHandlers[GlobalCommandTypeHandlers::GroupEnd] = handler_;
3027 break;
3028 case Message::Command::OOF:
3029 _globalCommandTypeHandlers[GlobalCommandTypeHandlers::OOF] = handler_;
3030 break;
3031 #endif
3032 case Message::Command::Ack:
3033 _globalCommandTypeHandlers[GlobalCommandTypeHandlers::Ack] = handler_;
3034 break;
3035 default:
3036 unsigned bits = 0;
3037 unsigned command = command_;
3038 while (command > 0)
3039 {
3040 ++bits;
3041 command >>= 1;
3042 }
3043 char errBuf[128];
3044 AMPS_snprintf(errBuf, sizeof(errBuf),
3045 "Invalid command '%.*s' passed to setGlobalCommandTypeHandler",
3046 CommandConstants<0>::Lengths[bits],
3047 CommandConstants<0>::Values[bits]);
3048 throw CommandException(errBuf);
3049 break;
3050 }
3051 }
3052
3053 void setGlobalCommandTypeMessageHandler(const GlobalCommandTypeHandlers handlerType_, const MessageHandler& handler_)
3054 {
3055 _globalCommandTypeHandlers[handlerType_] = handler_;
3056 }
3057
3058 void setFailedWriteHandler(FailedWriteHandler* handler_)
3059 {
3060 Lock<Mutex> l(_lock);
3061 _failedWriteHandler.reset(handler_);
3062 }
3063
3064 void setPublishStore(const Store& publishStore_)
3065 {
3066 Lock<Mutex> l(_lock);
3067 if (_connected)
3068 {
3069 throw AlreadyConnectedException("Setting a publish store on a connected client is undefined behavior");
3070 }
3071 _publishStore = publishStore_;
3072 }
3073
3074 void setBookmarkStore(const BookmarkStore& bookmarkStore_)
3075 {
3076 Lock<Mutex> l(_lock);
3077 if (_connected)
3078 {
3079 throw AlreadyConnectedException("Setting a bookmark store on a connected client is undefined behavior");
3080 }
3081 _bookmarkStore = bookmarkStore_;
3082 }
3083
3084 void setSubscriptionManager(SubscriptionManager* subscriptionManager_)
3085 {
3086 Lock<Mutex> l(_lock);
3087 if (_connected)
3088 {
3089 throw AlreadyConnectedException("Setting a subscription manager on a connected client is undefined behavior");
3090 }
3091 _subscriptionManager.reset(subscriptionManager_);
3092 captureSubscriptionManagerGenerationCount();
3093 }
3094
3095 SubscriptionManager* getSubscriptionManager() const
3096 {
3097 return const_cast<SubscriptionManager*>(_subscriptionManager.get());
3098 }
3099
3100 DisconnectHandler getDisconnectHandler() const
3101 {
3102 return _disconnectHandler;
3103 }
3104
3105 MessageHandler getDuplicateMessageHandler() const
3106 {
3107 return _globalCommandTypeHandlers[GlobalCommandTypeHandlers::DuplicateMessage];
3108 }
3109
3110 FailedWriteHandler* getFailedWriteHandler() const
3111 {
3112 return const_cast<FailedWriteHandler*>(_failedWriteHandler.get());
3113 }
3114
3115 Store getPublishStore() const
3116 {
3117 return _publishStore;
3118 }
3119
3120 BookmarkStore getBookmarkStore() const
3121 {
3122 return _bookmarkStore;
3123 }
3124
3125 amps_uint64_t publish(const char* topic_, size_t topicLen_, const char* data_, size_t dataLen_)
3126 {
3127 threadLocalMessage.reset();
3128 threadLocalMessage.setCommandEnum(Message::Command::Publish);
3129 if (!_publishStore.isValid())
3130 {
3131 threadLocalMessage.assignTopic(topic_, topicLen_);
3132 threadLocalMessage.assignData(data_, dataLen_);
3133 Lock<Mutex> l(_lock);
3134 _sendSequenced(threadLocalMessage, 0, 1);
3135 return 0;
3136 }
3137 else
3138 {
3139 return _publish(topic_, topicLen_, data_, dataLen_);
3140 }
3141 }
3142
3143 amps_uint64_t publish(const char* topic_, size_t topicLen_, const char* data_,
3144 size_t dataLen_, unsigned long expiration_)
3145 {
3146 threadLocalMessage.reset();
3147 char exprBuf[AMPS_NUMBER_BUFFER_LEN];
3148 size_t exprPos = convertToCharArray(exprBuf, expiration_);
3149 threadLocalMessage.setCommandEnum(Message::Command::Publish)
3150 .assignExpiration(exprBuf + exprPos,
3151 AMPS_NUMBER_BUFFER_LEN - exprPos);
3152 if (!_publishStore.isValid())
3153 {
3154 threadLocalMessage.assignTopic(topic_, topicLen_);
3155 threadLocalMessage.assignData(data_, dataLen_);
3156 Lock<Mutex> l(_lock);
3157 _sendSequenced(threadLocalMessage, 0, 1);
3158 return 0;
3159 }
3160 else
3161 {
3162 return _publish(topic_, topicLen_, data_, dataLen_);
3163 }
3164 }
3165
3166 class FlushAckHandler : ConnectionStateListener
3167 {
3168 private:
3169 ClientImpl* _pClient;
3170 Field _cmdId;
3171 #if __cplusplus >= 201100L || _MSC_VER >= 1900
3172 std::atomic<bool> _acked;
3173 std::atomic<bool> _disconnected;
3174 #else
3175 volatile bool _acked;
3176 volatile bool _disconnected;
3177 #endif
3178 public:
3179 FlushAckHandler(ClientImpl* pClient_)
3180 : _pClient(pClient_), _cmdId(), _acked(false), _disconnected(false)
3181 {
3182 pClient_->addConnectionStateListener(this);
3183 }
3184 ~FlushAckHandler()
3185 {
3186 _pClient->removeConnectionStateListener(this);
3187 _pClient->removeMessageHandler(_cmdId);
3188 _cmdId.clear();
3189 }
3190 void setCommandId(const Field& cmdId_)
3191 {
3192 _cmdId.deepCopy(cmdId_);
3193 }
3194 void invoke(const Message&)
3195 {
3196 _acked = true;
3197 }
3198 void connectionStateChanged(State state_)
3199 {
3200 if (state_ <= Shutdown)
3201 {
3202 _disconnected = true;
3203 }
3204 }
3205 bool acked()
3206 {
3207 return _acked;
3208 }
3209 bool done()
3210 {
3211 return _acked || _disconnected;
3212 }
3213 };
3214
3215 void publishFlush(long timeout_, unsigned ackType_)
3216 {
3217 static const char* processed = "processed"; // -V1096
3218 static const size_t processedLen = strlen(processed); // -V1096
3219 static const char* persisted = "persisted"; // -V1096
3220 static const size_t persistedLen = strlen(persisted); // -V1096
3221 static const char* flush = "flush"; // -V1096
3222 static const size_t flushLen = strlen(flush); // -V1096
3223 static const VersionInfo minPersisted("5.3.3.0"); // -V1096
3224 static const VersionInfo minFlush("4"); // -V1096
3225 if (ackType_ != Message::AckType::Processed
3226 && ackType_ != Message::AckType::Persisted)
3227 {
3228 throw CommandException("Flush can only be used with processed or persisted acks.");
3229 }
3230 FlushAckHandler flushHandler(this);
3231 if (_serverVersion >= minFlush)
3232 {
3233 Lock<Mutex> l(_lock);
3234 if (!_connected)
3235 {
3236 throw DisconnectedException("Not connected trying to flush");
3237 }
3238 threadLocalMessage.reset();
3239 threadLocalMessage.newCommandId();
3240 threadLocalMessage.assignCommand(flush, flushLen);
3241 if (_serverVersion < minPersisted
3242 || ackType_ == Message::AckType::Processed)
3243 {
3244 threadLocalMessage.assignAckType(processed, processedLen);
3245 }
3246 else
3247 {
3248 threadLocalMessage.assignAckType(persisted, persistedLen);
3249 }
3250 flushHandler.setCommandId(threadLocalMessage.getCommandId());
3251 addMessageHandler(threadLocalMessage.getCommandId(),
3252 std::bind(&FlushAckHandler::invoke,
3253 std::ref(flushHandler),
3254 std::placeholders::_1),
3255 ackType_, threadLocalMessage.getCommandEnum());
3256 NoDelay noDelay(_client);
3257 _send(threadLocalMessage);
3258 }
3259 if (_publishStore.isValid())
3260 {
3261 try
3262 {
3263 _publishStore.flush(timeout_);
3264 }
3265 catch (const AMPSException& ex)
3266 {
3267 AMPS_UNHANDLED_EXCEPTION(ex);
3268 throw;
3269 }
3270 }
3271 else if (_serverVersion < minFlush)
3272 {
3273 if (timeout_ > 0)
3274 {
3275 AMPS_USLEEP(timeout_ * 1000);
3276 }
3277 else
3278 {
3279 AMPS_USLEEP(1000 * 1000);
3280 }
3281 return;
3282 }
3283 if (timeout_)
3284 {
3285 Timer timer((double)timeout_);
3286 timer.start();
3287 while (!timer.check() && !flushHandler.done())
3288 {
3289 AMPS_USLEEP(10000);
3290 amps_invoke_waiting_function();
3291 }
3292 }
3293 else
3294 {
3295 while (!flushHandler.done())
3296 {
3297 AMPS_USLEEP(10000);
3298 amps_invoke_waiting_function();
3299 }
3300 }
3301 // No response or disconnect in timeout interval
3302 if (!flushHandler.done())
3303 {
3304 throw TimedOutException("Timed out waiting for flush");
3305 }
3306 // We got disconnected and there is no publish store
3307 if (!flushHandler.acked() && !_publishStore.isValid())
3308 {
3309 throw DisconnectedException("Disconnected waiting for flush");
3310 }
3311 }
3312
3313 amps_uint64_t deltaPublish(const char* topic_, size_t topicLength_,
3314 const char* data_, size_t dataLength_)
3315 {
3316 threadLocalMessage.reset();
3317 threadLocalMessage.setCommandEnum(Message::Command::DeltaPublish);
3318 if (!_publishStore.isValid())
3319 {
3320 threadLocalMessage.assignTopic(topic_, topicLength_);
3321 threadLocalMessage.assignData(data_, dataLength_);
3322 Lock<Mutex> l(_lock);
3323 _sendSequenced(threadLocalMessage, 0, 1);
3324 return 0;
3325 }
3326 else
3327 {
3328 return _publish(topic_, topicLength_, data_, dataLength_);
3329 }
3330 }
3331
3332 amps_uint64_t deltaPublish(const char* topic_, size_t topicLength_,
3333 const char* data_, size_t dataLength_,
3334 unsigned long expiration_)
3335 {
3336 threadLocalMessage.reset();
3337 char exprBuf[AMPS_NUMBER_BUFFER_LEN];
3338 size_t exprPos = convertToCharArray(exprBuf, expiration_);
3339 threadLocalMessage.setCommandEnum(Message::Command::DeltaPublish)
3340 .assignExpiration(exprBuf + exprPos,
3341 AMPS_NUMBER_BUFFER_LEN - exprPos);
3342 if (!_publishStore.isValid())
3343 {
3344 threadLocalMessage.assignTopic(topic_, topicLength_);
3345 threadLocalMessage.assignData(data_, dataLength_);
3346 Lock<Mutex> l(_lock);
3347 _sendSequenced(threadLocalMessage, 0, 1);
3348 return 0;
3349 }
3350 else
3351 {
3352 return _publish(topic_, topicLength_, data_, dataLength_);
3353 }
3354 }
3355
3356 amps_uint64_t _publish(const char* topic_, size_t topicLength_,
3357 const char* data_, size_t dataLength_)
3358 {
3359 threadLocalMessage.assignTopic(topic_, topicLength_)
3360 .setAckTypeEnum(Message::AckType::Persisted)
3361 .assignData(data_, dataLength_);
3362 amps_uint64_t haSequenceNumber = _publishStore.store(threadLocalMessage);
3363 char buf[AMPS_NUMBER_BUFFER_LEN];
3364 size_t pos = convertToCharArray(buf, haSequenceNumber);
3365 threadLocalMessage.assignSequence(buf + pos, AMPS_NUMBER_BUFFER_LEN - pos);
3366 {
3367 Lock<Mutex> l(_lock);
3368 _sendSequenced(threadLocalMessage, haSequenceNumber, 1);
3369 }
3370 return haSequenceNumber;
3371 }
3372
3373 virtual std::string logon(long timeout_, Authenticator& authenticator_,
3374 const char* options_ = NULL);
3375
3376 virtual std::string _logon(long timeout_, Authenticator& authenticator_,
3377 const char* options_ = NULL)
3378 {
3379 // _lock is already acquired
3380 threadLocalMessage.reset();
3381 threadLocalMessage.newCommandId();
3382 std::string newCommandId = threadLocalMessage.getCommandId();
3383 threadLocalMessage.setCommandEnum(Message::Command::Logon);
3384 threadLocalMessage.setClientName(_name);
3385 #ifdef AMPS_CLIENT_VERSION_WITH_LANGUAGE
3386 threadLocalMessage.assignVersion(AMPS_CLIENT_VERSION_WITH_LANGUAGE,
3387 strlen(AMPS_CLIENT_VERSION_WITH_LANGUAGE));
3388 #endif
3389 URI uri(_lastUri);
3390 if (uri.user().size())
3391 {
3392 threadLocalMessage.setUserId(uri.user());
3393 }
3394 if (uri.password().size())
3395 {
3396 threadLocalMessage.setPassword(uri.password());
3397 }
3398 if (uri.protocol() == "amps" && uri.messageType().size())
3399 {
3400 threadLocalMessage.setMessageType(uri.messageType());
3401 }
3402 if (uri.isTrue("pretty"))
3403 {
3404 threadLocalMessage.setOptions("pretty");
3405 }
3406
3407 threadLocalMessage.setPassword(authenticator_.authenticate(threadLocalMessage.getUserId(), threadLocalMessage.getPassword()));
3408 if (!_logonCorrelationData.empty())
3409 {
3410 threadLocalMessage.assignCorrelationId(_logonCorrelationData);
3411 }
3412 if (options_)
3413 {
3414 threadLocalMessage.setOptions(options_);
3415 }
3416 _username = threadLocalMessage.getUserId();
3417 try
3418 {
3419 AtomicFlagFlip pubFlip(&_logonInProgress);
3420 NoDelay noDelay(_client);
3421 while (true)
3422 {
3423 threadLocalMessage.setAckTypeEnum(Message::AckType::Processed);
3424 AckResponse ack = syncAckProcessing(timeout_, threadLocalMessage);
3425 if (ack.status() == "retry")
3426 {
3427 threadLocalMessage.setPassword(authenticator_.retry(ack.username(), ack.password()));
3428 _username = ack.username();
3429 threadLocalMessage.setUserId(_username);
3430 }
3431 else
3432 {
3433 authenticator_.completed(ack.username(), ack.password(), ack.reason());
3434 break;
3435 }
3436 }
3437 broadcastConnectionStateChanged(ConnectionStateListener::LoggedOn);
3438
3439 // Now re-send the heartbeat command if configured
3440 _sendHeartbeat();
3441 // Signal any threads waiting for _logonInProgress
3442 _lock.signalAll();
3443 }
3444 catch (const AMPSException& ex)
3445 {
3446 {
3447 Unlock<Mutex> u(_lock);
3448 setDisconnected();
3449 }
3450 _lock.signalAll();
3451 AMPS_UNHANDLED_EXCEPTION(ex);
3452 throw;
3453 }
3454 catch (...)
3455 {
3456 {
3457 Unlock<Mutex> u(_lock);
3458 setDisconnected();
3459 }
3460 _lock.signalAll();
3461 throw;
3462 }
3463
3464 if (_publishStore.isValid())
3465 {
3466 try
3467 {
3468 _publishStore.replay(_replayer);
3469 broadcastConnectionStateChanged(ConnectionStateListener::PublishReplayed);
3470 }
3471 catch (const PublishStoreGapException& ex)
3472 {
3473 {
3474 Unlock<Mutex> u(_lock);
3475 setDisconnected();
3476 }
3477 _lock.signalAll();
3478 AMPS_UNHANDLED_EXCEPTION(ex);
3479 throw;
3480 }
3481 catch (const StoreException& ex)
3482 {
3483 {
3484 Unlock<Mutex> u(_lock);
3485 setDisconnected();
3486 }
3487 _lock.signalAll();
3488 std::ostringstream os;
3489 os << "A local store exception occurred while logging on."
3490 << ex.toString();
3491 throw ConnectionException(os.str());
3492 }
3493 catch (const AMPSException& ex)
3494 {
3495 {
3496 Unlock<Mutex> u(_lock);
3497 setDisconnected();
3498 }
3499 _lock.signalAll();
3500 AMPS_UNHANDLED_EXCEPTION(ex);
3501 throw;
3502 }
3503 catch (const std::exception& ex)
3504 {
3505 {
3506 Unlock<Mutex> u(_lock);
3507 setDisconnected();
3508 }
3509 _lock.signalAll();
3510 AMPS_UNHANDLED_EXCEPTION(ex);
3511 throw;
3512 }
3513 catch (...)
3514 {
3515 {
3516 Unlock<Mutex> u(_lock);
3517 setDisconnected();
3518 }
3519 _lock.signalAll();
3520 throw;
3521 }
3522 }
3523 _lock.signalAll();
3524 return newCommandId;
3525 }
3526
3527 std::string subscribe(const MessageHandler& messageHandler_,
3528 const std::string& topic_,
3529 long timeout_,
3530 const std::string& filter_,
3531 const std::string& bookmark_,
3532 const std::string& options_,
3533 const std::string& subId_,
3534 bool isHASubscribe_ = true)
3535 {
3536 isHASubscribe_ &= (bool)_subscriptionManager;
3537 Lock<Mutex> l(_lock);
3538 threadLocalMessage.reset();
3539 threadLocalMessage.setCommandEnum(Message::Command::Subscribe);
3540 threadLocalMessage.newCommandId();
3541 std::string subId(subId_);
3542 if (subId.empty())
3543 {
3544 if (options_.find(AMPS_OPTIONS_REPLACE, 0, strlen(AMPS_OPTIONS_REPLACE) - 1) != std::string::npos)
3545 {
3546 throw ConnectionException("Cannot issue a replacement subscription; a valid subscription id is required.");
3547 }
3548
3549 subId = threadLocalMessage.getCommandId();
3550 }
3551 threadLocalMessage.setSubscriptionId(subId);
3552 AMPS::Message::Field subIdField = threadLocalMessage.getSubscriptionId();
3553 unsigned ackTypes = Message::AckType::Processed;
3554
3555 if (!bookmark_.empty() && _bookmarkStore.isValid())
3556 {
3557 ackTypes |= Message::AckType::Persisted;
3558 }
3559 threadLocalMessage.setTopic(topic_);
3560
3561 if (filter_.length())
3562 {
3563 threadLocalMessage.setFilter(filter_);
3564 }
3565 if (bookmark_.length())
3566 {
3567 if (bookmark_ == AMPS_BOOKMARK_RECENT)
3568 {
3569 // The returned Field is a deep copy, so use assign to get it cleared
3570 Message::Field mostRecent = _bookmarkStore.getMostRecent(subIdField);
3571 threadLocalMessage.assignOwnershipBookmark(mostRecent);
3572 }
3573 else
3574 {
3575 threadLocalMessage.setBookmark(bookmark_);
3576 if (_bookmarkStore.isValid())
3577 {
3578 if (bookmark_ != AMPS_BOOKMARK_NOW &&
3579 bookmark_ != AMPS_BOOKMARK_EPOCH)
3580 {
3581 _bookmarkStore.log(threadLocalMessage);
3582 _bookmarkStore.discard(threadLocalMessage);
3583 _bookmarkStore.persisted(subIdField, threadLocalMessage.getBookmark());
3584 }
3585 }
3586 }
3587 }
3588 if (options_.length())
3589 {
3590 threadLocalMessage.setOptions(options_);
3591 }
3592
3593 Message message = threadLocalMessage;
3594 AMPS_ATOMIC_BASE_TYPE generation = (AMPS_ATOMIC_BASE_TYPE)0;
3595 if (isHASubscribe_)
3596 {
3597 message = threadLocalMessage.deepCopy();
3598 {
3599 Unlock<Mutex> u(_lock);
3600 generation =
3601 _subscriptionManager->subscribe(messageHandler_, message,
3602 Message::AckType::None);
3603 }
3604 if (generation <= (AMPS_ATOMIC_BASE_TYPE)_subscriptionManagerGenerationCount)
3605 {
3606 return subId;
3607 }
3608 }
3609 if (!_routes.hasRoute(message.getSubscriptionId()))
3610 {
3611 _routes.addRoute(message.getSubscriptionId(), messageHandler_,
3612 Message::AckType::None, ackTypes, message.getCommandEnum());
3613 }
3614 message.setAckTypeEnum(ackTypes);
3615 if (!options_.empty())
3616 {
3617 message.setOptions(options_);
3618 }
3619 try
3620 {
3621 syncAckProcessing(timeout_, message, 0, generation);
3622 }
3623 catch (const DisconnectedException&)
3624 {
3625 if (!isHASubscribe_)
3626 {
3627 _routes.removeRoute(subIdField);
3628 throw;
3629 }
3630 else
3631 {
3632 AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(subIdField));
3633 throw;
3634 }
3635 }
3636 catch (const TimedOutException&)
3637 {
3638 AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(subIdField));
3639 throw;
3640 }
3641 catch (...)
3642 {
3643 if (isHASubscribe_)
3644 {
3645 // Have to unlock before calling into sub manager to avoid deadlock
3646 Unlock<Mutex> unlock(_lock);
3647 _subscriptionManager->unsubscribe(subIdField);
3648 }
3649 _routes.removeRoute(subIdField);
3650 throw;
3651 }
3652
3653 return subId;
3654 }
3655 std::string deltaSubscribe(const MessageHandler& messageHandler_,
3656 const std::string& topic_,
3657 long timeout_,
3658 const std::string& filter_,
3659 const std::string& bookmark_,
3660 const std::string& options_,
3661 const std::string& subId_ = "",
3662 bool isHASubscribe_ = true)
3663 {
3664 isHASubscribe_ &= (bool)_subscriptionManager;
3665 Lock<Mutex> l(_lock);
3666 threadLocalMessage.reset();
3667 threadLocalMessage.setCommandEnum(Message::Command::DeltaSubscribe);
3668 threadLocalMessage.newCommandId();
3669 std::string subId(subId_);
3670 if (subId.empty())
3671 {
3672 subId = threadLocalMessage.getCommandId();
3673 }
3674 threadLocalMessage.setSubscriptionId(subId);
3675 AMPS::Message::Field subIdField = threadLocalMessage.getSubscriptionId();
3676 unsigned ackTypes = Message::AckType::Processed;
3677
3678 if (!bookmark_.empty() && _bookmarkStore.isValid())
3679 {
3680 ackTypes |= Message::AckType::Persisted;
3681 }
3682 threadLocalMessage.setTopic(topic_);
3683 if (filter_.length())
3684 {
3685 threadLocalMessage.setFilter(filter_);
3686 }
3687 if (bookmark_.length())
3688 {
3689 if (bookmark_ == AMPS_BOOKMARK_RECENT)
3690 {
3691 // The returned Field is a deep copy, so use assign to get it cleared
3692 Message::Field mostRecent = _bookmarkStore.getMostRecent(subIdField);
3693 threadLocalMessage.assignOwnershipBookmark(mostRecent);
3694 }
3695 else
3696 {
3697 threadLocalMessage.setBookmark(bookmark_);
3698 if (_bookmarkStore.isValid())
3699 {
3700 if (bookmark_ != AMPS_BOOKMARK_NOW &&
3701 bookmark_ != AMPS_BOOKMARK_EPOCH)
3702 {
3703 _bookmarkStore.log(threadLocalMessage);
3704 _bookmarkStore.discard(threadLocalMessage);
3705 _bookmarkStore.persisted(subIdField, threadLocalMessage.getBookmark());
3706 }
3707 }
3708 }
3709 }
3710 if (options_.length())
3711 {
3712 threadLocalMessage.setOptions(options_);
3713 }
3714 AMPS_ATOMIC_BASE_TYPE generation = (AMPS_ATOMIC_BASE_TYPE)0;
3715 if (isHASubscribe_)
3716 {
3717 {
3718 Unlock<Mutex> u(_lock);
3719 generation =
3720 _subscriptionManager->subscribe(messageHandler_, threadLocalMessage.deepCopy(),
3721 Message::AckType::None);
3722 }
3723 if (generation <= (AMPS_ATOMIC_BASE_TYPE)_subscriptionManagerGenerationCount)
3724 {
3725 return subId;
3726 }
3727 }
3728 if (!_routes.hasRoute(threadLocalMessage.getSubscriptionId()))
3729 {
3730 _routes.addRoute(threadLocalMessage.getSubscriptionId(), messageHandler_,
3731 Message::AckType::None, ackTypes, threadLocalMessage.getCommandEnum());
3732 }
3733 threadLocalMessage.setAckTypeEnum(ackTypes);
3734 if (!options_.empty())
3735 {
3736 threadLocalMessage.setOptions(options_);
3737 }
3738 try
3739 {
3740 syncAckProcessing(timeout_, threadLocalMessage, 0, generation);
3741 }
3742 catch (const DisconnectedException&)
3743 {
3744 if (!isHASubscribe_)
3745 {
3746 _routes.removeRoute(subIdField);
3747 throw;
3748 }
3749 }
3750 catch (const TimedOutException&)
3751 {
3752 AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(subIdField));
3753 throw;
3754 }
3755 catch (...)
3756 {
3757 if (isHASubscribe_)
3758 {
3759 // Have to unlock before calling into sub manager to avoid deadlock
3760 Unlock<Mutex> unlock(_lock);
3761 _subscriptionManager->unsubscribe(subIdField);
3762 }
3763 _routes.removeRoute(subIdField);
3764 throw;
3765 }
3766 return subId;
3767 }
3768
3769 void unsubscribe(const std::string& id)
3770 {
3771 Lock<Mutex> l(_lock);
3772 unsubscribeInternal(id);
3773 }
3774
3775 void unsubscribe(void)
3776 {
3777 if (_subscriptionManager)
3778 {
3779 _subscriptionManager->clear();
3780 }
3781 {
3782 _routes.unsubscribeAll();
3783 Lock<Mutex> l(_lock);
3784 threadLocalMessage.reset();
3785 threadLocalMessage.setCommandEnum(Message::Command::Unsubscribe);
3786 threadLocalMessage.newCommandId();
3787 threadLocalMessage.setSubscriptionId("all");
3788 _sendWithoutRetry(threadLocalMessage);
3789 }
3790 deferredExecution(&amps_noOpFn, NULL);
3791 }
3792
3793 std::string sow(const MessageHandler& messageHandler_,
3794 const std::string& topic_,
3795 const std::string& filter_ = "",
3796 const std::string& orderBy_ = "",
3797 const std::string& bookmark_ = "",
3798 int batchSize_ = AMPS_DEFAULT_BATCH_SIZE,
3799 int topN_ = AMPS_DEFAULT_TOP_N,
3800 const std::string& options_ = "",
3801 long timeout_ = AMPS_DEFAULT_COMMAND_TIMEOUT)
3802 {
3803 Lock<Mutex> l(_lock);
3804 threadLocalMessage.reset();
3805 threadLocalMessage.setCommandEnum(Message::Command::SOW);
3806 threadLocalMessage.newCommandId();
3807 // need to keep our own copy of the command ID.
3808 std::string commandId = threadLocalMessage.getCommandId();
3809 threadLocalMessage.setQueryID(threadLocalMessage.getCommandId());
3810 unsigned ackTypes = Message::AckType::Processed | Message::AckType::Completed;
3811 threadLocalMessage.setAckTypeEnum(ackTypes);
3812 threadLocalMessage.setTopic(topic_);
3813 if (filter_.length())
3814 {
3815 threadLocalMessage.setFilter(filter_);
3816 }
3817 if (orderBy_.length())
3818 {
3819 threadLocalMessage.setOrderBy(orderBy_);
3820 }
3821 if (bookmark_.length())
3822 {
3823 threadLocalMessage.setBookmark(bookmark_);
3824 }
3825 threadLocalMessage.setBatchSize(AMPS::asString(batchSize_));
3826 if (topN_ != AMPS_DEFAULT_TOP_N)
3827 {
3828 threadLocalMessage.setTopNRecordsReturned(AMPS::asString(topN_));
3829 }
3830 if (options_.length())
3831 {
3832 threadLocalMessage.setOptions(options_);
3833 }
3834
3835 _routes.addRoute(threadLocalMessage.getQueryID(), messageHandler_,
3836 Message::AckType::None, ackTypes, threadLocalMessage.getCommandEnum());
3837
3838 try
3839 {
3840 syncAckProcessing(timeout_, threadLocalMessage);
3841 }
3842 catch (...)
3843 {
3844 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(commandId));
3845 throw;
3846 }
3847
3848 return commandId;
3849 }
3850
3851 std::string sow(const MessageHandler& messageHandler_,
3852 const std::string& topic_,
3853 long timeout_,
3854 const std::string& filter_ = "",
3855 int batchSize_ = AMPS_DEFAULT_BATCH_SIZE,
3856 int topN_ = AMPS_DEFAULT_TOP_N)
3857 {
3858 std::string notSet;
3859 return sow(messageHandler_,
3860 topic_,
3861 filter_,
3862 notSet, // orderBy
3863 notSet, // bookmark
3864 batchSize_,
3865 topN_,
3866 notSet,
3867 timeout_);
3868 }
3869
3870 std::string sowAndSubscribe(const MessageHandler& messageHandler_,
3871 const std::string& topic_,
3872 const std::string& filter_ = "",
3873 const std::string& orderBy_ = "",
3874 const std::string& bookmark_ = "",
3875 int batchSize_ = AMPS_DEFAULT_BATCH_SIZE,
3876 int topN_ = AMPS_DEFAULT_TOP_N,
3877 const std::string& options_ = "",
3878 long timeout_ = AMPS_DEFAULT_COMMAND_TIMEOUT,
3879 bool isHASubscribe_ = true)
3880 {
3881 isHASubscribe_ &= (bool)_subscriptionManager;
3882 unsigned ackTypes = Message::AckType::Processed;
3883 Lock<Mutex> l(_lock);
3884 threadLocalMessage.reset();
3885 threadLocalMessage.setCommandEnum(Message::Command::SOWAndSubscribe);
3886 threadLocalMessage.newCommandId();
3887 Field cid = threadLocalMessage.getCommandId();
3888 std::string subId = cid;
3889 threadLocalMessage.setQueryID(cid).setSubscriptionId(cid).setTopic(topic_);
3890 if (filter_.length())
3891 {
3892 threadLocalMessage.setFilter(filter_);
3893 }
3894 if (orderBy_.length())
3895 {
3896 threadLocalMessage.setOrderBy(orderBy_);
3897 }
3898 if (bookmark_.length())
3899 {
3900 threadLocalMessage.setBookmark(bookmark_);
3901 Message::Field bookmark = threadLocalMessage.getBookmark();
3902 if (_bookmarkStore.isValid())
3903 {
3904 ackTypes |= Message::AckType::Persisted;
3905 if (bookmark == AMPS_BOOKMARK_RECENT)
3906 {
3907 threadLocalMessage.assignOwnershipBookmark(_bookmarkStore.getMostRecent(threadLocalMessage.getSubscriptionId()));
3908 }
3909 else if (bookmark != AMPS_BOOKMARK_NOW &&
3910 bookmark != AMPS_BOOKMARK_EPOCH)
3911 {
3912 _bookmarkStore.log(threadLocalMessage);
3913 if (!BookmarkRange::isRange(bookmark))
3914 {
3915 _bookmarkStore.discard(threadLocalMessage);
3916 _bookmarkStore.persisted(threadLocalMessage.getSubscriptionId(),
3917 bookmark);
3918 }
3919 }
3920 }
3921 else if (bookmark == AMPS_BOOKMARK_RECENT)
3922 {
3923 threadLocalMessage.setBookmark(AMPS_BOOKMARK_EPOCH);
3924 }
3925 }
3926 threadLocalMessage.setBatchSize(AMPS::asString(batchSize_));
3927 if (topN_ != AMPS_DEFAULT_TOP_N)
3928 {
3929 threadLocalMessage.setTopNRecordsReturned(AMPS::asString(topN_));
3930 }
3931 if (options_.length())
3932 {
3933 threadLocalMessage.setOptions(options_);
3934 }
3935
3936 AMPS_ATOMIC_BASE_TYPE generation = (AMPS_ATOMIC_BASE_TYPE)0;
3937 if (isHASubscribe_)
3938 {
3939 {
3940 Unlock<Mutex> u(_lock);
3941 generation =
3942 _subscriptionManager->subscribe(messageHandler_, threadLocalMessage.deepCopy(),
3943 Message::AckType::None);
3944 }
3945 if (generation <= (AMPS_ATOMIC_BASE_TYPE)_subscriptionManagerGenerationCount)
3946 {
3947 return subId;
3948 }
3949 }
3950 _routes.addRoute(cid, messageHandler_,
3951 Message::AckType::None, ackTypes, threadLocalMessage.getCommandEnum());
3952 threadLocalMessage.setAckTypeEnum(ackTypes);
3953 if (!options_.empty())
3954 {
3955 threadLocalMessage.setOptions(options_);
3956 }
3957 try
3958 {
3959 syncAckProcessing(timeout_, threadLocalMessage, 0, generation);
3960 }
3961 catch (const DisconnectedException&)
3962 {
3963 if (!isHASubscribe_)
3964 {
3965 _routes.removeRoute(subId);
3966 throw;
3967 }
3968 }
3969 catch (const TimedOutException&)
3970 {
3971 AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(subId));
3972 throw;
3973 }
3974 catch (...)
3975 {
3976 if (isHASubscribe_)
3977 {
3978 // Have to unlock before calling into sub manager to avoid deadlock
3979 Unlock<Mutex> unlock(_lock);
3980 _subscriptionManager->unsubscribe(cid);
3981 }
3982 _routes.removeRoute(subId);
3983 throw;
3984 }
3985 return subId;
3986 }
3987
3988 std::string sowAndSubscribe(const MessageHandler& messageHandler_,
3989 const std::string& topic_,
3990 long timeout_,
3991 const std::string& filter_ = "",
3992 int batchSize_ = AMPS_DEFAULT_BATCH_SIZE,
3993 bool oofEnabled_ = false,
3994 int topN_ = AMPS_DEFAULT_TOP_N,
3995 bool isHASubscribe_ = true)
3996 {
3997 std::string notSet;
3998 return sowAndSubscribe(messageHandler_,
3999 topic_,
4000 filter_,
4001 notSet, // orderBy
4002 notSet, // bookmark
4003 batchSize_,
4004 topN_,
4005 (oofEnabled_ ? "oof" : ""),
4006 timeout_,
4007 isHASubscribe_);
4008 }
4009
4010 std::string sowAndDeltaSubscribe(const MessageHandler& messageHandler_,
4011 const std::string& topic_,
4012 const std::string& filter_ = "",
4013 const std::string& orderBy_ = "",
4014 int batchSize_ = AMPS_DEFAULT_BATCH_SIZE,
4015 int topN_ = AMPS_DEFAULT_TOP_N,
4016 const std::string& options_ = "",
4017 long timeout_ = AMPS_DEFAULT_COMMAND_TIMEOUT,
4018 bool isHASubscribe_ = true)
4019 {
4020 isHASubscribe_ &= (bool)_subscriptionManager;
4021 Lock<Mutex> l(_lock);
4022 threadLocalMessage.reset();
4023 threadLocalMessage.setCommandEnum(Message::Command::SOWAndDeltaSubscribe);
4024 threadLocalMessage.newCommandId();
4025 threadLocalMessage.setQueryID(threadLocalMessage.getCommandId());
4026 threadLocalMessage.setSubscriptionId(threadLocalMessage.getCommandId());
4027 std::string subId = threadLocalMessage.getSubscriptionId();
4028 threadLocalMessage.setTopic(topic_);
4029 if (filter_.length())
4030 {
4031 threadLocalMessage.setFilter(filter_);
4032 }
4033 if (orderBy_.length())
4034 {
4035 threadLocalMessage.setOrderBy(orderBy_);
4036 }
4037 threadLocalMessage.setBatchSize(AMPS::asString(batchSize_));
4038 if (topN_ != AMPS_DEFAULT_TOP_N)
4039 {
4040 threadLocalMessage.setTopNRecordsReturned(AMPS::asString(topN_));
4041 }
4042 if (options_.length())
4043 {
4044 threadLocalMessage.setOptions(options_);
4045 }
4046 AMPS_ATOMIC_BASE_TYPE generation = (AMPS_ATOMIC_BASE_TYPE)0;
4047 if (isHASubscribe_)
4048 {
4049 {
4050 Unlock<Mutex> u(_lock);
4051 generation =
4052 _subscriptionManager->subscribe(messageHandler_, threadLocalMessage.deepCopy(),
4053 Message::AckType::None);
4054 }
4055 if (generation <= (AMPS_ATOMIC_BASE_TYPE)_subscriptionManagerGenerationCount)
4056 {
4057 return subId;
4058 }
4059 }
4060 _routes.addRoute(threadLocalMessage.getQueryID(), messageHandler_,
4061 Message::AckType::None, Message::AckType::Processed, threadLocalMessage.getCommandEnum());
4062 threadLocalMessage.setAckTypeEnum(Message::AckType::Processed);
4063 if (!options_.empty())
4064 {
4065 threadLocalMessage.setOptions(options_);
4066 }
4067 try
4068 {
4069 syncAckProcessing(timeout_, threadLocalMessage, 0, generation);
4070 }
4071 catch (const DisconnectedException&)
4072 {
4073 if (!isHASubscribe_)
4074 {
4075 _routes.removeRoute(subId);
4076 throw;
4077 }
4078 }
4079 catch (const TimedOutException&)
4080 {
4081 AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(subId));
4082 throw;
4083 }
4084 catch (...)
4085 {
4086 if (isHASubscribe_)
4087 {
4088 // Have to unlock before calling into sub manager to avoid deadlock
4089 Unlock<Mutex> unlock(_lock);
4090 _subscriptionManager->unsubscribe(Field(subId));
4091 }
4092 _routes.removeRoute(subId);
4093 throw;
4094 }
4095 return subId;
4096 }
4097
4098 std::string sowAndDeltaSubscribe(const MessageHandler& messageHandler_,
4099 const std::string& topic_,
4100 long timeout_,
4101 const std::string& filter_ = "",
4102 int batchSize_ = AMPS_DEFAULT_BATCH_SIZE,
4103 bool oofEnabled_ = false,
4104 bool sendEmpties_ = false,
4105 int topN_ = AMPS_DEFAULT_TOP_N,
4106 bool isHASubscribe_ = true)
4107 {
4108 std::string notSet;
4109 Message::Options options;
4110 if (oofEnabled_)
4111 {
4112 options.setOOF();
4113 }
4114 if (sendEmpties_ == false)
4115 {
4116 options.setNoEmpties();
4117 }
4118 return sowAndDeltaSubscribe(messageHandler_,
4119 topic_,
4120 filter_,
4121 notSet, // orderBy
4122 batchSize_,
4123 topN_,
4124 options,
4125 timeout_,
4126 isHASubscribe_);
4127 }
4128
4129 std::string sowDelete(const MessageHandler& messageHandler_,
4130 const std::string& topic_,
4131 const std::string& filter_,
4132 long timeout_,
4133 Message::Field commandId_ = Message::Field())
4134 {
4135 if (_publishStore.isValid())
4136 {
4137 unsigned ackType = Message::AckType::Processed |
4138 Message::AckType::Stats |
4139 Message::AckType::Persisted;
4140 threadLocalMessage.reset();
4141 if (commandId_.empty())
4142 {
4143 threadLocalMessage.newCommandId();
4144 commandId_ = threadLocalMessage.getCommandId();
4145 }
4146 else
4147 {
4148 threadLocalMessage.setCommandId(commandId_.data(), commandId_.len());
4149 }
4150 threadLocalMessage.setCommandEnum(Message::Command::SOWDelete)
4151 .assignSubscriptionId(commandId_.data(), commandId_.len())
4152 .assignQueryID(commandId_.data(), commandId_.len())
4153 .setAckTypeEnum(ackType)
4154 .assignTopic(topic_.c_str(), topic_.length())
4155 .assignFilter(filter_.c_str(), filter_.length());
4156 amps_uint64_t haSequenceNumber = _publishStore.store(threadLocalMessage);
4157 char buf[AMPS_NUMBER_BUFFER_LEN];
4158 size_t pos = convertToCharArray(buf, haSequenceNumber);
4159 threadLocalMessage.assignSequence(buf + pos, AMPS_NUMBER_BUFFER_LEN - pos);
4160 {
4161 try
4162 {
4163 Lock<Mutex> l(_lock);
4164 _routes.addRoute(commandId_, messageHandler_,
4165 Message::AckType::Stats,
4166 Message::AckType::Processed | Message::AckType::Persisted,
4167 threadLocalMessage.getCommandEnum());
4168 syncAckProcessing(timeout_, threadLocalMessage,
4169 haSequenceNumber);
4170 }
4171 catch (const DisconnectedException&)
4172 { // -V565
4173 // Pass - it will get replayed upon reconnect
4174 }
4175 catch (...)
4176 {
4177 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(commandId_));
4178 throw;
4179 }
4180 }
4181 return (std::string)commandId_;
4182 }
4183 else
4184 {
4185 Lock<Mutex> l(_lock);
4186 threadLocalMessage.reset();
4187 if (commandId_.empty())
4188 {
4189 threadLocalMessage.newCommandId();
4190 commandId_ = threadLocalMessage.getCommandId();
4191 }
4192 else
4193 {
4194 threadLocalMessage.setCommandId(commandId_.data(), commandId_.len());
4195 }
4196 threadLocalMessage.setCommandEnum(Message::Command::SOWDelete)
4197 .assignSubscriptionId(commandId_.data(), commandId_.len())
4198 .assignQueryID(commandId_.data(), commandId_.len())
4199 .setAckTypeEnum(Message::AckType::Processed |
4200 Message::AckType::Stats)
4201 .assignTopic(topic_.c_str(), topic_.length())
4202 .assignFilter(filter_.c_str(), filter_.length());
4203 _routes.addRoute(commandId_, messageHandler_,
4204 Message::AckType::Stats,
4205 Message::AckType::Processed,
4206 threadLocalMessage.getCommandEnum());
4207 try
4208 {
4209 syncAckProcessing(timeout_, threadLocalMessage);
4210 }
4211 catch (...)
4212 {
4213 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(commandId_));
4214 throw;
4215 }
4216 return (std::string)commandId_;
4217 }
4218 }
4219
4220 std::string sowDeleteByData(const MessageHandler& messageHandler_,
4221 const std::string& topic_,
4222 const std::string& data_,
4223 long timeout_,
4224 Message::Field commandId_ = Message::Field())
4225 {
4226 if (_publishStore.isValid())
4227 {
4228 unsigned ackType = Message::AckType::Processed |
4229 Message::AckType::Stats |
4230 Message::AckType::Persisted;
4231 threadLocalMessage.reset();
4232 if (commandId_.empty())
4233 {
4234 threadLocalMessage.newCommandId();
4235 commandId_ = threadLocalMessage.getCommandId();
4236 }
4237 else
4238 {
4239 threadLocalMessage.setCommandId(commandId_.data(), commandId_.len());
4240 }
4241 threadLocalMessage.setCommandEnum(Message::Command::SOWDelete)
4242 .assignSubscriptionId(commandId_.data(), commandId_.len())
4243 .assignQueryID(commandId_.data(), commandId_.len())
4244 .setAckTypeEnum(ackType)
4245 .assignTopic(topic_.c_str(), topic_.length())
4246 .assignData(data_.c_str(), data_.length());
4247 amps_uint64_t haSequenceNumber = _publishStore.store(threadLocalMessage);
4248 char buf[AMPS_NUMBER_BUFFER_LEN];
4249 size_t pos = convertToCharArray(buf, haSequenceNumber);
4250 threadLocalMessage.assignSequence(buf + pos, AMPS_NUMBER_BUFFER_LEN - pos);
4251 {
4252 try
4253 {
4254 Lock<Mutex> l(_lock);
4255 _routes.addRoute(commandId_, messageHandler_,
4256 Message::AckType::Stats,
4257 Message::AckType::Processed | Message::AckType::Persisted,
4258 threadLocalMessage.getCommandEnum());
4259 syncAckProcessing(timeout_, threadLocalMessage,
4260 haSequenceNumber);
4261 }
4262 catch (const DisconnectedException&)
4263 { // -V565
4264 // Pass - it will get replayed upon reconnect
4265 }
4266 catch (...)
4267 {
4268 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(commandId_));
4269 throw;
4270 }
4271 }
4272 return (std::string)commandId_;
4273 }
4274 else
4275 {
4276 Lock<Mutex> l(_lock);
4277 threadLocalMessage.reset();
4278 if (commandId_.empty())
4279 {
4280 threadLocalMessage.newCommandId();
4281 commandId_ = threadLocalMessage.getCommandId();
4282 }
4283 else
4284 {
4285 threadLocalMessage.setCommandId(commandId_.data(), commandId_.len());
4286 }
4287 threadLocalMessage.setCommandEnum(Message::Command::SOWDelete)
4288 .assignSubscriptionId(commandId_.data(), commandId_.len())
4289 .assignQueryID(commandId_.data(), commandId_.len())
4290 .setAckTypeEnum(Message::AckType::Processed |
4291 Message::AckType::Stats)
4292 .assignTopic(topic_.c_str(), topic_.length())
4293 .assignData(data_.c_str(), data_.length());
4294 _routes.addRoute(commandId_, messageHandler_,
4295 Message::AckType::Stats,
4296 Message::AckType::Processed,
4297 threadLocalMessage.getCommandEnum());
4298 try
4299 {
4300 syncAckProcessing(timeout_, threadLocalMessage);
4301 }
4302 catch (...)
4303 {
4304 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(commandId_));
4305 throw;
4306 }
4307 return (std::string)commandId_;
4308 }
4309 }
4310
4311 std::string sowDeleteByKeys(const MessageHandler& messageHandler_,
4312 const std::string& topic_,
4313 const std::string& keys_,
4314 long timeout_,
4315 Message::Field commandId_ = Message::Field())
4316 {
4317 if (_publishStore.isValid())
4318 {
4319 unsigned ackType = Message::AckType::Processed |
4320 Message::AckType::Stats |
4321 Message::AckType::Persisted;
4322 threadLocalMessage.reset();
4323 if (commandId_.empty())
4324 {
4325 threadLocalMessage.newCommandId();
4326 commandId_ = threadLocalMessage.getCommandId();
4327 }
4328 else
4329 {
4330 threadLocalMessage.setCommandId(commandId_.data(), commandId_.len());
4331 }
4332 threadLocalMessage.setCommandEnum(Message::Command::SOWDelete)
4333 .assignSubscriptionId(commandId_.data(), commandId_.len())
4334 .assignQueryID(commandId_.data(), commandId_.len())
4335 .setAckTypeEnum(ackType)
4336 .assignTopic(topic_.c_str(), topic_.length())
4337 .assignSowKeys(keys_.c_str(), keys_.length());
4338 amps_uint64_t haSequenceNumber = _publishStore.store(threadLocalMessage);
4339 char buf[AMPS_NUMBER_BUFFER_LEN];
4340 size_t pos = convertToCharArray(buf, haSequenceNumber);
4341 threadLocalMessage.assignSequence(buf + pos, AMPS_NUMBER_BUFFER_LEN - pos);
4342 {
4343 try
4344 {
4345 Lock<Mutex> l(_lock);
4346 _routes.addRoute(commandId_, messageHandler_,
4347 Message::AckType::Stats,
4348 Message::AckType::Processed | Message::AckType::Persisted,
4349 threadLocalMessage.getCommandEnum());
4350 syncAckProcessing(timeout_, threadLocalMessage,
4351 haSequenceNumber);
4352 }
4353 catch (const DisconnectedException&)
4354 { // -V565
4355 // Pass - it will get replayed upon reconnect
4356 }
4357 catch (...)
4358 {
4359 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(commandId_));
4360 throw;
4361 }
4362 }
4363 return (std::string)commandId_;
4364 }
4365 else
4366 {
4367 Lock<Mutex> l(_lock);
4368 threadLocalMessage.reset();
4369 if (commandId_.empty())
4370 {
4371 threadLocalMessage.newCommandId();
4372 commandId_ = threadLocalMessage.getCommandId();
4373 }
4374 else
4375 {
4376 threadLocalMessage.setCommandId(commandId_.data(), commandId_.len());
4377 }
4378 threadLocalMessage.setCommandEnum(Message::Command::SOWDelete)
4379 .assignSubscriptionId(commandId_.data(), commandId_.len())
4380 .assignQueryID(commandId_.data(), commandId_.len())
4381 .setAckTypeEnum(Message::AckType::Processed |
4382 Message::AckType::Stats)
4383 .assignTopic(topic_.c_str(), topic_.length())
4384 .assignSowKeys(keys_.c_str(), keys_.length());
4385 _routes.addRoute(commandId_, messageHandler_,
4386 Message::AckType::Stats,
4387 Message::AckType::Processed,
4388 threadLocalMessage.getCommandEnum());
4389 try
4390 {
4391 syncAckProcessing(timeout_, threadLocalMessage);
4392 }
4393 catch (...)
4394 {
4395 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(commandId_));
4396 throw;
4397 }
4398 return (std::string)commandId_;
4399 }
4400 }
4401
4402 void startTimer(void)
4403 {
4404 if (_serverVersion >= "5.3.2.0")
4405 {
4406 throw CommandException("The start_timer command is deprecated.");
4407 }
4408 Lock<Mutex> l(_lock);
4409 threadLocalMessage.reset();
4410 threadLocalMessage.setCommandEnum(Message::Command::StartTimer);
4411
4412 _send(threadLocalMessage);
4413 }
4414
4415 std::string stopTimer(MessageHandler messageHandler_)
4416 {
4417 if (_serverVersion >= "5.3.2.0")
4418 {
4419 throw CommandException("The stop_timer command is deprecated.");
4420 }
4421 return executeAsync(Command("stop_timer").addAckType("completed"), messageHandler_);
4422 }
4423
4424 amps_handle getHandle(void)
4425 {
4426 return _client;
4427 }
4428
4436 void setExceptionListener(const std::shared_ptr<const ExceptionListener>& pListener_)
4437 {
4438 _pExceptionListener = pListener_;
4439 _exceptionListener = _pExceptionListener.get();
4440 }
4441
4442 void setExceptionListener(const ExceptionListener& listener_)
4443 {
4444 _exceptionListener = &listener_;
4445 }
4446
4447 const ExceptionListener& getExceptionListener(void) const
4448 {
4449 return *_exceptionListener;
4450 }
4451
4452 void setHeartbeat(unsigned heartbeatInterval_, unsigned readTimeout_)
4453 {
4454 if (readTimeout_ < heartbeatInterval_)
4455 {
4456 throw UsageException("The socket read timeout must be >= the heartbeat interval.");
4457 }
4458 Lock<Mutex> l(_lock);
4459 if (_heartbeatInterval != heartbeatInterval_ ||
4460 _readTimeout != readTimeout_)
4461 {
4462 _heartbeatInterval = heartbeatInterval_;
4463 _readTimeout = readTimeout_;
4464 _sendHeartbeat();
4465 }
4466 }
4467
4468 void _sendHeartbeat(void)
4469 {
4470 if (_connected && _heartbeatInterval != 0)
4471 {
4472 std::ostringstream options;
4473 options << "start," << _heartbeatInterval;
4474 _beatMessage.setOptions(options.str());
4475
4476 _heartbeatTimer.setTimeout(_heartbeatInterval * 1000.0);
4477 _heartbeatTimer.start();
4478 try
4479 {
4480 _sendWithoutRetry(_beatMessage);
4481 broadcastConnectionStateChanged(ConnectionStateListener::HeartbeatInitiated);
4482 }
4483 catch (ConnectionException& ex_)
4484 {
4485 // If we are disconnected when we attempt to send, that's OK;
4486 // we'll send this message after we re-connect (if we do).
4487 AMPS_UNHANDLED_EXCEPTION(ex_);
4488 }
4489 _beatMessage.setOptions("beat");
4490 }
4491 amps_result result = AMPS_E_OK;
4492 if (_readTimeout && _connected)
4493 {
4494 result = amps_client_set_read_timeout(_client, (int)_readTimeout);
4495 if (result != AMPS_E_OK && result != AMPS_E_DISCONNECTED)
4496 {
4497 AMPSException::throwFor(_client, result);
4498 }
4499 if (!_queueAckTimeout)
4500 {
4501 result = amps_client_set_idle_time(_client,
4502 (int)(_heartbeatInterval * 1000));
4503 if (result != AMPS_E_OK && result != AMPS_E_DISCONNECTED)
4504 {
4505 AMPSException::throwFor(_client, result);
4506 }
4507 }
4508 }
4509 }
4510
4511 void addConnectionStateListener(ConnectionStateListener* listener_)
4512 {
4513 Lock<Mutex> lock(_lock);
4514 _connectionStateListeners.insert(listener_);
4515 }
4516
4517 void removeConnectionStateListener(ConnectionStateListener* listener_)
4518 {
4519 Lock<Mutex> lock(_lock);
4520 _connectionStateListeners.erase(listener_);
4521 }
4522
4523 void clearConnectionStateListeners()
4524 {
4525 Lock<Mutex> lock(_lock);
4526 _connectionStateListeners.clear();
4527 }
4528
4529 void _registerHandler(Command& command_, Message::Field& cid_,
4530 MessageHandler& handler_, unsigned requestedAcks_,
4531 unsigned systemAddedAcks_, Message::Command::Type commandType_)
4532 {
4533 Message message = command_.getMessage();
4534 Message::Command::Type commandType = message.getCommandEnum();
4535 Message::Field subid = message.getSubscriptionId();
4536 Message::Field qid = message.getQueryID();
4537 // If we have an id, we're good, even if it's an existing route
4538 bool added = qid.len() || subid.len() || cid_.len();
4539 bool cidIsQid = cid_ == qid;
4540 bool cidUnique = !cidIsQid && cid_.len() > 0 && cid_ != subid;
4541 int addedCount = 0;
4542 if (subid.len() > 0)
4543 {
4544 // This can replace a non-subscribe with a matching id
4545 // with a subscription but not another subscription.
4546 addedCount += _routes.addRoute(subid, handler_, requestedAcks_,
4547 systemAddedAcks_, commandType_);
4548 if (!cidUnique
4549 && (commandType == Message::Command::Subscribe
4550 || commandType == Message::Command::DeltaSubscribe))
4551 {
4552 // We don't need to do anything else
4553 cid_ = subid;
4554 return;
4555 }
4556 }
4557 if (qid.len() > 0 && qid != subid
4558 && (commandType == Message::Command::SOW
4559 || commandType == Message::Command::SOWDelete
4560 || commandType == Message::Command::SOWAndSubscribe
4561 || commandType == Message::Command::SOWAndDeltaSubscribe))
4562 {
4563 while (_routes.hasRoute(qid))
4564 {
4565 message.newQueryId();
4566 if (cidIsQid)
4567 {
4568 cid_ = message.getQueryId();
4569 }
4570 qid = message.getQueryId();
4571 }
4572 if (addedCount == 0)
4573 {
4574 _routes.addRoute(qid, handler_, requestedAcks_,
4575 systemAddedAcks_, commandType_);
4576 }
4577 else
4578 {
4579 void* data = NULL;
4580 {
4581 Unlock<Mutex> u(_lock);
4582 data = amps_invoke_copy_route_function(handler_.userData());
4583 }
4584 if (!data)
4585 {
4586 _routes.addRoute(qid, handler_, requestedAcks_,
4587 systemAddedAcks_, commandType_);
4588 }
4589 else
4590 {
4591 _routes.addRoute(qid,
4592 MessageHandler(handler_.function(),
4593 data),
4594 requestedAcks_,
4595 systemAddedAcks_, commandType_);
4596 }
4597 }
4598 ++addedCount;
4599 }
4600 if (cidUnique && requestedAcks_ & ~Message::AckType::Persisted)
4601 {
4602 while (_routes.hasRoute(cid_))
4603 {
4604 cid_ = message.newCommandId().getCommandId();
4605 }
4606 if (addedCount == 0)
4607 {
4608 _routes.addRoute(cid_, handler_, requestedAcks_,
4609 systemAddedAcks_, commandType_);
4610 }
4611 else
4612 {
4613 void* data = NULL;
4614 {
4615 Unlock<Mutex> u(_lock);
4616 data = amps_invoke_copy_route_function(handler_.userData());
4617 }
4618 if (!data)
4619 {
4620 _routes.addRoute(cid_, handler_, requestedAcks_,
4621 systemAddedAcks_, commandType_);
4622 }
4623 else
4624 {
4625 _routes.addRoute(cid_,
4626 MessageHandler(handler_.function(),
4627 data),
4628 requestedAcks_,
4629 systemAddedAcks_, commandType_);
4630 }
4631 }
4632 }
4633 else if ((commandType == Message::Command::Publish ||
4634 commandType == Message::Command::DeltaPublish)
4635 && requestedAcks_ & ~Message::AckType::Persisted)
4636 {
4637 cid_ = command_.getMessage().newCommandId().getCommandId();
4638 _routes.addRoute(cid_, handler_, requestedAcks_,
4639 systemAddedAcks_, commandType_);
4640 added = true;
4641 }
4642 if (!added)
4643 {
4644 throw UsageException("To use a messagehandler, you must also supply a command or subscription ID.");
4645 }
4646 }
4647
4648 std::string executeAsyncNoLock(Command& command_, MessageHandler& handler_,
4649 bool isHASubscribe_ = true)
4650 {
4651 isHASubscribe_ &= (bool)_subscriptionManager;
4652 Message& message = command_.getMessage();
4653 unsigned systemAddedAcks = (handler_.isValid() || command_.hasProcessedAck()) ?
4654 Message::AckType::Processed : Message::AckType::None;
4655 unsigned requestedAcks = message.getAckTypeEnum();
4656 bool isPublishStore = _publishStore.isValid() && command_.needsSequenceNumber();
4657 Message::Command::Type commandType = message.getCommandEnum();
4658 if (commandType == Message::Command::StopTimer)
4659 {
4660 systemAddedAcks |= Message::AckType::Completed;
4661 }
4662 else if (commandType == Message::Command::Unsubscribe)
4663 {
4664 // Clear routes and sub manager
4665 const std::string subId = message.getSubscriptionId();
4666 if (subId == "all")
4667 {
4668 _routes.unsubscribeAll();
4669 if (_subscriptionManager)
4670 {
4671 Unlock<Mutex> unlock(_lock);
4672 _subscriptionManager->clear();
4673 }
4674 }
4675 else
4676 {
4677 _routes.removeRoute(subId);
4678 // Lock is already acquired
4679 if (_subscriptionManager)
4680 {
4681 // Have to unlock before calling into sub manager to avoid deadlock
4682 Unlock<Mutex> unlock(_lock);
4683 _subscriptionManager->unsubscribe(subId);
4684 }
4685 }
4686 // Make sure the clear gets processed by receive thread
4687 deferredExecution(&amps_noOpFn, NULL);
4688 }
4689 Message::Field cid = message.getCommandId();
4690 if (handler_.isValid() && cid.empty())
4691 {
4692 cid = message.newCommandId().getCommandId();
4693 }
4694 if (message.getBookmark().len() > 0)
4695 {
4696 if (command_.isSubscribe())
4697 {
4698 Message::Field bookmark = message.getBookmark();
4699 if (_bookmarkStore.isValid())
4700 {
4701 systemAddedAcks |= Message::AckType::Persisted;
4702 if (bookmark == AMPS_BOOKMARK_RECENT)
4703 {
4704 message.assignOwnershipBookmark(_bookmarkStore.getMostRecent(message.getSubscriptionId()));
4705 }
4706 else if (bookmark != AMPS_BOOKMARK_NOW &&
4707 bookmark != AMPS_BOOKMARK_EPOCH)
4708 {
4709 _bookmarkStore.log(message);
4710 if (!BookmarkRange::isRange(bookmark))
4711 {
4712 _bookmarkStore.discard(message);
4713 _bookmarkStore.persisted(message.getSubscriptionId(),
4714 bookmark);
4715 }
4716 }
4717 }
4718 else if (bookmark == AMPS_BOOKMARK_RECENT)
4719 {
4720 message.setBookmark(AMPS_BOOKMARK_EPOCH);
4721 }
4722 }
4723 }
4724 if (isPublishStore)
4725 {
4726 systemAddedAcks |= Message::AckType::Persisted;
4727 }
4728 bool isSubscribe = command_.isSubscribe();
4729 if (handler_.isValid() && !isSubscribe)
4730 {
4731 _registerHandler(command_, cid, handler_,
4732 requestedAcks, systemAddedAcks, commandType);
4733 }
4734 if (isPublishStore)
4735 {
4736 bool useSyncSend = cid.len() > 0 && command_.hasProcessedAck();
4737 amps_uint64_t haSequenceNumber = (amps_uint64_t)0;
4738 message.setAckTypeEnum(requestedAcks | systemAddedAcks);
4739 {
4740 Unlock<Mutex> u(_lock);
4741 haSequenceNumber = _publishStore.store(message);
4742 }
4743 message.setSequence(haSequenceNumber);
4744 try
4745 {
4746 if (useSyncSend)
4747 {
4748 syncAckProcessing((long)command_.getTimeout(), message,
4749 haSequenceNumber);
4750 }
4751 else
4752 {
4753 _sendSequenced(message, haSequenceNumber,
4754 commandType & (Message::Command::Publish
4755 | Message::Command::DeltaPublish));
4756 }
4757 }
4758 catch (const DisconnectedException&)
4759 {
4760 throw;
4761 }
4762 catch (...)
4763 {
4764 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(cid));
4765 throw;
4766 }
4767 }
4768 else
4769 {
4770 if (isSubscribe)
4771 {
4772 const Message::Field& subId = message.getSubscriptionId();
4773 AMPS_ATOMIC_BASE_TYPE generation = (AMPS_ATOMIC_BASE_TYPE)0;
4774 if (isHASubscribe_)
4775 {
4776 {
4777 Unlock<Mutex> u(_lock);
4778 generation =
4779 _subscriptionManager->subscribe(handler_,
4780 message.deepCopy(),
4781 requestedAcks);
4782 }
4783 if (generation <= (AMPS_ATOMIC_BASE_TYPE)_subscriptionManagerGenerationCount)
4784 {
4785 message.setAckTypeEnum(requestedAcks);
4786 return std::string(subId.data(), subId.len());
4787 }
4788 }
4789 if (handler_.isValid())
4790 {
4791 _registerHandler(command_, cid, handler_,
4792 requestedAcks, systemAddedAcks, commandType);
4793 }
4794 message.setAckTypeEnum(requestedAcks | systemAddedAcks);
4795 try
4796 {
4797 syncAckProcessing((long)command_.getTimeout(), message, 0, generation);
4798 }
4799 catch (const DisconnectedException&)
4800 {
4801 if (!isHASubscribe_)
4802 {
4803 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(cid));
4804 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(subId));
4805 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(message.getQueryId()));
4806 message.setAckTypeEnum(requestedAcks);
4807 throw;
4808 }
4809 }
4810 catch (const TimedOutException&)
4811 {
4812 AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(cid));
4813 AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(subId));
4814 AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(message.getQueryId()));
4815 message.setAckTypeEnum(requestedAcks);
4816 throw;
4817 }
4818 catch (...)
4819 {
4820 if (isHASubscribe_)
4821 {
4822 // Have to unlock before calling into sub manager to avoid deadlock
4823 Unlock<Mutex> unlock(_lock);
4824 _subscriptionManager->unsubscribe(subId);
4825 }
4826 if (message.getQueryID().len() > 0)
4827 {
4828 _routes.removeRoute(message.getQueryID());
4829 }
4830 _routes.removeRoute(cid);
4831 _routes.removeRoute(subId);
4832 message.setAckTypeEnum(requestedAcks);
4833 throw;
4834 }
4835 if (subId.len() > 0)
4836 {
4837 message.setAckTypeEnum(requestedAcks);
4838 return std::string(subId.data(), subId.len());
4839 }
4840 }
4841 else
4842 {
4843 // SOW, Flush, etc. should always be sync. Publish/delete may not be.
4844 bool useSyncSend = commandType & ~Message::Command::NoDataCommands
4845 || (cid.len() > 0 && command_.hasProcessedAck());
4846 message.setAckTypeEnum(requestedAcks | systemAddedAcks);
4847 try
4848 {
4849 if (useSyncSend)
4850 {
4851 syncAckProcessing((long)(command_.getTimeout()), message);
4852 }
4853 else
4854 {
4855 _sendSequenced(message, 0,
4856 commandType & (Message::Command::Publish
4857 | Message::Command::DeltaPublish));
4858 }
4859 }
4860 catch (const TimedOutException&)
4861 {
4862 AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(cid));
4863 AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(message.getQueryId()));
4864 message.setAckTypeEnum(requestedAcks);
4865 throw;
4866 }
4867 catch (const DisconnectedException&)
4868 {
4869 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(cid));
4870 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(message.getQueryId()));
4871 message.setAckTypeEnum(requestedAcks);
4872 throw;
4873 }
4874 catch (...)
4875 {
4876 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(cid));
4877 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(message.getQueryId()));
4878 message.setAckTypeEnum(requestedAcks);
4879 throw;
4880 }
4881 }
4882 }
4883 message.setAckTypeEnum(requestedAcks);
4884 return cid;
4885 }
4886
4887 MessageStream getEmptyMessageStream(void);
4888
4889 std::string executeAsync(Command& command_, MessageHandler& handler_,
4890 bool isHASubscribe_ = true)
4891 {
4892 Lock<Mutex> lock(_lock);
4893 return executeAsyncNoLock(command_, handler_, isHASubscribe_);
4894 }
4895
4896 // Queue Methods //
4897 void setAutoAck(bool isAutoAckEnabled_)
4898 {
4899 _isAutoAckEnabled = isAutoAckEnabled_;
4900 }
4901 bool getAutoAck(void) const
4902 {
4903 return _isAutoAckEnabled;
4904 }
4905 void setAckBatchSize(const unsigned batchSize_)
4906 {
4907 _ackBatchSize = batchSize_;
4908 if (!_queueAckTimeout)
4909 {
4910 _queueAckTimeout = AMPS_DEFAULT_QUEUE_ACK_TIMEOUT;
4911 amps_client_set_idle_time(_client, _queueAckTimeout);
4912 }
4913 }
4914 unsigned getAckBatchSize(void) const
4915 {
4916 return _ackBatchSize;
4917 }
4918 int getAckTimeout(void) const
4919 {
4920 return _queueAckTimeout;
4921 }
4922 void setAckTimeout(const int ackTimeout_)
4923 {
4924 amps_client_set_idle_time(_client, ackTimeout_);
4925 _queueAckTimeout = ackTimeout_;
4926 }
4927 size_t _ack(QueueBookmarks& queueBookmarks_)
4928 {
4929 if (queueBookmarks_._bookmarkCount)
4930 {
4931 threadLocalMessage.reset();
4932 threadLocalMessage.setCommandEnum(Message::Command::SOWDelete)
4933 .setTopic(queueBookmarks_._topic)
4934 .setBookmark(queueBookmarks_._data)
4935 .setCommandId("AMPS-queue-ack");
4936 amps_uint64_t haSequenceNumber = 0;
4937 if (_publishStore.isValid())
4938 {
4939 haSequenceNumber = _publishStore.store(threadLocalMessage);
4940 threadLocalMessage.setAckType("persisted")
4941 .setSequence(haSequenceNumber);
4942 queueBookmarks_._data.erase();
4943 queueBookmarks_._bookmarkCount = 0;
4944 }
4945 _sendSequenced(threadLocalMessage, haSequenceNumber, 0);
4946 if (!_publishStore.isValid())
4947 {
4948 queueBookmarks_._data.erase();
4949 queueBookmarks_._bookmarkCount = 0;
4950 }
4951 return 1;
4952 }
4953 return 0;
4954 }
4955 void ack(const Field& topic_, const Field& bookmark_, const char* options_ = NULL)
4956 {
4957 if (_isAutoAckEnabled)
4958 {
4959 return;
4960 }
4961 _ack(topic_, bookmark_, options_);
4962 }
4963 void _ack(const Field& topic_, const Field& bookmark_, const char* options_ = NULL)
4964 {
4965 if (bookmark_.len() == 0)
4966 {
4967 return;
4968 }
4969 Lock<Mutex> lock(_lock);
4970 if (_ackBatchSize < 2 || options_ != NULL)
4971 {
4972 threadLocalMessage.reset();
4973 threadLocalMessage.setCommandEnum(Message::Command::SOWDelete)
4974 .setCommandId("AMPS-queue-ack")
4975 .setTopic(topic_).setBookmark(bookmark_);
4976 if (options_)
4977 {
4978 threadLocalMessage.setOptions(options_);
4979 }
4980 amps_uint64_t haSequenceNumber = 0;
4981 if (_publishStore.isValid())
4982 {
4983 haSequenceNumber = _publishStore.store(threadLocalMessage);
4984 threadLocalMessage.setAckType("persisted")
4985 .setSequence(haSequenceNumber);
4986 }
4987 _sendSequenced(threadLocalMessage, haSequenceNumber, 0);
4988 return;
4989 }
4990 // have we acked anything for this hash
4991 topic_hash hash = CRC<0>::crcNoSSE(topic_.data(), topic_.len());
4992 TopicHashMap::iterator it = _topicHashMap.find(hash);
4993 if (it == _topicHashMap.end())
4994 {
4995 // add a new one to the map
4996 it = _topicHashMap.emplace(TopicHashMap::value_type(hash, QueueBookmarks(topic_))).first;
4997 }
4998 QueueBookmarks& queueBookmarks = it->second;
4999 if (queueBookmarks._data.length())
5000 {
5001 queueBookmarks._data.append(",");
5002 }
5003 else
5004 {
5005 queueBookmarks._oldestTime = amps_now();
5006 }
5007 queueBookmarks._data.append(bookmark_);
5008 if (++queueBookmarks._bookmarkCount >= _ackBatchSize)
5009 {
5010 _ack(queueBookmarks);
5011 }
5012 }
5013 void flushAcks(void)
5014 {
5015 size_t sendCount = 0;
5016 if (!_connected)
5017 {
5018 return;
5019 }
5020 else
5021 {
5022 Lock<Mutex> lock(_lock);
5023 typedef TopicHashMap::iterator iterator;
5024 for (iterator it = _topicHashMap.begin(), end = _topicHashMap.end(); it != end; ++it)
5025 {
5026 QueueBookmarks& queueBookmarks = it->second;
5027 sendCount += _ack(queueBookmarks);
5028 }
5029 }
5030 if (sendCount && _connected)
5031 {
5032 publishFlush(0, Message::AckType::Processed);
5033 }
5034 }
5035 // called when there's idle time, to see if we need to flush out any "acks"
5036 void checkQueueAcks(void)
5037 {
5038 if (!_topicHashMap.size())
5039 {
5040 return;
5041 }
5042 Lock<Mutex> lock(_lock);
5043 try
5044 {
5045 amps_uint64_t threshold = amps_now()
5046 - (amps_uint64_t)_queueAckTimeout;
5047 typedef TopicHashMap::iterator iterator;
5048 for (iterator it = _topicHashMap.begin(), end = _topicHashMap.end(); it != end; ++it)
5049 {
5050 QueueBookmarks& queueBookmarks = it->second;
5051 if (queueBookmarks._bookmarkCount && queueBookmarks._oldestTime < threshold)
5052 {
5053 _ack(queueBookmarks);
5054 }
5055 }
5056 }
5057 catch (std::exception& ex)
5058 {
5059 AMPS_UNHANDLED_EXCEPTION(ex);
5060 }
5061 }
5062
5063 void deferredExecution(DeferredExecutionFunc func_, void* userData_)
5064 {
5065 Lock<Mutex> lock(_deferredExecutionLock);
5066 _deferredExecutionList.emplace_back(
5067 DeferredExecutionRequest(func_, userData_));
5068 }
5069
5070 inline void processDeferredExecutions(void)
5071 {
5072 if (_deferredExecutionList.size())
5073 {
5074 Lock<Mutex> lock(_deferredExecutionLock);
5075 DeferredExecutionList::iterator it = _deferredExecutionList.begin();
5076 DeferredExecutionList::iterator end = _deferredExecutionList.end();
5077 for (; it != end; ++it)
5078 {
5079 try
5080 {
5081 it->_func(it->_userData);
5082 }
5083 catch (...)
5084 { // -V565
5085 // Intentionally ignore errors
5086 }
5087 }
5088 _deferredExecutionList.clear();
5089 _routes.invalidateCache();
5090 _routeCache.invalidateCache();
5091 }
5092 }
5093
5094 bool getRetryOnDisconnect(void) const
5095 {
5096 return _isRetryOnDisconnect;
5097 }
5098
5099 void setRetryOnDisconnect(bool isRetryOnDisconnect_)
5100 {
5101 _isRetryOnDisconnect = isRetryOnDisconnect_;
5102 }
5103
5104 void setDefaultMaxDepth(unsigned maxDepth_)
5105 {
5106 _defaultMaxDepth = maxDepth_;
5107 }
5108
5109 unsigned getDefaultMaxDepth(void) const
5110 {
5111 return _defaultMaxDepth;
5112 }
5113
5114 void setTransportFilterFunction(amps_transport_filter_function filter_,
5115 void* userData_)
5116 {
5117 amps_client_set_transport_filter_function(_client, filter_, userData_);
5118 }
5119
5120 void setThreadCreatedCallback(amps_thread_created_callback callback_,
5121 void* userData_)
5122 {
5123 amps_client_set_thread_created_callback(_client, callback_, userData_);
5124 }
5125
5126 void setPublishBatching(amps_uint64_t batchSizeBytes_, amps_uint64_t batchTimeoutMillis_)
5127 {
5128 amps_client_set_batch_send(_client, batchSizeBytes_, batchTimeoutMillis_);
5129 }
5130 }; // class ClientImpl
5131
5206
5207 class MessageStream
5208 {
5209 RefHandle<MessageStreamImpl> _body;
5210 public:
5215 class iterator
5216 {
5217 MessageStream* _pStream;
5218 Message _current;
5219 inline void advance(void);
5220
5221 public:
5222 iterator() // end
5223 : _pStream(NULL)
5224 {;}
5225 iterator(MessageStream* pStream_)
5226 : _pStream(pStream_)
5227 {
5228 advance();
5229 }
5230
5231 bool operator==(const iterator& rhs) const
5232 {
5233 return _pStream == rhs._pStream;
5234 }
5235 bool operator!=(const iterator& rhs) const
5236 {
5237 return _pStream != rhs._pStream;
5238 }
5239 void operator++(void)
5240 {
5241 advance();
5242 }
5243 Message operator*(void)
5244 {
5245 return _current;
5246 }
5247 Message* operator->(void)
5248 {
5249 return &_current;
5250 }
5251 };
5253 bool isValid() const
5254 {
5255 return _body.isValid();
5256 }
5257
5260 iterator begin(void)
5261 {
5262 if (!_body.isValid())
5263 {
5264 throw UsageException("This MessageStream is not valid and cannot be iterated.");
5265 }
5266 return iterator(this);
5267 }
5270 // For non-SOW queries, the end is never reached.
5271 iterator end(void)
5272 {
5273 return iterator();
5274 }
5275 inline MessageStream(void) = default;
5276
5282 MessageStream timeout(unsigned timeout_);
5283
5287 MessageStream conflate(void);
5293 MessageStream maxDepth(unsigned maxDepth_);
5296 unsigned getMaxDepth(void) const;
5299 unsigned getDepth(void) const;
5300
5301 private:
5302 inline MessageStream(const Client& client_);
5303 inline MessageStream(RefHandle<MessageStreamImpl> body_);
5304 inline void setSOWOnly(const std::string& commandId_,
5305 const std::string& queryId_ = "");
5306 inline void setSubscription(const std::string& subId_,
5307 const std::string& commandId_ = "",
5308 const std::string& queryId_ = "");
5309 inline void setStatsOnly(const std::string& commandId_,
5310 const std::string& queryId_ = "");
5311 inline void setAcksOnly(const std::string& commandId_, unsigned acks_);
5312
5313 inline operator MessageHandler(void);
5314
5315 inline static MessageStream fromExistingHandler(const MessageHandler& handler);
5316
5317 friend class Client;
5318 friend class ClientImpl;
5319
5320 };
5321
5341 class Client // -V553
5342 {
5343 protected:
5344 BorrowRefHandle<ClientImpl> _body;
5345 public:
5346 static const int DEFAULT_COMMAND_TIMEOUT = AMPS_DEFAULT_COMMAND_TIMEOUT;
5347 static const int DEFAULT_BATCH_SIZE = AMPS_DEFAULT_BATCH_SIZE;
5348 static const int DEFAULT_TOP_N = AMPS_DEFAULT_TOP_N;
5349
5358 Client(const std::string& clientName = "")
5359 : _body(new ClientImpl(clientName), true)
5360 {;}
5361
5362 Client(ClientImpl* existingClient)
5363 : _body(existingClient, true)
5364 {;}
5365
5366 Client(ClientImpl* existingClient, bool isRef)
5367 : _body(existingClient, isRef)
5368 {;}
5369
5370 Client(const Client& rhs) : _body(rhs._body) {;}
5371 virtual ~Client(void) {;}
5372
5373 Client& operator=(const Client& rhs)
5374 {
5375 _body = rhs._body;
5376 return *this;
5377 }
5378
5379 bool isValid()
5380 {
5381 return _body.isValid();
5382 }
5383
5396 void setName(const std::string& name)
5397 {
5398 _body.get().setName(name);
5399 }
5400
5403 const std::string& getName() const
5404 {
5405 return _body.get().getName();
5406 }
5407
5411 const std::string& getNameHash() const
5412 {
5413 return _body.get().getNameHash();
5414 }
5415
5419 const amps_uint64_t getNameHashValue() const
5420 {
5421 return _body.get().getNameHashValue();
5422 }
5423
5427 const std::string& getConnectionName() const
5428 {
5429 return _body.get().getConnectionName();
5430 }
5431
5438 void setLogonCorrelationData(const std::string& logonCorrelationData_)
5439 {
5440 _body.get().setLogonCorrelationData(logonCorrelationData_);
5441 }
5442
5445 const std::string& getLogonCorrelationData() const
5446 {
5447 return _body.get().getLogonCorrelationData();
5448 }
5449
5453 void addHttpPreflightHeader(const std::string& header_)
5454 {
5455 _body.get().addHttpPreflightHeader(header_);
5456 }
5457
5462 void addHttpPreflightHeader(const std::string& key_, const std::string& value_)
5463 {
5464 _body.get().addHttpPreflightHeader(key_, value_);
5465 }
5466
5468 void clearHttpPreflightHeaders()
5469 {
5470 _body.get().clearHttpPreflightHeaders();
5471 }
5472
5476 template<class T>
5477 void setHttpPreflightHeaders(const T& headers_)
5478 {
5479 _body.get().setHttpPreflightHeaders(headers_);
5480 }
5481
5490 size_t getServerVersion() const
5491 {
5492 return _body.get().getServerVersion();
5493 }
5494
5501 VersionInfo getServerVersionInfo() const
5502 {
5503 return _body.get().getServerVersionInfo();
5504 }
5505
5515 static size_t convertVersionToNumber(const std::string& version_)
5516 {
5517 return AMPS::convertVersionToNumber(version_.c_str(), version_.length());
5518 }
5519
5530 static size_t convertVersionToNumber(const char* data_, size_t len_)
5531 {
5532 return AMPS::convertVersionToNumber(data_, len_);
5533 }
5534
5537 const std::string& getURI() const
5538 {
5539 return _body.get().getURI();
5540 }
5541
5548
5550
5561 void connect(const std::string& uri)
5562 {
5563 _body.get().connect(uri);
5564 }
5565
5568 void disconnect()
5569 {
5570 _body.get().disconnect();
5571 }
5572
5586 void send(const Message& message)
5587 {
5588 _body.get().send(message);
5589 }
5590
5599 void addMessageHandler(const Field& commandId_,
5600 const AMPS::MessageHandler& messageHandler_,
5601 unsigned requestedAcks_, bool isSubscribe_)
5602 {
5603 Message::Command::Type commandType = isSubscribe_ ? Message::Command::Subscribe : Message::Command::SOW;
5604 _body.get().addMessageHandler(commandId_, messageHandler_,
5605 requestedAcks_, commandType);
5606 }
5607
5616 void addMessageHandler(const Field& commandId_,
5617 const AMPS::MessageHandler& messageHandler_,
5618 unsigned requestedAcks_, Message::Command::Type commandType_)
5619 {
5620 _body.get().addMessageHandler(commandId_, messageHandler_,
5621 requestedAcks_, commandType_);
5622 }
5623
5627 bool removeMessageHandler(const Field& commandId_)
5628 {
5629 return _body.get().removeMessageHandler(commandId_);
5630 }
5631
5656 std::string send(const MessageHandler& messageHandler_,
5657 Message& message_,
5658 int timeout_ = 0,
5659 AMPS_ATOMIC_BASE_TYPE subscribeGeneration_ = 0)
5660 {
5661 return _body.get().send(messageHandler_, message_, timeout_, subscribeGeneration_);
5662 }
5663
5677 #if defined(_WIN32) || __cplusplus >= 201402L
5678 [[deprecated("Use HAClient for automatic reconnection and a ConnectionStateListener to monitor connection state.")]]
5679 #endif
5680 virtual void setDisconnectHandler(const DisconnectHandler& disconnectHandler)
5681 {
5682 _body.get().setDisconnectHandler(disconnectHandler);
5683 }
5684
5691 #if defined(_WIN32) || __cplusplus >= 201402L
5692 [[deprecated("Use HAClient for automatic reconnection and a ConnectionStateListener to monitor connection state.")]]
5693 #endif
5694 DisconnectHandler getDisconnectHandler(void) const
5695 {
5696 return _body.get().getDisconnectHandler();
5697 }
5698
5703 virtual ConnectionInfo getConnectionInfo() const
5704 {
5705 return _body.get().getConnectionInfo();
5706 }
5707
5716 void setBookmarkStore(const BookmarkStore& bookmarkStore_)
5717 {
5718 _body.get().setBookmarkStore(bookmarkStore_);
5719 }
5720
5724 BookmarkStore getBookmarkStore()
5725 {
5726 return _body.get().getBookmarkStore();
5727 }
5728
5732 SubscriptionManager* getSubscriptionManager()
5733 {
5734 return _body.get().getSubscriptionManager();
5735 }
5736
5744 void setSubscriptionManager(SubscriptionManager* subscriptionManager_)
5745 {
5746 _body.get().setSubscriptionManager(subscriptionManager_);
5747 }
5748
5768 void setPublishStore(const Store& publishStore_)
5769 {
5770 _body.get().setPublishStore(publishStore_);
5771 }
5772
5776 Store getPublishStore()
5777 {
5778 return _body.get().getPublishStore();
5779 }
5780
5784 void setDuplicateMessageHandler(const MessageHandler& duplicateMessageHandler_)
5785 {
5786 _body.get().setGlobalCommandTypeMessageHandler(ClientImpl::GlobalCommandTypeHandlers::DuplicateMessage,
5787 duplicateMessageHandler_);
5788 }
5789
5799 MessageHandler getDuplicateMessageHandler(void)
5800 {
5801 return _body.get().getDuplicateMessageHandler();
5802 }
5803
5813 void setFailedWriteHandler(FailedWriteHandler* handler_)
5814 {
5815 _body.get().setFailedWriteHandler(handler_);
5816 }
5817
5821 FailedWriteHandler* getFailedWriteHandler()
5822 {
5823 return _body.get().getFailedWriteHandler();
5824 }
5825
5826
5844 amps_uint64_t publish(const std::string& topic_, const std::string& data_)
5845 {
5846 return _body.get().publish(topic_.c_str(), topic_.length(),
5847 data_.c_str(), data_.length());
5848 }
5849
5869 amps_uint64_t publish(const char* topic_, size_t topicLength_,
5870 const char* data_, size_t dataLength_)
5871 {
5872 return _body.get().publish(topic_, topicLength_, data_, dataLength_);
5873 }
5874
5893 amps_uint64_t publish(const std::string& topic_, const std::string& data_,
5894 unsigned long expiration_)
5895 {
5896 return _body.get().publish(topic_.c_str(), topic_.length(),
5897 data_.c_str(), data_.length(), expiration_);
5898 }
5899
5920 amps_uint64_t publish(const char* topic_, size_t topicLength_,
5921 const char* data_, size_t dataLength_,
5922 unsigned long expiration_)
5923 {
5924 return _body.get().publish(topic_, topicLength_,
5925 data_, dataLength_, expiration_);
5926 }
5927
5966 void publishFlush(long timeout_ = 0, unsigned ackType_ = Message::AckType::Processed)
5967 {
5968 _body.get().publishFlush(timeout_, ackType_);
5969 }
5970
5971
5987 amps_uint64_t deltaPublish(const std::string& topic_, const std::string& data_)
5988 {
5989 return _body.get().deltaPublish(topic_.c_str(), topic_.length(),
5990 data_.c_str(), data_.length());
5991 }
5992
6010 amps_uint64_t deltaPublish(const char* topic_, size_t topicLength_,
6011 const char* data_, size_t dataLength_)
6012 {
6013 return _body.get().deltaPublish(topic_, topicLength_,
6014 data_, dataLength_);
6015 }
6016
6033 amps_uint64_t deltaPublish(const std::string& topic_, const std::string& data_,
6034 unsigned long expiration_)
6035 {
6036 return _body.get().deltaPublish(topic_.c_str(), topic_.length(),
6037 data_.c_str(), data_.length(),
6038 expiration_);
6039 }
6040
6059 amps_uint64_t deltaPublish(const char* topic_, size_t topicLength_,
6060 const char* data_, size_t dataLength_,
6061 unsigned long expiration_)
6062 {
6063 return _body.get().deltaPublish(topic_, topicLength_,
6064 data_, dataLength_, expiration_);
6065 }
6066
6082 std::string logon(int timeout_ = 0,
6083 Authenticator& authenticator_ = DefaultAuthenticator::instance(),
6084 const char* options_ = NULL)
6085 {
6086 return _body.get().logon(timeout_, authenticator_, options_);
6087 }
6101 std::string logon(const char* options_, int timeout_ = 0)
6102 {
6103 return _body.get().logon(timeout_, DefaultAuthenticator::instance(),
6104 options_);
6105 }
6106
6120 std::string logon(const std::string& options_, int timeout_ = 0)
6121 {
6122 return _body.get().logon(timeout_, DefaultAuthenticator::instance(),
6123 options_.c_str());
6124 }
6125
6145 std::string subscribe(const MessageHandler& messageHandler_,
6146 const std::string& topic_,
6147 long timeout_ = 0,
6148 const std::string& filter_ = "",
6149 const std::string& options_ = "",
6150 const std::string& subId_ = "")
6151 {
6152 return _body.get().subscribe(messageHandler_, topic_, timeout_,
6153 filter_, "", options_, subId_);
6154 }
6155
6171 MessageStream subscribe(const std::string& topic_,
6172 long timeout_ = 0, const std::string& filter_ = "",
6173 const std::string& options_ = "",
6174 const std::string& subId_ = "")
6175 {
6176 MessageStream result(*this);
6177 if (_body.get().getDefaultMaxDepth())
6178 {
6179 result.maxDepth(_body.get().getDefaultMaxDepth());
6180 }
6181 result.setSubscription(_body.get().subscribe(
6182 result.operator MessageHandler(),
6183 topic_, timeout_, filter_, "",
6184 options_, subId_, false));
6185 return result;
6186 }
6187
6203 MessageStream subscribe(const char* topic_,
6204 long timeout_ = 0, const std::string& filter_ = "",
6205 const std::string& options_ = "",
6206 const std::string& subId_ = "")
6207 {
6208 MessageStream result(*this);
6209 if (_body.get().getDefaultMaxDepth())
6210 {
6211 result.maxDepth(_body.get().getDefaultMaxDepth());
6212 }
6213 result.setSubscription(_body.get().subscribe(
6214 result.operator MessageHandler(),
6215 topic_, timeout_, filter_, "",
6216 options_, subId_, false));
6217 return result;
6218 }
6219
6232 std::string deltaSubscribe(const MessageHandler& messageHandler_,
6233 const std::string& topic_,
6234 long timeout_,
6235 const std::string& filter_ = "",
6236 const std::string& options_ = "",
6237 const std::string& subId_ = "")
6238 {
6239 return _body.get().deltaSubscribe(messageHandler_, topic_, timeout_,
6240 filter_, "", options_, subId_);
6241 }
6250 MessageStream deltaSubscribe(const std::string& topic_,
6251 long timeout_, const std::string& filter_ = "",
6252 const std::string& options_ = "",
6253 const std::string& subId_ = "")
6254 {
6255 MessageStream result(*this);
6256 if (_body.get().getDefaultMaxDepth())
6257 {
6258 result.maxDepth(_body.get().getDefaultMaxDepth());
6259 }
6260 result.setSubscription(_body.get().deltaSubscribe(
6261 result.operator MessageHandler(),
6262 topic_, timeout_, filter_, "",
6263 options_, subId_, false));
6264 return result;
6265 }
6266
6268 MessageStream deltaSubscribe(const char* topic_,
6269 long timeout_, const std::string& filter_ = "",
6270 const std::string& options_ = "",
6271 const std::string& subId_ = "")
6272 {
6273 MessageStream result(*this);
6274 if (_body.get().getDefaultMaxDepth())
6275 {
6276 result.maxDepth(_body.get().getDefaultMaxDepth());
6277 }
6278 result.setSubscription(_body.get().deltaSubscribe(
6279 result.operator MessageHandler(),
6280 topic_, timeout_, filter_, "",
6281 options_, subId_, false));
6282 return result;
6283 }
6284
6310 std::string bookmarkSubscribe(const MessageHandler& messageHandler_,
6311 const std::string& topic_,
6312 long timeout_,
6313 const std::string& bookmark_,
6314 const std::string& filter_ = "",
6315 const std::string& options_ = "",
6316 const std::string& subId_ = "")
6317 {
6318 return _body.get().subscribe(messageHandler_, topic_, timeout_,
6319 filter_, bookmark_, options_, subId_);
6320 }
6338 MessageStream bookmarkSubscribe(const std::string& topic_,
6339 long timeout_,
6340 const std::string& bookmark_,
6341 const std::string& filter_ = "",
6342 const std::string& options_ = "",
6343 const std::string& subId_ = "")
6344 {
6345 MessageStream result(*this);
6346 if (_body.get().getDefaultMaxDepth())
6347 {
6348 result.maxDepth(_body.get().getDefaultMaxDepth());
6349 }
6350 result.setSubscription(_body.get().subscribe(
6351 result.operator MessageHandler(),
6352 topic_, timeout_, filter_,
6353 bookmark_, options_,
6354 subId_, false));
6355 return result;
6356 }
6357
6359 MessageStream bookmarkSubscribe(const char* topic_,
6360 long timeout_,
6361 const std::string& bookmark_,
6362 const std::string& filter_ = "",
6363 const std::string& options_ = "",
6364 const std::string& subId_ = "")
6365 {
6366 MessageStream result(*this);
6367 if (_body.get().getDefaultMaxDepth())
6368 {
6369 result.maxDepth(_body.get().getDefaultMaxDepth());
6370 }
6371 result.setSubscription(_body.get().subscribe(
6372 result.operator MessageHandler(),
6373 topic_, timeout_, filter_,
6374 bookmark_, options_,
6375 subId_, false));
6376 return result;
6377 }
6378
6387 void unsubscribe(const std::string& commandId)
6388 {
6389 return _body.get().unsubscribe(commandId);
6390 }
6391
6399 void unsubscribe()
6400 {
6401 return _body.get().unsubscribe();
6402 }
6403
6404
6434 std::string sow(const MessageHandler& messageHandler_,
6435 const std::string& topic_,
6436 const std::string& filter_ = "",
6437 const std::string& orderBy_ = "",
6438 const std::string& bookmark_ = "",
6439 int batchSize_ = DEFAULT_BATCH_SIZE,
6440 int topN_ = DEFAULT_TOP_N,
6441 const std::string& options_ = "",
6442 long timeout_ = DEFAULT_COMMAND_TIMEOUT)
6443 {
6444 return _body.get().sow(messageHandler_, topic_, filter_, orderBy_,
6445 bookmark_, batchSize_, topN_, options_,
6446 timeout_);
6447 }
6472 MessageStream sow(const std::string& topic_,
6473 const std::string& filter_ = "",
6474 const std::string& orderBy_ = "",
6475 const std::string& bookmark_ = "",
6476 int batchSize_ = DEFAULT_BATCH_SIZE,
6477 int topN_ = DEFAULT_TOP_N,
6478 const std::string& options_ = "",
6479 long timeout_ = DEFAULT_COMMAND_TIMEOUT)
6480 {
6481 MessageStream result(*this);
6482 if (_body.get().getDefaultMaxDepth())
6483 {
6484 result.maxDepth(_body.get().getDefaultMaxDepth());
6485 }
6486 result.setSOWOnly(_body.get().sow(result.operator MessageHandler(),
6487 topic_, filter_, orderBy_, bookmark_,
6488 batchSize_, topN_, options_, timeout_));
6489 return result;
6490 }
6491
6493 MessageStream sow(const char* topic_,
6494 const std::string& filter_ = "",
6495 const std::string& orderBy_ = "",
6496 const std::string& bookmark_ = "",
6497 int batchSize_ = DEFAULT_BATCH_SIZE,
6498 int topN_ = DEFAULT_TOP_N,
6499 const std::string& options_ = "",
6500 long timeout_ = DEFAULT_COMMAND_TIMEOUT)
6501 {
6502 MessageStream result(*this);
6503 if (_body.get().getDefaultMaxDepth())
6504 {
6505 result.maxDepth(_body.get().getDefaultMaxDepth());
6506 }
6507 result.setSOWOnly(_body.get().sow(result.operator MessageHandler(),
6508 topic_, filter_, orderBy_, bookmark_,
6509 batchSize_, topN_, options_, timeout_));
6510 return result;
6511 }
6534 std::string sow(const MessageHandler& messageHandler_,
6535 const std::string& topic_,
6536 long timeout_,
6537 const std::string& filter_ = "",
6538 int batchSize_ = DEFAULT_BATCH_SIZE,
6539 int topN_ = DEFAULT_TOP_N)
6540 {
6541 return _body.get().sow(messageHandler_, topic_, timeout_, filter_,
6542 batchSize_, topN_);
6543 }
6566 std::string sowAndSubscribe(const MessageHandler& messageHandler_,
6567 const std::string& topic_,
6568 long timeout_,
6569 const std::string& filter_ = "",
6570 int batchSize_ = DEFAULT_BATCH_SIZE,
6571 bool oofEnabled_ = false,
6572 int topN_ = DEFAULT_TOP_N)
6573 {
6574 return _body.get().sowAndSubscribe(messageHandler_, topic_, timeout_,
6575 filter_, batchSize_, oofEnabled_,
6576 topN_);
6577 }
6578
6598 MessageStream sowAndSubscribe(const std::string& topic_,
6599 long timeout_,
6600 const std::string& filter_ = "",
6601 int batchSize_ = DEFAULT_BATCH_SIZE,
6602 bool oofEnabled_ = false,
6603 int topN_ = DEFAULT_TOP_N)
6604 {
6605 MessageStream result(*this);
6606 if (_body.get().getDefaultMaxDepth())
6607 {
6608 result.maxDepth(_body.get().getDefaultMaxDepth());
6609 }
6610 result.setSubscription(_body.get().sowAndSubscribe(
6611 result.operator MessageHandler(),
6612 topic_, timeout_, filter_,
6613 batchSize_, oofEnabled_,
6614 topN_, false));
6615 return result;
6616 }
6636 MessageStream sowAndSubscribe(const char* topic_,
6637 long timeout_,
6638 const std::string& filter_ = "",
6639 int batchSize_ = DEFAULT_BATCH_SIZE,
6640 bool oofEnabled_ = false,
6641 int topN_ = DEFAULT_TOP_N)
6642 {
6643 MessageStream result(*this);
6644 if (_body.get().getDefaultMaxDepth())
6645 {
6646 result.maxDepth(_body.get().getDefaultMaxDepth());
6647 }
6648 result.setSubscription(_body.get().sowAndSubscribe(
6649 result.operator MessageHandler(),
6650 topic_, timeout_, filter_,
6651 batchSize_, oofEnabled_,
6652 topN_, false));
6653 return result;
6654 }
6655
6656
6684 std::string sowAndSubscribe(const MessageHandler& messageHandler_,
6685 const std::string& topic_,
6686 const std::string& filter_ = "",
6687 const std::string& orderBy_ = "",
6688 const std::string& bookmark_ = "",
6689 int batchSize_ = DEFAULT_BATCH_SIZE,
6690 int topN_ = DEFAULT_TOP_N,
6691 const std::string& options_ = "",
6692 long timeout_ = DEFAULT_COMMAND_TIMEOUT)
6693 {
6694 return _body.get().sowAndSubscribe(messageHandler_, topic_, filter_,
6695 orderBy_, bookmark_, batchSize_,
6696 topN_, options_, timeout_);
6697 }
6698
6723 MessageStream sowAndSubscribe(const std::string& topic_,
6724 const std::string& filter_ = "",
6725 const std::string& orderBy_ = "",
6726 const std::string& bookmark_ = "",
6727 int batchSize_ = DEFAULT_BATCH_SIZE,
6728 int topN_ = DEFAULT_TOP_N,
6729 const std::string& options_ = "",
6730 long timeout_ = DEFAULT_COMMAND_TIMEOUT)
6731 {
6732 MessageStream result(*this);
6733 if (_body.get().getDefaultMaxDepth())
6734 {
6735 result.maxDepth(_body.get().getDefaultMaxDepth());
6736 }
6737 result.setSubscription(_body.get().sowAndSubscribe(
6738 result.operator MessageHandler(),
6739 topic_, filter_, orderBy_,
6740 bookmark_, batchSize_, topN_,
6741 options_, timeout_, false));
6742 return result;
6743 }
6744
6746 MessageStream sowAndSubscribe(const char* topic_,
6747 const std::string& filter_ = "",
6748 const std::string& orderBy_ = "",
6749 const std::string& bookmark_ = "",
6750 int batchSize_ = DEFAULT_BATCH_SIZE,
6751 int topN_ = DEFAULT_TOP_N,
6752 const std::string& options_ = "",
6753 long timeout_ = DEFAULT_COMMAND_TIMEOUT)
6754 {
6755 MessageStream result(*this);
6756 if (_body.get().getDefaultMaxDepth())
6757 {
6758 result.maxDepth(_body.get().getDefaultMaxDepth());
6759 }
6760 result.setSubscription(_body.get().sowAndSubscribe(
6761 result.operator MessageHandler(),
6762 topic_, filter_, orderBy_,
6763 bookmark_, batchSize_, topN_,
6764 options_, timeout_, false));
6765 return result;
6766 }
6767
6792 std::string sowAndDeltaSubscribe(const MessageHandler& messageHandler_,
6793 const std::string& topic_,
6794 const std::string& filter_ = "",
6795 const std::string& orderBy_ = "",
6796 int batchSize_ = DEFAULT_BATCH_SIZE,
6797 int topN_ = DEFAULT_TOP_N,
6798 const std::string& options_ = "",
6799 long timeout_ = DEFAULT_COMMAND_TIMEOUT)
6800 {
6801 return _body.get().sowAndDeltaSubscribe(messageHandler_, topic_,
6802 filter_, orderBy_, batchSize_,
6803 topN_, options_, timeout_);
6804 }
6825 MessageStream sowAndDeltaSubscribe(const std::string& topic_,
6826 const std::string& filter_ = "",
6827 const std::string& orderBy_ = "",
6828 int batchSize_ = DEFAULT_BATCH_SIZE,
6829 int topN_ = DEFAULT_TOP_N,
6830 const std::string& options_ = "",
6831 long timeout_ = DEFAULT_COMMAND_TIMEOUT)
6832 {
6833 MessageStream result(*this);
6834 if (_body.get().getDefaultMaxDepth())
6835 {
6836 result.maxDepth(_body.get().getDefaultMaxDepth());
6837 }
6838 result.setSubscription(_body.get().sowAndDeltaSubscribe(
6839 result.operator MessageHandler(),
6840 topic_, filter_, orderBy_,
6841 batchSize_, topN_, options_,
6842 timeout_, false));
6843 return result;
6844 }
6845
6847 MessageStream sowAndDeltaSubscribe(const char* topic_,
6848 const std::string& filter_ = "",
6849 const std::string& orderBy_ = "",
6850 int batchSize_ = DEFAULT_BATCH_SIZE,
6851 int topN_ = DEFAULT_TOP_N,
6852 const std::string& options_ = "",
6853 long timeout_ = DEFAULT_COMMAND_TIMEOUT)
6854 {
6855 MessageStream result(*this);
6856 if (_body.get().getDefaultMaxDepth())
6857 {
6858 result.maxDepth(_body.get().getDefaultMaxDepth());
6859 }
6860 result.setSubscription(_body.get().sowAndDeltaSubscribe(
6861 result.operator MessageHandler(),
6862 topic_, filter_, orderBy_,
6863 batchSize_, topN_, options_,
6864 timeout_, false));
6865 return result;
6866 }
6867
6892 std::string sowAndDeltaSubscribe(const MessageHandler& messageHandler_,
6893 const std::string& topic_,
6894 long timeout_,
6895 const std::string& filter_ = "",
6896 int batchSize_ = DEFAULT_BATCH_SIZE,
6897 bool oofEnabled_ = false,
6898 bool sendEmpties_ = false,
6899 int topN_ = DEFAULT_TOP_N)
6900 {
6901 return _body.get().sowAndDeltaSubscribe(messageHandler_, topic_,
6902 timeout_, filter_, batchSize_,
6903 oofEnabled_, sendEmpties_,
6904 topN_);
6905 }
6906
6928 MessageStream sowAndDeltaSubscribe(const std::string& topic_,
6929 long timeout_,
6930 const std::string& filter_ = "",
6931 int batchSize_ = DEFAULT_BATCH_SIZE,
6932 bool oofEnabled_ = false,
6933 bool sendEmpties_ = false,
6934 int topN_ = DEFAULT_TOP_N)
6935 {
6936 MessageStream result(*this);
6937 if (_body.get().getDefaultMaxDepth())
6938 {
6939 result.maxDepth(_body.get().getDefaultMaxDepth());
6940 }
6941 result.setSubscription(_body.get().sowAndDeltaSubscribe(
6942 result.operator MessageHandler(),
6943 topic_, timeout_, filter_,
6944 batchSize_, oofEnabled_,
6945 sendEmpties_, topN_, false));
6946 return result;
6947 }
6969 MessageStream sowAndDeltaSubscribe(const char* topic_,
6970 long timeout_,
6971 const std::string& filter_ = "",
6972 int batchSize_ = DEFAULT_BATCH_SIZE,
6973 bool oofEnabled_ = false,
6974 bool sendEmpties_ = false,
6975 int topN_ = DEFAULT_TOP_N)
6976 {
6977 MessageStream result(*this);
6978 if (_body.get().getDefaultMaxDepth())
6979 {
6980 result.maxDepth(_body.get().getDefaultMaxDepth());
6981 }
6982 result.setSubscription(_body.get().sowAndDeltaSubscribe(
6983 result.operator MessageHandler(),
6984 topic_, timeout_, filter_,
6985 batchSize_, oofEnabled_,
6986 sendEmpties_, topN_, false));
6987 return result;
6988 }
7008 std::string sowDelete(const MessageHandler& messageHandler,
7009 const std::string& topic,
7010 const std::string& filter,
7011 long timeout)
7012 {
7013 return _body.get().sowDelete(messageHandler, topic, filter, timeout);
7014 }
7031 Message sowDelete(const std::string& topic_, const std::string& filter_,
7032 long timeout_ = 0)
7033 {
7034 MessageStream stream(*this);
7035 stream.timeout((unsigned int)timeout_);
7036 char buf[Message::IdentifierLength + 1];
7037 buf[Message::IdentifierLength] = 0;
7038 AMPS_snprintf(buf, Message::IdentifierLength + 1, "%lx", MessageImpl::newId());
7039 Field cid(buf);
7040 try
7041 {
7042 stream.setStatsOnly(cid);
7043 _body.get().sowDelete(stream.operator MessageHandler(), topic_, filter_, timeout_, cid);
7044 return *(stream.begin());
7045 }
7046 catch (const DisconnectedException&)
7047 {
7048 removeMessageHandler(cid);
7049 throw;
7050 }
7051 catch (const TimedOutException&)
7052 {
7053 removeMessageHandler(cid);
7054 throw;
7055 }
7056 }
7057
7062 void startTimer()
7063 {
7064 _body.get().startTimer();
7065 }
7066
7073 std::string stopTimer(const MessageHandler& messageHandler)
7074 {
7075 return _body.get().stopTimer(messageHandler);
7076 }
7077
7099 std::string sowDeleteByKeys(const MessageHandler& messageHandler_,
7100 const std::string& topic_,
7101 const std::string& keys_,
7102 long timeout_ = 0)
7103 {
7104 return _body.get().sowDeleteByKeys(messageHandler_, topic_, keys_, timeout_);
7105 }
7126 Message sowDeleteByKeys(const std::string& topic_, const std::string& keys_,
7127 long timeout_ = 0)
7128 {
7129 MessageStream stream(*this);
7130 stream.timeout((unsigned int)timeout_);
7131 char buf[Message::IdentifierLength + 1];
7132 buf[Message::IdentifierLength] = 0;
7133 AMPS_snprintf(buf, Message::IdentifierLength + 1, "%lx", MessageImpl::newId());
7134 Field cid(buf);
7135 try
7136 {
7137 stream.setStatsOnly(cid);
7138 _body.get().sowDeleteByKeys(stream.operator MessageHandler(), topic_, keys_, timeout_, cid);
7139 return *(stream.begin());
7140 }
7141 catch (const DisconnectedException&)
7142 {
7143 removeMessageHandler(cid);
7144 throw;
7145 }
7146 catch (const TimedOutException&)
7147 {
7148 removeMessageHandler(cid);
7149 throw;
7150 }
7151 }
7152
7167 std::string sowDeleteByData(const MessageHandler& messageHandler_,
7168 const std::string& topic_, const std::string& data_,
7169 long timeout_ = 0)
7170 {
7171 return _body.get().sowDeleteByData(messageHandler_, topic_, data_, timeout_);
7172 }
7173
7188 Message sowDeleteByData(const std::string& topic_, const std::string& data_,
7189 long timeout_ = 0)
7190 {
7191 MessageStream stream(*this);
7192 stream.timeout((unsigned int)timeout_);
7193 char buf[Message::IdentifierLength + 1];
7194 buf[Message::IdentifierLength] = 0;
7195 AMPS_snprintf(buf, Message::IdentifierLength + 1, "%lx", MessageImpl::newId());
7196 Field cid(buf);
7197 try
7198 {
7199 stream.setStatsOnly(cid);
7200 _body.get().sowDeleteByData(stream.operator MessageHandler(), topic_, data_, timeout_, cid);
7201 return *(stream.begin());
7202 }
7203 catch (const DisconnectedException&)
7204 {
7205 removeMessageHandler(cid);
7206 throw;
7207 }
7208 catch (const TimedOutException&)
7209 {
7210 removeMessageHandler(cid);
7211 throw;
7212 }
7213 }
7214
7218 amps_handle getHandle()
7219 {
7220 return _body.get().getHandle();
7221 }
7222
7231 void setExceptionListener(const std::shared_ptr<const ExceptionListener>& pListener_)
7232 {
7233 _body.get().setExceptionListener(pListener_);
7234 }
7235
7244 #if defined(_WIN32) || __cplusplus >= 201402L
7245 [[deprecated("Use setExceptionListener(std::shared_ptr<const ExceptionListener>&)")]]
7246 #endif
7247 void setExceptionListener(const ExceptionListener& listener_)
7248 {
7249 _body.get().setExceptionListener(listener_);
7250 }
7251
7254 const ExceptionListener& getExceptionListener(void) const
7255 {
7256 return _body.get().getExceptionListener();
7257 }
7258
7266 // type of message) from the server for the specified interval (plus a grace period),
7280 void setHeartbeat(unsigned heartbeatTime_, unsigned readTimeout_)
7281 {
7282 _body.get().setHeartbeat(heartbeatTime_, readTimeout_);
7283 }
7284
7292 // type of message) from the server for the specified interval (plus a grace period),
7304 void setHeartbeat(unsigned heartbeatTime_)
7305 {
7306 _body.get().setHeartbeat(heartbeatTime_, 2 * heartbeatTime_);
7307 }
7308
7310 #if defined(_WIN32) || __cplusplus >= 201402L
7311 [[deprecated("Use setLastChanceMessageHandler.")]]
7312 #endif
7313 void setUnhandledMessageHandler(const AMPS::MessageHandler& messageHandler)
7314 {
7315 setLastChanceMessageHandler(messageHandler);
7316 }
7317
7320 void setLastChanceMessageHandler(const AMPS::MessageHandler& messageHandler)
7321 {
7322 _body.get().setGlobalCommandTypeMessageHandler(ClientImpl::GlobalCommandTypeHandlers::LastChance,
7323 messageHandler);
7324 }
7325
7346 void setGlobalCommandTypeMessageHandler(const std::string& command_, const MessageHandler& handler_)
7347 {
7348 _body.get().setGlobalCommandTypeMessageHandler(command_, handler_);
7349 }
7350
7371 void setGlobalCommandTypeMessageHandler(const Message::Command::Type command_, const MessageHandler& handler_)
7372 {
7373 _body.get().setGlobalCommandTypeMessageHandler(command_, handler_);
7374 }
7375
7381 static const char* BOOKMARK_NOW()
7382 {
7383 return AMPS_BOOKMARK_NOW;
7384 }
7390 static const char* NOW()
7391 {
7392 return AMPS_BOOKMARK_NOW;
7393 }
7394
7400 static const char* BOOKMARK_EPOCH()
7401 {
7402 return AMPS_BOOKMARK_EPOCH;
7403 }
7404
7410 static const char* EPOCH()
7411 {
7412 return AMPS_BOOKMARK_EPOCH;
7413 }
7414
7421 static const char* BOOKMARK_MOST_RECENT()
7422 {
7423 return AMPS_BOOKMARK_RECENT;
7424 }
7425
7432 static const char* MOST_RECENT()
7433 {
7434 return AMPS_BOOKMARK_RECENT;
7435 }
7436
7443 static const char* BOOKMARK_RECENT()
7444 {
7445 return AMPS_BOOKMARK_RECENT;
7446 }
7447
7448
7454 void addConnectionStateListener(ConnectionStateListener* listener)
7455 {
7456 _body.get().addConnectionStateListener(listener);
7457 }
7458
7462 void removeConnectionStateListener(ConnectionStateListener* listener)
7463 {
7464 _body.get().removeConnectionStateListener(listener);
7465 }
7466
7469 void clearConnectionStateListeners()
7470 {
7471 _body.get().clearConnectionStateListeners();
7472 }
7473
7499 std::string executeAsync(Command& command_, MessageHandler handler_)
7500 {
7501 return _body.get().executeAsync(command_, handler_);
7502 }
7503
7533 std::string executeAsyncNoResubscribe(Command& command_,
7534 MessageHandler handler_)
7535 {
7536 std::string id;
7537 try
7538 {
7539 if (command_.isSubscribe())
7540 {
7541 Message& message = command_.getMessage();
7542 Field subId = message.getSubscriptionId();
7543 bool useExistingHandler = !subId.empty() && !message.getOptions().empty() && message.getOptions().contains("replace", 7);
7544 if (useExistingHandler)
7545 {
7546 MessageHandler existingHandler;
7547 if (_body.get()._routes.getRoute(subId, existingHandler))
7548 {
7549 // we found an existing handler.
7550 _body.get().executeAsync(command_, existingHandler, false);
7551 return id; // empty string indicates existing
7552 }
7553 }
7554 }
7555 id = _body.get().executeAsync(command_, handler_, false);
7556 }
7557 catch (const DisconnectedException&)
7558 {
7559 removeMessageHandler(command_.getMessage().getCommandId());
7560 if (command_.isSubscribe())
7561 {
7562 removeMessageHandler(command_.getMessage().getSubscriptionId());
7563 }
7564 if (command_.isSow())
7565 {
7566 removeMessageHandler(command_.getMessage().getQueryID());
7567 }
7568 throw;
7569 }
7570 return id;
7571 }
7572
7585 MessageStream execute(Command& command_);
7586
7595 void ack(Field& topic_, Field& bookmark_, const char* options_ = NULL)
7596 {
7597 _body.get().ack(topic_, bookmark_, options_);
7598 }
7599
7607 void ack(Message& message_, const char* options_ = NULL)
7608 {
7609 _body.get().ack(message_.getTopic(), message_.getBookmark(), options_);
7610 }
7619 void ack(const std::string& topic_, const std::string& bookmark_,
7620 const char* options_ = NULL)
7621 {
7622 _body.get().ack(Field(topic_.data(), topic_.length()), Field(bookmark_.data(), bookmark_.length()), options_);
7623 }
7624
7630 void ackDeferredAutoAck(Field& topic_, Field& bookmark_, const char* options_ = NULL)
7631 {
7632 _body.get()._ack(topic_, bookmark_, options_);
7633 }
7643 void flushAcks(void)
7644 {
7645 _body.get().flushAcks();
7646 }
7647
7652 bool getAutoAck(void) const
7653 {
7654 return _body.get().getAutoAck();
7655 }
7662 void setAutoAck(bool isAutoAckEnabled_)
7663 {
7664 _body.get().setAutoAck(isAutoAckEnabled_);
7665 }
7670 unsigned getAckBatchSize(void) const
7671 {
7672 return _body.get().getAckBatchSize();
7673 }
7680 void setAckBatchSize(const unsigned ackBatchSize_)
7681 {
7682 _body.get().setAckBatchSize(ackBatchSize_);
7683 }
7684
7691 int getAckTimeout(void) const
7692 {
7693 return _body.get().getAckTimeout();
7694 }
7703 void setAckTimeout(const int ackTimeout_)
7704 {
7705 if (!ackTimeout_ && _body.get().getAckBatchSize() > 1)
7706 {
7707 throw UsageException("Ack timeout must be > 0 when ack batch size > 1");
7708 }
7709 _body.get().setAckTimeout(ackTimeout_);
7710 }
7711
7712
7721 void setRetryOnDisconnect(bool isRetryOnDisconnect_)
7722 {
7723 _body.get().setRetryOnDisconnect(isRetryOnDisconnect_);
7724 }
7725
7730 bool getRetryOnDisconnect(void) const
7731 {
7732 return _body.get().getRetryOnDisconnect();
7733 }
7734
7739 void setDefaultMaxDepth(unsigned maxDepth_)
7740 {
7741 _body.get().setDefaultMaxDepth(maxDepth_);
7742 }
7743
7748 unsigned getDefaultMaxDepth(void) const
7749 {
7750 return _body.get().getDefaultMaxDepth();
7751 }
7752
7759 void setTransportFilterFunction(amps_transport_filter_function filter_,
7760 void* userData_)
7761 {
7762 return _body.get().setTransportFilterFunction(filter_, userData_);
7763 }
7764
7773 void setThreadCreatedCallback(amps_thread_created_callback callback_,
7774 void* userData_)
7775 {
7776 return _body.get().setThreadCreatedCallback(callback_, userData_);
7777 }
7778
7785 void setPublishBatching(size_t batchSize_, amps_uint64_t batchTimeoutMillis_)
7786 {
7787 return _body.get().setPublishBatching(batchSize_, batchTimeoutMillis_);
7788 }
7789
7795 void deferredExecution(DeferredExecutionFunc func_, void* userData_)
7796 {
7797 _body.get().deferredExecution(func_, userData_);
7798 }
7802 };
7803
7804 inline void
7805 ClientImpl::lastChance(AMPS::Message& message)
7806 {
7807 AMPS_CALL_EXCEPTION_WRAPPER(_globalCommandTypeHandlers[GlobalCommandTypeHandlers::LastChance].invoke(message));
7808 }
7809
7810 inline unsigned
7811 ClientImpl::persistedAck(AMPS::Message& message)
7812 {
7813 unsigned deliveries = 0;
7814 try
7815 {
7816 /*
7817 * Best Practice: If you don't care about the dupe acks that
7818 * occur during failover or rapid disconnect/reconnect, then just
7819 * ignore them. We could discard each duplicate from the
7820 * persisted store, but the storage costs of doing 1 record
7821 * discards is heavy. In most scenarios we'll just quickly blow
7822 * through the duplicates and get back to processing the
7823 * non-dupes.
7824 */
7825 const char* data = NULL;
7826 size_t len = 0;
7827 const char* status = NULL;
7828 size_t statusLen = 0;
7829 amps_handle messageHandle = message.getMessage();
7830 const size_t NotEntitled = 12, Duplicate = 9, Failure = 7;
7831 amps_message_get_field_value(messageHandle, AMPS_Reason, &data, &len);
7832 amps_message_get_field_value(messageHandle, AMPS_Status, &status, &statusLen);
7833 if (len == NotEntitled || len == Duplicate ||
7834 (statusLen == Failure && status[0] == 'f'))
7835 {
7836 if (_failedWriteHandler)
7837 {
7838 if (_publishStore.isValid())
7839 {
7840 amps_uint64_t sequence =
7841 amps_message_get_field_uint64(messageHandle, AMPS_Sequence);
7842 FailedWriteStoreReplayer replayer(this, data, len);
7843 AMPS_CALL_EXCEPTION_WRAPPER(_publishStore.replaySingle(
7844 replayer, sequence));
7845 }
7846 else // Call the handler with what little we have
7847 {
7848 Message emptyMessage;
7849 emptyMessage.setSequence(message.getSequence());
7850 AMPS_CALL_EXCEPTION_WRAPPER(
7851 _failedWriteHandler->failedWrite(emptyMessage,
7852 data, len));
7853 }
7854 ++deliveries;
7855 }
7856 }
7857 if (_publishStore.isValid())
7858 {
7859 // Ack for publisher will have sequence while
7860 // ack for bookmark subscribe won't
7861 amps_uint64_t seq = amps_message_get_field_uint64(messageHandle,
7862 AMPS_Sequence);
7863 if (seq > 0)
7864 {
7865 ++deliveries;
7866 AMPS_CALL_EXCEPTION_WRAPPER(_publishStore.discardUpTo(seq));
7867 }
7868 }
7869
7870 if (!deliveries && _bookmarkStore.isValid())
7871 {
7872 amps_message_get_field_value(messageHandle, AMPS_SubscriptionId,
7873 &data, &len);
7874 if (len > 0)
7875 {
7876 Message::Field subId(data, len);
7877 const char* bookmarkData = NULL;
7878 size_t bookmarkLen = 0;
7879 amps_message_get_field_value(messageHandle,
7880 AMPS_Bookmark,
7881 &bookmarkData,
7882 &bookmarkLen);
7883 // Everything is there and not unsubscribed AC-912
7884 if (bookmarkLen > 0 && _routes.hasRoute(subId))
7885 {
7886 ++deliveries;
7887 _bookmarkStore.persisted(subId, Message::Field(bookmarkData, bookmarkLen));
7888 }
7889 }
7890 }
7891 }
7892 catch (std::exception& ex)
7893 {
7894 AMPS_UNHANDLED_EXCEPTION(ex);
7895 }
7896 return deliveries;
7897 }
7898
7899 inline unsigned
7900 ClientImpl::processedAck(Message& message)
7901 {
7902 unsigned deliveries = 0;
7903 AckResponse ack;
7904 const char* data = NULL;
7905 size_t len = 0;
7906 amps_handle messageHandle = message.getMessage();
7907 amps_message_get_field_value(messageHandle, AMPS_CommandId, &data, &len);
7908 Lock<Mutex> l(_lock);
7909 if (data && len)
7910 {
7911 Lock<Mutex> guard(_ackMapLock);
7912 AckMap::iterator i = _ackMap.find(std::string(data, len));
7913 if (i != _ackMap.end())
7914 {
7915 ++deliveries;
7916 ack = i->second;
7917 _ackMap.erase(i);
7918 }
7919 }
7920 if (deliveries)
7921 {
7922 amps_message_get_field_value(messageHandle, AMPS_Status, &data, &len);
7923 ack.setStatus(data, len);
7924 amps_message_get_field_value(messageHandle, AMPS_Reason, &data, &len);
7925 ack.setReason(data, len);
7926 amps_message_get_field_value(messageHandle, AMPS_UserId, &data, &len);
7927 ack.setUsername(data, len);
7928 amps_message_get_field_value(messageHandle, AMPS_Password, &data, &len);
7929 ack.setPassword(data, len);
7930 amps_message_get_field_value(messageHandle, AMPS_Version, &data, &len);
7931 ack.setServerVersion(data, len);
7932 amps_message_get_field_value(messageHandle, AMPS_Options, &data, &len);
7933 ack.setOptions(data, len);
7934 // This sets bookmark, nameHashValue, and sequenceNo
7935 ack.setBookmark(message.getBookmark());
7936 ack.setResponded();
7937 _lock.signalAll();
7938 }
7939 return deliveries;
7940 }
7941
7942 inline void
7943 ClientImpl::checkAndSendHeartbeat(bool force)
7944 {
7945 if (force || _heartbeatTimer.check())
7946 {
7947 _heartbeatTimer.start();
7948 try
7949 {
7950 sendWithoutRetry(_beatMessage);
7951 }
7952 catch (const AMPSException&)
7953 {
7954 ;
7955 }
7956 }
7957 }
7958
7959 inline ConnectionInfo ClientImpl::getConnectionInfo() const
7960 {
7961 ConnectionInfo info;
7962 std::ostringstream writer;
7963
7964 info["client.uri"] = _lastUri;
7965 info["client.name"] = _name;
7966 info["client.username"] = _username;
7967 if (_publishStore.isValid())
7968 {
7969 writer << _publishStore.unpersistedCount();
7970 info["publishStore.unpersistedCount"] = writer.str();
7971 writer.clear();
7972 writer.str("");
7973 }
7974
7975 return info;
7976 }
7977
7978 inline std::string ClientImpl::logon(long timeout_, Authenticator& authenticator_,
7979 const char* options_)
7980 {
7981 Lock<Mutex> l(_lock);
7982 std::string cmdId = _logon(timeout_, authenticator_, options_);
7983 // Resubscribe
7984 if (_subscriptionManager)
7985 {
7986 {
7987 Client wrapper(this, false);
7988 // Have to release the lock here or receive thread can't
7989 // invoke the message handler.
7990 Unlock<Mutex> unlock(_lock);
7991 _subscriptionManager->resubscribe(wrapper);
7992 }
7993 broadcastConnectionStateChanged(ConnectionStateListener::Resubscribed);
7994 }
7995 return cmdId;
7996 }
7997
7998 inline amps_result
7999 ClientImpl::ClientImplMessageHandler(amps_handle messageHandle_, void* userData_)
8000 {
8001 const unsigned SOWMask = Message::Command::SOW | Message::Command::GroupBegin | Message::Command::GroupEnd;
8002 const unsigned PublishMask = Message::Command::OOF | Message::Command::Publish | Message::Command::DeltaPublish;
8003 ClientImpl* me = (ClientImpl*) userData_;
8004 AMPS_CALL_EXCEPTION_WRAPPER_2(me, me->processDeferredExecutions());
8005 if (!messageHandle_)
8006 {
8007 if (me->_queueAckTimeout)
8008 {
8009 me->checkQueueAcks();
8010 }
8011 me->checkAndSendHeartbeat();
8012 return AMPS_E_OK;
8013 }
8014
8015 me->_readMessage.replace(messageHandle_);
8016 Message& message = me->_readMessage;
8017 Message::Command::Type commandType = message.getCommandEnum();
8018 if (commandType & SOWMask)
8019 {
8020 #if 0 // Not currently implemented, to avoid an extra branch in delivery
8021 // A small cheat here to get the right handler, using knowledge of the
8022 // Command values of SOW (8), GroupBegin (8192), and GroupEnd (16384)
8023 // and their GlobalCommandTypeHandlers values 1, 2, 3.
8024 AMPS_CALL_EXCEPTION_WRAPPER_2(me,
8025 me->_globalCommandTypeHandlers[1 + (commandType / 8192)].invoke(message));
8026 #endif
8027 AMPS_CALL_EXCEPTION_WRAPPER_2(me, me->_routes.deliverData(message,
8028 message.getQueryID()));
8029 }
8030 else if (commandType & PublishMask)
8031 {
8032 #if 0 // Not currently implemented, to avoid an extra branch in delivery
8033 AMPS_CALL_EXCEPTION_WRAPPER_2(me,
8034 me->_globalCommandTypeHandlers[(commandType == Message::Command::Publish ?
8035 GlobalCommandTypeHandlers::Publish :
8036 GlobalCommandTypeHandlers::OOF)].invoke(message));
8037 #endif
8038 const char* subIds = NULL;
8039 size_t subIdsLen = 0;
8040 // Publish command, send to subscriptions
8041 amps_message_get_field_value(messageHandle_, AMPS_SubscriptionIds,
8042 &subIds, &subIdsLen);
8043 size_t subIdCount = me->_routes.parseRoutes(AMPS::Field(subIds, subIdsLen), me->_routeCache);
8044 for (size_t i = 0; i < subIdCount; ++i)
8045 {
8046 MessageRouter::RouteCache::value_type& lookupResult = me->_routeCache[i];
8047 MessageHandler& handler = lookupResult.handler;
8048 if (handler.isValid())
8049 {
8050 amps_message_set_field_value(messageHandle_,
8051 AMPS_SubscriptionId,
8052 subIds + lookupResult.idOffset,
8053 lookupResult.idLength);
8054 Message::Field bookmark = message.getBookmark();
8055 bool isMessageQueue = message.getLeasePeriod().len() != 0;
8056 bool isAutoAck = me->_isAutoAckEnabled;
8057
8058 if (!isMessageQueue && !bookmark.empty() &&
8059 me->_bookmarkStore.isValid())
8060 {
8061 if (me->_bookmarkStore.isDiscarded(me->_readMessage))
8062 {
8063 //Call duplicate message handler in handlers map
8064 if (me->_globalCommandTypeHandlers[GlobalCommandTypeHandlers::DuplicateMessage].isValid())
8065 {
8066 AMPS_CALL_EXCEPTION_WRAPPER_2(me, me->_globalCommandTypeHandlers[GlobalCommandTypeHandlers::DuplicateMessage].invoke(message));
8067 }
8068 }
8069 else
8070 {
8071 me->_bookmarkStore.log(me->_readMessage);
8072 AMPS_CALL_EXCEPTION_WRAPPER_2(me,
8073 handler.invoke(message));
8074 }
8075 }
8076 else
8077 {
8078 if (isMessageQueue && isAutoAck)
8079 {
8080 try
8081 {
8082 AMPS_CALL_EXCEPTION_WRAPPER_STREAM_FULL_2(me, handler.invoke(message));
8083 if (!message.getIgnoreAutoAck())
8084 {
8085 AMPS_CALL_EXCEPTION_WRAPPER_2(me,
8086 me->_ack(message.getTopic(), message.getBookmark()));
8087 }
8088 }
8089 catch (std::exception& ex)
8090 {
8091 if (!message.getIgnoreAutoAck())
8092 {
8093 AMPS_CALL_EXCEPTION_WRAPPER_2(me,
8094 me->_ack(message.getTopic(), message.getBookmark(), "cancel"));
8095 }
8096 AMPS_UNHANDLED_EXCEPTION_2(me, ex);
8097 }
8098 }
8099 else
8100 {
8101 AMPS_CALL_EXCEPTION_WRAPPER_2(me,
8102 handler.invoke(message));
8103 }
8104 }
8105 }
8106 else
8107 {
8108 me->lastChance(message);
8109 }
8110 } // for (subidsEnd)
8111 }
8112 else if (commandType == Message::Command::Ack)
8113 {
8114 unsigned ackType = message.getAckTypeEnum();
8115 unsigned deliveries = 0U;
8116 switch (ackType)
8117 {
8118 case Message::AckType::Persisted:
8119 deliveries += me->persistedAck(message);
8120 break;
8121 case Message::AckType::Processed: // processed
8122 deliveries += me->processedAck(message);
8123 break;
8124 }
8125 MessageHandler ackHandler = me->_globalCommandTypeHandlers[GlobalCommandTypeHandlers::Ack];
8126 if (ackHandler.isValid())
8127 {
8128 AMPS_CALL_EXCEPTION_WRAPPER_2(me, ackHandler.invoke(message));
8129 ++deliveries;
8130 }
8131 AMPS_CALL_EXCEPTION_WRAPPER_2(me, deliveries += me->_routes.deliverAck(message, ackType));
8132 if (deliveries == 0)
8133 {
8134 me->lastChance(message);
8135 }
8136 }
8137 else if (commandType == Message::Command::Heartbeat)
8138 {
8139 AMPS_CALL_EXCEPTION_WRAPPER_2(me,
8140 me->_globalCommandTypeHandlers[GlobalCommandTypeHandlers::Heartbeat].invoke(message));
8141 if (me->_heartbeatTimer.getTimeout() != 0.0) // -V550
8142 {
8143 me->checkAndSendHeartbeat(true);
8144 }
8145 else
8146 {
8147 me->lastChance(message);
8148 }
8149 return AMPS_E_OK;
8150 }
8151 else if (!message.getCommandId().empty())
8152 {
8153 unsigned deliveries = 0U;
8154 try
8155 {
8156 while (me->_connected) // Keep sending heartbeats when stream is full
8157 {
8158 try
8159 {
8160 deliveries = me->_routes.deliverData(message, message.getCommandId());
8161 break;
8162 }
8163 #ifdef _WIN32
8164 catch (MessageStreamFullException&)
8165 #else
8166 catch (MessageStreamFullException& msfEx_)
8167 #endif
8168 {
8169 try
8170 {
8171 me->checkAndSendHeartbeat(false);
8172 }
8173 #ifdef _WIN32
8174 catch (std::exception&)
8175 #else
8176 catch (std::exception& stdEx_)
8177 #endif
8178 {
8179 ;
8180 }
8181 }
8182 }
8183 }
8184 catch (std::exception& stdEx_)
8185 {
8186 try
8187 {
8188 me->_exceptionListener->exceptionThrown(stdEx_);
8189 }
8190 catch (...)
8191 {
8192 ;
8193 }
8194 }
8195 if (deliveries == 0)
8196 {
8197 me->lastChance(message);
8198 }
8199 }
8200 me->checkAndSendHeartbeat();
8201 return AMPS_E_OK;
8202 }
8203
8204 inline void
8205 ClientImpl::ClientImplPreDisconnectHandler(amps_handle /*client*/, unsigned failedConnectionVersion, void* userData)
8206 {
8207 ClientImpl* me = (ClientImpl*) userData;
8208 //Client wrapper(me);
8209 // Go ahead and signal any waiters if they are around...
8210 me->clearAcks(failedConnectionVersion);
8211 }
8212
8213 inline amps_result
8214 ClientImpl::ClientImplDisconnectHandler(amps_handle /*client*/, void* userData)
8215 {
8216 ClientImpl* me = (ClientImpl*) userData;
8217 Lock<Mutex> l(me->_lock);
8218 Client wrapper(me, false);
8219 if (me->_connected)
8220 {
8221 me->broadcastConnectionStateChanged(ConnectionStateListener::Disconnected);
8222 }
8223 me->captureSubscriptionManagerGenerationCount();
8224 bool retryInProgress = false;
8225 try
8226 {
8227 me->_connected = false;
8228 me->_lock.signalAll();
8229 // Have to release the lock here or receive thread can't
8230 // invoke the message handler.
8231 Unlock<Mutex> unlock(me->_lock);
8232 me->_disconnectHandler.invoke(wrapper);
8233 }
8234 #ifdef _WIN32
8235 catch (const RetryOperationException&)
8236 #else
8237 catch (const RetryOperationException& ex)
8238 #endif
8239 {
8240 retryInProgress = true;
8241 }
8242 catch (const std::exception& ex)
8243 {
8244 AMPS_UNHANDLED_EXCEPTION_2(me, ex);
8245 }
8246 me->_lock.signalAll();
8247
8248 if (!me->_connected)
8249 {
8250 if (retryInProgress)
8251 {
8252 AMPS_UNHANDLED_EXCEPTION_2(me, RetryOperationException("Reconnect in progress."));
8253 }
8254 else
8255 {
8256 me->broadcastConnectionStateChanged(ConnectionStateListener::Shutdown);
8257 AMPS_UNHANDLED_EXCEPTION_2(me, DisconnectedException("Reconnect failed."));
8258 }
8259 return AMPS_E_DISCONNECTED;
8260 }
8261 return AMPS_E_OK;
8262 }
8263
8264 inline const char*
8265 ClientImpl::ClientImplGetHttpPreflightMessage(void* userData_)
8266 {
8267 ClientImpl* me = (ClientImpl*)userData_;
8268 std::ostringstream os;
8269 // [transport]://[user[:password]@][host]:port[/path][?uri_params]
8270 // firstColon is after transport
8271 size_t firstColon = me->_lastUri.find(':');
8272 // pathEnd is start of uri_params or npos
8273 size_t pathEnd = me->_lastUri.find('?');
8274 // lastColon separates host and port, last before pathEnd
8275 size_t lastColon = me->_lastUri.rfind(':', pathEnd);
8276 // at ends user/password and precedes host
8277 size_t at = me->_lastUri.rfind('@', lastColon);
8278 // hostStart is either after at or following firstColon ://
8279 size_t hostStart = at == std::string::npos ? firstColon + 3 : at + 1;
8280 size_t hostLen = lastColon - hostStart;
8281 // pathStart follows port
8282 size_t pathStart = me->_lastUri.find('/', lastColon);
8283 size_t pathLen = pathEnd;
8284 if (pathEnd != std::string::npos)
8285 {
8286 pathLen = pathEnd - pathStart;
8287 }
8288 os << "GET " << me->_lastUri.substr(pathStart, pathLen)
8289 << " HTTP/1.1\r\nHost: " << me->_lastUri.substr(hostStart, hostLen)
8290 << "\r\nConnection: upgrade\r\nUpgrade: "
8291 << me->_lastUri.substr(0, firstColon) << "\r\n";
8292 for (auto& header : me->_httpPreflightHeaders)
8293 {
8294 os << header << "\r\n";
8295 }
8296 os << "\r\n";
8297 me->_preflightMessage = os.str();
8298 return me->_preflightMessage.c_str();
8299 }
8300
8301 class FIX
8302 {
8303 const char* _data;
8304 size_t _len;
8305 char _fieldSep;
8306 public:
8307 class iterator
8308 {
8309 const char* _data;
8310 size_t _len;
8311 size_t _pos;
8312 char _fieldSep;
8313 iterator(const char* data_, size_t len_, size_t pos_, char fieldSep_)
8314 : _data(data_), _len(len_), _pos(pos_), _fieldSep(fieldSep_)
8315 {
8316 while (_pos != _len && _data[_pos] == _fieldSep)
8317 {
8318 ++_pos;
8319 }
8320 }
8321 public:
8322 typedef void* difference_type;
8323 typedef std::forward_iterator_tag iterator_category;
8324 typedef std::pair<Message::Field, Message::Field> value_type;
8325 typedef value_type* pointer;
8326 typedef value_type& reference;
8327 bool operator==(const iterator& rhs) const
8328 {
8329 return _pos == rhs._pos;
8330 }
8331 bool operator!=(const iterator& rhs) const
8332 {
8333 return _pos != rhs._pos;
8334 }
8335 iterator& operator++()
8336 {
8337 // Skip through the data
8338 while (_pos != _len && _data[_pos] != _fieldSep)
8339 {
8340 ++_pos;
8341 }
8342 // Skip through any field separators
8343 while (_pos != _len && _data[_pos] == _fieldSep)
8344 {
8345 ++_pos;
8346 }
8347 return *this;
8348 }
8349
8350 value_type operator*() const
8351 {
8352 value_type result;
8353 size_t i = _pos, keyLength = 0, valueStart = 0, valueLength = 0;
8354 for (; i < _len && _data[i] != '='; ++i)
8355 {
8356 ++keyLength;
8357 }
8358
8359 result.first.assign(_data + _pos, keyLength);
8360
8361 if (i < _len && _data[i] == '=')
8362 {
8363 ++i;
8364 valueStart = i;
8365 for (; i < _len && _data[i] != _fieldSep; ++i)
8366 {
8367 valueLength++;
8368 }
8369 }
8370 result.second.assign(_data + valueStart, valueLength);
8371 return result;
8372 }
8373
8374 friend class FIX;
8375 };
8376 class reverse_iterator
8377 {
8378 const char* _data;
8379 size_t _len;
8380 const char* _pos;
8381 char _fieldSep;
8382 public:
8383 typedef std::pair<Message::Field, Message::Field> value_type;
8384 reverse_iterator(const char* data, size_t len, const char* pos, char fieldsep)
8385 : _data(data), _len(len), _pos(pos), _fieldSep(fieldsep)
8386 {
8387 if (_pos)
8388 {
8389 // skip past meaningless trailing fieldseps
8390 while (_pos >= _data && *_pos == _fieldSep)
8391 {
8392 --_pos;
8393 }
8394 while (_pos > _data && *_pos != _fieldSep)
8395 {
8396 --_pos;
8397 }
8398 // if we stopped before the 0th character, it's because
8399 // it's a field sep. advance one to point to the first character
8400 // of a key.
8401 if (_pos > _data || (_pos == _data && *_pos == _fieldSep))
8402 {
8403 ++_pos;
8404 }
8405 if (_pos < _data)
8406 {
8407 _pos = 0;
8408 }
8409 }
8410 }
8411 bool operator==(const reverse_iterator& rhs) const
8412 {
8413 return _pos == rhs._pos;
8414 }
8415 bool operator!=(const reverse_iterator& rhs) const
8416 {
8417 return _pos != rhs._pos;
8418 }
8419 reverse_iterator& operator++()
8420 {
8421 if (_pos == _data)
8422 {
8423 _pos = 0;
8424 }
8425 else
8426 {
8427 // back up 1 to a field separator
8428 --_pos;
8429 // keep backing up through field separators
8430 while (_pos >= _data && *_pos == _fieldSep)
8431 {
8432 --_pos;
8433 }
8434 // now back up to the beginning of this field
8435 while (_pos > _data && *_pos != _fieldSep)
8436 {
8437 --_pos;
8438 }
8439 if (_pos > _data || (_pos == _data && *_pos == _fieldSep))
8440 {
8441 ++_pos;
8442 }
8443 if (_pos < _data)
8444 {
8445 _pos = 0;
8446 }
8447 }
8448 return *this;
8449 }
8450 value_type operator*() const
8451 {
8452 value_type result;
8453 size_t keyLength = 0, valueStart = 0, valueLength = 0;
8454 size_t i = (size_t)(_pos - _data);
8455 for (; i < _len && _data[i] != '='; ++i)
8456 {
8457 ++keyLength;
8458 }
8459 result.first.assign(_pos, keyLength);
8460 if (i < _len && _data[i] == '=')
8461 {
8462 ++i;
8463 valueStart = i;
8464 for (; i < _len && _data[i] != _fieldSep; ++i)
8465 {
8466 valueLength++;
8467 }
8468 }
8469 result.second.assign(_data + valueStart, valueLength);
8470 return result;
8471 }
8472 };
8473 FIX(const Message::Field& data, char fieldSeparator = 1)
8474 : _data(data.data()), _len(data.len()),
8475 _fieldSep(fieldSeparator)
8476 {
8477 }
8478
8479 FIX(const char* data, size_t len, char fieldSeparator = 1)
8480 : _data(data), _len(len), _fieldSep(fieldSeparator)
8481 {
8482 }
8483
8484 iterator begin() const
8485 {
8486 return iterator(_data, _len, 0, _fieldSep);
8487 }
8488 iterator end() const
8489 {
8490 return iterator(_data, _len, _len, _fieldSep);
8491 }
8492
8493
8494 reverse_iterator rbegin() const
8495 {
8496 return reverse_iterator(_data, _len, _data + (_len - 1), _fieldSep);
8497 }
8498
8499 reverse_iterator rend() const
8500 {
8501 return reverse_iterator(_data, _len, 0, _fieldSep);
8502 }
8503 };
8504
8505
8518
8519 template <class T>
8520 class _FIXBuilder
8521 {
8522 std::stringstream _data;
8523 char _fs;
8524 public:
8530 _FIXBuilder(char fieldSep_ = (char)1) : _fs(fieldSep_) {;}
8531
8539 void append(const T& tag, const char* value, size_t offset, size_t length)
8540 {
8541 _data << tag << '=';
8542 _data.write(value + offset, (std::streamsize)length);
8543 _data << _fs;
8544 }
8550 void append(const T& tag, const std::string& value)
8551 {
8552 _data << tag << '=' << value << _fs;
8553 }
8554
8557 std::string getString() const
8558 {
8559 return _data.str();
8560 }
8561 operator std::string() const
8562 {
8563 return _data.str();
8564 }
8565
8567 void reset()
8568 {
8569 _data.str(std::string());
8570 }
8571 };
8572
8576
8577 typedef _FIXBuilder<int> FIXBuilder;
8578
8582
8583 typedef _FIXBuilder<std::string> NVFIXBuilder;
8584
8585
8593
8594 class FIXShredder
8595 {
8596 char _fs;
8597 public:
8602 FIXShredder(char fieldSep_ = (char)1) : _fs(fieldSep_) {;}
8603
8606 typedef std::map<Message::Field, Message::Field> map_type;
8607
8613 map_type toMap(const Message::Field& data)
8614 {
8615 FIX fix(data, _fs);
8616 map_type retval;
8617 for (FIX::iterator a = fix.begin(); a != fix.end(); ++a)
8618 {
8619 retval.insert(*a);
8620 }
8621
8622 return retval;
8623 }
8624 };
8625
8626 #define AMPS_MESSAGE_STREAM_CACHE_MAX 128
8627 class MessageStreamImpl : public AMPS::RefBody, AMPS::ConnectionStateListener
8628 {
8629 Mutex _lock;
8630 std::deque<Message> _q;
8631 std::deque<Message> _cache;
8632 std::string _commandId;
8633 std::string _subId;
8634 std::string _queryId;
8635 Client _client;
8636 unsigned _timeout;
8637 unsigned _maxDepth;
8638 unsigned _requestedAcks;
8639 size_t _cacheMax;
8640 Message::Field _previousTopic;
8641 Message::Field _previousBookmark;
8642 typedef enum : unsigned int { Unset = 0x0, Running = 0x10, Subscribe = 0x11, SOWOnly = 0x12, AcksOnly = 0x13, Conflate = 0x14, Closed = 0x1, Disconnected = 0x2 } State;
8643 #if __cplusplus >= 201100L || _MSC_VER >= 1900
8644 std::atomic<State> _state;
8645 #else
8646 volatile State _state;
8647 #endif
8648 typedef std::map<std::string, Message*> SOWKeyMap;
8649 SOWKeyMap _sowKeyMap;
8650 public:
8651 MessageStreamImpl(const Client& client_)
8652 : _client(client_),
8653 _timeout(0),
8654 _maxDepth((unsigned)~0),
8655 _requestedAcks(0),
8656 _cacheMax(AMPS_MESSAGE_STREAM_CACHE_MAX),
8657 _state(Unset)
8658 {
8659 if (_client.isValid())
8660 {
8661 _client.addConnectionStateListener(this);
8662 }
8663 }
8664
8665 MessageStreamImpl(ClientImpl* client_)
8666 : _client(client_, false),
8667 _timeout(0),
8668 _maxDepth((unsigned)~0),
8669 _requestedAcks(0),
8670 _state(Unset)
8671 {
8672 if (_client.isValid())
8673 {
8674 _client.addConnectionStateListener(this);
8675 }
8676 }
8677
8678 ~MessageStreamImpl()
8679 {
8680 }
8681
8682 virtual void destroy()
8683 {
8684 try
8685 {
8686 close();
8687 }
8688 catch (std::exception& e)
8689 {
8690 try
8691 {
8692 if (_client.isValid())
8693 {
8694 _client.getExceptionListener().exceptionThrown(e);
8695 }
8696 }
8697 catch (...) {/*Ignore exception listener exceptions*/} // -V565
8698 }
8699 if (_client.isValid())
8700 {
8701 _client.removeConnectionStateListener(this);
8702 Client c = _client;
8703 _client = Client((ClientImpl*)NULL);
8704 c.deferredExecution(MessageStreamImpl::destroyer, this);
8705 }
8706 else
8707 {
8708 delete this;
8709 }
8710 }
8711
8712 static void destroyer(void* vpMessageStreamImpl_)
8713 {
8714 delete ((MessageStreamImpl*)vpMessageStreamImpl_);
8715 }
8716
8717 void setSubscription(const std::string& subId_,
8718 const std::string& commandId_ = "",
8719 const std::string& queryId_ = "")
8720 {
8721 Lock<Mutex> lock(_lock);
8722 _subId = subId_;
8723 if (!commandId_.empty() && commandId_ != subId_)
8724 {
8725 _commandId = commandId_;
8726 }
8727 if (!queryId_.empty() && queryId_ != subId_ && queryId_ != commandId_)
8728 {
8729 _queryId = queryId_;
8730 }
8731 // It's possible to disconnect between creation/registration and here.
8732 if (Disconnected == _state)
8733 {
8734 return;
8735 }
8736 assert(Unset == _state);
8737 _state = Subscribe;
8738 }
8739
8740 void setSOWOnly(const std::string& commandId_,
8741 const std::string& queryId_ = "")
8742 {
8743 Lock<Mutex> lock(_lock);
8744 _commandId = commandId_;
8745 if (!queryId_.empty() && queryId_ != commandId_)
8746 {
8747 _queryId = queryId_;
8748 }
8749 // It's possible to disconnect between creation/registration and here.
8750 if (Disconnected == _state)
8751 {
8752 return;
8753 }
8754 assert(Unset == _state);
8755 _state = SOWOnly;
8756 }
8757
8758 void setStatsOnly(const std::string& commandId_,
8759 const std::string& queryId_ = "")
8760 {
8761 Lock<Mutex> lock(_lock);
8762 _commandId = commandId_;
8763 if (!queryId_.empty() && queryId_ != commandId_)
8764 {
8765 _queryId = queryId_;
8766 }
8767 // It's possible to disconnect between creation/registration and here.
8768 if (Disconnected == _state)
8769 {
8770 return;
8771 }
8772 assert(Unset == _state);
8773 _state = AcksOnly;
8774 _requestedAcks = Message::AckType::Stats;
8775 }
8776
8777 void setAcksOnly(const std::string& commandId_, unsigned acks_)
8778 {
8779 Lock<Mutex> lock(_lock);
8780 _commandId = commandId_;
8781 // It's possible to disconnect between creation/registration and here.
8782 if (Disconnected == _state)
8783 {
8784 return;
8785 }
8786 assert(Unset == _state);
8787 _state = AcksOnly;
8788 _requestedAcks = acks_;
8789 }
8790
8791 void connectionStateChanged(ConnectionStateListener::State state_)
8792 {
8793 Lock<Mutex> lock(_lock);
8794 if (state_ == AMPS::ConnectionStateListener::Disconnected)
8795 {
8796 _state = Disconnected;
8797 close();
8798 }
8799 else if (state_ == AMPS::ConnectionStateListener::Connected
8800 && _commandId.empty()
8801 && _subId.empty()
8802 && _queryId.empty())
8803 {
8804 // AC-1331 Reconnect before command was sent, so Unset
8805 _state = Unset;
8806 }
8807 _lock.signalAll();
8808 }
8809
8810 void timeout(unsigned timeout_)
8811 {
8812 _timeout = timeout_;
8813 }
8814 void conflate(void)
8815 {
8816 if (_state == Subscribe)
8817 {
8818 _state = Conflate;
8819 }
8820 }
8821 void maxDepth(unsigned maxDepth_)
8822 {
8823 if (maxDepth_)
8824 {
8825 _maxDepth = maxDepth_;
8826 }
8827 else
8828 {
8829 _maxDepth = (unsigned)~0;
8830 }
8831 }
8832 unsigned getMaxDepth(void) const
8833 {
8834 return _maxDepth;
8835 }
8836 unsigned getDepth(void) const
8837 {
8838 return (unsigned)(_q.size());
8839 }
8840
8841 bool next(Message& current_)
8842 {
8843 Lock<Mutex> lock(_lock);
8844 if (!_previousTopic.empty() && !_previousBookmark.empty())
8845 {
8846 try
8847 {
8848 if (_client.isValid())
8849 {
8850 _client.ackDeferredAutoAck(_previousTopic, _previousBookmark);
8851 }
8852 }
8853 #ifdef _WIN32
8854 catch (AMPSException&)
8855 #else
8856 catch (AMPSException& e)
8857 #endif
8858 {
8859 current_.invalidate();
8860 _previousTopic.clear();
8861 _previousBookmark.clear();
8862 return false;
8863 }
8864 _previousTopic.clear();
8865 _previousBookmark.clear();
8866 }
8867 // Don't wait to wait more than 1s at a time
8868 long minWaitTime = (_timeout && _timeout < 1000) ? _timeout : 1000;
8869 Timer timer((double)_timeout);
8870 timer.start();
8871 while (_q.empty() && _state & Running)
8872 {
8873 // Using timeout so python can interrupt
8874 _lock.wait(minWaitTime);
8875 {
8876 Unlock<Mutex> unlck(_lock);
8877 amps_invoke_waiting_function();
8878 }
8879 if (_timeout)
8880 {
8881 // In case we woke up early, see how much longer to wait
8882 if (timer.checkAndGetRemaining(&minWaitTime))
8883 {
8884 // No time left
8885 break;
8886 }
8887 // Adjust next wait time
8888 minWaitTime = (minWaitTime < 1000) ? minWaitTime : 1000;
8889 }
8890 }
8891 if (current_.isValid() && _cache.size() < _cacheMax)
8892 {
8893 current_.reset();
8894 _cache.push_back(current_);
8895 }
8896 if (!_q.empty())
8897 {
8898 current_ = _q.front();
8899 if (_q.size() == _maxDepth)
8900 {
8901 _lock.signalAll();
8902 }
8903 _q.pop_front();
8904 if (_state == Conflate)
8905 {
8906 std::string sowKey = current_.getSowKey();
8907 if (sowKey.length())
8908 {
8909 _sowKeyMap.erase(sowKey);
8910 }
8911 }
8912 else if (_state == AcksOnly)
8913 {
8914 _requestedAcks &= ~(current_.getAckTypeEnum());
8915 }
8916 if ((_state == AcksOnly && _requestedAcks == 0) ||
8917 (_state == SOWOnly && current_.getCommand() == "group_end"))
8918 {
8919 _state = Closed;
8920 }
8921 else if (current_.isValid()
8922 && current_.getCommandEnum() == Message::Command::Publish
8923 && _client.isValid() && _client.getAutoAck()
8924 && !current_.getLeasePeriod().empty()
8925 && !current_.getBookmark().empty())
8926 {
8927 _previousTopic = current_.getTopic().deepCopy();
8928 _previousBookmark = current_.getBookmark().deepCopy();
8929 }
8930 return true;
8931 }
8932 if (_state == Disconnected)
8933 {
8934 throw DisconnectedException("Connection closed.");
8935 }
8936 current_.invalidate();
8937 if (_state == Closed)
8938 {
8939 return false;
8940 }
8941 return _timeout != 0;
8942 }
8943 void close(void)
8944 {
8945 if (_client.isValid())
8946 {
8947 if (_state == SOWOnly || _state == Subscribe) //not delete
8948 {
8949 if (!_commandId.empty())
8950 {
8951 _client.unsubscribe(_commandId);
8952 }
8953 if (!_subId.empty())
8954 {
8955 _client.unsubscribe(_subId);
8956 }
8957 if (!_queryId.empty())
8958 {
8959 _client.unsubscribe(_queryId);
8960 }
8961 }
8962 else
8963 {
8964 if (!_commandId.empty())
8965 {
8966 _client.removeMessageHandler(_commandId);
8967 }
8968 if (!_subId.empty())
8969 {
8970 _client.removeMessageHandler(_subId);
8971 }
8972 if (!_queryId.empty())
8973 {
8974 _client.removeMessageHandler(_queryId);
8975 }
8976 }
8977 }
8978 if (_state == SOWOnly || _state == Subscribe || _state == Unset)
8979 {
8980 _state = Closed;
8981 }
8982 }
8983 static void _messageHandler(const Message& message_, MessageStreamImpl* this_)
8984 {
8985 Lock<Mutex> lock(this_->_lock);
8986 if (this_->_state != Conflate)
8987 {
8988 AMPS_TESTING_SLOW_MESSAGE_STREAM
8989 if (this_->_q.size() >= this_->_maxDepth)
8990 {
8991 // We throw here so that heartbeats can be sent. The exception
8992 // will be handled internally only, and the same Message will
8993 // come back to try again. Make sure to signal.
8994 this_->_lock.signalAll();
8995 throw MessageStreamFullException("Stream is currently full.");
8996 }
8997 if (!this_->_cache.empty())
8998 {
8999 this_->_cache.front().deepCopy(message_);
9000 this_->_q.push_back(this_->_cache.front());
9001 this_->_cache.pop_front();
9002 }
9003 else
9004 {
9005 this_->_q.emplace_back(message_.deepCopy());
9006 }
9007 if (message_.getCommandEnum() == Message::Command::Publish &&
9008 this_->_client.isValid() && this_->_client.getAutoAck() &&
9009 !message_.getLeasePeriod().empty() &&
9010 !message_.getBookmark().empty())
9011 {
9012 message_.setIgnoreAutoAck();
9013 }
9014 }
9015 else
9016 {
9017 std::string sowKey = message_.getSowKey();
9018 if (sowKey.length())
9019 {
9020 SOWKeyMap::iterator it = this_->_sowKeyMap.find(sowKey);
9021 if (it != this_->_sowKeyMap.end())
9022 {
9023 it->second->deepCopy(message_);
9024 }
9025 else
9026 {
9027 if (this_->_q.size() >= this_->_maxDepth)
9028 {
9029 // We throw here so that heartbeats can be sent. The
9030 // exception will be handled internally only, and the
9031 // same Message will come back to try again. Make sure
9032 // to signal.
9033 this_->_lock.signalAll();
9034 throw MessageStreamFullException("Stream is currently full.");
9035 }
9036 if (!this_->_cache.empty())
9037 {
9038 this_->_cache.front().deepCopy(message_);
9039 this_->_q.push_back(this_->_cache.front());
9040 this_->_cache.pop_front();
9041 }
9042 else
9043 {
9044 this_->_q.emplace_back(message_.deepCopy());
9045 }
9046 this_->_sowKeyMap[sowKey] = &(this_->_q.back());
9047 }
9048 }
9049 else
9050 {
9051 if (this_->_q.size() >= this_->_maxDepth)
9052 {
9053 // We throw here so that heartbeats can be sent. The exception
9054 // will be handled internally only, and the same Message will
9055 // come back to try again. Make sure to signal.
9056 this_->_lock.signalAll();
9057 throw MessageStreamFullException("Stream is currently full.");
9058 }
9059 if (!this_->_cache.empty())
9060 {
9061 this_->_cache.front().deepCopy(message_);
9062 this_->_q.push_back(this_->_cache.front());
9063 this_->_cache.pop_front();
9064 }
9065 else
9066 {
9067 this_->_q.emplace_back(message_.deepCopy());
9068 }
9069 if (message_.getCommandEnum() == Message::Command::Publish &&
9070 this_->_client.isValid() && this_->_client.getAutoAck() &&
9071 !message_.getLeasePeriod().empty() &&
9072 !message_.getBookmark().empty())
9073 {
9074 message_.setIgnoreAutoAck();
9075 }
9076 }
9077 }
9078 this_->_lock.signalAll();
9079 }
9080 };
9081 inline MessageStream::MessageStream(const Client& client_)
9082 : _body(new MessageStreamImpl(client_))
9083 {
9084 }
9085 inline MessageStream::MessageStream(RefHandle<MessageStreamImpl> body_)
9086 : _body(body_)
9087 {
9088 }
9089 inline void MessageStream::iterator::advance(void)
9090 {
9091 _pStream = _pStream->_body->next(_current) ? _pStream : NULL;
9092 }
9093 inline MessageStream::operator MessageHandler(void)
9094 {
9095 return MessageHandler((void(*)(const Message&, void*))MessageStreamImpl::_messageHandler, &_body.get());
9096 }
9097 inline MessageStream MessageStream::fromExistingHandler(const MessageHandler& handler_)
9098 {
9099 MessageStream result;
9100 if (handler_._func == (MessageHandler::FunctionType)MessageStreamImpl::_messageHandler)
9101 {
9102 result._body = (MessageStreamImpl*)(handler_._userData);
9103 }
9104 return result;
9105 }
9106
9107 inline void MessageStream::setSOWOnly(const std::string& commandId_,
9108 const std::string& queryId_)
9109 {
9110 _body->setSOWOnly(commandId_, queryId_);
9111 }
9112 inline void MessageStream::setSubscription(const std::string& subId_,
9113 const std::string& commandId_,
9114 const std::string& queryId_)
9115 {
9116 _body->setSubscription(subId_, commandId_, queryId_);
9117 }
9118 inline void MessageStream::setStatsOnly(const std::string& commandId_,
9119 const std::string& queryId_)
9120 {
9121 _body->setStatsOnly(commandId_, queryId_);
9122 }
9123 inline void MessageStream::setAcksOnly(const std::string& commandId_,
9124 unsigned acks_)
9125 {
9126 _body->setAcksOnly(commandId_, acks_);
9127 }
9128 inline MessageStream MessageStream::timeout(unsigned timeout_)
9129 {
9130 _body->timeout(timeout_);
9131 return *this;
9132 }
9133 inline MessageStream MessageStream::conflate(void)
9134 {
9135 _body->conflate();
9136 return *this;
9137 }
9138 inline MessageStream MessageStream::maxDepth(unsigned maxDepth_)
9139 {
9140 _body->maxDepth(maxDepth_);
9141 return *this;
9142 }
9143 inline unsigned MessageStream::getMaxDepth(void) const
9144 {
9145 return _body->getMaxDepth();
9146 }
9147 inline unsigned MessageStream::getDepth(void) const
9148 {
9149 return _body->getDepth();
9150 }
9151
9152 inline MessageStream ClientImpl::getEmptyMessageStream(void)
9153 {
9154 return MessageStream(_pEmptyMessageStream.get()->_body);
9155 }
9156
9157 inline MessageStream Client::execute(Command& command_)
9158 {
9159 // If the command is sow and has a sub_id, OR
9160 // if the command has a replace option, return the existing
9161 // messagestream, don't create a new one.
9162 ClientImpl& body = _body.get();
9163 Message& message = command_.getMessage();
9164 Field subId = message.getSubscriptionId();
9165 unsigned ackTypes = message.getAckTypeEnum();
9166 bool useExistingHandler = !subId.empty() && ((!message.getOptions().empty() && message.getOptions().contains("replace", 7)) || message.getCommandEnum() == Message::Command::SOW);
9167 if (useExistingHandler)
9168 {
9169 // Try to find the existing message handler.
9170 if (!subId.empty())
9171 {
9172 MessageHandler existingHandler;
9173 if (body._routes.getRoute(subId, existingHandler))
9174 {
9175 // we found an existing handler. It might not be a message stream, but that's okay.
9176 body.executeAsync(command_, existingHandler, false);
9177 return MessageStream::fromExistingHandler(existingHandler);
9178 }
9179 }
9180 // fall through; we'll a new handler altogether.
9181 }
9182 // Make sure something will be returned to the stream or use the empty one
9183 // Check that: it's a command that doesn't normally return data, and there
9184 // are no acks requested for the cmd id
9185 Message::Command::Type command = message.getCommandEnum();
9186 if ((command & Message::Command::NoDataCommands)
9187 && (ackTypes == Message::AckType::Persisted
9188 || ackTypes == Message::AckType::None))
9189 {
9190 executeAsync(command_, MessageHandler());
9191 if (!body._pEmptyMessageStream)
9192 {
9193 body._pEmptyMessageStream.reset(new MessageStream((ClientImpl*)0));
9194 body._pEmptyMessageStream.get()->_body->close();
9195 }
9196 return body.getEmptyMessageStream();
9197 }
9198 MessageStream stream(*this);
9199 if (body.getDefaultMaxDepth())
9200 {
9201 stream.maxDepth(body.getDefaultMaxDepth());
9202 }
9203 MessageHandler handler = stream.operator MessageHandler();
9204 std::string commandID = body.executeAsync(command_, handler, false);
9205 if (command_.hasStatsAck())
9206 {
9207 stream.setStatsOnly(commandID, command_.getMessage().getQueryId());
9208 }
9209 else if (command_.isSow())
9210 {
9211 if (command_.getAckTypeEnum() & Message::AckType::Completed)
9212 {
9213 stream.setAcksOnly(commandID,
9214 ackTypes);
9215 }
9216 else
9217 {
9218 stream.setSOWOnly(commandID, command_.getMessage().getQueryId());
9219 }
9220 }
9221 else if (command_.isSubscribe())
9222 {
9223 stream.setSubscription(commandID,
9224 command_.getMessage().getCommandId(),
9225 command_.getMessage().getQueryId());
9226 }
9227 else
9228 {
9229 // Persisted acks for writes don't come back with command id
9230 if (command == Message::Command::Publish ||
9231 command == Message::Command::DeltaPublish ||
9232 command == Message::Command::SOWDelete)
9233 {
9234 stream.setAcksOnly(commandID,
9235 ackTypes & (unsigned)~Message::AckType::Persisted);
9236 }
9237 else
9238 {
9239 stream.setAcksOnly(commandID, ackTypes);
9240 }
9241 }
9242 return stream;
9243 }
9244
9245 // This is here because it uses api from Client.
9246 inline void Message::ack(const char* options_) const
9247 {
9248 ClientImpl* pClient = _body.get().clientImpl();
9249 Message::Field bookmark = getBookmark();
9250 if (pClient && bookmark.len() &&
9251 !pClient->getAutoAck())
9252 //(!pClient->getAutoAck() || getIgnoreAutoAck()))
9253 {
9254 pClient->ack(getTopic(), bookmark, options_);
9255 }
9256 }
9257 }// end namespace AMPS
9258 #endif
amps.h
Core type and function declarations for the AMPS C client.
amps_client_disconnect
AMPSDLL void amps_client_disconnect(amps_handle handle)
Disconnects from the AMPS server, if connected.
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_client_destroy
AMPSDLL void amps_client_destroy(amps_handle handle)
Disconnects and destroys an AMPS client object.
amps_client_connect
AMPSDLL amps_result amps_client_connect(amps_handle handle, const amps_char *uri)
Connects to the AMPS server specified in uri.
amps_client_send
AMPSDLL amps_result amps_client_send(amps_handle client, amps_handle message)
Sends a message to the AMPS server.
amps_message_get_field_uint64
AMPSDLL amps_uint64_t amps_message_get_field_uint64(amps_handle message, FieldId field)
Gets the unsigned 64-bit int value of a header field in an AMPS message.
amps_client_set_name
AMPSDLL amps_result amps_client_set_name(amps_handle handle, const amps_char *clientName)
Sets the name on an amps client object.
amps_result
amps_result
Return values from amps_xxx functions.
Definition: amps.h:217
AMPS_E_RETRY
@ AMPS_E_RETRY
The operation has not succeeded, but ought to be retried.
Definition: amps.h:245
AMPS_E_OK
@ AMPS_E_OK
Success.
Definition: amps.h:221
AMPS_E_DISCONNECTED
@ AMPS_E_DISCONNECTED
The client and server are disconnected.
Definition: amps.h:249
amps_client_attempt_reconnect
AMPSDLL amps_result amps_client_attempt_reconnect(amps_handle client, unsigned version)
Manually invokes the user-supplied disconnect handler for this client.
amps_client_set_read_timeout
AMPSDLL amps_result amps_client_set_read_timeout(amps_handle client, int readTimeout)
Sets a read timeout (seconds), in which if no message is received, the connection is presumed dead.
amps_thread_created_callback
amps_result(* amps_thread_created_callback)(AMPS_THREAD_T, void *)
Prototype for a user-supplied callback function to allow thread attributes to be set when a new threa...
Definition: amps.h:696
amps_client_set_transport_filter_function
AMPSDLL amps_result amps_client_set_transport_filter_function(amps_handle client, amps_transport_filter_function filter, void *userData)
Sets a user-supplied callback function for filtering data before it is sent and after it is received.
amps_client_set_thread_created_callback
AMPSDLL amps_result amps_client_set_thread_created_callback(amps_handle client, amps_thread_created_callback callback, void *userData)
Sets a user-supplied callback function to allow thread attributes to set when a new thread is created...
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_client_set_predisconnect_handler
AMPSDLL void amps_client_set_predisconnect_handler(amps_handle client, amps_predisconnect_handler predisconnectHandler, void *userData)
Sets the predisconnect handler function to be called when a disconnect occurs.
amps_client_set_message_handler
AMPSDLL void amps_client_set_message_handler(amps_handle client, amps_handler messageHandler, void *userData)
Sets the message handler function for this client.
amps_client_send_batch
AMPSDLL amps_result amps_client_send_batch(amps_handle client, amps_handle message, unsigned *version_out, int addToBatch)
Adds a message to the send cache, possibly sending the cache.
amps_client_get_socket
AMPSDLL AMPS_SOCKET amps_client_get_socket(amps_handle client)
Returns the socket from the underlying transport in client, or NULL if no transport is associated wit...
amps_handle
void * amps_handle
Opaque handle type used to refer to objects in the AMPS api.
Definition: amps.h:211
amps_transport_filter_function
void(* amps_transport_filter_function)(const unsigned char *, size_t, short, void *)
Prototype for a user-supplied callback function for filtering data before it is sent and after it is ...
Definition: amps.h:670
amps_client_set_idle_time
AMPSDLL amps_result amps_client_set_idle_time(amps_handle client, int idleTime)
Sets an idle-time (milliseconds).
amps_client_set_batch_send
AMPSDLL void amps_client_set_batch_send(amps_handle client_, amps_uint64_t batchSizeBytes_, amps_uint64_t batchTimeout_)
Sets a byte size batchSizeBytes and timeout for using batch sends of publish and delta_publish messag...
amps_client_create
AMPSDLL amps_handle amps_client_create(const amps_char *clientName)
Functions for creation of an AMPS client.
amps_client_set_disconnect_handler
AMPSDLL void amps_client_set_disconnect_handler(amps_handle client, amps_handler disconnectHandler, void *userData)
Sets the disconnect handler function to be called when a disconnect occurs.
amps_client_set_http_preflight_callback
AMPSDLL amps_result amps_client_set_http_preflight_callback(amps_handle client, amps_http_preflight_callback callback, void *userData)
Sets a user-supplied callback function for when a connection is established and the provided uri incl...
AMPS::DangerousFlushPublishStoreResizeHandler
bool DangerousFlushPublishStoreResizeHandler(Store store_, size_t, void *data_)
PublishStoreResizeHandler that will block up to the timeout specified in user data milliseconds tryin...
Definition: ampsplusplus.hpp:1411
AMPS::PublishStoreResizeHandler
bool(* PublishStoreResizeHandler)(Store store_, size_t size_, void *userData_)
Function type for PublishStore resize events The store_ param is store which is resizing.
Definition: ampsplusplus.hpp:1091
AMPS::AMPSException
Base class for all exceptions in AMPS.
Definition: AMPSException.hpp:41
AMPS::AMPSException::throwFor
static void throwFor(const T &context_, amps_result result_)
Constructs and throws the appropriate exception corresponding to a given result.
AMPS::Authenticator
The interface for handling authentication with the AMPS server.
Definition: ampsplusplus.hpp:1007
AMPS::Authenticator::authenticate
virtual std::string authenticate(const std::string &userName_, const std::string &password_)=0
Called by Client just before the logon command is sent.
AMPS::Authenticator::completed
virtual void completed(const std::string &userName_, const std::string &password_, const std::string &reason_)=0
Called by Client once a logon completes successfully.
AMPS::Authenticator::retry
virtual std::string retry(const std::string &userName_, const std::string &password_)=0
Called by Client when a logon ack is received with a status of retry.
AMPS::BookmarkStore
Interface for BookmarkStoreImpl classes.
Definition: BookmarkStore.hpp:229
AMPS::Client
Client represents a connection to an AMPS server, but does not provide failover or reconnection behav...
Definition: ampsplusplus.hpp:5342
AMPS::Client::sow
MessageStream sow(const char *topic_, const std::string &filter_="", const std::string &orderBy_="", const std::string &bookmark_="", int batchSize_=DEFAULT_BATCH_SIZE, int topN_=DEFAULT_TOP_N, const std::string &options_="", long timeout_=DEFAULT_COMMAND_TIMEOUT)
Query the SOW cache of a topic.
Definition: ampsplusplus.hpp:6493
AMPS::Client::publish
amps_uint64_t publish(const char *topic_, size_t topicLength_, const char *data_, size_t dataLength_)
Publish a message to an AMPS topic, returning the sequence number assigned by the publish store if on...
Definition: ampsplusplus.hpp:5869
AMPS::Client::setExceptionListener
void setExceptionListener(const std::shared_ptr< const ExceptionListener > &pListener_)
Sets the exception listener for exceptions that are not thrown back to the user (for example,...
Definition: ampsplusplus.hpp:7231
AMPS::Client::logon
std::string logon(const char *options_, int timeout_=0)
Logon to the server, providing the client name, credentials (if available) client information (such a...
Definition: ampsplusplus.hpp:6101
AMPS::Client::flushAcks
void flushAcks(void)
Sends any queued message queue ack messages to the server immediately.
Definition: ampsplusplus.hpp:7643
AMPS::Client::startTimer
void startTimer()
Definition: ampsplusplus.hpp:7062
AMPS::Client::subscribe
MessageStream subscribe(const std::string &topic_, long timeout_=0, const std::string &filter_="", const std::string &options_="", const std::string &subId_="")
Subscribe to a topic.
Definition: ampsplusplus.hpp:6171
AMPS::Client::setBookmarkStore
void setBookmarkStore(const BookmarkStore &bookmarkStore_)
Set the bookmark store to be used by the client.
Definition: ampsplusplus.hpp:5716
AMPS::Client::getURI
const std::string & getURI() const
Returns the last URI this client is connected to.
Definition: ampsplusplus.hpp:5537
AMPS::Client::removeConnectionStateListener
void removeConnectionStateListener(ConnectionStateListener *listener)
Attempts to remove listener from self's set of ConnectionStateListeners.
Definition: ampsplusplus.hpp:7462
AMPS::Client::publish
amps_uint64_t publish(const std::string &topic_, const std::string &data_, unsigned long expiration_)
Publish a message to an AMPS topic, returning the sequence number assigned by the publish store (if a...
Definition: ampsplusplus.hpp:5893
AMPS::Client::BOOKMARK_EPOCH
static const char * BOOKMARK_EPOCH()
Convenience method for returning the special value to start a subscription at the beginning of the tr...
Definition: ampsplusplus.hpp:7400
AMPS::Client::deltaSubscribe
std::string deltaSubscribe(const MessageHandler &messageHandler_, const std::string &topic_, long timeout_, const std::string &filter_="", const std::string &options_="", const std::string &subId_="")
Delta Subscribe to a topic.
Definition: ampsplusplus.hpp:6232
AMPS::Client::getSubscriptionManager
SubscriptionManager * getSubscriptionManager()
Get the subscription manager being used by the client.
Definition: ampsplusplus.hpp:5732
AMPS::Client::setPublishBatching
void setPublishBatching(size_t batchSize_, amps_uint64_t batchTimeoutMillis_)
Sets the max bytes to cache and max timeout in millis for caching delta_publish and publish commands.
Definition: ampsplusplus.hpp:7785
AMPS::Client::bookmarkSubscribe
std::string bookmarkSubscribe(const MessageHandler &messageHandler_, const std::string &topic_, long timeout_, const std::string &bookmark_, const std::string &filter_="", const std::string &options_="", const std::string &subId_="")
Subscribe to a topic using a bookmark.
Definition: ampsplusplus.hpp:6310
AMPS::Client::addHttpPreflightHeader
void addHttpPreflightHeader(const std::string &key_, const std::string &value_)
Adds a given key/value pair as an HTTP header line as "key: value" to the end of the headers that wil...
Definition: ampsplusplus.hpp:5462
AMPS::Client::unsubscribe
void unsubscribe()
Unsubscribe from all topics.
Definition: ampsplusplus.hpp:6399
AMPS::Client::setHeartbeat
void setHeartbeat(unsigned heartbeatTime_)
Requests heartbeating with the AMPS server.
Definition: ampsplusplus.hpp:7304
AMPS::Client::setLastChanceMessageHandler
void setLastChanceMessageHandler(const AMPS::MessageHandler &messageHandler)
Sets the message handler called when no other handler matches.
Definition: ampsplusplus.hpp:7320
AMPS::Client::BOOKMARK_MOST_RECENT
static const char * BOOKMARK_MOST_RECENT()
Convenience method for returning the special value to start a subscription at a recovery point based ...
Definition: ampsplusplus.hpp:7421
AMPS::Client::getHandle
amps_handle getHandle()
Returns the underlying amps_handle for this client, to be used with amps_client_* functions from the ...
Definition: ampsplusplus.hpp:7218
AMPS::Client::setFailedWriteHandler
void setFailedWriteHandler(FailedWriteHandler *handler_)
Set the handler that is invoked to report when a publish fails, for example if the publisher is not e...
Definition: ampsplusplus.hpp:5813
AMPS::Client::setName
void setName(const std::string &name)
Sets the name of this client, assuming no name was provided previously.
Definition: ampsplusplus.hpp:5396
AMPS::Client::deltaPublish
amps_uint64_t deltaPublish(const std::string &topic_, const std::string &data_)
Publish the changed fields of a message to an AMPS topic.
Definition: ampsplusplus.hpp:5987
AMPS::Client::getFailedWriteHandler
FailedWriteHandler * getFailedWriteHandler()
Get the handler that is invoked to report on failed writes.
Definition: ampsplusplus.hpp:5821
AMPS::Client::setHttpPreflightHeaders
void setHttpPreflightHeaders(const T &headers_)
Sets the given HTTP header lines to be sent for the HTTP GET Upgrade request.
Definition: ampsplusplus.hpp:5477
AMPS::Client::setDefaultMaxDepth
void setDefaultMaxDepth(unsigned maxDepth_)
Sets a default max depth on all subsequently created MessageStream objects.
Definition: ampsplusplus.hpp:7739
AMPS::Client::publish
amps_uint64_t publish(const std::string &topic_, const std::string &data_)
Publish a message to an AMPS topic, returning the sequence number assigned by the publish store if on...
Definition: ampsplusplus.hpp:5844
AMPS::Client::sowAndDeltaSubscribe
MessageStream sowAndDeltaSubscribe(const char *topic_, const std::string &filter_="", const std::string &orderBy_="", int batchSize_=DEFAULT_BATCH_SIZE, int topN_=DEFAULT_TOP_N, const std::string &options_="", long timeout_=DEFAULT_COMMAND_TIMEOUT)
Query the SOW cache of a topic and initiates a new delta subscription on it.
Definition: ampsplusplus.hpp:6847
AMPS::Client::setAckBatchSize
void setAckBatchSize(const unsigned ackBatchSize_)
Sets the queue ack batch size setting.
Definition: ampsplusplus.hpp:7680
AMPS::Client::disconnect
void disconnect()
Disconnect from an AMPS server.
Definition: ampsplusplus.hpp:5568
AMPS::Client::NOW
static const char * NOW()
Convenience method for returning the special value to start a subscription at the end of the transact...
Definition: ampsplusplus.hpp:7390
AMPS::Client::clearHttpPreflightHeaders
void clearHttpPreflightHeaders()
Clears all previously set HTTP header lines.
Definition: ampsplusplus.hpp:5468
AMPS::Client::deltaPublish
amps_uint64_t deltaPublish(const char *topic_, size_t topicLength_, const char *data_, size_t dataLength_)
Publish the changed fields of a message to an AMPS topic.
Definition: ampsplusplus.hpp:6010
AMPS::Client::convertVersionToNumber
static size_t convertVersionToNumber(const char *data_, size_t len_)
Converts a string version, such as "3.8.1.5" into the same numeric form used internally and returned ...
Definition: ampsplusplus.hpp:5530
AMPS::Client::Client
Client(const std::string &clientName="")
Constructs a new client with a given client name.
Definition: ampsplusplus.hpp:5358
AMPS::Client::getExceptionListener
const ExceptionListener & getExceptionListener(void) const
Returns the exception listener set on this Client.
Definition: ampsplusplus.hpp:7254
AMPS::Client::unsubscribe
void unsubscribe(const std::string &commandId)
Unsubscribe from a topic.
Definition: ampsplusplus.hpp:6387
AMPS::Client::setExceptionListener
void setExceptionListener(const ExceptionListener &listener_)
Definition: ampsplusplus.hpp:7247
AMPS::Client::getDisconnectHandler
DisconnectHandler getDisconnectHandler(void) const
Definition: ampsplusplus.hpp:5694
AMPS::Client::getAckTimeout
int getAckTimeout(void) const
Returns the current value of the message queue ack timeout setting – that is, the amount of time afte...
Definition: ampsplusplus.hpp:7691
AMPS::Client::bookmarkSubscribe
MessageStream bookmarkSubscribe(const std::string &topic_, long timeout_, const std::string &bookmark_, const std::string &filter_="", const std::string &options_="", const std::string &subId_="")
Subscribe to a topic using a bookmark.
Definition: ampsplusplus.hpp:6338
AMPS::Client::getRetryOnDisconnect
bool getRetryOnDisconnect(void) const
Returns true if automatic retry of a command to AMPS after a reconnect is enabled.
Definition: ampsplusplus.hpp:7730
AMPS::Client::getBookmarkStore
BookmarkStore getBookmarkStore()
Get the bookmark store being used by the client.
Definition: ampsplusplus.hpp:5724
AMPS::Client::subscribe
MessageStream subscribe(const char *topic_, long timeout_=0, const std::string &filter_="", const std::string &options_="", const std::string &subId_="")
Subscribe to a topic.
Definition: ampsplusplus.hpp:6203
AMPS::Client::getLogonCorrelationData
const std::string & getLogonCorrelationData() const
Returns the currently set logon correlation data for the client.
Definition: ampsplusplus.hpp:5445
AMPS::Client::logon
std::string logon(int timeout_=0, Authenticator &authenticator_=DefaultAuthenticator::instance(), const char *options_=NULL)
Logon to the server, providing the client name, credentials (if available), and client information (s...
Definition: ampsplusplus.hpp:6082
AMPS::Client::getNameHashValue
const amps_uint64_t getNameHashValue() const
Returns the numeric name hash of this client as generated by the server and returned when the client ...
Definition: ampsplusplus.hpp:5419
AMPS::Client::setGlobalCommandTypeMessageHandler
void setGlobalCommandTypeMessageHandler(const Message::Command::Type command_, const MessageHandler &handler_)
Sets a handler for all messages of a particular type: currently supported types are heartbeat message...
Definition: ampsplusplus.hpp:7371
AMPS::Client::getName
const std::string & getName() const
Returns the name of this client passed in the constructor.
Definition: ampsplusplus.hpp:5403
AMPS::Client::getAutoAck
bool getAutoAck(void) const
Returns the value of the queue auto-ack setting.
Definition: ampsplusplus.hpp:7652
AMPS::Client::getDefaultMaxDepth
unsigned getDefaultMaxDepth(void) const
Returns the default max depth for returned MessageStream objects.
Definition: ampsplusplus.hpp:7748
AMPS::Client::setSubscriptionManager
void setSubscriptionManager(SubscriptionManager *subscriptionManager_)
Set the subscription manager to be used by the client.
Definition: ampsplusplus.hpp:5744
AMPS::Client::ack
void ack(Field &topic_, Field &bookmark_, const char *options_=NULL)
Acknowledge a message queue message by supplying a topic and bookmark: this adds the ack to the curre...
Definition: ampsplusplus.hpp:7595
AMPS::Client::clearConnectionStateListeners
void clearConnectionStateListeners()
Clear all listeners from self's set of ConnectionStateListeners.
Definition: ampsplusplus.hpp:7469
AMPS::Client::BOOKMARK_NOW
static const char * BOOKMARK_NOW()
Convenience method for returning the special value to start a subscription at the end of the transact...
Definition: ampsplusplus.hpp:7381
AMPS::Client::bookmarkSubscribe
MessageStream bookmarkSubscribe(const char *topic_, long timeout_, const std::string &bookmark_, const std::string &filter_="", const std::string &options_="", const std::string &subId_="")
Subscribe to a topic using a bookmark.
Definition: ampsplusplus.hpp:6359
AMPS::Client::executeAsyncNoResubscribe
std::string executeAsyncNoResubscribe(Command &command_, MessageHandler handler_)
Execute the provided command and, once AMPS acknowledges the command, process messages in response to...
Definition: ampsplusplus.hpp:7533
AMPS::Client::getServerVersion
size_t getServerVersion() const
Returns the server version retrieved during logon.
Definition: ampsplusplus.hpp:5490
AMPS::Client::EPOCH
static const char * EPOCH()
Convenience method for returning the special value to start a subscription at the beginning of the tr...
Definition: ampsplusplus.hpp:7410
AMPS::Client::send
void send(const Message &message)
Sends a Message to the connected AMPS server, performing only minimal validation and bypassing client...
Definition: ampsplusplus.hpp:5586
AMPS::Client::publishFlush
void publishFlush(long timeout_=0, unsigned ackType_=Message::AckType::Processed)
Ensure that AMPS messages are sent and have been processed by the AMPS server.
Definition: ampsplusplus.hpp:5966
AMPS::Client::deltaPublish
amps_uint64_t deltaPublish(const char *topic_, size_t topicLength_, const char *data_, size_t dataLength_, unsigned long expiration_)
Publish the changed fields of a message to an AMPS topic.
Definition: ampsplusplus.hpp:6059
AMPS::Client::ack
void ack(const std::string &topic_, const std::string &bookmark_, const char *options_=NULL)
Acknowledge a message queue message by supplying a topic and bookmark string: this adds the ack to th...
Definition: ampsplusplus.hpp:7619
AMPS::Client::deltaSubscribe
MessageStream deltaSubscribe(const char *topic_, long timeout_, const std::string &filter_="", const std::string &options_="", const std::string &subId_="")
Delta Subscribe to a topic.
Definition: ampsplusplus.hpp:6268
AMPS::Client::setUnhandledMessageHandler
void setUnhandledMessageHandler(const AMPS::MessageHandler &messageHandler)
Definition: ampsplusplus.hpp:7313
AMPS::Client::addMessageHandler
void addMessageHandler(const Field &commandId_, const AMPS::MessageHandler &messageHandler_, unsigned requestedAcks_, bool isSubscribe_)
Adds a MessageHandler to be invoked for Messages with the given CommandId as their command id,...
Definition: ampsplusplus.hpp:5599
AMPS::Client::deltaPublish
amps_uint64_t deltaPublish(const std::string &topic_, const std::string &data_, unsigned long expiration_)
Publish the changed fields of a message to an AMPS topic.
Definition: ampsplusplus.hpp:6033
AMPS::Client::addMessageHandler
void addMessageHandler(const Field &commandId_, const AMPS::MessageHandler &messageHandler_, unsigned requestedAcks_, Message::Command::Type commandType_)
Adds a MessageHandler to be invoked for Messages with the given CommandId as their command id,...
Definition: ampsplusplus.hpp:5616
AMPS::Client::executeAsync
std::string executeAsync(Command &command_, MessageHandler handler_)
Execute the provided command and, once AMPS acknowledges the command, process messages in response to...
Definition: ampsplusplus.hpp:7499
AMPS::Client::getAckBatchSize
unsigned getAckBatchSize(void) const
Returns the value of the queue ack batch size setting.
Definition: ampsplusplus.hpp:7670
AMPS::Client::setAutoAck
void setAutoAck(bool isAutoAckEnabled_)
Sets the queue auto-ack setting on this client.
Definition: ampsplusplus.hpp:7662
AMPS::Client::addHttpPreflightHeader
void addHttpPreflightHeader(const std::string &header_)
Adds a given HTTP header line to the end of the headers that will be sent for the HTTP GET Upgrade re...
Definition: ampsplusplus.hpp:5453
AMPS::Client::execute
MessageStream execute(Command &command_)
Execute the provided command and return messages received in response in a MessageStream.
Definition: ampsplusplus.hpp:9157
AMPS::Client::connect
void connect(const std::string &uri)
Connect to an AMPS server.
Definition: ampsplusplus.hpp:5561
AMPS::Client::getNameHash
const std::string & getNameHash() const
Returns the name hash string of this client as generated by the server and returned when the client l...
Definition: ampsplusplus.hpp:5411
AMPS::Client::MOST_RECENT
static const char * MOST_RECENT()
Convenience method for returning the special value to start a subscription at a recovery point based ...
Definition: ampsplusplus.hpp:7432
AMPS::Client::removeMessageHandler
bool removeMessageHandler(const Field &commandId_)
Removes a MessageHandler for a given ComandId from self.
Definition: ampsplusplus.hpp:5627
AMPS::Client::setRetryOnDisconnect
void setRetryOnDisconnect(bool isRetryOnDisconnect_)
Enables or disables automatic retry of a command to AMPS after a reconnect.
Definition: ampsplusplus.hpp:7721
AMPS::Client::setGlobalCommandTypeMessageHandler
void setGlobalCommandTypeMessageHandler(const std::string &command_, const MessageHandler &handler_)
Sets a handler for all messages of a particular type, or for messages that would be delivered to a pa...
Definition: ampsplusplus.hpp:7346
AMPS::Client::setTransportFilterFunction
void setTransportFilterFunction(amps_transport_filter_function filter_, void *userData_)
Sets a filter function on the transport that is called with all raw data sent or received.
Definition: ampsplusplus.hpp:7759
AMPS::Client::getConnectionName
const std::string & getConnectionName() const
Returns the connection name string of this client as generated by the server and returned when the cl...
Definition: ampsplusplus.hpp:5427
AMPS::Client::getServerVersionInfo
VersionInfo getServerVersionInfo() const
Returns the server version retrieved during logon.
Definition: ampsplusplus.hpp:5501
AMPS::Client::setPublishStore
void setPublishStore(const Store &publishStore_)
Set the publish store to be used by the client.
Definition: ampsplusplus.hpp:5768
AMPS::Client::setHeartbeat
void setHeartbeat(unsigned heartbeatTime_, unsigned readTimeout_)
Requests heartbeating with the AMPS server.
Definition: ampsplusplus.hpp:7280
AMPS::Client::deltaSubscribe
MessageStream deltaSubscribe(const std::string &topic_, long timeout_, const std::string &filter_="", const std::string &options_="", const std::string &subId_="")
Delta Subscribe to a topic.
Definition: ampsplusplus.hpp:6250
AMPS::Client::setLogonCorrelationData
void setLogonCorrelationData(const std::string &logonCorrelationData_)
Sets the logon correlation data for the client.
Definition: ampsplusplus.hpp:5438
AMPS::Client::addConnectionStateListener
void addConnectionStateListener(ConnectionStateListener *listener)
Adds a ConnectionStateListener to self's set of listeners.
Definition: ampsplusplus.hpp:7454
AMPS::Client::setThreadCreatedCallback
void setThreadCreatedCallback(amps_thread_created_callback callback_, void *userData_)
Sets a callback function on the transport that is called when a new thread is created to receive data...
Definition: ampsplusplus.hpp:7773
AMPS::Client::subscribe
std::string subscribe(const MessageHandler &messageHandler_, const std::string &topic_, long timeout_=0, const std::string &filter_="", const std::string &options_="", const std::string &subId_="")
Subscribe to a topic.
Definition: ampsplusplus.hpp:6145
AMPS::Client::setDisconnectHandler
virtual void setDisconnectHandler(const DisconnectHandler &disconnectHandler)
Definition: ampsplusplus.hpp:5680
AMPS::Client::setAckTimeout
void setAckTimeout(const int ackTimeout_)
Sets the message queue ack timeout value.
Definition: ampsplusplus.hpp:7703
AMPS::Client::getDuplicateMessageHandler
MessageHandler getDuplicateMessageHandler(void)
Returns the callback function that is invoked when a duplicate message is detected.
Definition: ampsplusplus.hpp:5799
AMPS::Client::getConnectionInfo
virtual ConnectionInfo getConnectionInfo() const
Get the connection information for the current connection.
Definition: ampsplusplus.hpp:5703
AMPS::Client::send
std::string send(const MessageHandler &messageHandler_, Message &message_, int timeout_=0, AMPS_ATOMIC_BASE_TYPE subscribeGeneration_=0)
Sends a Message to the connected AMPS server, performing only minimal validation and bypassing client...
Definition: ampsplusplus.hpp:5656
AMPS::Client::setDuplicateMessageHandler
void setDuplicateMessageHandler(const MessageHandler &duplicateMessageHandler_)
Sets a callback function that is invoked when a duplicate message is detected.
Definition: ampsplusplus.hpp:5784
AMPS::Client::sowAndSubscribe
MessageStream sowAndSubscribe(const char *topic_, const std::string &filter_="", const std::string &orderBy_="", const std::string &bookmark_="", int batchSize_=DEFAULT_BATCH_SIZE, int topN_=DEFAULT_TOP_N, const std::string &options_="", long timeout_=DEFAULT_COMMAND_TIMEOUT)
Query the SOW cache of a topic and initiates a new subscription on it.
Definition: ampsplusplus.hpp:6746
AMPS::Client::BOOKMARK_RECENT
static const char * BOOKMARK_RECENT()
Convenience method for returning the special value to start a subscription at a recovery point based ...
Definition: ampsplusplus.hpp:7443
AMPS::Client::stopTimer
std::string stopTimer(const MessageHandler &messageHandler)
Definition: ampsplusplus.hpp:7073
AMPS::Client::ack
void ack(Message &message_, const char *options_=NULL)
Acknowledge a message queue message by supplying the message directly: this adds the ack to the curre...
Definition: ampsplusplus.hpp:7607
AMPS::Client::convertVersionToNumber
static size_t convertVersionToNumber(const std::string &version_)
Converts a string version, such as "3.8.1.5" into the same numeric form used internally and returned ...
Definition: ampsplusplus.hpp:5515
AMPS::Client::publish
amps_uint64_t publish(const char *topic_, size_t topicLength_, const char *data_, size_t dataLength_, unsigned long expiration_)
Publish a message to an AMPS topic, returning the sequence number assigned by the publish store if on...
Definition: ampsplusplus.hpp:5920
AMPS::Client::getPublishStore
Store getPublishStore()
Get the publish store used by the client.
Definition: ampsplusplus.hpp:5776
AMPS::Client::logon
std::string logon(const std::string &options_, int timeout_=0)
Logon to the server, providing the client name, credentials (if available) client information (such a...
Definition: ampsplusplus.hpp:6120
AMPS::Command
Command is an encapsulation of a single AMPS command sent by the client.
Definition: ampsplusplus.hpp:473
AMPS::Command::setData
Command & setData(const char *data_, size_t dataLen_)
Sets the data for this command.
Definition: ampsplusplus.hpp:837
AMPS::Command::setTopN
Command & setTopN(unsigned topN_)
Definition: ampsplusplus.hpp:857
AMPS::Command::setSequence
Command & setSequence(const char *seq_, size_t seqLen_)
Definition: ampsplusplus.hpp:810
AMPS::Command::setTopic
Command & setTopic(const std::string &topic_)
Definition: ampsplusplus.hpp:679
AMPS::Command::setAckType
Command & setAckType(unsigned ackType_)
Definition: ampsplusplus.hpp:931
AMPS::Command::setOrderBy
Command & setOrderBy(const std::string &orderBy_)
Definition: ampsplusplus.hpp:705
AMPS::Command::setOptions
Command & setOptions(const char *options_, size_t optionsLen_)
Sets the options string for this command: see Message.Options for a helper class for constructing the...
Definition: ampsplusplus.hpp:797
AMPS::Command::setSowKey
Command & setSowKey(const char *sowKey_, size_t sowKeyLen_)
Sets the SowKey field of the command, typically used for a publish command to a topic in the state of...
Definition: ampsplusplus.hpp:625
AMPS::Command::setAckType
Command & setAckType(const std::string &ackType_)
Definition: ampsplusplus.hpp:909
AMPS::Command::setTimeout
Command & setTimeout(unsigned timeout_)
Sets the client-side timeout for this command.
Definition: ampsplusplus.hpp:851
AMPS::Command::setBookmark
Command & setBookmark(const char *bookmark_, size_t bookmarkLen_)
Set the bookmark to be used this command.
Definition: ampsplusplus.hpp:759
AMPS::Command::setQueryId
Command & setQueryId(const char *queryId_, size_t queryIdLen_)
Definition: ampsplusplus.hpp:738
AMPS::Command::getAckType
std::string getAckType() const
Definition: ampsplusplus.hpp:953
AMPS::Command::reset
Command & reset(Message::Command::Type command_)
Resets the fields of self, and sets the command to command_.
Definition: ampsplusplus.hpp:600
AMPS::Command::setBatchSize
Command & setBatchSize(unsigned batchSize_)
Sets the batch size for this command, which controls how many records are sent together in the result...
Definition: ampsplusplus.hpp:873
AMPS::Command::setSubId
Command & setSubId(const char *subId_, size_t subIdLen_)
Definition: ampsplusplus.hpp:725
AMPS::Command::setSequence
Command & setSequence(const amps_uint64_t seq_)
Definition: ampsplusplus.hpp:816
AMPS::Command::setOptions
Command & setOptions(const std::string &options_)
Sets the options string for this command: see Message.Options for a helper class for constructing the...
Definition: ampsplusplus.hpp:789
AMPS::Command::setSowKeys
Command & setSowKeys(const char *sowKeys_, size_t sowKeysLen_)
Sets the SowKeys for the command.
Definition: ampsplusplus.hpp:660
AMPS::Command::setSubId
Command & setSubId(const std::string &subId_)
Definition: ampsplusplus.hpp:718
AMPS::Command::setCommandId
Command & setCommandId(const std::string &cmdId_)
Definition: ampsplusplus.hpp:666
AMPS::Command::setSequence
Command & setSequence(const std::string &seq_)
Definition: ampsplusplus.hpp:803
AMPS::Command::setCommandId
Command & setCommandId(const char *cmdId_, size_t cmdIdLen_)
Definition: ampsplusplus.hpp:673
AMPS::Command::Command
Command(Message::Command::Type command_)
Creates an object to represent the given AMPS command, such as "sow" or "subscribe".
Definition: ampsplusplus.hpp:575
AMPS::Command::setOrderBy
Command & setOrderBy(const char *orderBy_, size_t orderByLen_)
Definition: ampsplusplus.hpp:712
AMPS::Command::setData
Command & setData(const std::string &data_)
Sets the data for this command from an existing string.
Definition: ampsplusplus.hpp:829
AMPS::Command::setQueryId
Command & setQueryId(const std::string &queryId_)
Definition: ampsplusplus.hpp:731
AMPS::Command::Command
Command(const std::string &command_)
Creates an object to represent the given AMPS command, such as "sow" or "subscribe".
Definition: ampsplusplus.hpp:560
AMPS::Command::setTopic
Command & setTopic(const char *topic_, size_t topicLen_)
Definition: ampsplusplus.hpp:686
AMPS::Command::setFilter
Command & setFilter(const std::string &filter_)
Definition: ampsplusplus.hpp:692
AMPS::Command::setCorrelationId
Command & setCorrelationId(const std::string &correlationId_)
Set the correlation ID for this command.
Definition: ampsplusplus.hpp:770
AMPS::Command::setFilter
Command & setFilter(const char *filter_, size_t filterLen_)
Definition: ampsplusplus.hpp:699
AMPS::Command::Command
Command(const char *command_, size_t commandLen_)
Creates an object to represent the given AMPS command, such as "sow" or "subscribe".
Definition: ampsplusplus.hpp:568
AMPS::Command::setCorrelationId
Command & setCorrelationId(const char *correlationId_, size_t correlationIdLen_)
Set the correlation ID for this command.
Definition: ampsplusplus.hpp:782
AMPS::Command::getAckTypeEnum
unsigned getAckTypeEnum() const
Definition: ampsplusplus.hpp:958
AMPS::Command::setSowKey
Command & setSowKey(const std::string &sowKey_)
Sets the SowKey field of the command, typically used for a publish command to a topic in the state of...
Definition: ampsplusplus.hpp:612
AMPS::Command::setSowKeys
Command & setSowKeys(const std::string &sowKeys_)
Sets the SowKeys for the command.
Definition: ampsplusplus.hpp:642
AMPS::Command::addAckType
Command & addAckType(const std::string &ackType_)
Definition: ampsplusplus.hpp:895
AMPS::Command::reset
Command & reset(const char *command_, size_t commandLen_)
Resets the fields of self, and sets the command to command_.
Definition: ampsplusplus.hpp:592
AMPS::Command::setExpiration
Command & setExpiration(unsigned expiration_)
Set the expiration time for a publish command.
Definition: ampsplusplus.hpp:889
AMPS::Command::setBookmark
Command & setBookmark(const std::string &bookmark_)
Set the bookmark to be used this command.
Definition: ampsplusplus.hpp:748
AMPS::Command::reset
Command & reset(const std::string &command_)
Resets the fields of self, and sets the command to command_.
Definition: ampsplusplus.hpp:583
AMPS::ConnectionStateListener
Abstract base class for connection state listeners.
Definition: ampsplusplus.hpp:1511
AMPS::ConnectionStateListener::State
State
Constants for the state of the connection.
Definition: ampsplusplus.hpp:1514
AMPS::ConnectionStateListener::connectionStateChanged
virtual void connectionStateChanged(State newState_)=0
Pure virtual method for receiving the change in connection state.
AMPS::DefaultAuthenticator
A default implementation of Authenticator that only uses an unchanged password and does not implement...
Definition: ampsplusplus.hpp:1038
AMPS::DefaultAuthenticator::authenticate
std::string authenticate(const std::string &, const std::string &password_)
A simple implementation that returns an unmodified password.
Definition: ampsplusplus.hpp:1043
AMPS::DefaultAuthenticator::instance
static Authenticator & instance()
Static function to return a static instance used when no Authenticator is supplied to a Client.
Definition: ampsplusplus.hpp:1060
AMPS::DefaultAuthenticator::retry
std::string retry(const std::string &, const std::string &)
Throws an AuthenticationException because retry is not implemented.
Definition: ampsplusplus.hpp:1050
AMPS::DefaultAuthenticator::completed
void completed(const std::string &, const std::string &, const std::string &)
Called by Client once a logon completes successfully.
Definition: ampsplusplus.hpp:1055
AMPS::ExceptionListener
Exception listener for unhandled exceptions.
Definition: ampsplusplus.hpp:205
AMPS::FIXShredder
Class for parsing a FIX format message into a std::map of keys and values, where the keys and values ...
Definition: ampsplusplus.hpp:8595
AMPS::FIXShredder::map_type
std::map< Message::Field, Message::Field > map_type
Convenience defintion for the std::map specialization used for this class.
Definition: ampsplusplus.hpp:8606
AMPS::FIXShredder::toMap
map_type toMap(const Message::Field &data)
Returns the key/value pairs within the message, represented as AMPS::Field objects that contain point...
Definition: ampsplusplus.hpp:8613
AMPS::FIXShredder::FIXShredder
FIXShredder(char fieldSep_=(char) 1)
Construct an instance of FIXShredder using the specified value as the delimiter between fields.
Definition: ampsplusplus.hpp:8602
AMPS::FailedResubscribeHandler
Abstract base class where you can implement handling of exceptions that occur when a SubscriptionMana...
Definition: ampsplusplus.hpp:1440
AMPS::FailedResubscribeHandler::failure
virtual bool failure(const Message &message_, const MessageHandler &handler_, unsigned requestedAckTypes_, const AMPSException &exception_)=0
Implement this function to return true if the subscription should be removed from the SubscriptionMan...
AMPS::FailedWriteHandler
Class to handle when a client receives a duplicate publish message, or not entitled message.
Definition: ampsplusplus.hpp:1382
AMPS::FailedWriteHandler::failedWrite
virtual void failedWrite(const Message &message_, const char *reason_, size_t reasonLength_)=0
Called when the server indicates a message could not be written.
AMPS::Field
Field represents the value of a single field in a Message.
Definition: Field.hpp:88
AMPS::Field::empty
bool empty() const
Returns 'true' if empty, 'false' otherwise.
Definition: Field.hpp:129
AMPS::Field::len
size_t len() const
Returns the length of the data underlying this field.
Definition: Field.hpp:286
AMPS::Handler< DisconnectHandlerFunc, Client & >
AMPS::MessageStream::iterator
Represents an iterator over messages in an AMPS topic.
Definition: ampsplusplus.hpp:5216
AMPS::MessageStream
An iterable object representing the results of an AMPS subscription and/or query.
Definition: ampsplusplus.hpp:5208
AMPS::MessageStream::begin
iterator begin(void)
Returns an iterator representing the beginning of the topic or subscription.
Definition: ampsplusplus.hpp:5260
AMPS::MessageStream::isValid
bool isValid() const
Returns true if self is a valid stream that may be iterated.
Definition: ampsplusplus.hpp:5253
AMPS::MessageStream::conflate
MessageStream conflate(void)
Sets self to conflation mode, where a new update for a matching sow key will replace the previous one...
Definition: ampsplusplus.hpp:9133
AMPS::MessageStream::maxDepth
MessageStream maxDepth(unsigned maxDepth_)
Sets the maximum number of messages that can be held in the underlying queue.
Definition: ampsplusplus.hpp:9138
AMPS::MessageStream::timeout
MessageStream timeout(unsigned timeout_)
Sets the maximum time to wait for the next message in milliseconds; if no message is available within...
Definition: ampsplusplus.hpp:9128
AMPS::MessageStream::getDepth
unsigned getDepth(void) const
Gets the current number of messages held in the underlying queue.
Definition: ampsplusplus.hpp:9147
AMPS::MessageStream::getMaxDepth
unsigned getMaxDepth(void) const
Gets the maximum number of messages that can be held in the underlying queue.
Definition: ampsplusplus.hpp:9143
AMPS::MessageStream::end
iterator end(void)
Returns an iterator representing the end of the topic or subscription.
Definition: ampsplusplus.hpp:5271
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::assignExpiration
Message & assignExpiration(const std::string &v)
Assigns the value of the Expiration header for this Message without copying.
Definition: Message.hpp:1347
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::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::setQueryID
Message & setQueryID(const std::string &v)
Sets the value of the QueryID header for this Message.
Definition: Message.hpp:1459
AMPS::Message::getCommandId
Field getCommandId() const
Retrieves the value of the CommandId header of the Message as a Field which references the underlying...
Definition: Message.hpp:1344
AMPS::Message::setExpiration
Message & setExpiration(const std::string &v)
Sets the value of the Expiration header for this Message.
Definition: Message.hpp:1347
AMPS::Message::setMessageType
Message & setMessageType(const std::string &v)
Sets the value of the MessageType header for this Message.
Definition: Message.hpp:1354
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::assignOwnershipBookmark
Message & assignOwnershipBookmark(const Field &f)
Assigns the value of the Bookmark header for this Message without copying and makes this Message resp...
Definition: Message.hpp:1236
AMPS::Message::assignAckType
Message & assignAckType(const std::string &v)
Assigns the value of the AckType header for this Message without copying.
Definition: Message.hpp:1172
AMPS::Message::assignSubscriptionId
Message & assignSubscriptionId(const std::string &v)
Assigns the value of the SubscriptionId header for this Message without copying.
Definition: Message.hpp:1469
AMPS::Message::deepCopy
Message deepCopy(void) const
Returns a deep copy of self.
Definition: Message.hpp:567
AMPS::Message::newCommandId
Message & newCommandId()
Creates and sets a new sequential value for the CommandId header for this Message.
Definition: Message.hpp:1344
AMPS::Message::getSequence
Field getSequence() const
Retrieves the value of the Sequence header of the Message as a Field which references the underlying ...
Definition: Message.hpp:1464
AMPS::Message::setBookmark
Message & setBookmark(const std::string &v)
Sets the value of the Bookmark header for this Message.
Definition: Message.hpp:1236
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::setClientName
Message & setClientName(const std::string &v)
Sets the value of the ClientName header for this Message.
Definition: Message.hpp:1345
AMPS::Message::assignFilter
Message & assignFilter(const std::string &v)
Assigns the value of the Filter header for this Message without copying.
Definition: Message.hpp:1348
AMPS::Message::setSequence
Message & setSequence(const std::string &v)
Sets the value of the Sequence header for this Message.
Definition: Message.hpp:1464
AMPS::Message::setOrderBy
Message & setOrderBy(const std::string &v)
Sets the value of the OrderBy header for this Message.
Definition: Message.hpp:1457
AMPS::Message::setUserId
Message & setUserId(const std::string &v)
Sets the value of the UserId header for this Message.
Definition: Message.hpp:1495
AMPS::Message::setPassword
Message & setPassword(const std::string &v)
Sets the value of the Password header for this Message.
Definition: Message.hpp:1458
AMPS::Message::setCommandId
Message & setCommandId(const std::string &v)
Sets the value of the CommandId header for this Message.
Definition: Message.hpp:1344
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::assignVersion
Message & assignVersion(const std::string &v)
Assigns the value of the Version header for this Message without copying.
Definition: Message.hpp:1494
AMPS::Message::assignCommand
Message & assignCommand(const std::string &v)
Assigns the value of the Command header for this Message without copying.
Definition: Message.hpp:1237
AMPS::Message::setQueryId
Message & setQueryId(const std::string &v)
Sets the value of the QueryID header for this Message.
Definition: Message.hpp:1459
AMPS::Message::setTopNRecordsReturned
Message & setTopNRecordsReturned(const std::string &v)
Sets the value of the TopNRecordsReturned header for this Message.
Definition: Message.hpp:1493
AMPS::Message::assignSowKeys
Message & assignSowKeys(const std::string &v)
Assigns the value of the SowKeys header for this Message without copying.
Definition: Message.hpp:1467
AMPS::Message::setAckType
Message & setAckType(const std::string &v)
Sets the value of the AckType header for this Message.
Definition: Message.hpp:1172
AMPS::Message::setSowKey
Message & setSowKey(const std::string &v)
Sets the value of the SowKey header for this Message.
Definition: Message.hpp:1466
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::getPassword
Field getPassword() const
Retrieves the value of the Password header of the Message as a Field which references the underlying ...
Definition: Message.hpp:1458
AMPS::Message::assignQueryID
Message & assignQueryID(const std::string &v)
Assigns the value of the QueryID header for this Message without copying.
Definition: Message.hpp:1459
AMPS::Message::assignTopic
Message & assignTopic(const std::string &v)
Assigns the value of the Topic header for this Message without copying.
Definition: Message.hpp:1491
AMPS::Message::setTopic
Message & setTopic(const std::string &v)
Sets the value of the Topic header for this Message.
Definition: Message.hpp:1491
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::setSubscriptionId
Message & setSubscriptionId(const std::string &v)
Sets the value of the SubscriptionId header for this Message.
Definition: Message.hpp:1469
AMPS::Message::setFilter
Message & setFilter(const std::string &v)
Sets the value of the Filter header for this Message.
Definition: Message.hpp:1348
AMPS::Message::getAckType
Field getAckType() const
Retrieves the value of the AckType header of the Message as a Field which references the underlying b...
Definition: Message.hpp:1172
AMPS::Message::getQueryId
Field getQueryId() const
Retrieves the value of the QueryID header of the Message as a Field which references the underlying b...
Definition: Message.hpp:1459
AMPS::Message::assignSequence
Message & assignSequence(const std::string &v)
Assigns the value of the Sequence header for this Message without copying.
Definition: Message.hpp:1464
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::assignCorrelationId
Message & assignCorrelationId(const std::string &v)
Assigns the value of the CorrelationId header for this Message without copying.
Definition: Message.hpp:1346
AMPS::Message::getQueryID
Field getQueryID() const
Retrieves the value of the QueryID header of the Message as a Field which references the underlying b...
Definition: Message.hpp:1459
AMPS::Message::setSowKeys
Message & setSowKeys(const std::string &v)
Sets the value of the SowKeys header for this Message.
Definition: Message.hpp:1467
AMPS::Message::setBatchSize
Message & setBatchSize(const std::string &v)
Sets the value of the BatchSize header for this Message.
Definition: Message.hpp:1235
AMPS::Message::setCommand
Message & setCommand(const std::string &v)
Sets the value of the Command header for this Message.
Definition: Message.hpp:1237
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::setCorrelationId
Message & setCorrelationId(const std::string &v)
Sets the value of the CorrelationId header for this Message.
Definition: Message.hpp:1346
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::StoreImpl
Abstract base class for storing published messages for an HA publisher client.
Definition: ampsplusplus.hpp:1098
AMPS::StoreImpl::getLowestUnpersisted
virtual amps_uint64_t getLowestUnpersisted() const =0
Get the oldest unpersisted message sequence in the store.
AMPS::StoreImpl::unpersistedCount
virtual size_t unpersistedCount() const =0
Method to return how many messages are in the store that have not been discarded, indicating that the...
AMPS::StoreImpl::getUnsetSequence
static amps_uint64_t getUnsetSequence()
Method to return the value used to represent no such sequence.
Definition: ampsplusplus.hpp:1167
AMPS::StoreImpl::getLastPersisted
virtual amps_uint64_t getLastPersisted()=0
Get the last persisted sequence number.
AMPS::StoreImpl::getUnsetPosition
static size_t getUnsetPosition()
Method to return the value used to represent not found or unset.
Definition: ampsplusplus.hpp:1160
AMPS::StoreImpl::discardUpTo
virtual void discardUpTo(amps_uint64_t index_)=0
Called by Client to indicate that all messages up to and including.
AMPS::StoreImpl::flush
virtual void flush(long timeout_)=0
Method to wait for the Store to discard everything that has been stored up to the point in time when ...
AMPS::StoreImpl::store
virtual amps_uint64_t store(const Message &message_)=0
Called by Client to store a message being published.
AMPS::StoreImpl::replay
virtual void replay(StoreReplayer &replayer_)=0
Called by Client to get all stored and non-discarded messages replayed by the store onto the StoreRep...
AMPS::StoreImpl::replaySingle
virtual bool replaySingle(StoreReplayer &replayer_, amps_uint64_t index_)=0
Called by Client to get a single message replayed by the store onto the StoreReplayer.
AMPS::StoreImpl::StoreImpl
StoreImpl(bool errorOnPublishGap_=false)
Default constructor.
Definition: ampsplusplus.hpp:1105
AMPS::StoreImpl::setResizeHandler
virtual void setResizeHandler(PublishStoreResizeHandler handler_, void *userData_)
Set a handler to be called if the Store needs to resize in order to keep storing messages.
Definition: ampsplusplus.hpp:1191
AMPS::StoreReplayer
Abstract base class for replaying a publish message.
Definition: ampsplusplus.hpp:1070
AMPS::StoreReplayer::execute
virtual void execute(Message &message_)=0
Called by implementations of Store to replay a message from the store.
AMPS::Store
Handle class for StoreImpl classes that track publish messages.
Definition: ampsplusplus.hpp:1224
AMPS::Store::replay
void replay(StoreReplayer &replayer_)
Called by Client to get all stored and non-discarded messages replayed by the store onto the StoreRep...
Definition: ampsplusplus.hpp:1259
AMPS::Store::get
StoreImpl * get()
Used to get a pointer to the implementation.
Definition: ampsplusplus.hpp:1363
AMPS::Store::getErrorOnPublishGap
bool getErrorOnPublishGap() const
Called to check if the Store will throw PublishStoreGapException.
Definition: ampsplusplus.hpp:1355
AMPS::Store::unpersistedCount
size_t unpersistedCount() const
Method to return how many messages are in the store that have not been discarded, indicating that the...
Definition: ampsplusplus.hpp:1280
AMPS::Store::isValid
bool isValid() const
Method to return if there is an underlying implementation for the Store.
Definition: ampsplusplus.hpp:1288
AMPS::Store::getLastPersisted
amps_uint64_t getLastPersisted()
Get the last persisted message sequence in the store.
Definition: ampsplusplus.hpp:1317
AMPS::Store::discardUpTo
void discardUpTo(amps_uint64_t index_)
Called by Client to indicate that all messages up to and including.
Definition: ampsplusplus.hpp:1250
AMPS::Store::getLowestUnpersisted
amps_uint64_t getLowestUnpersisted()
Get the oldest unpersisted message sequence in the store.
Definition: ampsplusplus.hpp:1309
AMPS::Store::flush
void flush(long timeout_=0)
Method to wait for the Store to discard everything that has been stored up to the point in time when ...
Definition: ampsplusplus.hpp:1301
AMPS::Store::setResizeHandler
void setResizeHandler(PublishStoreResizeHandler handler_, void *userData_)
Set a handler to be called if the Store needs to resize in order to keep storing messages.
Definition: ampsplusplus.hpp:1331
AMPS::Store::replaySingle
bool replaySingle(StoreReplayer &replayer_, amps_uint64_t index_)
Called by Client to get a single message replayed by the store onto the StoreReplayer.
Definition: ampsplusplus.hpp:1271
AMPS::Store::setErrorOnPublishGap
void setErrorOnPublishGap(bool errorOnPublishGap_)
Called to enable or disable throwing PublishStoreGapException.
Definition: ampsplusplus.hpp:1346
AMPS::Store::store
amps_uint64_t store(const Message &message_)
Called by Client to store a message being published.
Definition: ampsplusplus.hpp:1239
AMPS::SubscriptionManager
Abstract base class to manage all subscriptions placed on a client so that they can be re-established...
Definition: ampsplusplus.hpp:1462
AMPS::SubscriptionManager::resubscribe
virtual void resubscribe(Client &client_)=0
Called by Client to get all subscriptions placed again.
AMPS::SubscriptionManager::getGenerationCount
virtual AMPS_ATOMIC_BASE_TYPE getGenerationCount() const
Gets the current subscription manager generation count.
Definition: ampsplusplus.hpp:1479
AMPS::SubscriptionManager::setFailedResubscribeHandler
virtual void setFailedResubscribeHandler(std::shared_ptr< FailedResubscribeHandler > handler_)
Set a handler to deal with failing subscriptions after a failover event.
Definition: ampsplusplus.hpp:1498
AMPS::SubscriptionManager::unsubscribe
virtual void unsubscribe(const Message::Field &subId_)=0
Called by Client when a subscription is unsubscribed.
AMPS::SubscriptionManager::clear
virtual void clear()=0
Clear subscriptions and reset to the initial state.
AMPS::SubscriptionManager::subscribe
virtual AMPS_ATOMIC_BASE_TYPE subscribe(MessageHandler messageHandler_, const Message &message_, unsigned requestedAckTypes_)=0
Called by Client when a subscription is placed.
AMPS::_FIXBuilder
Provides a convenient way of building messages in FIX format, typically referenced using the typedefs...
Definition: ampsplusplus.hpp:8521
AMPS::_FIXBuilder::_FIXBuilder
_FIXBuilder(char fieldSep_=(char) 1)
Construct an instance of _FIXBuilder, using the specified separator between fields.
Definition: ampsplusplus.hpp:8530
AMPS::_FIXBuilder::getString
std::string getString() const
Returns the current contents of this builder as a string.
Definition: ampsplusplus.hpp:8557
AMPS::_FIXBuilder::append
void append(const T &tag, const std::string &value)
Write a field with the provided tag and value to the message being constructed.
Definition: ampsplusplus.hpp:8550
AMPS::_FIXBuilder::reset
void reset()
Clear all data from the builder.
Definition: ampsplusplus.hpp:8567
AMPS::_FIXBuilder::append
void append(const T &tag, const char *value, size_t offset, size_t length)
Write a field with the provided tag and value to the message being constructed.
Definition: ampsplusplus.hpp:8539
AMPS::Client::sowDelete
std::string sowDelete(const MessageHandler &messageHandler, const std::string &topic, const std::string &filter, long timeout)
Deletes one or more messages from a topic's SOW cache.
Definition: ampsplusplus.hpp:7008
AMPS::Client::sowDeleteByData
Message sowDeleteByData(const std::string &topic_, const std::string &data_, long timeout_=0)
Deletes the message whose keys match the message data provided.
Definition: ampsplusplus.hpp:7188
AMPS::Client::sowAndSubscribe
std::string sowAndSubscribe(const MessageHandler &messageHandler_, const std::string &topic_, const std::string &filter_="", const std::string &orderBy_="", const std::string &bookmark_="", int batchSize_=DEFAULT_BATCH_SIZE, int topN_=DEFAULT_TOP_N, const std::string &options_="", long timeout_=DEFAULT_COMMAND_TIMEOUT)
Query the SOW cache of a topic and initiates a new subscription on it.
Definition: ampsplusplus.hpp:6684
AMPS::Client::sow
std::string sow(const MessageHandler &messageHandler_, const std::string &topic_, long timeout_, const std::string &filter_="", int batchSize_=DEFAULT_BATCH_SIZE, int topN_=DEFAULT_TOP_N)
Query the SOW cache of a topic.
Definition: ampsplusplus.hpp:6534
AMPS::Client::sowAndDeltaSubscribe
std::string sowAndDeltaSubscribe(const MessageHandler &messageHandler_, const std::string &topic_, long timeout_, const std::string &filter_="", int batchSize_=DEFAULT_BATCH_SIZE, bool oofEnabled_=false, bool sendEmpties_=false, int topN_=DEFAULT_TOP_N)
Query the SOW cache of a topic and initiates a new delta subscription on it.
Definition: ampsplusplus.hpp:6892
AMPS::Client::sowDeleteByData
std::string sowDeleteByData(const MessageHandler &messageHandler_, const std::string &topic_, const std::string &data_, long timeout_=0)
Deletes the message whose keys match the message data provided.
Definition: ampsplusplus.hpp:7167
AMPS::Client::sowAndSubscribe
std::string sowAndSubscribe(const MessageHandler &messageHandler_, const std::string &topic_, long timeout_, const std::string &filter_="", int batchSize_=DEFAULT_BATCH_SIZE, bool oofEnabled_=false, int topN_=DEFAULT_TOP_N)
Query the SOW cache of a topic and initiates a new subscription on it.
Definition: ampsplusplus.hpp:6566
AMPS::Client::sowDeleteByKeys
std::string sowDeleteByKeys(const MessageHandler &messageHandler_, const std::string &topic_, const std::string &keys_, long timeout_=0)
Deletes messages that match SOW keys from a topic's SOW cache.
Definition: ampsplusplus.hpp:7099
AMPS::Client::sowAndDeltaSubscribe
std::string sowAndDeltaSubscribe(const MessageHandler &messageHandler_, const std::string &topic_, const std::string &filter_="", const std::string &orderBy_="", int batchSize_=DEFAULT_BATCH_SIZE, int topN_=DEFAULT_TOP_N, const std::string &options_="", long timeout_=DEFAULT_COMMAND_TIMEOUT)
Query the SOW cache of a topic and initiates a new delta subscription on it.
Definition: ampsplusplus.hpp:6792
AMPS::Client::sow
std::string sow(const MessageHandler &messageHandler_, const std::string &topic_, const std::string &filter_="", const std::string &orderBy_="", const std::string &bookmark_="", int batchSize_=DEFAULT_BATCH_SIZE, int topN_=DEFAULT_TOP_N, const std::string &options_="", long timeout_=DEFAULT_COMMAND_TIMEOUT)
Query a State-of-the-World topic.
Definition: ampsplusplus.hpp:6434
AMPS_BOOKMARK_NOW
#define AMPS_BOOKMARK_NOW
Start the subscription at the point in time when AMPS processes the subscription.
Definition: BookmarkStore.hpp:55
AMPS_BOOKMARK_EPOCH
#define AMPS_BOOKMARK_EPOCH
Start the subscription at the beginning of the journal.
Definition: BookmarkStore.hpp:51
AMPS_BOOKMARK_RECENT
#define AMPS_BOOKMARK_RECENT
Start the subscription at the first undiscarded message in the bookmark store, or at the end of the b...
Definition: BookmarkStore.hpp:47
AMPS::BookmarkStore::get
BookmarkStoreImpl * get()
Used to get a pointer to the implementation.
Definition: BookmarkStore.hpp:465
AMPS::Client::sowAndDeltaSubscribe
MessageStream sowAndDeltaSubscribe(const char *topic_, long timeout_, const std::string &filter_="", int batchSize_=DEFAULT_BATCH_SIZE, bool oofEnabled_=false, bool sendEmpties_=false, int topN_=DEFAULT_TOP_N)
Query the SOW cache of a topic and initiates a new delta subscription on it.
Definition: ampsplusplus.hpp:6969
AMPS::Client::sowAndDeltaSubscribe
MessageStream sowAndDeltaSubscribe(const std::string &topic_, const std::string &filter_="", const std::string &orderBy_="", int batchSize_=DEFAULT_BATCH_SIZE, int topN_=DEFAULT_TOP_N, const std::string &options_="", long timeout_=DEFAULT_COMMAND_TIMEOUT)
Query the SOW cache of a topic and initiates a new delta subscription on it.
Definition: ampsplusplus.hpp:6825
AMPS::Client::sowDeleteByKeys
Message sowDeleteByKeys(const std::string &topic_, const std::string &keys_, long timeout_=0)
Deletes messages that match SOW keys from a topic's SOW cache.
Definition: ampsplusplus.hpp:7126
AMPS::Client::sow
MessageStream sow(const std::string &topic_, const std::string &filter_="", const std::string &orderBy_="", const std::string &bookmark_="", int batchSize_=DEFAULT_BATCH_SIZE, int topN_=DEFAULT_TOP_N, const std::string &options_="", long timeout_=DEFAULT_COMMAND_TIMEOUT)
Query the SOW cache of a topic.
Definition: ampsplusplus.hpp:6472
AMPS::Client::sowAndSubscribe
MessageStream sowAndSubscribe(const std::string &topic_, long timeout_, const std::string &filter_="", int batchSize_=DEFAULT_BATCH_SIZE, bool oofEnabled_=false, int topN_=DEFAULT_TOP_N)
Query the SOW cache of a topic and initiates a new subscription on it.
Definition: ampsplusplus.hpp:6598
AMPS::Client::sowAndSubscribe
MessageStream sowAndSubscribe(const char *topic_, long timeout_, const std::string &filter_="", int batchSize_=DEFAULT_BATCH_SIZE, bool oofEnabled_=false, int topN_=DEFAULT_TOP_N)
Query the SOW cache of a topic and initiates a new subscription on it.
Definition: ampsplusplus.hpp:6636
AMPS::Client::sowDelete
Message sowDelete(const std::string &topic_, const std::string &filter_, long timeout_=0)
Deletes one or more messages from a topic's SOW cache.
Definition: ampsplusplus.hpp:7031
AMPS::Client::sowAndDeltaSubscribe
MessageStream sowAndDeltaSubscribe(const std::string &topic_, long timeout_, const std::string &filter_="", int batchSize_=DEFAULT_BATCH_SIZE, bool oofEnabled_=false, bool sendEmpties_=false, int topN_=DEFAULT_TOP_N)
Query the SOW cache of a topic and initiates a new delta subscription on it.
Definition: ampsplusplus.hpp:6928
AMPS::Client::sowAndSubscribe
MessageStream sowAndSubscribe(const std::string &topic_, const std::string &filter_="", const std::string &orderBy_="", const std::string &bookmark_="", int batchSize_=DEFAULT_BATCH_SIZE, int topN_=DEFAULT_TOP_N, const std::string &options_="", long timeout_=DEFAULT_COMMAND_TIMEOUT)
Query the SOW cache of a topic and initiates a new subscription on it.
Definition: ampsplusplus.hpp:6723