26 #ifndef _RECOVERYPOINTADAPTER_H_ 27 #define _RECOVERYPOINTADAPTER_H_ 35 #include <unordered_map> 45 class RecoveryPointAdapter;
62 virtual void purge() = 0;
67 virtual void close() = 0;
83 if (!_pAdapter || !_pAdapter->
next(_current))
94 : _pAdapter(pAdapter_)
99 bool operator==(
const iterator& rhs)
101 return _pAdapter == rhs._pAdapter;
103 bool operator!=(
const iterator& rhs)
105 return _pAdapter != rhs._pAdapter;
107 void operator++(
void)
123 : _body(body_, isRef_) { }
136 return iterator(&(_body.get()));
150 _body.get().update(recoveryPoint_);
163 _body.get().purge(subId_);
181 return _body.isValid();
184 BorrowRefHandle<RecoveryPointAdapterImpl> _body;
208 const std::shared_ptr<RecoveryPointAdapterImpl>& delegate_,
209 unsigned updateThreshold_ = 10,
210 double timeoutMillis_ = 2000.0,
211 long updateIntervalMillis_ = 2000
214 , _delegate(delegate_)
215 , _updateThreshold(updateThreshold_)
216 , _timeoutMillis(timeoutMillis_)
217 , _updateIntervalMillis(updateIntervalMillis_)
221 _thread = std::thread(&ConflatingRecoveryPointAdapter::updateThread,
229 for (UpdateIter purged = _latestUpdates.begin();
230 purged != _latestUpdates.end(); ++purged)
232 Field clearableSubId = purged->first;
233 purged->second.
clear();
234 clearableSubId.
clear();
244 return _delegate->next(current_);
252 Lock<Mutex> lock(_lock);
253 UpdateIter lastUpdate = _latestUpdates.find(subId);
254 if (lastUpdate == _latestUpdates.end())
258 _latestUpdates[subId] = recoveryPoint_.
deepCopy();
260 if (_timeoutMillis != 0.0)
262 Timer timer(_timeoutMillis);
264 _timers[subId] = timer;
270 lastUpdate->second.deepCopy(recoveryPoint_);
272 if (++_counts[subId] >= _updateThreshold)
284 Lock<Mutex> lock(_lock);
287 for (UpdateIter purged = _latestUpdates.begin();
288 purged != _latestUpdates.end(); ++purged)
290 Field clearableSubId = purged->first;
291 purged->second.
clear();
292 clearableSubId.
clear();
294 _latestUpdates.clear();
301 _delegate->purge(subId_);
302 Lock<Mutex> lock(_lock);
303 UpdateIter purged = _latestUpdates.find(subId_);
304 if (purged != _latestUpdates.end())
306 Field clearableSubId = purged->first;
307 purged->second.
clear();
308 _latestUpdates.erase(purged);
309 _counts.erase(subId_);
310 _timers.erase(subId_);
311 clearableSubId.
clear();
324 Lock<Mutex> lock(_lock);
332 while (!_counts.empty())
345 Lock<Mutex> lock(_lock);
355 std::vector<SavedUpdate> _queuedUpdates;
358 DeferLock<Mutex> lock(_lock);
361 _lock.wait(_updateIntervalMillis);
364 for (TimerMap::iterator timer = _timers.begin();
365 timer != _timers.end(); )
367 if (timer->second.check())
369 UpdateIter
update = _latestUpdates.find(timer->first);
370 if (update != _latestUpdates.end())
374 _queuedUpdates.push_back(*update);
375 _counts.erase(update->first);
376 timer = _timers.erase(timer);
377 _latestUpdates.erase(update);
392 volatile bool updateAll = _updateAll;
394 for (CountMap::iterator count = _counts.begin();
395 count != _counts.end(); )
397 if (updateAll || _timeoutMillis == 0.0
398 || count->second >= _updateThreshold)
400 UpdateIter
update = _latestUpdates.find(count->first);
401 if (update != _latestUpdates.end())
405 _queuedUpdates.push_back(*update);
406 count = _counts.erase(count);
407 _timers.erase(update->first);
408 _latestUpdates.erase(update);
427 for (std::vector<SavedUpdate>::iterator
update = _queuedUpdates.begin(), end = _queuedUpdates.end();
update != end; ++
update)
429 _delegate->update(
update->second);
431 clearableSubId.
clear();
434 _queuedUpdates.clear();
444 volatile bool _closed;
447 std::shared_ptr<RecoveryPointAdapterImpl> _delegate;
453 typedef std::unordered_map<Field, RecoveryPoint, Field::FieldHash> UpdateMap;
454 typedef std::pair<Field, RecoveryPoint> SavedUpdate;
455 typedef UpdateMap::value_type Update;
456 typedef UpdateMap::iterator UpdateIter;
457 typedef std::unordered_map<Field, Timer, Field::FieldHash> TimerMap;
458 typedef TimerMap::iterator TimerIter;
459 typedef std::unordered_map<Field, unsigned, Field::FieldHash> CountMap;
460 typedef CountMap::iterator CountIter;
463 UpdateMap _latestUpdates;
475 unsigned _updateThreshold;
478 double _timeoutMillis;
481 long _updateIntervalMillis;
484 volatile bool _updateAll;
489 #endif //_RECOVERYPOINTADAPTER_H_ virtual void close()
Take any necessary actions to close the associated storage.
Definition: RecoveryPointAdapter.hpp:316
void purge()
Remove all data from the storage.
Definition: RecoveryPointAdapter.hpp:154
virtual void purge(const Field &subId_)
Remove the specified subId_ from the storage.
Definition: RecoveryPointAdapter.hpp:299
virtual void purge()=0
Remove all data from the storage.
void prune()
Take any necessary actions to close the associated storage.
Definition: RecoveryPointAdapter.hpp:173
RecoveryPointAdapter implementation that delegates storage to another RecoveryPointAdapter but provid...
Definition: RecoveryPointAdapter.hpp:190
void purge(const Field &subId_)
Remove the specified subId_ from the storage.
Definition: RecoveryPointAdapter.hpp:161
virtual void _runUpdateAll()
Lock is already held.
Definition: RecoveryPointAdapter.hpp:329
virtual void purge()
Remove all data from the storage.
Definition: RecoveryPointAdapter.hpp:281
void clear()
Deletes the data associated with this Field, should only be used on Fields that were created as deepC...
Definition: Field.hpp:246
virtual void prune()
Take any necessary actions to reduce associated storage size.
Definition: RecoveryPointAdapter.hpp:69
RecoveryPointAdapter a handle class for implementing external storage of subscription recovery points...
Definition: RecoveryPointAdapter.hpp:74
virtual void update(RecoveryPoint &recoveryPoint_)=0
Update the storage information with the given recovery point.
virtual bool next(RecoveryPoint ¤t_)=0
Recovery is done by iteration over elements in storage.
bool isValid() const
Return if this has a valid implementation.
Definition: RecoveryPointAdapter.hpp:179
ConflatingRecoveryPointAdapter(const std::shared_ptr< RecoveryPointAdapterImpl > &delegate_, unsigned updateThreshold_=10, double timeoutMillis_=2000.0, long updateIntervalMillis_=2000)
Conflate updates to delegate_ where they will only be processed every updateIntervalMillis_ for subsc...
Definition: RecoveryPointAdapter.hpp:207
Provides access to the subId and bookmark needed to restart a subscription.
Definition: RecoveryPoint.hpp:67
Defines the AMPS::Field class, which represents the value of a field in a message.
RecoveryPoint deepCopy()
Return a deep copy of self.
Definition: RecoveryPoint.hpp:97
Provides AMPS::RecoveryPoint, AMPS::RecoveryPointFactory, AMPS::FixedRecoveryPoint, and AMPS::DynamicRecoveryPoint.
void update(RecoveryPoint &recoveryPoint_)
Update the storage information with the given recovery point.
Definition: RecoveryPointAdapter.hpp:148
virtual void close()=0
Take any necessary actions to close the associated storage.
virtual bool next(RecoveryPoint ¤t_)
Recovery is done by iteration over elements in storage.
Definition: RecoveryPointAdapter.hpp:242
iterator begin()
To recover from an adapter, iterate over the adapter from begin() to end() with a RecoveryPointIterat...
Definition: RecoveryPointAdapter.hpp:134
Field represents the value of a single field in a Message.
Definition: Field.hpp:85
virtual void updateAll()
Push all updates to underlying adapter.
Definition: RecoveryPointAdapter.hpp:322
void close()
Take any necessary actions to close the associated storage.
Definition: RecoveryPointAdapter.hpp:167
RecoveryPointAdapterImpl virtual base class for implementing external storage of subscription recover...
Definition: RecoveryPointAdapter.hpp:49
void deepCopy(const Field &orig_)
Makes self a deep copy of the original field.
Definition: Field.hpp:217
Definition: ampsplusplus.hpp:103
iterator end()
Return the end of recovery marker.
Definition: RecoveryPointAdapter.hpp:141
const Field & getSubId() const
Get the sub id for this recovery point.
Definition: RecoveryPoint.hpp:84
virtual void update(RecoveryPoint &recoveryPoint_)
Update the storage information with the given recovery point.
Definition: RecoveryPointAdapter.hpp:249