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

MemorySubscriptionManager.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
26 #ifndef _MEMORYSUBSCRIPTIONMANAGER_H_
27 #define _MEMORYSUBSCRIPTIONMANAGER_H_
28
29 #include <amps/ampsplusplus.hpp>
30 #include <amps/Field.hpp>
31 #include <algorithm>
32 #include <forward_list>
33 #include <list>
34 #include <map>
35 #include <memory>
36 #include <set>
37
42
43 namespace AMPS
44 {
45
49 class MemorySubscriptionManager : public SubscriptionManager
50 {
51 protected:
52
53 class SubscriptionInfo
54 {
55 public:
56 SubscriptionInfo(MessageHandler messageHandler_,
57 const Message& message_,
58 unsigned requestedAckTypes_)
59 : _handler(messageHandler_)
60 , _m(message_)
61 , _subId(message_.getSubscriptionId())
62 , _requestedAckTypes(requestedAckTypes_)
63 , _useBookmark(!message_.getBookmark().empty())
64 , _clearSubId(false)
65 {
66 std::string options = _m.getOptions();
67 size_t replace = options.find("replace");
68 // AMPS should be ok if options contains ,,
69 static const size_t replaceLen = 7; // -V1096
70 if (replace != std::string::npos)
71 {
72 options.erase(replace, replaceLen);
73 _m.setOptions(options);
74 }
75 _paused = (options.find("pause") != std::string::npos);
76 }
77
78 ~SubscriptionInfo()
79 {
80 if (_clearSubId)
81 {
82 _m.getSubscriptionId().clear();
83 }
84 }
85
86 void resubscribe(Client& client_, AMPS_ATOMIC_BASE_TYPE generation_,
87 int timeout_, void* userData_)
88 {
89 // A previous NotEntitledException could have set userid on
90 // the message and that field will no longer be valid.
91 _m.setUserId((const char*)0, 0);
92 if (_useBookmark)
93 {
94 // Use the same bookmark for all members of a pause group
95 if (_paused && !_recent.empty())
96 {
97 _m.setBookmark(_recent);
98 }
99 else
100 {
101 _m.assignOwnershipBookmark(client_.getBookmarkStore().getMostRecent(_subId));
102 }
103 }
104 _m.newCommandId();
105 _m.setAckTypeEnum(_requestedAckTypes);
106 if (!userData_)
107 {
108 client_.send(_handler, _m, timeout_, generation_);
109 }
110 else
111 {
112 MessageHandler handler(_handler.function(), userData_);
113 // Empty string indicates not registered due to generation count
114 if (client_.send(handler, _m, timeout_, generation_).empty())
115 {
116 amps_invoke_remove_route_function(userData_);
117 }
118 }
119 }
120
121 Message message() const
122 {
123 return _m;
124 }
125
126 MessageHandler messageHandler() const
127 {
128 return _handler;
129 }
130
131 const Message::Field& subId() const
132 {
133 return _subId;
134 }
135
136 unsigned requestedAcks() const
137 {
138 return _requestedAckTypes;
139 }
140
141 // Returns true if the last subId is removed, false otherwise
142 bool removeSubId(const Message::Field& subId_)
143 {
144 size_t subIdLen = subId_.len();
145 const char* subIdData = subId_.data();
146 while (subIdLen && *subIdData == ',')
147 {
148 ++subIdData;
149 --subIdLen;
150 }
151 while (subIdLen && subIdData[subIdLen - 1] == ',')
152 {
153 --subIdLen;
154 }
155 if (subIdLen == 0 || subIdLen > _subId.len())
156 {
157 return _subId.empty();
158 }
159 bool match = true;
160 size_t matchStart = 0;
161 size_t matchCount = 0;
162 for (size_t i = 0; i < _subId.len(); ++i)
163 {
164 if (_subId.data()[i] == ',')
165 {
166 if (matchCount == subIdLen)
167 {
168 break;
169 }
170 matchStart = i + 1;
171 matchCount = 0;
172 match = true;
173 }
174 else if (match)
175 {
176 if (_subId.data()[i] == subIdData[matchCount])
177 {
178 ++matchCount;
179 }
180 else
181 {
182 matchCount = 0;
183 match = false;
184 }
185 }
186 }
187 if (match && matchCount == subIdLen)
188 {
189 size_t newLen = _subId.len() - matchCount;
190 if (newLen > 1) // More than just ,
191 {
192 while (matchStart + matchCount < _subId.len() &&
193 _subId.data()[matchStart + matchCount] == ',')
194 {
195 ++matchCount;
196 --newLen;
197 }
198 char* buffer = (char*)malloc(newLen);
199 if (!buffer)
200 {
201 throw AMPSException("Memory allocation failed in removeSubscription", AMPS_E_MEMORY);
202 }
203 // Match is not first
204 if (matchStart > 0)
205 {
206 memcpy(buffer, _subId.data(), matchStart);
207 }
208 // Match is not last
209 if (matchStart + matchCount < _subId.len())
210 {
211 memcpy(buffer + matchStart,
212 _subId.data() + matchStart + matchCount,
213 _subId.len() - matchStart - matchCount);
214 }
215 if (_clearSubId)
216 {
217 _m.getSubscriptionId().clear();
218 }
219 else
220 {
221 _clearSubId = true;
222 }
223 _m.assignSubscriptionId(buffer, newLen);
224 _subId = _m.getSubscriptionId();
225 return false;
226 }
227 else
228 {
229 if (_clearSubId)
230 {
231 _m.getSubscriptionId().clear();
232 _clearSubId = false;
233 }
234 else
235 {
236 _m.getSubscriptionId().assign(NULL, 0);
237 }
238 _subId = _m.getSubscriptionId();
239 return true;
240 }
241 }
242 return _subId.empty();
243 }
244
245 bool paused() const
246 {
247 return _paused;
248 }
249
250 void pause()
251 {
252 if (_paused)
253 {
254 return;
255 }
256 std::string opts(Message::Options::Pause());
257 opts.append(_m.getOptions());
258 _m.setOptions(opts);
259 _paused = true;
260 }
261
262 std::string getMostRecent(Client& client_)
263 {
264 if (!_recent.empty())
265 {
266 return _recent;
267 }
268 std::map<amps_uint64_t, amps_uint64_t> publishers;
269 const char* start = _subId.data();
270 const char* end = _subId.data() + _subId.len();
271 while (start < end)
272 {
273 const char* comma = (const char*)memchr(start, ',',
274 (size_t)(end - start));
275 // No more commas found, just use start->end
276 if (!comma)
277 {
278 comma = end;
279 }
280 if (comma == start)
281 {
282 start = comma + 1;
283 continue;
284 }
285 Message::Field sid(start, (size_t)(comma - start));
286 Message::Field sidRecent = client_.getBookmarkStore().getMostRecent(sid);
287 const char* sidRecentStart = sidRecent.data();
288 const char* sidRecentEnd = sidRecent.data() + sidRecent.len();
289 while (sidRecentStart < sidRecentEnd)
290 {
291 const char* sidRecentComma = (const char*)
292 memchr(sidRecentStart, ',',
293 (size_t)(sidRecentEnd - sidRecentStart));
294 // No more commas found, just use start->end
295 if (!sidRecentComma)
296 {
297 sidRecentComma = sidRecentEnd;
298 }
299 if (sidRecentComma == sidRecentStart)
300 {
301 sidRecentStart = sidRecentComma + 1;
302 continue;
303 }
304 Message::Field bookmark(sidRecentStart,
305 (size_t)(sidRecentComma - sidRecentStart));
306 amps_uint64_t publisher = (amps_uint64_t)0;
307 amps_uint64_t seq = (amps_uint64_t)0;
308 Field::parseBookmark(bookmark, publisher, seq);
309 if (publishers.count(publisher) == 0
310 || publishers[publisher] > seq)
311 {
312 publishers[publisher] = seq;
313 }
314 // Move past comma
315 sidRecentStart = sidRecentComma + 1;
316 }
317 // Move past comma
318 start = comma + 1;
319 sidRecent.clear();
320 }
321 std::ostringstream os;
322 for (std::map<amps_uint64_t, amps_uint64_t>::iterator i = publishers.begin();
323 i != publishers.end();
324 ++i)
325 {
326 if (i->first == 0 && i->second == 0)
327 {
328 continue;
329 }
330 if (os.tellp() > 0)
331 {
332 os << ',';
333 }
334 os << i->first << '|' << i->second << "|";
335 }
336 _recent = os.str();
337 return _recent;
338 }
339
340 void setMostRecent(const std::string& recent_)
341 {
342 _recent = recent_;
343 }
344
345 private:
346 std::string _recent;
347 MessageHandler _handler;
348 Message _m;
349 Message::Field _subId;
350 unsigned _requestedAckTypes;
351 bool _useBookmark;
352 bool _paused;
353 bool _clearSubId;
354
355 };//class SubscriptionInfo
356
357 typedef std::map<SubscriptionInfo*, AMPSException> FailedResubscribeMap;
358
359 class Resubscriber
360 {
361 Client& _client;
362 MemorySubscriptionManager* _pManager;
363 AMPS_ATOMIC_BASE_TYPE _generation;
364 int _timeout;
365 public:
366 FailedResubscribeMap* _failures;
367
368 Resubscriber(Client& client_, MemorySubscriptionManager* pManager_)
369 : _client(client_)
370 , _pManager(pManager_)
371 , _generation(0)
372 , _timeout(pManager_->getResubscriptionTimeout())
373 {
374 _failures = new FailedResubscribeMap();
375 }
376 // We want the same pointer
377 Resubscriber(const Resubscriber& rhs_)
378 : _client(rhs_._client)
379 , _pManager(rhs_._pManager)
380 , _generation(rhs_._generation)
381 , _timeout(rhs_._timeout)
382 , _failures(rhs_._failures)
383 { }
384 Resubscriber& operator=(const Resubscriber& rhs_)
385 {
386 _client = rhs_._client;
387 _pManager = rhs_._pManager;
388 _generation = rhs_._generation;
389 _timeout = rhs_._timeout;
390 _failures = rhs_._failures;
391 return *this;
392 }
393 void setGenerationCount(void)
394 {
395 _generation = _pManager->getGenerationCount();
396 }
397 AMPS_ATOMIC_BASE_TYPE getGenerationCount(void)
398 {
399 return _generation;
400 }
401 void operator()(SubscriptionInfo* iter_)
402 {
403 try
404 {
405 if (_generation == _pManager->getGenerationCount())
406 {
407 void* data = amps_invoke_copy_route_function(iter_->messageHandler().userData());
408 // This can silently not send if client disconnects, but connectAndLogonInternal
409 // will handle that via DisconnectDisabler
410 iter_->resubscribe(_client, _generation, _timeout, data);
411 }
412 }
413 catch (const DisconnectedException&)
414 {
415 // We don't report the disconnect to FailedResubscribeHandler.
416 // Stop trying anything else
417 throw;
418 }
419 catch (const AMPSException& ex)
420 {
421 _failures->insert(std::make_pair(iter_, ex));
422 }
423 }
424 };
425
426 class Deleter
427 {
428 bool _clearSubId;
429 public:
430 Deleter(bool clearSubId_ = false)
431 : _clearSubId(clearSubId_)
432 { }
433 void operator()(std::pair<Message::Field, SubscriptionInfo*> iter_)
434 {
435 if (_clearSubId)
436 {
437 iter_.first.clear();
438 }
439 else
440 {
441 amps_invoke_remove_route_function(iter_.second->messageHandler().userData());
442 delete iter_.second;
443 }
444 }
445 void operator()(SubscriptionInfo* iter_)
446 {
447 delete iter_;
448 }
449 };
450
451 virtual SubscriptionInfo* createSubscriptionInfo(MessageHandler messageHandler_,
452 const Message& message_,
453 unsigned requestedAckTypes_)
454 {
455 return new SubscriptionInfo(messageHandler_, message_,
456 requestedAckTypes_);
457 }
458
459 public:
460 typedef std::map<Message::Field, SubscriptionInfo*, Message::Field::FieldHash> SubscriptionMap;
461
462 MemorySubscriptionManager()
463 : _generationCount(0)
464 , _resubscribing(0)
465 , _resubscriptionTimeout(getDefaultResubscriptionTimeout())
466 { ; }
467
468 ~MemorySubscriptionManager()
469 {
470 _clear();
471 }
472
479 AMPS_ATOMIC_BASE_TYPE subscribe(MessageHandler messageHandler_,
480 const Message& message_, unsigned requestedAckTypes_)
481 {
482 const Message::Field& subId = message_.getSubscriptionId();
483 if (!subId.empty())
484 {
485 Lock<Mutex> l(_lock);
486 while (_resubscribing != 0)
487 {
488 _lock.wait(10);
489 }
490 std::string options = message_.getOptions();
491 if (options.find("resume") != std::string::npos)
492 {
493 // For a resume, we store each sub id with a single Subscription
494 SubscriptionInfo* subInfo = createSubscriptionInfo(MessageHandler(),
495 message_,
496 requestedAckTypes_);
497 bool saved = false;
498 Field fullSubId = subInfo->subId();
499 const char* start = fullSubId.data();
500 const char* end = fullSubId.data() + fullSubId.len();
501 while (start < end)
502 {
503 const char* comma = (const char*)memchr(start, ',',
504 (size_t)(end - start));
505 // No more commas found, just use start->end
506 if (!comma)
507 {
508 comma = end;
509 }
510 if (comma == start)
511 {
512 start = comma + 1;
513 continue;
514 }
515 Message::Field sid = Message::Field(start,
516 (size_t)(comma - start));
517 // Calling resume on something already resumed is ignored,
518 // so don't update anything that exists.
519 if (_resumed.find(sid) == _resumed.end())
520 {
521 _resumed[sid.deepCopy()] = subInfo;
522 saved = true;
523 }
524 // Move past comma
525 start = comma + 1;
526 }
527 if (saved)
528 {
529 _resumedSet.insert(subInfo);
530 }
531 else
532 {
533 delete subInfo;
534 }
535 }
536 else if (options.find("pause") != std::string::npos)
537 {
538 const char* start = subId.data();
539 const char* end = subId.data() + subId.len();
540 while (start < end)
541 {
542 MessageHandler messageHandler = messageHandler_;
543 const char* comma = (const char*)memchr(start, ',',
544 (size_t)(end - start));
545 // No more commas found, just use start->end
546 if (!comma)
547 {
548 comma = end;
549 }
550 if (comma == start)
551 {
552 start = comma + 1;
553 continue;
554 }
555 Message::Field sid = Message::Field(start,
556 (size_t)(comma - start));
557 SubscriptionMap::iterator resume = _resumed.find(sid);
558 if (resume != _resumed.end())
559 {
560 SubscriptionInfo* subPtr = resume->second;
561 Message::Field subField(resume->first);
562 _resumed.erase(resume); // Remove mapping for sid
563 subField.clear();
564 // If last subId, remove completely
565 if (subPtr->removeSubId(sid))
566 {
567 _resumedSet.erase(subPtr);
568 delete subPtr;
569 }
570 }
571 // Move past comma
572 start = comma + 1;
573 SubscriptionMap::iterator item = _active.find(sid);
574 if (item != _active.end())
575 {
576 if (options.find("replace") != std::string::npos)
577 {
578 messageHandler = item->second->messageHandler();
579 delete item->second;
580 _active.erase(item);
581 }
582 else
583 {
584 item->second->pause();
585 continue; // Leave current one
586 }
587 }
588 else
589 {
590 Unlock<Mutex> u(_lock);
591 void* data = amps_invoke_copy_route_function(
592 messageHandler_.userData());
593 if (data)
594 {
595 messageHandler = MessageHandler(messageHandler_.function(), data);
596 }
597 }
598 Message m = message_.deepCopy();
599 m.setSubscriptionId(sid.data(), sid.len());
600 SubscriptionInfo* s = createSubscriptionInfo(messageHandler, m,
601 requestedAckTypes_);
602 // Insert using the subId from s, which is deep copy of original
603 _active[s->subId()] = s;
604 }
605 }
606 else // Not a pause or resume
607 {
608 MessageHandler messageHandler = messageHandler_;
609 SubscriptionMap::iterator item = _active.find(subId);
610 if (item != _active.end())
611 {
612 messageHandler = item->second->messageHandler();
613 delete item->second;
614 _active.erase(item);
615 }
616 else
617 {
618 Unlock<Mutex> u(_lock);
619 void* data = amps_invoke_copy_route_function(
620 messageHandler_.userData());
621 if (data)
622 {
623 messageHandler = MessageHandler(messageHandler_.function(), data);
624 }
625 }
626 SubscriptionInfo* s = createSubscriptionInfo(messageHandler,
627 message_,
628 requestedAckTypes_);
629 // Insert using the subId from s, which is deep copy of original
630 _active[s->subId()] = s;
631 }
632 }
633 return (AMPS_ATOMIC_BASE_TYPE)_generationCount;
634 }
635
639 void unsubscribe(const Message::Field& subId_)
640 {
641 Lock<Mutex> l(_lock);
642 SubscriptionMap::iterator item = _active.find(subId_);
643 if (item != _active.end())
644 {
645 SubscriptionInfo* subPtr = item->second;
646 _active.erase(item);
647 while (_resubscribing != 0)
648 {
649 _lock.wait(10);
650 }
651 Unlock<Mutex> u(_lock);
652 amps_invoke_remove_route_function(subPtr->messageHandler().userData());
653 delete subPtr;
654 }
655 item = _resumed.find(subId_);
656 if (item != _resumed.end())
657 {
658 SubscriptionInfo* subPtr = item->second;
659 Message::Field subField(item->first);
660 _resumed.erase(item);
661 subField.clear();
662 // If last subId, remove completely
663 if (subPtr->removeSubId(subId_))
664 {
665 _resumedSet.erase(subPtr);
666 while (_resubscribing != 0)
667 {
668 _lock.wait(10);
669 }
670 delete subPtr;
671 }
672 }
673 }
674
677 void clear()
678 {
679 _clear();
680 }
681
684 void _clear()
685 {
686 Lock<Mutex> l(_lock);
687 while (_resubscribing != 0)
688 {
689 _lock.wait(10);
690 }
691 // Settting _resubscribing keeps other threads from touching data
692 // even if lock isn't held. Don't want to hold lock when
693 // amps_invoke_remove_route_function is called.
694 AtomicFlagFlip resubFlip(&_resubscribing);
695 {
696 Unlock<Mutex> u(_lock);
697 std::for_each(_active.begin(), _active.end(), Deleter());
698 std::for_each(_resumedSet.begin(), _resumedSet.end(), Deleter());
699 std::for_each(_resumed.begin(), _resumed.end(), Deleter(true));
700 }
701 _active.clear();
702 _resumed.clear();
703 _resumedSet.clear();
704 }
705
709 void resubscribe(Client& client_)
710 {
711 std::forward_list<SubscriptionInfo*> subscriptions;
712 Resubscriber resubscriber(client_, this);
713 // Only one resubscribe at a time, if one is already happening
714 // it should end quickly with gen count bumped
715 Lock<Mutex> l(_lock);
716 AMPS_FETCH_ADD(&_generationCount, 1);
717 while (_resubscribing != 0)
718 {
719 _lock.wait(10);
720 }
721 try
722 {
723 AtomicFlagFlip resubFlip(&_resubscribing);
724 {
725 resubscriber.setGenerationCount();
726 subscriptions.assign(_resumedSet.begin(), _resumedSet.end());
727 for (SubscriptionMap::iterator iter = _active.begin();
728 iter != _active.end(); ++iter)
729 {
730 SubscriptionInfo* sub = iter->second;
731 if (sub->paused())
732 {
733 SubscriptionMap::iterator resIter = _resumed.find(sub->subId());
734 // All pause subs resuming together should be sent with
735 // bookmark as list of the resumes' most recents
736 if (resIter != _resumed.end())
737 {
738 sub->setMostRecent(resIter->second->getMostRecent(client_));
739 }
740 }
741 subscriptions.push_front(iter->second);
742 }
743 }
744 Unlock<Mutex> unlock(_lock);
745 bool throwExcept = false;
746 AMPSException except("None", AMPS_E_OK);
747 try
748 {
749 std::for_each(subscriptions.begin(), subscriptions.end(),
750 resubscriber);
751 }
752 catch (const DisconnectedException& disEx_)
753 {
754 throwExcept = true;
755 except = disEx_;
756 }
757 if (resubscriber.getGenerationCount() != getGenerationCount())
758 {
759 // We need to throw here because disconnect handling is disabled when
760 // this is called by the HAClient in connectAndLogonInternal
761 throw DisconnectedException("Disconnect occurred during resubscribe");
762 }
763 std::vector<SubscriptionInfo*> removals;
764 if (_failedResubscribeHandler)
765 {
766 try
767 {
768 for (auto failedSub = resubscriber._failures->begin();
769 failedSub != resubscriber._failures->end(); ++failedSub)
770 {
771 SubscriptionInfo* pSubInfo = failedSub->first;
772 if (_failedResubscribeHandler->failure(pSubInfo->message(),
773 pSubInfo->messageHandler(),
774 pSubInfo->requestedAcks(),
775 failedSub->second))
776 {
777 removals.push_back(pSubInfo);
778 }
779 else // We'll rethrow an exception for failure left in place
780 {
781 if (!throwExcept)
782 {
783 except = failedSub->second;
784 throwExcept = true;
785 }
786 }
787 }
788 }
789 catch (const AMPSException& ex_)
790 {
791 if (!throwExcept)
792 {
793 except = ex_;
794 throwExcept = true;
795 }
796 }
797 catch (const std::exception& ex_)
798 {
799 if (!throwExcept)
800 {
801 except = AMPSException(ex_.what(), AMPS_E_RETRY);
802 throwExcept = true;
803 }
804 }
805 catch (...)
806 {
807 if (!throwExcept)
808 {
809 except = AMPSException("Unknown Exception thrown by FailedResubscribeHandler", AMPS_E_RETRY);
810 throwExcept = true;
811 }
812 }
813 }
814 else if (!throwExcept)
815 {
816 throwExcept = !resubscriber._failures->empty();
817 if (throwExcept)
818 {
819 except = resubscriber._failures->begin()->second;
820 }
821 }
822 // Remove any failiures that should be removed
823 if (_failedResubscribeHandler && !removals.empty())
824 {
825 Lock<Mutex> relock(_lock);
826 for (std::vector<SubscriptionInfo*>::iterator pSubInfo = removals.begin();
827 pSubInfo != removals.end(); ++pSubInfo)
828 {
829 _active.erase((*pSubInfo)->subId());
830 _resumed.erase((*pSubInfo)->subId());
831 _resumedSet.erase(*pSubInfo);
832 delete *pSubInfo;
833 }
834 }
835 // Throw an exception not removed
836 if (throwExcept)
837 {
838 throw except;
839 }
840 delete resubscriber._failures;
841 resubscriber._failures = 0;
842 }
843 catch (const AMPSException&)
844 {
845 delete resubscriber._failures;
846 resubscriber._failures = 0;
847 throw;
848 }
849 catch (const std::exception&)
850 {
851 delete resubscriber._failures;
852 resubscriber._failures = 0;
853 throw;
854 }
855 }
856
860 void setResubscriptionTimeout(int timeout_)
861 {
862 if (timeout_ >= 0)
863 {
864 _resubscriptionTimeout = timeout_;
865 }
866 }
867
871 int getResubscriptionTimeout(void)
872 {
873 return _resubscriptionTimeout;
874 }
875
880 static int setDefaultResubscriptionTimeout(int timeout_)
881 {
882 static int _defaultResubscriptionTimeout = // -V1096
883 AMPS_SUBSCRIPTION_MANAGER_DEFAULT_TIMEOUT;
884 if (timeout_ >= 0)
885 {
886 _defaultResubscriptionTimeout = timeout_;
887 }
888 return _defaultResubscriptionTimeout;
889 }
890
895 static int getDefaultResubscriptionTimeout(void)
896 {
897 return setDefaultResubscriptionTimeout(-1);
898 }
899
904 AMPS_ATOMIC_BASE_TYPE getGenerationCount() const
905 {
906 return (AMPS_ATOMIC_BASE_TYPE)_generationCount;
907 }
908
909 private:
910
911 SubscriptionMap _active;
912 SubscriptionMap _resumed;
913 std::set<SubscriptionInfo*> _resumedSet;
914 Mutex _lock;
915 AMPS_ATOMIC_TYPE _generationCount;
916 AMPS_ATOMIC_TYPE_8 _resubscribing;
917 int _resubscriptionTimeout;
918 }; //class MemorySubscriptionManager
919
920 } // namespace AMPS
921
922 #endif //_MEMORYSUBSCRIPTIONMANAGER_H_
Field.hpp
Defines the AMPS::Field class, which represents the value of a field in a message.
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_MEMORY
@ AMPS_E_MEMORY
A memory error occurred.
Definition: amps.h:225
ampsplusplus.hpp
Core type, function, and class declarations for the AMPS C++ client.
AMPS::AMPSException
Base class for all exceptions in AMPS.
Definition: AMPSException.hpp:41
AMPS::Client
Client represents a connection to an AMPS server, but does not provide failover or reconnection behav...
Definition: ampsplusplus.hpp:5507
AMPS::Client::getBookmarkStore
BookmarkStore getBookmarkStore()
Get the bookmark store being used by the client.
Definition: ampsplusplus.hpp:5889
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:5751
AMPS::Field
Field represents the value of a single field in a Message.
Definition: Field.hpp:88
AMPS::Field::clear
void clear()
Deletes the data associated with this Field, should only be used on Fields that were created as deepC...
Definition: Field.hpp:266
AMPS::Field::empty
bool empty() const
Returns 'true' if empty, 'false' otherwise.
Definition: Field.hpp:129
AMPS::Field::deepCopy
void deepCopy(const Field &orig_)
Makes self a deep copy of the original field.
Definition: Field.hpp:219
AMPS::Field::len
size_t len() const
Returns the length of the data underlying this field.
Definition: Field.hpp:286
AMPS::Field::data
const char * data() const
Returns the (non-null-terminated) data underlying this field.
Definition: Field.hpp:279
AMPS::Handler< MessageHandlerFunc, const Message & >
AMPS::MemorySubscriptionManager
A SubscriptionManager implementation that maintains subscriptions placed in memory so that they can b...
Definition: MemorySubscriptionManager.hpp:50
AMPS::MemorySubscriptionManager::setResubscriptionTimeout
void setResubscriptionTimeout(int timeout_)
Sets the timeout used when trying to resubscribe after disconnect.
Definition: MemorySubscriptionManager.hpp:860
AMPS::MemorySubscriptionManager::clear
void clear()
Clear all subscriptions from the manager.
Definition: MemorySubscriptionManager.hpp:677
AMPS::MemorySubscriptionManager::unsubscribe
void unsubscribe(const Message::Field &subId_)
Remove the subscription from the manager.
Definition: MemorySubscriptionManager.hpp:639
AMPS::MemorySubscriptionManager::resubscribe
void resubscribe(Client &client_)
Place all saved subscriptions on the provided Client.
Definition: MemorySubscriptionManager.hpp:709
AMPS::MemorySubscriptionManager::getGenerationCount
AMPS_ATOMIC_BASE_TYPE getGenerationCount() const
Gets the current generation count, which is incremented each time resubcribe is called.
Definition: MemorySubscriptionManager.hpp:904
AMPS::MemorySubscriptionManager::_clear
void _clear()
Clear all subscriptions from the manager.
Definition: MemorySubscriptionManager.hpp:684
AMPS::MemorySubscriptionManager::subscribe
AMPS_ATOMIC_BASE_TYPE subscribe(MessageHandler messageHandler_, const Message &message_, unsigned requestedAckTypes_)
Save a subscription so it can be placed again if a disconnect occurs.
Definition: MemorySubscriptionManager.hpp:479
AMPS::MemorySubscriptionManager::getDefaultResubscriptionTimeout
static int getDefaultResubscriptionTimeout(void)
Gets the default timeout used by new MemorySubscriptionManager objects when trying to resubscribe aft...
Definition: MemorySubscriptionManager.hpp:895
AMPS::MemorySubscriptionManager::getResubscriptionTimeout
int getResubscriptionTimeout(void)
Gets the timeout used when trying to resubscribe after disconnect.
Definition: MemorySubscriptionManager.hpp:871
AMPS::MemorySubscriptionManager::setDefaultResubscriptionTimeout
static int setDefaultResubscriptionTimeout(int timeout_)
Sets the default timeout used by new MemorySubscriptionManager objects when trying to resubscribe aft...
Definition: MemorySubscriptionManager.hpp:880
AMPS::Message
Message encapsulates a single message sent to or received from an AMPS server, and provides methods f...
Definition: Message.hpp:518
AMPS::Message::getBookmark
Field getBookmark() const
Retrieves the value of the Bookmark header of the Message as a Field which references the underlying ...
Definition: Message.hpp:1234
AMPS::Message::deepCopy
Message deepCopy(void) const
Returns a deep copy of self.
Definition: Message.hpp:565
AMPS::Message::getSubscriptionId
Field getSubscriptionId() const
Retrieves the value of the SubscriptionId header of the Message as a Field which references the under...
Definition: Message.hpp:1467
AMPS::Message::setSubscriptionId
Message & setSubscriptionId(const std::string &v)
Sets the value of the SubscriptionId header for this Message.
Definition: Message.hpp:1467
AMPS::Message::getOptions
Field getOptions() const
Retrieves the value of the Options header of the Message as a Field which references the underlying b...
Definition: Message.hpp:1356
AMPS::SubscriptionManager
Abstract base class to manage all subscriptions placed on a client so that they can be re-established...
Definition: ampsplusplus.hpp:1462
AMPS::BookmarkStore::getMostRecent
Message::Field getMostRecent(const Message::Field &subId_)
Returns the most recent bookmark from the log that ought to be used for (re-)subscriptions.
Definition: BookmarkStore.hpp:310