26 #ifndef _HYBRIDPUBLISHSTORE_H_
27 #define _HYBRIDPUBLISHSTORE_H_
32 #if __cplusplus >= 201100L || _MSC_VER >= 1900
60 : _store(NULL), _handler(NULL), _data(NULL)
73 #if __cplusplus >= 201100L || _MSC_VER >= 1900
74 std::atomic<size_t> _entries;
75 std::atomic<size_t> _errorCount;
76 std::atomic<amps_uint64_t> _lastIndex;
78 volatile size_t _entries;
79 volatile size_t _errorCount;
80 volatile amps_uint64_t _lastIndex;
83 SwappingOutReplayer(
PublishStore* pStore_,
size_t entries_)
84 : _pStore(pStore_), _entries(entries_)
85 , _errorCount(0), _lastIndex(0)
93 amps_uint64_t lastIndex()
100 if (_entries > 0 && _errorCount == 0)
105 _pStore->
store(message_,
false);
108 message_.getMessage(),
132 bool errorOnPublishGap_ =
false)
134 , _memStore(maxMemoryCapacity_, errorOnPublishGap_)
135 , _fileStore(fileName_, errorOnPublishGap_)
136 , _cap(maxMemoryCapacity_)
137 , _lowWatermark((size_t)((double)maxMemoryCapacity_ * 0.5))
138 , _lowestIndexInMemory(0)
139 , _holdSwapping(false)
141 _handlerData._store =
this;
157 bool errorOnPublishGap_ =
false)
159 , _memStore(maxMemoryCapacity_, errorOnPublishGap_)
160 , _fileStore(fileName_, errorOnPublishGap_)
161 , _cap(maxMemoryCapacity_)
162 , _lowWatermark((size_t)((double)maxMemoryCapacity_ * 0.5))
163 , _lowestIndexInMemory(0)
164 , _holdSwapping(false)
166 _handlerData._store =
this;
179 Lock<Mutex> guard(_lock);
180 _lowWatermark = lowWatermark_;
191 Lock<Mutex> guard(_lock);
192 return _lowWatermark;
200 Lock<Mutex> guard(_lock);
201 while (_holdSwapping)
203 if (!_lock.wait(1000))
205 Unlock<Mutex> u(_lock);
206 amps_invoke_waiting_function();
210 FlagFlip flip(&_holdSwapping);
212 Unlock<Mutex> u(_lock);
216 Lock<Mutex> l(_lock);
221 if (_lowestIndexInMemory <= index_)
224 _lowestIndexInMemory = index_ + 1;
236 Lock<Mutex> guard(_lock);
237 while (_holdSwapping)
239 if (!_lock.wait(1000))
241 amps_invoke_waiting_function();
245 FlagFlip flip(&_holdSwapping);
247 Unlock<Mutex> u(_lock);
248 _fileStore.
replay(replayer_);
249 _memStore.
replay(replayer_);
271 Lock<Mutex> guard(_lock);
272 amps_uint64_t waitFor = _getHybridHighestUnpersisted();
275 if (waitFor == unset)
281 bool timedOut =
false;
282 long waitTime = (timeout_ < 1000) ? timeout_ : 1000;
283 AMPS_START_TIMER(timeout_)
285 while (!timedOut && waitFor >= _getHybridLowestUnpersisted() &&
286 _getHybridLowestUnpersisted() != unset)
288 if (!_lock.wait(waitTime))
291 AMPS_RESET_TIMER(timedOut, timeout_);
292 waitTime = (timeout_ < 1000) ? timeout_ : 1000;
293 Unlock<Mutex> unlck(_lock);
294 amps_invoke_waiting_function();
298 if (timedOut && waitFor >= _getHybridLowestUnpersisted() &&
299 _getHybridLowestUnpersisted() != unset)
301 throw TimedOutException(
"Timed out waiting to flush publish store.");
306 while (waitFor >= _getHybridLowestUnpersisted() &&
307 _getHybridLowestUnpersisted() != unset)
311 Unlock<Mutex> unlck(_lock);
312 amps_invoke_waiting_function();
319 amps_uint64_t lowestIndexInMemory;
321 Lock<Mutex> guard(_lock);
322 lowestIndexInMemory = _lowestIndexInMemory;
324 if (index_ < lowestIndexInMemory)
339 Lock<Mutex> guard(_lock);
340 while (_holdSwapping)
342 if (!_lock.wait(1000))
344 Unlock<Mutex> u(_lock);
345 amps_invoke_waiting_function();
351 FlagFlip flip(&_holdSwapping);
352 SwappingOutReplayer swapper(&_fileStore,
355 Unlock<Mutex> u(_lock);
356 _memStore.
replay(swapper);
359 if (swapper.getErrors() == 0)
361 _lowestIndexInMemory = swapper.lastIndex();
365 return _memStore.
store(message_);
370 _handlerData.init(handler_, data_);
372 (
void*)&_handlerData);
377 return _handlerData._handler;
382 Lock<Mutex> guard(_lock);
383 return _getHybridLowestUnpersisted();
386 amps_uint64_t getHighestUnpersisted()
const
388 Lock<Mutex> guard(_lock);
389 return _getHybridHighestUnpersisted();
394 Lock<Mutex> guard(_lock);
395 return _getHybridLastPersisted();
398 inline virtual void setErrorOnPublishGap(
bool errorOnPublishGap_)
400 StoreImpl::setErrorOnPublishGap(errorOnPublishGap_);
401 _memStore.setErrorOnPublishGap(errorOnPublishGap_);
402 _fileStore.setErrorOnPublishGap(errorOnPublishGap_);
408 static bool resizeHandler(Store store_,
size_t size_,
void* data_)
410 HandlerData* handlerData = (HandlerData*)data_;
412 return handlerData->_handler(store_, size_, handlerData->_data);
416 amps_uint64_t _getHybridLowestUnpersisted()
const
420 if (filemin == AMPS_UNSET_SEQUENCE)
424 if (memmin == AMPS_UNSET_SEQUENCE || memmin > filemin)
433 amps_uint64_t _getHybridHighestUnpersisted()
const
435 amps_uint64_t filemax = _fileStore.getHighestUnpersisted();
436 amps_uint64_t memmax = _memStore.getHighestUnpersisted();
437 if (filemax == AMPS_UNSET_SEQUENCE)
441 if (memmax == AMPS_UNSET_SEQUENCE || memmax < filemax)
449 amps_uint64_t _getHybridLastPersisted()
452 if (!_lowestIndexInMemory &&
460 return (memLast < fileLast) ? memLast : fileLast;
463 MemoryPublishStore _memStore;
464 PublishStore _fileStore;
466 size_t _lowWatermark;
467 amps_uint64_t _lowestIndexInMemory;
469 HandlerData _handlerData;
470 #if __cplusplus >= 201100L || _MSC_VER >= 1900
471 std::atomic<bool> _holdSwapping;
473 volatile bool _holdSwapping;
Provides AMPS::MemoryPublishStore, a publish store that holds messages in memory.
Provides AMPS::PublishStore, a publish store that uses memory-mapped files to provide a publish store...
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.
Core type, function, and class declarations for the AMPS C++ client.
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:1083
virtual void discardUpTo(amps_uint64_t index_)
Remove all messages with an index up to and including index_.
Definition: BlockPublishStore.hpp:498
void replay(StoreReplayer &replayer_)
Replay all messages in the Store onto the given StoreReplayer.
Definition: BlockPublishStore.hpp:561
amps_uint64_t getLowestUnpersisted() const
Get the oldest unpersisted message sequence in the store.
Definition: BlockPublishStore.hpp:699
size_t unpersistedCount() const
Method to return the count of messages that currently in the Store because they have not been discard...
Definition: BlockPublishStore.hpp:642
bool replaySingle(StoreReplayer &replayer_, amps_uint64_t index_)
Replay one message in the Store onto the given StoreReplayer.
Definition: BlockPublishStore.hpp:593
amps_uint64_t getLastPersisted(void)
Get the last persisted sequence number.
Definition: BlockPublishStore.hpp:711
virtual amps_uint64_t store(const Message &message_)
Store a given message that will be delivered to AMPS.
Definition: BlockPublishStore.hpp:251
An implementation of StoreImpl for publication.
Definition: HybridPublishStore.hpp:51
size_t unpersistedCount() const
The number of messages in the Store that have not been discarded.
Definition: HybridPublishStore.hpp:256
HybridPublishStore(const char *fileName_, size_t maxMemoryCapacity_, bool errorOnPublishGap_=false)
Create a HybridPublishStore that will use fileName_ as its file storage and stores at most maxMemoryC...
Definition: HybridPublishStore.hpp:131
virtual void flush(long timeout_)
Method to wait for the Store to discard everything that has been stored up to the point in time when ...
Definition: HybridPublishStore.hpp:269
HybridPublishStore(const std::string &fileName_, size_t maxMemoryCapacity_, bool errorOnPublishGap_=false)
Create a HybridPublishStore that will use fileName_ as its file storage and stores at most maxMemoryC...
Definition: HybridPublishStore.hpp:156
void setResizeHandler(PublishStoreResizeHandler handler_, void *data_)
Set a handler to be called if the Store needs to resize in order to keep storing messages.
Definition: HybridPublishStore.hpp:368
void setLowWatermark(size_t lowWatermark_)
Set how many messags remain in memory after messages get offlined.
Definition: HybridPublishStore.hpp:177
amps_uint64_t store(const Message &message_)
Used internally by Client to put messages into the Store.
Definition: HybridPublishStore.hpp:337
amps_uint64_t getLastPersisted(void)
Get the last persisted sequence number.
Definition: HybridPublishStore.hpp:392
size_t getLowWatermark()
Get how many messags remain in memory after messages get offlined.
Definition: HybridPublishStore.hpp:189
void replay(StoreReplayer &replayer_)
Used internally by Client to replay messages in the store to AMPS after a successful connection.
Definition: HybridPublishStore.hpp:234
void discardUpTo(amps_uint64_t index_)
Discard all messages in the store up to and including index_.
Definition: HybridPublishStore.hpp:198
amps_uint64_t getLowestUnpersisted() const
Get the oldest unpersisted message sequence in the store.
Definition: HybridPublishStore.hpp:380
bool replaySingle(StoreReplayer &replayer_, amps_uint64_t index_)
Called by Client to get a single message replayed by the store onto the StoreReplayer.
Definition: HybridPublishStore.hpp:317
Message encapsulates a single message sent to or received from an AMPS server, and provides methods f...
Definition: Message.hpp:532
A StoreImpl implementation that uses a memory-mapped file to provide a publish store that persists ac...
Definition: PublishStore.hpp:47
Abstract base class for storing published messages for an HA publisher client.
Definition: ampsplusplus.hpp:1090
static amps_uint64_t getUnsetSequence()
Method to return the value used to represent no such sequence.
Definition: ampsplusplus.hpp:1159
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:1183
Abstract base class for replaying a publish message.
Definition: ampsplusplus.hpp:1062