26 #ifndef _SOWRECOVERYPOINTADAPTER_H_ 27 #define _SOWRECOVERYPOINTADAPTER_H_ 39 #define AMPS_SOW_STORE_DEFAULT_TOPIC "/ADMIN/bookmark_store" 40 #define AMPS_SOW_STORE_DEFAULT_CLIENT_FIELD "clientName" 41 #define AMPS_SOW_STORE_DEFAULT_SUB_FIELD "subId" 42 #define AMPS_SOW_STORE_DEFAULT_BOOKMARK_FIELD "bookmark" 44 #define SOW_RECOVERY_HANDLE_EXCEPTION(x) \ 45 catch (const AMPSException& ex_) \ 47 std::ostringstream os; \ 48 os << x << ": AMPSException " << ex_.what(); \ 49 StoreException ex(os.str()); \ 50 if (!_throwNotListen && _pExceptionListener) \ 52 _pExceptionListener->exceptionThrown(ex); \ 56 catch (const std::exception& ex_) \ 58 std::ostringstream os; \ 59 os << x << ": std::exception " << ex_.what(); \ 60 StoreException ex(os.str()); \ 61 if (!_throwNotListen && _pExceptionListener) \ 63 _pExceptionListener->exceptionThrown(ex); \ 69 std::ostringstream os; \ 70 os << x << ": Unknown exception"; \ 71 StoreException ex(os.str()); \ 72 if (!_throwNotListen && _pExceptionListener) \ 74 _pExceptionListener->exceptionThrown(ex); \ 120 const string& trackedClientName_,
121 unsigned timeoutMillis_ = 5000,
122 bool useTimestamp_ =
false,
123 bool closeClient_ =
true,
124 bool updateFailureThrows_ =
false,
125 const string& topic_ = AMPS_SOW_STORE_DEFAULT_TOPIC,
126 const string& clientNameField_ = AMPS_SOW_STORE_DEFAULT_CLIENT_FIELD,
127 const string& subIdField_ = AMPS_SOW_STORE_DEFAULT_SUB_FIELD,
128 const string& bookmarkField_ = AMPS_SOW_STORE_DEFAULT_BOOKMARK_FIELD
132 , _serializeBuffer(0)
134 , _deserializeBuffer(0)
135 , _client(storeClient_)
136 , _trackedName(trackedClientName_)
138 , _nameField(clientNameField_)
139 , _subIdField(subIdField_)
140 , _bookmarkField(bookmarkField_)
141 , _timeoutMillis(timeoutMillis_)
143 , _closeClient(closeClient_)
145 , _throwNotListen(updateFailureThrows_)
146 , _useTimestamp(useTimestamp_)
149 _initSerialization();
156 delete[] _serializeBuffer;
157 delete[] _deserializeBuffer;
166 static Field emptyField;
173 .
setFilter(
"/" + _nameField +
"='" + _trackedName +
"'")
177 cmd.
setOptions(
"select=[-/,+/" + _subIdField +
",+/" 178 + _bookmarkField +
"],timestamp");
182 cmd.
setOptions(
"select=[-/,+/" + _subIdField +
",+/" 183 + _bookmarkField +
"]");
186 _msIter = _stream.
begin();
205 return next(current_);
229 SOW_RECOVERY_HANDLE_EXCEPTION(
"SOWRecoveryPoint::next")
239 Field data = serialize(recoveryPoint_);
243 SOW_RECOVERY_HANDLE_EXCEPTION(
"SOWRecoveryPoint::update")
252 +
"='" + _trackedName +
"'");
254 SOW_RECOVERY_HANDLE_EXCEPTION(
"SOWRecoveryPoint::purge")
264 + _trackedName +
"' and /" 268 SOW_RECOVERY_HANDLE_EXCEPTION(
"SOWRecoveryPoint::purge(subId)")
286 _pExceptionListener = pListener_;
291 if (_closed || !_client.isValid())
299 SOW_RECOVERY_HANDLE_EXCEPTION(
"SOWRecoveryPoint::close publishFlush")
309 SOW_RECOVERY_HANDLE_EXCEPTION(
"SOWRecoveryPoint::close disconnect")
312 void _initSerialization()
317 if (_serializeLen == 0)
319 _serializeLen = (size_t) (_nameField.length()
320 + _trackedName.length()
321 + _subIdField.length()
322 + _bookmarkField.length()
323 + (AMPS_MAX_BOOKMARK_LEN * 4UL)
324 + SUBID_LEN + JSON_EXTRA);
325 _serializeLen += (128 - (_serializeLen % 128));
327 _serializeBuffer =
new char[_serializeLen];
328 AMPS_snprintf(_serializeBuffer, _serializeLen,
329 "{\"%s\":\"%s\",\"%s\":\"", _nameField.c_str()
330 , _trackedName.c_str()
331 , _subIdField.c_str());
332 _serializeStart = JSON_START + _nameField.length()
333 + _trackedName.length() + _subIdField.length();
335 SOW_RECOVERY_HANDLE_EXCEPTION(
"SOWRecoveryPoint::initSerialization")
340 virtual void initSerialization()
342 _initSerialization();
350 const Field& timestamp_)
358 const char* start = (
const char*)memchr((
const void*)data_.
data(),
359 (int)
':', data_.
len());
362 throw StoreException(
"Failure parsing json RecoveryPoint subId, no :");
364 size_t remain = data_.
len() - (size_t)(start - data_.
data());
365 start = (
const char*)memchr((
const void*)start, (int)
'"', remain);
368 throw StoreException(
"Failure parsing json RecoveryPoint subId, no start \"");
371 remain = data_.
len() - (size_t)(start - data_.
data());
372 const char* end = (
const char*)memchr((
const void*)start,
376 throw StoreException(
"Failure parsing json RecoveryPoint subId, no end \"");
378 size_t len = (size_t)(end - start);
379 subId =
Field(start, len);
380 start = (
const char*)memchr((
const void*)start, (int)
':', data_.
len());
383 throw StoreException(
"Failure parsing json RecoveryPoint bookmark, no :");
385 remain = data_.
len() - (size_t)(start - data_.
data());
386 start = (
const char*)memchr((
const void*)start, (int)
'"', remain);
389 throw StoreException(
"Failure parsing json RecoveryPoint bookmark, no start \"");
392 remain = data_.
len() - (size_t)(start - data_.
data());
393 end = (
const char*)memchr((
const void*)start, (int)
'"', remain);
396 throw StoreException(
"Failure parsing json RecoveryPoint bookmark, no end \"");
398 len = (size_t)(end - start);
399 if (_useTimestamp && !timestamp_.
empty())
401 if (_deserializeLen < len + timestamp_.
len())
403 delete[] _deserializeBuffer;
404 _deserializeBuffer = 0;
406 if (!_deserializeBuffer)
408 _deserializeLen = len + timestamp_.
len() + 1;
409 _deserializeBuffer =
new char[_deserializeLen];
411 memcpy((
void*)_deserializeBuffer, (
const void*)start, len);
412 _deserializeBuffer[len] =
',';
413 memcpy((
void*)(_deserializeBuffer + len + 1),
414 (
const void*)timestamp_.
data(), timestamp_.
len());
415 bookmark =
Field(_deserializeBuffer, _deserializeLen);
419 bookmark =
Field(start, len);
422 SOW_RECOVERY_HANDLE_EXCEPTION(
"SOWRecoveryPoint::deserialize")
434 size_t fullLen = _serializeStart + subId.
len()
435 + _bookmarkField.length() + bookmark.
len() + JSON_END;
436 if (fullLen >= _serializeLen)
438 _serializeLen = fullLen + (128 - (fullLen % 128));
439 delete[] _serializeBuffer;
443 AMPS_snprintf(_serializeBuffer + _serializeStart,
444 _serializeLen - _serializeStart,
445 "%.*s\",\"%s\":\"%.*s\"}", (
int)subId.
len()
447 , _bookmarkField.c_str()
448 , (int)bookmark.
len()
450 _serializeField.assign(_serializeBuffer, fullLen);
452 SOW_RECOVERY_HANDLE_EXCEPTION(
"SOWRecoveryPoint::serialize")
453 return _serializeField;
456 enum Constants :
size_t 465 size_t _serializeLen;
466 size_t _serializeStart;
467 Field _serializeField;
468 char* _serializeBuffer;
469 size_t _deserializeLen;
470 char* _deserializeBuffer;
472 std::string _trackedName;
474 std::string _nameField;
475 std::string _subIdField;
476 std::string _bookmarkField;
477 unsigned _timeoutMillis;
481 std::shared_ptr<const ExceptionListener> _pExceptionListener;
484 bool _throwNotListen;
489 #endif //_SOWRECOVERYPOINTADAPTER_H_ Command & setOptions(const std::string &options_)
Sets the options string for this command: see Message.Options for a helper class for constructing the...
Definition: ampsplusplus.hpp:758
virtual void update(RecoveryPoint &recoveryPoint_)
Update the storage information with the given recovery point.
Definition: SOWRecoveryPointAdapter.hpp:235
Message encapsulates a single message sent to or received from an AMPS server, and provides methods f...
Definition: Message.hpp:531
iterator begin(void)
Returns an iterator representing the beginning of the topic or subscription.
Definition: ampsplusplus.hpp:4996
const char * data() const
Returns the (non-null-terminated) data underlying this field.
Definition: Field.hpp:260
Field getCommand() const
Retrieves the value of the Command header of the Message as a new Field.
Definition: Message.hpp:1195
MessageStream execute(Command &command_)
Execute the provided command and return messages received in response in a MessageStream.
Definition: ampsplusplus.hpp:8724
RecoveryPointAdapter virtual base class for implementing external storage of subscription recovery po...
Definition: SOWRecoveryPointAdapter.hpp:84
Command & setData(const std::string &data_)
Sets the data for this command from an existing string.
Definition: ampsplusplus.hpp:798
Command & setFilter(const std::string &filter_)
Definition: ampsplusplus.hpp:661
Client represents a connection to an AMPS server, but does not provide failover or reconnection behav...
Definition: ampsplusplus.hpp:5075
Provides access to the subId and bookmark needed to restart a subscription.
Definition: RecoveryPoint.hpp:67
RecoveryPointImpl virtual base class provides access to the subId and bookmark needed to restart a su...
Definition: RecoveryPoint.hpp:49
Command & setTimeout(unsigned timeout_)
Sets the client-side timeout for this command.
Definition: ampsplusplus.hpp:820
virtual void purge()
Remove all data from the storage.
Definition: SOWRecoveryPointAdapter.hpp:247
virtual bool next(RecoveryPoint ¤t_)
Recovery is done by iteration over elements in storage.
Definition: SOWRecoveryPointAdapter.hpp:164
bool empty() const
Returns 'true' if empty, 'false' otherwise.
Definition: Field.hpp:128
Defines the AMPS::Field class, which represents the value of a field in a message.
Core type, function, and class declarations for the AMPS C++ client.
size_t len() const
Returns the length of the data underlying this field.
Definition: Field.hpp:267
Provides AMPS::RecoveryPointAdapter, an iterface for implementing external storage of bookmark subscr...
virtual void purge(const Field &subId_)
Remove the specified subId_ from the storage.
Definition: SOWRecoveryPointAdapter.hpp:259
void setExceptionListener(const std::shared_ptr< const ExceptionListener > &pListener_)
Set an exception listener on this adapter that will be notified of all exceptions that occur rather t...
Definition: SOWRecoveryPointAdapter.hpp:284
SOWRecoveryPointAdapter(const Client &storeClient_, const string &trackedClientName_, unsigned timeoutMillis_=5000, bool useTimestamp_=false, bool closeClient_=true, bool updateFailureThrows_=false, const string &topic_=AMPS_SOW_STORE_DEFAULT_TOPIC, const string &clientNameField_=AMPS_SOW_STORE_DEFAULT_CLIENT_FIELD, const string &subIdField_=AMPS_SOW_STORE_DEFAULT_SUB_FIELD, const string &bookmarkField_=AMPS_SOW_STORE_DEFAULT_BOOKMARK_FIELD)
Create a SOWRecoveryPointAdapter for a BookmarkStore that writes updated RecoveryPoints to the server...
Definition: SOWRecoveryPointAdapter.hpp:119
Provides AMPS::RecoveryPoint, AMPS::RecoveryPointFactory, AMPS::FixedRecoveryPoint, and AMPS::DynamicRecoveryPoint.
Represents an iterator over messages in an AMPS topic.
Definition: ampsplusplus.hpp:4951
void publishFlush(long timeout_=0, unsigned ackType_=Message::AckType::Processed)
Ensure that AMPS messages are sent and have been processed by the AMPS server.
Definition: ampsplusplus.hpp:5626
virtual void close()
Take any necessary actions to close the associated storage.
Definition: SOWRecoveryPointAdapter.hpp:272
Field represents the value of a single field in a Message.
Definition: Field.hpp:86
An iterable object representing the results of an AMPS subscription and/or query. ...
Definition: ampsplusplus.hpp:4943
Field getTimestamp() const
Retrieves the value of the Timestamp header of the Message as a new Field.
Definition: Message.hpp:1430
Field getData() const
Returns the data from this message.
Definition: Message.hpp:1460
FixedRecoveryPoint is a RecoveryPoint implementation where subId and bookmark are set explicitly...
Definition: RecoveryPoint.hpp:133
Command & setTopic(const std::string &topic_)
Definition: ampsplusplus.hpp:648
void disconnect()
Disconnect from an AMPS server.
Definition: ampsplusplus.hpp:5262
RecoveryPointAdapterImpl virtual base class for implementing external storage of subscription recover...
Definition: RecoveryPointAdapter.hpp:52
Definition: ampsplusplus.hpp:106
const Field & getSubId() const
Get the sub id for this recovery point.
Definition: RecoveryPoint.hpp:84
MessageStream timeout(unsigned timeout_)
Sets the maximum time to wait for the next message in milliseconds; if no message is available within...
Definition: ampsplusplus.hpp:8695
Command is an encapsulation of a single AMPS command sent by the client.
Definition: ampsplusplus.hpp:441
std::string sowDelete(const MessageHandler &messageHandler, const std::string &topic, const std::string &filter, long timeout)
Deletes one or more messages from a topic's SOW cache.
Definition: ampsplusplus.hpp:6668
const Field & getBookmark() const
Get the bookmark for this recovery point.
Definition: RecoveryPoint.hpp:91