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

MemoryBookmarkStore.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 _MEMORYBOOKMARKSTORE_H_
27 #define _MEMORYBOOKMARKSTORE_H_
28
29 #include <amps/BookmarkStore.hpp>
30 #include <amps/Field.hpp>
31 #include <amps/Message.hpp>
32 #include <amps/RecoveryPoint.hpp>
33 #include <amps/RecoveryPointAdapter.hpp>
34 #include <atomic>
35 #include <functional>
36 #include <map>
37 #include <sstream>
38 #include <vector>
39 #include <stdlib.h>
40 #include <assert.h>
41
42 #define AMPS_MIN_BOOKMARK_LEN 3
43 #define AMPS_INITIAL_MEMORY_BOOKMARK_SIZE 16384UL
44
49
50 namespace AMPS
51 {
52
58 class MemoryBookmarkStore : public BookmarkStoreImpl
59 {
60 protected:
61 class Subscription
62 {
63 public:
64 typedef std::map<Message::Field, size_t, Message::Field::FieldHash> RecoveryMap;
65 typedef std::map<amps_uint64_t, amps_uint64_t> PublisherMap;
66 typedef std::map<Message::Field, size_t, Message::Field::FieldHash>::iterator
67 RecoveryIterator;
68 typedef std::map<amps_uint64_t, amps_uint64_t>::iterator PublisherIterator;
69
70 // Start sequence at 1 so that 0 can be used during file subclasses
71 // recovery as an indicator that a message wasn't logged because
72 // isDiscarded() was true.
73 Subscription(MemoryBookmarkStore* store_, const Message::Field& id_)
74 : _current(1), _currentBase(0), _least(1), _leastBase(0)
75 , _recoveryMin(AMPS_UNSET_INDEX), _recoveryBase(AMPS_UNSET_INDEX)
76 , _recoveryMax(AMPS_UNSET_INDEX), _recoveryMaxBase(AMPS_UNSET_INDEX)
77 , _entriesLength(AMPS_INITIAL_MEMORY_BOOKMARK_SIZE), _entries(NULL)
78 , _store(store_)
79 {
80 // Need our own memory for the sub id
81 _id.deepCopy(id_);
82 _store->resize(_id, (char**)&_entries,
83 sizeof(Entry)*AMPS_INITIAL_MEMORY_BOOKMARK_SIZE, false);
84 setLastPersistedToEpoch();
85 }
86
87 ~Subscription()
88 {
89 Lock<Mutex> guard(_subLock);
90 if (_entries)
91 {
92 for (size_t i = 0; i < _entriesLength; ++i)
93 {
94 _entries[i]._val.clear();
95 }
96 // resize to 0 will free _entries
97 _store->resize(_id, (char**)&_entries, 0);
98 }
99 _id.clear();
100 _recent.clear();
101 _lastPersisted.clear();
102 _recentList.clear();
103 _range.clear();
104 _recoveryTimestamp.clear();
105 }
106
107 size_t log(const Message::Field& bookmark_)
108 {
109 if (bookmark_ == AMPS_BOOKMARK_NOW)
110 {
111 return 0;
112 }
113 Lock<Mutex> guard(_subLock);
114 // Either relog the recovery or log it
115 size_t index = recover(bookmark_, true);
116 if (index == AMPS_UNSET_INDEX)
117 {
118 // Check for wrap
119 if (_current >= _entriesLength)
120 {
121 _current = 0;
122 _currentBase += _entriesLength;
123 }
124 // Check for resize
125 // If list is too small, double it
126 if ((_current == _least && _leastBase < _currentBase) ||
127 (_current == _recoveryMin && _recoveryBase < _currentBase))
128 {
129 if (!_store->resize(_id, (char**)&_entries,
130 sizeof(Entry) * _entriesLength * 2))
131 {
132 //Try again
133 return log(bookmark_);
134 }
135 // Length was doubled
136 _entriesLength *= 2;
137 }
138
139 // Add this entry to the end of our list
140 /*
141 if (bookmark_ == AMPS_BOOKMARK_NOW)
142 {
143 // Save a now timestamp bookmark
144 char* nowTimestamp = (char*)malloc(AMPS_TIMESTAMP_LEN);
145 struct tm timeInfo;
146 time_t now;
147 time(&now);
148 #ifdef _WIN32
149 gmtime_s(&timeInfo, &now);
150 #else
151 gmtime_r(&now, &timeInfo);
152 #endif
153 strftime(nowTimestamp, AMPS_TIMESTAMP_LEN,
154 "%Y%m%dT%H%M%S", &timeInfo);
155 nowTimestamp[AMPS_TIMESTAMP_LEN-1] = 'Z';
156 _entries[_current]._val.assign(nowTimestamp,
157 AMPS_TIMESTAMP_LEN);
158 _entries[_current]._active = false;
159 index = _current++;
160 return index + _currentBase;
161 }
162 else
163 */
164 {
165 // Is this an attempt at a range?
166 bool isRange = BookmarkRange::isRange(bookmark_);
167 bool startExclusive = false;
168 if (isRange)
169 {
170 _range.set(bookmark_);
171 // Stricter check on range syntax
172 if (!_range.isValid())
173 {
174 _range.clear();
175 throw CommandException("Invalid bookmark range specified.");
176 }
177 startExclusive = !_range.isStartInclusive();
178 }
179 if (isRange || _publishers.empty())
180 {
181 // Check validity and init publishers map for a range
182 Message::Field parseable = isRange ? _range.getStart() : bookmark_;
183 amps_uint64_t publisher, sequence;
184 std::vector<Field> bmList = Field::parseBookmarkList(parseable);
185 if (bmList.empty() && !Field::isTimestamp(parseable))
186 {
187 if (isRange)
188 {
189 _range.clear();
190 }
191 return 0;
192 }
193 for (std::vector<Field>::iterator bkmk = bmList.begin(); bkmk != bmList.end(); ++bkmk)
194 {
195 parseBookmark(*bkmk, publisher, sequence);
196 if (publisher != (amps_uint64_t)0)
197 {
198 if (isRange && startExclusive)
199 {
200 // Compare it to our publishers map
201 PublisherIterator pub = _publishers.find(publisher);
202 if (pub == _publishers.end() || pub->second < sequence)
203 {
204 _publishers[publisher] = sequence;
205 }
206 }
207 }
208 else if (!Field::isTimestamp(*bkmk))
209 {
210 // This is an invalid bookmark, so don't save anything
211 if (isRange)
212 {
213 _range.clear();
214 if (startExclusive)
215 {
216 _publishers.clear();
217 }
218 }
219 return 0;
220 }
221 }
222 }
223 if (!isRange)
224 {
225 _entries[_current]._val.deepCopy(bookmark_);
226 }
227 else
228 {
229 Unlock<Mutex> unlock(_subLock);
230 _store->updateAdapter(this);
231 // Don't actually log a range
232 return 0;
233 }
234 }
235 _entries[_current]._active = true;
236 index = _current++;
237 }
238 return index + _currentBase;
239 }
240
241 bool discard(size_t index_)
242 {
243 Lock<Mutex> guard(_subLock);
244 return _discard(index_);
245 }
246
247 bool discard(const Message::Field& bookmark_)
248 {
249 // These are discarded when logged or not logged
250 if (bookmark_ == AMPS_BOOKMARK_NOW)
251 {
252 return false;
253 }
254 Lock<Mutex> guard(_subLock);
255 size_t search = _least;
256 size_t searchBase = _leastBase;
257 size_t searchMax = _current;
258 size_t searchMaxBase = _currentBase;
259 if (_least + _leastBase == _current + _currentBase)
260 {
261 if (_recoveryMin != AMPS_UNSET_INDEX)
262 {
263 search = _recoveryMin;
264 searchBase = _recoveryBase;
265 searchMax = _recoveryMax;
266 searchMaxBase = _recoveryMaxBase;
267 }
268 else // Store is empty, so nothing to do
269 {
270 return false;
271 }
272 }
273 assert(searchMax != AMPS_UNSET_INDEX);
274 assert(searchMaxBase != AMPS_UNSET_INDEX);
275 assert(search != AMPS_UNSET_INDEX);
276 assert(searchBase != AMPS_UNSET_INDEX);
277 // Search while we don't find the provided bookmark and we're in valid range
278 while (search + searchBase < searchMax + searchMaxBase)
279 {
280 if (_entries[search]._val == bookmark_)
281 {
282 return _discard(search + searchBase);
283 }
284 if (++search == _entriesLength)
285 {
286 // Least has now loooped around
287 searchBase += _entriesLength;
288 search = 0;
289 }
290 }
291 return false;
292 }
293
294 // Get sequence number from a Field that is a bookmark
295 static void parseBookmark(const Message::Field& field_,
296 amps_uint64_t& publisherId_,
297 amps_uint64_t& sequenceNumber_)
298 {
299 Message::Field::parseBookmark(field_, publisherId_, sequenceNumber_);
300 }
301
302 // Check to see if this message is older than the most recent one seen,
303 // and if it is, check if it discarded.
304 bool isDiscarded(const Message::Field& bookmark_)
305 {
306 Lock<Mutex> guard(_subLock);
307 if (BookmarkRange::isRange(bookmark_))
308 {
309 return false;
310 }
311
312 amps_uint64_t publisher, sequence;
313 parseBookmark(bookmark_, publisher, sequence);
314 // Bookmarks like EPOCH, NOW, or invalid bookmarks we ignore
315 // A timestamp could be logged as the first bookmark when starting
316 // a new subscription.
317 if (publisher == 0 && !Field::isTimestamp(bookmark_))
318 {
319 return true;
320 }
321 // Check if we've already recovered this bookmark
322 size_t recoveredIdx = recover(bookmark_, false);
323 // Compare it to our publishers map
324 PublisherIterator pub = _publishers.find(publisher);
325 if (pub == _publishers.end() || pub->second < sequence)
326 {
327 _publishers[publisher] = sequence;
328 if (recoveredIdx == AMPS_UNSET_INDEX)
329 {
330 return false;
331 }
332 }
333 if (recoveredIdx != AMPS_UNSET_INDEX)
334 {
335 if (!_entries[recoveredIdx]._active)
336 {
337 _recovered.erase(bookmark_);
338 return true;
339 }
340 return false;
341 }
342 // During recovery, we don't really care if it's been discarded
343 // or not. We just want _publishers updated. No need for the
344 // costly linear search.
345 if (_store->_recovering)
346 {
347 return false;
348 }
349 // During failure and recovery scenarios, we'll see out of order
350 // bookmarks arrive, either because (a) we're replaying or (b)
351 // a publisher has cut over, and we've cut over to a new server.
352 // Scan the list to see if we have a match.
353 size_t base = _leastBase;
354 for (size_t i = _least; i + base < _current + _currentBase; i++)
355 {
356 if ( i >= _entriesLength )
357 {
358 i = 0;
359 base = _currentBase;
360 }
361 if (_entries[i]._val == bookmark_)
362 {
363 return !_entries[i]._active;
364 }
365 }
366
367 return true; // message is totally discarded
368 }
369
370 bool empty(void) const
371 {
372 if (_least == AMPS_UNSET_INDEX ||
373 ((_least + _leastBase) == (_current + _currentBase) &&
374 _recoveryMin == AMPS_UNSET_INDEX))
375 {
376 return true;
377 }
378 return false;
379 }
380
381 void updateMostRecent()
382 {
383 Lock<Mutex> guard(_subLock);
384 _updateMostRecent();
385 }
386
387 const BookmarkRange& getRange() const
388 {
389 return _range;
390 }
391
392 Message::Field getMostRecentList(bool usePublishersList_ = true)
393 {
394 Lock<Mutex> guard(_subLock);
395 bool useLastPersisted = !_lastPersisted.empty() &&
396 _lastPersisted.len() > 1;
397 // when this is called, we'll take a moment to update the list
398 // of things recovered,
399 // so we don't accidentally log anything we ought not to.
400 _updateMostRecent();
401 bool useRecent = !_recent.empty() && _recent.len() > 1;
402 amps_uint64_t lastPublisher = 0;
403 amps_uint64_t lastSeq = 0;
404 amps_uint64_t recentPublisher = 0;
405 amps_uint64_t recentSeq = 0;
406 if (useLastPersisted)
407 {
408 parseBookmark(_lastPersisted, lastPublisher, lastSeq);
409 }
410 if (useRecent)
411 {
412 parseBookmark(_recent, recentPublisher, recentSeq);
413 if (empty() && useLastPersisted)
414 {
415 useRecent = false;
416 }
417 else
418 {
419 if (useLastPersisted && lastPublisher == recentPublisher)
420 {
421 if (lastSeq <= recentSeq)
422 {
423 useRecent = false;
424 }
425 else
426 {
427 useLastPersisted = false;
428 }
429 }
430 }
431 }
432 // Set size for all bookmarks that will be used
433 size_t totalLen = (useLastPersisted ? _lastPersisted.len() + 1 : 0);
434 if (useRecent)
435 {
436 totalLen += _recent.len() + 1;
437 }
438 // If we don't have a non-EPOCH persisted ack and we don't have a
439 // non-EPOCH most recent bookmark, OR we have a range
440 // we can build a list based on all the publishers instead.
441 if (usePublishersList_
442 && ((!useLastPersisted && !useRecent)
443 || _lastPersisted == AMPS_BOOKMARK_EPOCH))
444 {
445 std::ostringstream os;
446 for (PublisherIterator pub = _publishers.begin();
447 pub != _publishers.end(); ++pub)
448 {
449 if (pub->first == 0 && pub->second == 0)
450 {
451 continue;
452 }
453 if (pub->first == recentPublisher && recentSeq < pub->second)
454 {
455 os << recentPublisher << '|' << recentSeq << "|,";
456 }
457 else
458 {
459 os << pub->first << '|' << pub->second << "|,";
460 }
461 }
462 std::string recent = os.str();
463 if (!recent.empty())
464 {
465 totalLen = recent.length();
466 if (!_recoveryTimestamp.empty())
467 {
468 totalLen += _recoveryTimestamp.len();
469 recent += std::string(_recoveryTimestamp);
470 }
471 else
472 {
473 // Remove trailing ,
474 recent.erase(--totalLen);
475 }
476 // Reset _recentList to new value and return it
477 _recentList.clear();
478 _recentList = Message::Field(recent).deepCopy();
479 if (_range.isValid())
480 {
481 if (_range.getStart() != recent
482 && _recentList != AMPS_BOOKMARK_EPOCH)
483 {
484 _range.replaceStart(_recentList, true);
485 }
486 else if (_range.isStartInclusive())
487 {
488 amps_uint64_t publisher, sequence;
489 parseBookmark(_range.getStart(), publisher,
490 sequence);
491 PublisherIterator pub = _publishers.find(publisher);
492 if (pub != _publishers.end()
493 && pub->second >= sequence)
494 {
495 _range.makeStartExclusive();
496 }
497 }
498 return _range.deepCopy();
499 }
500 return _recentList.deepCopy();
501 }
502 if (_range.isValid())
503 {
504 return _range.deepCopy();
505 }
506 }
507 if (!_recoveryTimestamp.empty() && !_range.isValid())
508 {
509 totalLen += _recoveryTimestamp.len() + 1;
510 }
511 // If we have nothing discarded, return EPOCH
512 if (totalLen == 0
513 || (_recent.len() < 2 && !empty()))
514 {
515 if (_range.isValid())
516 {
517 return _range.deepCopy();
518 }
519 if (!useRecent)
520 {
521 return Field::stringCopy(AMPS_BOOKMARK_EPOCH);
522 }
523 _setLastPersistedToEpoch();
524 return _lastPersisted.deepCopy();
525 }
526 // Remove the trailing , from the length
527 totalLen -= 1;
528 char* field = (char*)malloc(totalLen);
529 if (!field)
530 {
531 throw AMPSException("Failed to allocate memory for most recent bookmark.", AMPS_E_MEMORY);
532 }
533 size_t len = 0;
534 if (useRecent)
535 {
536 len = _recent.len();
537 memcpy(field, _recent.data(), len);
538 if (len < totalLen)
539 {
540 field[len++] = ',';
541 }
542 }
543 if (useLastPersisted)
544 {
545 memcpy(field + len, _lastPersisted.data(), _lastPersisted.len());
546 len += _lastPersisted.len();
547 if (len < totalLen)
548 {
549 field[len++] = ',';
550 }
551 }
552 if (!_recoveryTimestamp.empty() && !_range.isValid())
553 {
554 memcpy(field + len, _recoveryTimestamp.data(),
555 _recoveryTimestamp.len());
556 // If more is to be written after this, uncomment the following
557 //len += _lastPersisted.len();
558 //if (len < totalLen) field[len++] = ',';
559 }
560 // _recentList clear will delete[] current buffer and assign will get cleared
561 _recentList.clear();
562 _recentList.assign(field, totalLen);
563 if (_range.isValid())
564 {
565 if (_recentList != AMPS_BOOKMARK_EPOCH)
566 {
567 if (_range.getStart() != _recentList)
568 {
569 _range.replaceStart(_recentList, true);
570 }
571 else if (_range.isStartInclusive())
572 {
573 amps_uint64_t publisher, sequence;
574 parseBookmark(_range.getStart(), publisher,
575 sequence);
576 PublisherIterator pub = _publishers.find(publisher);
577 if (pub != _publishers.end()
578 && pub->second >= sequence)
579 {
580 _range.makeStartExclusive();
581 }
582 }
583 }
584 return _range.deepCopy();
585 }
586 return _recentList.deepCopy();
587 }
588
589 Message::Field getMostRecent(bool update_ = false)
590 {
591 Lock<Mutex> guard(_subLock);
592 // Return the same as last time if nothing's changed
593 // _recent is the most recent bookmark.
594 if (update_ && _store->_recentChanged)
595 {
596 _updateMostRecent();
597 }
598 if (_recent.empty())
599 {
600 return Message::Field(AMPS_BOOKMARK_EPOCH);
601 }
602 else
603 {
604 return _recent;
605 }
606 }
607
608 Message::Field getLastPersisted()
609 {
610 Lock<Mutex> guard(_subLock);
611 return _lastPersisted;
612 }
613
614 void setMostRecent(const Message::Field& recent_)
615 {
616 _recent.clear();
617 _recent.deepCopy(recent_);
618 }
619
620 void setRecoveryTimestamp(const char* recoveryTimestamp_,
621 size_t len_ = 0)
622 {
623 _recoveryTimestamp.clear();
624 size_t len = (len_ == 0) ? AMPS_TIMESTAMP_LEN : len_;
625 char* ts = (char*)malloc(len);
626 if (!ts)
627 {
628 throw AMPSException("Failed to allocate memory for recovery timestamp bookmark.", AMPS_E_MEMORY);
629 }
630 memcpy((void*)ts, (const void*)recoveryTimestamp_, len);
631 _recoveryTimestamp.assign(ts, len);
632 }
633
634 void moveEntries(char* old_, char* new_, size_t newSize_)
635 {
636 size_t least = _least;
637 size_t leastBase = _leastBase;
638 if (_recoveryMin != AMPS_UNSET_INDEX)
639 {
640 least = _recoveryMin;
641 leastBase = _recoveryBase;
642 }
643 // First check if we grew in place, if so, just move current after least
644 if (old_ == new_)
645 {
646 if (newSize_ - (sizeof(Entry)*_entriesLength) > sizeof(Entry)*least)
647 {
648 memcpy(new_ + (sizeof(Entry)*_entriesLength),
649 old_, (sizeof(Entry)*least));
650 // Clear the beginning where those entries were
651 memset(old_, 0, sizeof(Entry)*least);
652 }
653 else // We have to use an intermediate buffer
654 {
655 Entry* buffer = new Entry[least];
656 memcpy((void*)buffer, (void*)old_, sizeof(Entry)*least);
657 //Put the beginning entries at the start of the new buffer
658 memcpy((void*)new_, (void*)((char*)old_ + (sizeof(Entry)*least)),
659 (_entriesLength - least)*sizeof(Entry));
660 //Put the end entries after the beginning entries
661 memcpy((void*)((char*)new_ + ((_entriesLength - least)*sizeof(Entry))),
662 (void*)buffer, least * sizeof(Entry));
663 // Least is now at 0 so base must be increased
664 leastBase += least;
665 least = 0;
666 delete [] buffer;
667 }
668 }
669 else
670 {
671 //Put the beginning entries at the start of the new buffer
672 memcpy((void*)new_, (void*)((char*)old_ + (sizeof(Entry)*least)),
673 (_entriesLength - least)*sizeof(Entry));
674 //Put the end entries after the beginning entries
675 memcpy((void*)((char*)new_ + ((_entriesLength - least)*sizeof(Entry))),
676 (void*)old_, least * sizeof(Entry));
677 // Least is now at 0 so base must be increased
678 leastBase += least;
679 least = 0;
680 }
681 if (_recoveryMin != AMPS_UNSET_INDEX)
682 {
683 _least = least + (_least + _leastBase) - (_recoveryMin + _recoveryBase);
684 _recoveryMax = least + (_recoveryMax + _recoveryMaxBase) -
685 (_recoveryMin + _recoveryBase);
686 _recoveryMaxBase = leastBase;
687 _recoveryMin = least;
688 _recoveryBase = leastBase;
689 }
690 else
691 {
692 _least = least;
693 }
694 _leastBase = leastBase;
695 // Current is now after everything and using the same base
696 _currentBase = _leastBase;
697 _current = least + _entriesLength;
698 }
699
700 inline size_t getOldestBookmarkSeq()
701 {
702 Lock<Mutex> guard(_subLock);
703 // If there is nothing in the store, return -1, otherwise return lowest
704 return ((_least + _leastBase) == (_current + _currentBase)) ? AMPS_UNSET_INDEX :
705 _least + _leastBase;
706 }
707
708 bool lastPersisted(const Message::Field& bookmark_)
709 {
710 // These shouldn't be persisted
711 if (bookmark_ == AMPS_BOOKMARK_NOW
712 || BookmarkRange::isRange(bookmark_))
713 {
714 return false;
715 }
716 Lock<Mutex> guard(_subLock);
717 return _setLastPersisted(bookmark_);
718 }
719
720 bool _setLastPersisted(const Message::Field& bookmark_)
721 {
722 if (!_lastPersisted.empty())
723 {
724 amps_uint64_t publisher, publisher_lastPersisted;
725 amps_uint64_t sequence, sequence_lastPersisted;
726 parseBookmark(bookmark_, publisher, sequence);
727 parseBookmark(_lastPersisted, publisher_lastPersisted,
728 sequence_lastPersisted);
729 if (publisher == publisher_lastPersisted &&
730 sequence <= sequence_lastPersisted)
731 {
732 return false;
733 }
734 }
735 // deepCopy will clear what's in _lastPersisted
736 _lastPersisted.deepCopy(bookmark_);
737 _store->_recentChanged = true;
738 _recoveryTimestamp.clear();
739 return true;
740 }
741
742 Message::Field lastPersisted(size_t bookmark_)
743 {
744 Lock<Mutex> guard(_subLock);
745 Message::Field& bookmark = _entries[bookmark_]._val;
746 // These shouldn't be persisted
747 if (bookmark == AMPS_BOOKMARK_NOW
748 || BookmarkRange::isRange(bookmark))
749 {
750 return bookmark;
751 }
752 _setLastPersisted(bookmark);
753 return bookmark;
754 }
755
756 // Returns the index of the recovered item, either the index where it
757 // was first stored prior to getMostRecent, or the new index if it is
758 // relogged either because this is called from log() or because it was
759 // not active but also not persisted.
760 size_t recover(const Message::Field& bookmark_, bool relogIfNotDiscarded)
761 {
762 size_t retVal = AMPS_UNSET_INDEX;
763 if (_recovered.empty() || _recoveryBase == AMPS_UNSET_INDEX)
764 {
765 return retVal;
766 }
767 // Check if this is a recovered bookmark.
768 // If so, copy the existing one to the new location
769 RecoveryIterator item = _recovered.find(bookmark_);
770 if (item != _recovered.end())
771 {
772 size_t seqNo = item->second;
773 size_t index = (seqNo - _recoveryBase) % _entriesLength;
774 // If we only have recovery entries and isDiscarded is
775 // checking on an already discarded entry, update recent.
776 if (_least + _leastBase == _current + _currentBase &&
777 !_entries[index]._active)
778 {
779 _store->_recentChanged = true;
780 _recent.clear();
781 _recent = _entries[index]._val.deepCopy();
782 retVal = moveEntry(index);
783 if (retVal == AMPS_UNSET_INDEX)
784 {
785 recover(bookmark_, relogIfNotDiscarded);
786 }
787 _least = _current;
788 _leastBase = _currentBase;
789 }
790 else if (!_entries[index]._active || relogIfNotDiscarded)
791 {
792 retVal = moveEntry(index);
793 if (retVal == AMPS_UNSET_INDEX)
794 {
795 recover(bookmark_, relogIfNotDiscarded);
796 }
797 }
798 else
799 {
800 return index;
801 }
802 _recovered.erase(item);
803 if (_recovered.empty())
804 {
805 _recoveryMin = AMPS_UNSET_INDEX;
806 _recoveryBase = AMPS_UNSET_INDEX;
807 _recoveryMax = AMPS_UNSET_INDEX;
808 _recoveryMaxBase = AMPS_UNSET_INDEX;
809 }
810 else if (index == _recoveryMin)
811 {
812 while (_entries[_recoveryMin]._val.empty() &&
813 (_recoveryMin + _recoveryBase) < (_recoveryMax + _recoveryMaxBase))
814 {
815 if (++_recoveryMin == _entriesLength)
816 {
817 _recoveryMin = 0;
818 _recoveryBase += _entriesLength;
819 }
820 }
821 }
822 }
823 return retVal;
824 }
825
826 // Return the id of this Subscription
827 Message::Field id() const
828 {
829 return _id;
830 }
831
832 struct Entry
833 {
834 Message::Field _val; //16
835 bool _active; //17
836 char _padding[32 - sizeof(Message::Field) - sizeof(bool)]; //32
837
838 Entry() : _active(false) // -V730
839 {
840 ;
841 }
842 };
843
844 typedef std::vector<Entry*> EntryPtrList;
845
846 void getRecoveryEntries(EntryPtrList& list_)
847 {
848 if (_recoveryMin == AMPS_UNSET_INDEX ||
849 _recoveryMax == AMPS_UNSET_INDEX)
850 {
851 return;
852 }
853 size_t base = _recoveryBase;
854 size_t max = _recoveryMax + _recoveryMaxBase;
855 for (size_t i = _recoveryMin; i + base < max; ++i)
856 {
857 if (i == _entriesLength)
858 {
859 i = 0;
860 base = _recoveryMaxBase;
861 }
862 //list_.insert(&(_entries[i]));
863 list_.push_back(&(_entries[i]));
864 }
865 return;
866 }
867
868 void getActiveEntries(EntryPtrList& list_)
869 {
870 size_t base = _leastBase;
871 for (size_t i = _least; i + base < _current + _currentBase; ++i)
872 {
873 if (i >= _entriesLength)
874 {
875 i = 0;
876 base = _currentBase;
877 }
878 //list_.insert(&(_entries[i]));
879 list_.push_back(&(_entries[i]));
880 }
881 return;
882 }
883
884 Entry* getEntryByIndex(size_t index_)
885 {
886 Lock<Mutex> guard(_subLock);
887 size_t base = (_recoveryBase == AMPS_UNSET_INDEX ||
888 index_ >= _least + _leastBase)
889 ? _leastBase : _recoveryBase;
890 // Return NULL if not a valid index
891 size_t min = (_recoveryMin == AMPS_UNSET_INDEX ?
892 _least + _leastBase :
893 _recoveryMin + _recoveryBase);
894 if (index_ >= _current + _currentBase || index_ < min)
895 {
896 return NULL;
897 }
898 return &(_entries[(index_ - base) % _entriesLength]);
899 }
900
901 void justRecovered()
902 {
903 Lock<Mutex> guard(_subLock);
904 _updateMostRecent();
905 EntryPtrList list;
906 getRecoveryEntries(list);
907 setPublishersToDiscarded(&list, &_publishers);
908 }
909
910 void setPublishersToDiscarded(EntryPtrList* recovered_,
911 PublisherMap* publishers_)
912 {
913 // Need to reset publishers to only have up to the last
914 // discarded sequence number. Messages that were in transit
915 // during previous run but not discarded should be considered
916 // new and not duplicate after a restart/recovery.
917 for (EntryPtrList::iterator i = recovered_->begin();
918 i != recovered_->end(); ++i)
919 {
920 if ((*i)->_val.empty())
921 {
922 continue;
923 }
924 amps_uint64_t publisher = (amps_uint64_t)0;
925 amps_uint64_t sequence = (amps_uint64_t)0;
926 parseBookmark((*i)->_val, publisher, sequence);
927 if (publisher && sequence && (*i)->_active &&
928 (*publishers_)[publisher] >= sequence)
929 {
930 (*publishers_)[publisher] = sequence - 1;
931 }
932 }
933 }
934
935 void clearLastPersisted()
936 {
937 Lock<Mutex> guard(_subLock);
938 _lastPersisted.clear();
939 }
940
941 void setLastPersistedToEpoch()
942 {
943 Lock<Mutex> guard(_subLock);
944 _setLastPersistedToEpoch();
945 }
946
947 private:
948 Subscription(const Subscription&);
949 Subscription& operator=(const Subscription&);
950
951 size_t moveEntry(size_t index_)
952 {
953 // Check for wrap
954 if (_current >= _entriesLength)
955 {
956 _current = 0;
957 _currentBase += _entriesLength;
958 }
959 // Check for resize
960 // If list is too small, double it
961 if ((_current == _least % _entriesLength &&
962 _leastBase < _currentBase) ||
963 (_current == _recoveryMin && _recoveryBase < _currentBase))
964 {
965 if (!_store->resize(_id, (char**)&_entries,
966 sizeof(Entry) * _entriesLength * 2))
967 {
968 return AMPS_UNSET_INDEX;
969 }
970 // Length was doubled
971 _entriesLength *= 2;
972 }
973 _entries[_current]._val = _entries[index_]._val;
974 _entries[_current]._active = _entries[index_]._active;
975 // No need to clear Field, just set it to empty
976 _entries[index_]._val.assign(NULL, 0);
977 _entries[index_]._active = false;
978 return _current++;
979 }
980
981 void _setLastPersistedToEpoch()
982 {
983 size_t fieldLen = strlen(AMPS_BOOKMARK_EPOCH);
984 char* field = (char*)malloc(fieldLen);
985 if (!field)
986 {
987 throw AMPSException("Failed to allocate memory for last persisted bookmark.", AMPS_E_MEMORY);
988 }
989 memcpy(field, AMPS_BOOKMARK_EPOCH, fieldLen);
990 _lastPersisted.clear();
991 _lastPersisted.assign(field, fieldLen);
992 }
993
994 bool _discard(size_t index_)
995 {
996 bool retVal = false;
997 // Lock should already be held
998 assert((_recoveryBase == AMPS_UNSET_INDEX && _recoveryMin == AMPS_UNSET_INDEX) ||
999 (_recoveryBase != AMPS_UNSET_INDEX && _recoveryMin != AMPS_UNSET_INDEX));
1000 size_t base = (_recoveryBase == AMPS_UNSET_INDEX
1001 || index_ >= _least + _leastBase)
1002 ? _leastBase : _recoveryBase;
1003 // discard of a record not in the log is a no-op
1004 size_t min = (_recoveryMin == AMPS_UNSET_INDEX ? _least + _leastBase :
1005 _recoveryMin + _recoveryBase);
1006 if (index_ >= _current + _currentBase || index_ < min)
1007 {
1008 return retVal;
1009 }
1010
1011 // log that this one is discarded, then
1012 // recalculate what the most recent entry is.
1013 Entry& e = _entries[(index_ - base) % _entriesLength];
1014 e._active = false;
1015
1016 size_t index = index_;
1017 if (_recoveryMin != AMPS_UNSET_INDEX &&
1018 index_ == _recoveryMin + _recoveryBase)
1019 {
1020 // Find all to discard
1021 size_t j = _recoveryMin;
1022 while (j + _recoveryBase < _recoveryMax + _recoveryMaxBase &&
1023 !_entries[j]._active)
1024 {
1025 // This index might be left-over from a slow discard and we
1026 // may have reconnected. We have a few possibilities at this point.
1027 // 1. If we re-logged this bookmark, this index will point at an
1028 // empty bookmark. This could happen if the discard thread was slow
1029 // and the reconnect was fast. We wouldn't report the
1030 // the re-arrival of the bookmark as a duplicate because it
1031 // hadn't been marked as discarded. In this case, we have to
1032 // simply move past this in the recovery area.
1033 // 2. This bookmark should become _recent because we haven't
1034 // yet received anything since our last call to getMostRecent.
1035 // In this case, we need to take it out of recovered but not
1036 // clear it. The publishers map should report it as duplicate.
1037 // 3. This is the 'oldest' recovered, but we have received new
1038 // bookmarks since we got this one. We can clear it because the
1039 // publishers map should report it as a duplicate if/when it
1040 // does arrive again. Move the _recoveryMin ahead and remove it
1041 // from recovered.
1042 Message::Field& bookmark = _entries[j]._val;
1043 // Option 1 skips this and just moves on
1044 if (!bookmark.empty())
1045 {
1046 _recovered.erase(bookmark);
1047 // Make sure our publishers map will mark it discarded
1048 amps_uint64_t publisher, sequence;
1049 parseBookmark(bookmark, publisher, sequence);
1050 PublisherIterator pub = _publishers.find(publisher);
1051 if (pub == _publishers.end() || pub->second < sequence)
1052 {
1053 _publishers[publisher] = sequence;
1054 }
1055 if (_least + _leastBase == _current + _currentBase ||
1056 ((_least + _leastBase) % _entriesLength) ==
1057 ((_recoveryMin + _recoveryBase + 1)) % _entriesLength)
1058 {
1059 // Option 2, reset recent
1060 retVal = true;
1061 _store->_recentChanged = true;
1062 _recoveryTimestamp.clear();
1063 _recent.clear();
1064 _recent = bookmark;
1065 bookmark.assign(NULL, 0);
1066 }
1067 else
1068 {
1069 // Option 3, simply clear this one
1070 bookmark.clear();
1071 }
1072 }
1073 // If we reach the buffer end,
1074 // keep checking from the beginning
1075 if (++j == _entriesLength)
1076 {
1077 // Least has now loooped around
1078 _recoveryBase += _entriesLength;
1079 j = 0;
1080 }
1081 }
1082 assert(j + _recoveryBase != _recoveryMax + _recoveryMaxBase ||
1083 _recovered.empty());
1084 if (_recovered.empty())
1085 {
1086 _recoveryMin = AMPS_UNSET_INDEX;
1087 _recoveryBase = AMPS_UNSET_INDEX;
1088 _recoveryMax = AMPS_UNSET_INDEX;
1089 _recoveryMaxBase = AMPS_UNSET_INDEX;
1090 // Cleared recovered, want to check onward
1091 index = _least + _leastBase;
1092 }
1093 else
1094 {
1095 _recoveryMin = j;
1096 }
1097 }
1098 // if this is the first item in the list, discard all inactive ones
1099 // as long as recovery also says its okay
1100 if (index == _least + _leastBase)
1101 {
1102 // Find all to discard
1103 size_t j = _least;
1104 while (j + _leastBase < _current + _currentBase &&
1105 !_entries[j]._active)
1106 {
1107 //Must free associated memory
1108 _recent.clear();
1109 _recent = _entries[j]._val;
1110 _entries[j]._val.assign(NULL, 0);
1111 _store->_recentChanged = true;
1112 retVal = true;
1113 _recoveryTimestamp.clear();
1114 // If we reach the buffer end,
1115 // keep checking from the beginning
1116 if (++j == _entriesLength)
1117 {
1118 // Least has now loooped around
1119 _leastBase += _entriesLength;
1120 j = 0;
1121 }
1122 }
1123 _least = j;
1124 }
1125 return retVal;
1126 }
1127
1128 void _updateMostRecent()
1129 {
1130 // Lock is already held
1131 _recovered.clear();
1132 assert((_recoveryBase == AMPS_UNSET_INDEX && _recoveryMin == AMPS_UNSET_INDEX) ||
1133 (_recoveryBase != AMPS_UNSET_INDEX && _recoveryMin != AMPS_UNSET_INDEX));
1134 size_t base = (_recoveryMin == AMPS_UNSET_INDEX) ? _leastBase : _recoveryBase;
1135 size_t start = (_recoveryMin == AMPS_UNSET_INDEX) ? _least : _recoveryMin;
1136 _recoveryMin = AMPS_UNSET_INDEX;
1137 _recoveryBase = AMPS_UNSET_INDEX;
1138 _recoveryMax = AMPS_UNSET_INDEX;
1139 _recoveryMaxBase = AMPS_UNSET_INDEX;
1140 for (size_t i = start; i + base < _current + _currentBase; i++)
1141 {
1142 if ( i >= _entriesLength )
1143 {
1144 i = 0;
1145 base = _currentBase;
1146 }
1147 if (i >= _recoveryMax + _recoveryBase && i < _least + _leastBase)
1148 {
1149 continue;
1150 }
1151 Entry& entry = _entries[i];
1152 if (!entry._val.empty())
1153 {
1154 _recovered[entry._val] = i + base;
1155 if (_recoveryMin == AMPS_UNSET_INDEX)
1156 {
1157 _recoveryMin = i;
1158 _recoveryBase = base;
1159 _recoveryMax = _current;
1160 _recoveryMaxBase = _currentBase;
1161 }
1162 }
1163 }
1164 if (_current == _entriesLength)
1165 {
1166 _current = 0;
1167 _currentBase += _entriesLength;
1168 }
1169 _least = _current;
1170 _leastBase = _currentBase;
1171 }
1172
1173 Message::Field _id;
1174 Message::Field _recent;
1175 Message::Field _lastPersisted;
1176 Message::Field _recentList;
1177 BookmarkRange _range;
1178 Message::Field _recoveryTimestamp;
1179 size_t _current;
1180 size_t _currentBase;
1181 size_t _least;
1182 size_t _leastBase;
1183 size_t _recoveryMin;
1184 size_t _recoveryBase;
1185 size_t _recoveryMax;
1186 size_t _recoveryMaxBase;
1187 size_t _entriesLength;
1188 Entry* _entries;
1189 MemoryBookmarkStore* _store;
1190 Mutex _subLock;
1191 RecoveryMap _recovered;
1192 public:
1193 PublisherMap _publishers;
1194 };
1195
1196 public:
1199 MemoryBookmarkStore() : BookmarkStoreImpl(),
1200 _subsLock(),
1201 _lock(),
1202 _serverVersion(AMPS_DEFAULT_MIN_VERSION),
1203 _recentChanged(true),
1204 _recovering(false),
1205 _recoveryPointAdapter(NULL),
1206 _recoveryPointFactory(NULL),
1207 _adapterSequence((amps_uint64_t)0),
1208 _nextAdapterUpdate((amps_uint64_t)0)
1209 { ; }
1210
1211 typedef RecoveryPointAdapter::iterator RecoveryIterator;
1212
1219 MemoryBookmarkStore(const RecoveryPointAdapter& adapter_,
1220 RecoveryPointFactory factory_ = NULL)
1221 : BookmarkStoreImpl()
1222 , _subsLock()
1223 , _lock()
1224 , _serverVersion(AMPS_DEFAULT_MIN_VERSION)
1225 , _recentChanged(true)
1226 , _recovering(true)
1227 , _recoveryPointAdapter(adapter_)
1228 , _recoveryPointFactory(factory_)
1229 , _adapterSequence((amps_uint64_t)0)
1230 , _nextAdapterUpdate((amps_uint64_t)0)
1231 {
1232 Message msg;
1233 if (!_recoveryPointFactory)
1234 {
1235 _recoveryPointFactory = &FixedRecoveryPoint::create;
1236 }
1237 for (RecoveryIterator recoveryPoint = _recoveryPointAdapter.begin();
1238 recoveryPoint != _recoveryPointAdapter.end();
1239 ++recoveryPoint)
1240 {
1241 Field subId(recoveryPoint->getSubId());
1242 msg.setSubscriptionHandle(static_cast<amps_subscription_handle>(0));
1243 msg.setSubId(subId);
1244 Field bookmark = recoveryPoint->getBookmark();
1245 if (BookmarkRange::isRange(bookmark))
1246 {
1247 msg.setBookmark(bookmark);
1248 _log(msg);
1249 }
1250 else
1251 {
1252 std::vector<Field> bmList = Field::parseBookmarkList(bookmark);
1253 for (std::vector<Field>::iterator bkmk = bmList.begin(); bkmk != bmList.end(); ++bkmk)
1254 {
1255 if (Field::isTimestamp(*bkmk))
1256 {
1257 find(subId)->setRecoveryTimestamp(bkmk->data(), bkmk->len());
1258 }
1259 else
1260 {
1261 msg.assignBookmark(bkmk->data(), bkmk->len());
1262 _isDiscarded(msg);
1263 _log(msg);
1264 _discard(msg);
1265 }
1266 }
1267 // Reset to original bookmark
1268 msg.setBookmark(bookmark);
1269 }
1270 }
1271 _recovering = false;
1272 }
1273
1274 virtual ~MemoryBookmarkStore()
1275 {
1276 if (_recoveryPointAdapter.isValid())
1277 {
1278 _recoveryPointAdapter.close();
1279 }
1280 __purge();
1281 }
1282
1288 virtual size_t log(Message& message_)
1289 {
1290 Lock<Mutex> guard(_lock);
1291 return _log(message_);
1292 }
1293
1299 virtual void discard(const Message& message_)
1300 {
1301 Lock<Mutex> guard(_lock);
1302 (void)_discard(message_);
1303 }
1304
1312 virtual void discard(const Message::Field& subId_, size_t bookmarkSeqNo_)
1313 {
1314 Lock<Mutex> guard(_lock);
1315 (void)_discard(subId_, bookmarkSeqNo_);
1316 }
1317
1323 virtual Message::Field getMostRecent(const Message::Field& subId_)
1324 {
1325 Lock<Mutex> guard(_lock);
1326 return _getMostRecent(subId_);
1327 }
1328
1337 virtual bool isDiscarded(Message& message_)
1338 {
1339 Lock<Mutex> guard(_lock);
1340 return _isDiscarded(message_);
1341 }
1342
1348 virtual void purge()
1349 {
1350 Lock<Mutex> guard(_lock);
1351 _purge();
1352 }
1353
1359 virtual void purge(const Message::Field& subId_)
1360 {
1361 Lock<Mutex> guard(_lock);
1362 _purge(subId_);
1363 }
1364
1369 virtual size_t getOldestBookmarkSeq(const Message::Field& subId_)
1370 {
1371 Lock<Mutex> guard(_lock);
1372 return _getOldestBookmarkSeq(subId_);
1373 }
1374
1380 virtual void persisted(const Message::Field& subId_,
1381 const Message::Field& bookmark_)
1382 {
1383 Lock<Mutex> guard(_lock);
1384 _persisted(find(subId_), bookmark_);
1385 }
1386
1393 virtual Message::Field persisted(const Message::Field& subId_,
1394 size_t bookmark_)
1395 {
1396 Lock<Mutex> guard(_lock);
1397 return _persisted(find(subId_), bookmark_);
1398 }
1399
1404 void setServerVersion(const VersionInfo& version_)
1405 {
1406 setServerVersion(version_.getOldStyleVersion());
1407 }
1408
1413 void setServerVersion(size_t version_)
1414 {
1415 Lock<Mutex> guard(_subsLock);
1416 _serverVersion = version_;
1417 }
1418
1419 inline bool isWritableBookmark(size_t length)
1420 {
1421 return length >= AMPS_MIN_BOOKMARK_LEN;
1422 }
1423
1424 typedef Subscription::EntryPtrList EntryPtrList;
1425
1426 protected:
1427
1428 // Called once lock is acquired
1429 size_t _log(Message& message_)
1430 {
1431 Message::Field bookmark = message_.getBookmark();
1432 Subscription* pSub = (Subscription*)(message_.getSubscriptionHandle());
1433 if (!pSub)
1434 {
1435 Message::Field subId = message_.getSubscriptionId();
1436 if (subId.empty())
1437 {
1438 subId = message_.getSubscriptionIds();
1439 }
1440 pSub = find(subId);
1441 message_.setSubscriptionHandle(
1442 static_cast<amps_subscription_handle>(pSub));
1443 }
1444 size_t retVal = pSub->log(bookmark);
1445 message_.setBookmarkSeqNo(retVal);
1446 return retVal;
1447 }
1448
1449 // Called once lock is acquired, or from ctor
1450 bool _discard(const Message& message_)
1451 {
1452 size_t bookmarkSeqNo = message_.getBookmarkSeqNo();
1453 Subscription* pSub = (Subscription*)(message_.getSubscriptionHandle());
1454 if (!pSub)
1455 {
1456 Message::Field subId = message_.getSubscriptionId();
1457 if (subId.empty())
1458 {
1459 subId = message_.getSubscriptionIds();
1460 }
1461 pSub = find(subId);
1462 }
1463 bool retVal = pSub->discard(bookmarkSeqNo);
1464 if (retVal)
1465 {
1466 updateAdapter(pSub);
1467 }
1468 return retVal;
1469 }
1470
1471 // Called once lock is acquired
1472 bool _discard(const Message::Field& subId_, size_t bookmarkSeqNo_)
1473 {
1474 Subscription* pSub = find(subId_);
1475 bool retVal = pSub->discard(bookmarkSeqNo_);
1476 if (retVal)
1477 {
1478 updateAdapter(pSub);
1479 }
1480 return retVal;
1481 }
1482
1483 // Called once lock is acquired
1484 Message::Field _getMostRecent(const Message::Field& subId_,
1485 bool usePublishersList_ = true)
1486 {
1487 Subscription* pSub = find(subId_);
1488 return pSub->getMostRecentList(usePublishersList_);
1489 }
1490
1491 // Called once lock is acquired
1492 bool _isDiscarded(Message& message_)
1493 {
1494 Message::Field subId = message_.getSubscriptionId();
1495 if (subId.empty())
1496 {
1497 subId = message_.getSubscriptionIds();
1498 }
1499 Subscription* pSub = find(subId);
1500 message_.setSubscriptionHandle(
1501 static_cast<amps_subscription_handle>(pSub));
1502 return pSub->isDiscarded(message_.getBookmark());
1503 }
1504
1505 // Called once lock is acquired
1506 size_t _getOldestBookmarkSeq(const Message::Field& subId_)
1507 {
1508 Subscription* pSub = find(subId_);
1509 return pSub->getOldestBookmarkSeq();
1510 }
1511
1512 // Called once lock is acquired
1513 virtual void _persisted(Subscription* pSub_,
1514 const Message::Field& bookmark_)
1515 {
1516 if (pSub_->lastPersisted(bookmark_))
1517 {
1518 updateAdapter(pSub_);
1519 }
1520 }
1521
1522 // Called once lock is acquired
1523 virtual Message::Field _persisted(Subscription* pSub_, size_t bookmark_)
1524 {
1525 return pSub_->lastPersisted(bookmark_);
1526 }
1527
1528 // Called once lock is acquired
1529 void _purge()
1530 {
1531 if (_recoveryPointAdapter.isValid())
1532 {
1533 _recoveryPointAdapter.purge();
1534 }
1535 __purge();
1536 }
1537
1538 // Called once lock is acquired
1539 void __purge()
1540 {
1541 // Walk through list and clear Fields before calling clear
1542 while (!_subs.empty())
1543 {
1544 SubscriptionMap::iterator iter = _subs.begin();
1545 //The subId key is cleared when deleting the Subscription, which shares
1546 //the _data pointer in its id field.
1547 const_cast<Message::Field&>(iter->first).clear();
1548 delete (iter->second);
1549 _subs.erase(iter);
1550 }
1551 _subs.clear();
1552 }
1553
1554 // Called once lock is acquired
1555 virtual void _purge(const Message::Field& subId_)
1556 {
1557 if (_recoveryPointAdapter.isValid())
1558 {
1559 _recoveryPointAdapter.purge(subId_);
1560 }
1561 __purge(subId_);
1562 }
1563
1564 // Called once lock is acquired
1565 virtual void __purge(const Message::Field& subId_)
1566 {
1567 Lock<Mutex> guard(_subsLock);
1568 SubscriptionMap::iterator iter = _subs.find(subId_);
1569 if (iter == _subs.end())
1570 {
1571 return;
1572 }
1573 const_cast<Message::Field&>(iter->first).clear();
1574 delete (iter->second);
1575 _subs.erase(iter);
1576 }
1577
1578 // Can be used by subclasses during recovery
1579 void setMostRecent(const Message::Field& subId_,
1580 const Message::Field& recent_)
1581 {
1582 find(subId_)->setMostRecent(recent_);
1583 }
1584
1585 Mutex _subsLock;
1586 Mutex _lock;
1587 static const char ENTRY_BOOKMARK = 'b';
1588 static const char ENTRY_DISCARD = 'd';
1589 static const char ENTRY_PERSISTED = 'p';
1590
1591 virtual Subscription* find(const Message::Field& subId_)
1592 {
1593 if (subId_.empty())
1594 {
1595 throw StoreException("A valid subscription ID must be provided to the Bookmark Store");
1596 }
1597 Lock<Mutex> guard(_subsLock);
1598 if (_subs.count(subId_) == 0)
1599 {
1600 // Subscription will be created
1601 Message::Field id;
1602 id.deepCopy(subId_);
1603 _subs[id] = new Subscription(this, id);
1604 return _subs[id];
1605 }
1606 return _subs[subId_];
1607 }
1608
1609 virtual bool resize(const Message::Field& subId_, char** newBuffer_, size_t size_,
1610 bool callResizeHandler_ = true)
1611 {
1612 if(!newBuffer_)
1613 {
1614 throw AMPSException("Resize called with null pointer", AMPS_E_MEMORY);
1615 }
1616 if (size_ == 0) // Delete the buffer
1617 {
1618 if (*newBuffer_)
1619 {
1620 free(*newBuffer_);
1621 *newBuffer_ = NULL;
1622 }
1623 return true;
1624 }
1625 if (callResizeHandler_ && !callResizeHandler(subId_, size_))
1626 {
1627 return false;
1628 }
1629 char* allocBuffer = (char*)malloc(size_);
1630 if (!allocBuffer)
1631 {
1632 throw AMPSException("Failed to allocate memory for BookmarkStore resize.", AMPS_E_MEMORY);
1633 }
1634 memset(allocBuffer, 0, size_);
1635 if (*newBuffer_)
1636 {
1637 find(subId_)->moveEntries(*newBuffer_, allocBuffer, size_);
1638 free(*newBuffer_);
1639 }
1640 *newBuffer_ = allocBuffer;
1641 return true;
1642 }
1643
1644 protected:
1645 void updateAdapter(Subscription* pSub_)
1646 {
1647 if (_recovering || !_recentChanged || !_recoveryPointAdapter.isValid())
1648 {
1649 return;
1650 }
1651 Field bookmark = pSub_->getMostRecentList(false);
1652 RecoveryPoint update = _recoveryPointFactory(pSub_->id(), bookmark);
1653 // Use atomic seq to keep updates in order
1654 amps_uint64_t seq = _adapterSequence.fetch_add(1);
1655 Unlock<Mutex> unlock(_lock);
1656 while (_nextAdapterUpdate.load() < seq)
1657 {
1658 AMPS_USLEEP(100);
1659 }
1660 try
1661 {
1662 _recoveryPointAdapter.update(update);
1663 }
1664 catch (const std::exception&)
1665 {
1666 _nextAdapterUpdate.fetch_add(1);
1667 // Free memory, the bookmark needs to be cleared
1668 bookmark.clear();
1669 throw;
1670 }
1671 _nextAdapterUpdate.fetch_add(1);
1672 // Free memory, the bookmark needs to be cleared
1673 bookmark.clear();
1674 }
1675
1676 typedef std::map<Message::Field, Subscription*, Message::Field::FieldHash> SubscriptionMap;
1677 SubscriptionMap _subs;
1678 size_t _serverVersion;
1679 bool _recentChanged;
1680 bool _recovering;
1681 typedef std::set<Subscription*> SubscriptionSet;
1682 RecoveryPointAdapter _recoveryPointAdapter;
1683 RecoveryPointFactory _recoveryPointFactory;
1684 std::atomic<amps_uint64_t> _adapterSequence;
1685 std::atomic<amps_uint64_t> _nextAdapterUpdate;
1686 };
1687
1688 } // end namespace AMPS
1689
1690 #endif //_MEMORYBOOKMARKSTORE_H_
1691
Field.hpp
Defines the AMPS::Field class, which represents the value of a field in a message.
Message.hpp
Defines the AMPS::Message class and related classes.
RecoveryPointAdapter.hpp
Provides AMPS::RecoveryPointAdapter, an iterface for implementing external storage of bookmark subscr...
RecoveryPoint.hpp
Provides AMPS::RecoveryPoint, AMPS::RecoveryPointFactory, AMPS::FixedRecoveryPoint,...
AMPS::RecoveryPointFactory
RecoveryPoint(* RecoveryPointFactory)(const Field &subId_, const Field &bookmark_)
RecoveryPointFactory is a function type for producing a RecoveryPoint that is sent to a RecoveryPoint...
Definition: RecoveryPoint.hpp:126
AMPS_E_MEMORY
@ AMPS_E_MEMORY
A memory error occurred.
Definition: amps.h:225
AMPS::AMPSException
Base class for all exceptions in AMPS.
Definition: AMPSException.hpp:41
AMPS::BookmarkStoreImpl
Abstract base class for storing received bookmarks for HA clients.
Definition: BookmarkStore.hpp:78
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::stringCopy
static Field stringCopy(const char *str_)
Makes a copy of str_ in a new Field.
Definition: Field.hpp:253
AMPS::FixedRecoveryPoint::create
static RecoveryPoint create(const Field &subId_, const Field &bookmark_)
Use this function in BookmarkStore::setRecoveryPointFactory( std::bind(&FixedRecoveryPoint::create,...
Definition: RecoveryPoint.hpp:139
AMPS::MemoryBookmarkStore
A BookmarkStoreImpl implementation that stores bookmarks in memory.
Definition: MemoryBookmarkStore.hpp:59
AMPS::MemoryBookmarkStore::discard
virtual void discard(const Message &message_)
Log a discard-bookmark entry to the persistent log based on a bookmark sequence number.
Definition: MemoryBookmarkStore.hpp:1299
AMPS::MemoryBookmarkStore::purge
virtual void purge(const Message::Field &subId_)
Called to purge the contents of this store for particular subId.
Definition: MemoryBookmarkStore.hpp:1359
AMPS::MemoryBookmarkStore::purge
virtual void purge()
Called to purge the contents of this store.
Definition: MemoryBookmarkStore.hpp:1348
AMPS::MemoryBookmarkStore::MemoryBookmarkStore
MemoryBookmarkStore()
Creates a MemoryBookmarkStore.
Definition: MemoryBookmarkStore.hpp:1199
AMPS::MemoryBookmarkStore::persisted
virtual void persisted(const Message::Field &subId_, const Message::Field &bookmark_)
Mark the bookmark provided as replicated to all sync replication destinations for the given subscript...
Definition: MemoryBookmarkStore.hpp:1380
AMPS::MemoryBookmarkStore::discard
virtual void discard(const Message::Field &subId_, size_t bookmarkSeqNo_)
Log a discard-bookmark entry to the persistent log based on a bookmark sequence number.
Definition: MemoryBookmarkStore.hpp:1312
AMPS::MemoryBookmarkStore::MemoryBookmarkStore
MemoryBookmarkStore(const RecoveryPointAdapter &adapter_, RecoveryPointFactory factory_=NULL)
Creates a MemoryBookmarkStore.
Definition: MemoryBookmarkStore.hpp:1219
AMPS::MemoryBookmarkStore::setServerVersion
void setServerVersion(const VersionInfo &version_)
Internally used to set the server version so the store knows how to deal with persisted acks and call...
Definition: MemoryBookmarkStore.hpp:1404
AMPS::MemoryBookmarkStore::persisted
virtual Message::Field persisted(const Message::Field &subId_, size_t bookmark_)
Mark the bookmark provided as replicated to all sync replication destinations for the given subscript...
Definition: MemoryBookmarkStore.hpp:1393
AMPS::MemoryBookmarkStore::setServerVersion
void setServerVersion(size_t version_)
Internally used to set the server version so the store knows how to deal with persisted acks and call...
Definition: MemoryBookmarkStore.hpp:1413
AMPS::MemoryBookmarkStore::log
virtual size_t log(Message &message_)
Log a bookmark to the persistent log and return the corresponding sequence number for this bookmark.
Definition: MemoryBookmarkStore.hpp:1288
AMPS::MemoryBookmarkStore::getOldestBookmarkSeq
virtual size_t getOldestBookmarkSeq(const Message::Field &subId_)
Called to find the oldest bookmark in the store.
Definition: MemoryBookmarkStore.hpp:1369
AMPS::MemoryBookmarkStore::getMostRecent
virtual Message::Field getMostRecent(const Message::Field &subId_)
Returns the most recent bookmark from the log that ought to be used for (re-)subscriptions.
Definition: MemoryBookmarkStore.hpp:1323
AMPS::MemoryBookmarkStore::isDiscarded
virtual bool isDiscarded(Message &message_)
Called for each arriving message to determine if the application has already seen this bookmark and s...
Definition: MemoryBookmarkStore.hpp:1337
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::setSubId
Message & setSubId(const std::string &v)
Sets the value of the SubscriptionId header for this Message.
Definition: Message.hpp:1467
AMPS::Message::setBookmark
Message & setBookmark(const std::string &v)
Sets the value of the Bookmark header for this Message.
Definition: Message.hpp:1234
AMPS::Message::assignBookmark
Message & assignBookmark(const std::string &v)
Assigns the value of the Bookmark header for this Message without copying.
Definition: Message.hpp:1234
AMPS::RecoveryPointAdapter
RecoveryPointAdapter a handle class for implementing external storage of subscription recovery points...
Definition: RecoveryPointAdapter.hpp:78
AMPS::RecoveryPointAdapter::begin
iterator begin()
To recover from an adapter, iterate over the adapter from begin() to end() with a RecoveryPointIterat...
Definition: RecoveryPointAdapter.hpp:137
AMPS::RecoveryPointAdapter::update
void update(RecoveryPoint &recoveryPoint_)
Update the storage information with the given recovery point.
Definition: RecoveryPointAdapter.hpp:151
AMPS::RecoveryPointAdapter::isValid
bool isValid() const
Return if this has a valid implementation.
Definition: RecoveryPointAdapter.hpp:182
AMPS::RecoveryPointAdapter::purge
void purge()
Remove all data from the storage.
Definition: RecoveryPointAdapter.hpp:157
AMPS::RecoveryPointAdapter::close
void close()
Take any necessary actions to close the associated storage.
Definition: RecoveryPointAdapter.hpp:170
AMPS::RecoveryPointAdapter::end
iterator end()
Return the end of recovery marker.
Definition: RecoveryPointAdapter.hpp:144
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