26 #ifndef _SOWRECOVERYPOINTADAPTER_H_ 27 #define _SOWRECOVERYPOINTADAPTER_H_ 38 #define AMPS_SOW_STORE_DEFAULT_TOPIC "/ADMIN/bookmark_store" 39 #define AMPS_SOW_STORE_DEFAULT_CLIENT_FIELD "clientName" 40 #define AMPS_SOW_STORE_DEFAULT_SUB_FIELD "subId" 41 #define AMPS_SOW_STORE_DEFAULT_BOOKMARK_FIELD "bookmark" 43 #define SOW_RECOVERY_HANDLE_EXCEPTION(x) \ 44 catch (const AMPSException& ex_) \ 46 std::ostringstream os; \ 47 os << x << ": AMPSException " << ex_.what(); \ 48 StoreException ex(os.str()); \ 49 if (!_throwNotListen && _pExceptionListener) \ 51 _pExceptionListener->exceptionThrown(ex); \ 55 catch (const std::exception& ex_) \ 57 std::ostringstream os; \ 58 os << x << ": std::exception " << ex_.what(); \ 59 StoreException ex(os.str()); \ 60 if (!_throwNotListen && _pExceptionListener) \ 62 _pExceptionListener->exceptionThrown(ex); \ 68 std::ostringstream os; \ 69 os << x << ": Unknown exception"; \ 70 StoreException ex(os.str()); \ 71 if (!_throwNotListen && _pExceptionListener) \ 73 _pExceptionListener->exceptionThrown(ex); \ 119 const string& trackedClientName_,
120 unsigned timeoutMillis_ = 5000,
121 bool useTimestamp_ =
false,
122 bool closeClient_ =
true,
123 bool updateFailureThrows_ =
false,
124 const string& topic_ = AMPS_SOW_STORE_DEFAULT_TOPIC,
125 const string& clientNameField_ = AMPS_SOW_STORE_DEFAULT_CLIENT_FIELD,
126 const string& subIdField_ = AMPS_SOW_STORE_DEFAULT_SUB_FIELD,
127 const string& bookmarkField_ = AMPS_SOW_STORE_DEFAULT_BOOKMARK_FIELD
131 , _serializeBuffer(0)
133 , _deserializeBuffer(0)
134 , _client(storeClient_)
135 , _trackedName(trackedClientName_)
137 , _nameField(clientNameField_)
138 , _subIdField(subIdField_)
139 , _bookmarkField(bookmarkField_)
140 , _timeoutMillis(timeoutMillis_)
142 , _closeClient(closeClient_)
144 , _throwNotListen(updateFailureThrows_)
145 , _useTimestamp(useTimestamp_)
148 _initSerialization();
155 delete[] _serializeBuffer;
156 delete[] _deserializeBuffer;
165 static Field emptyField;
172 .
setFilter(
"/" + _nameField +
"='" + _trackedName +
"'")
176 cmd.
setOptions(
"select=[-/,+/" + _subIdField +
",+/" 177 + _bookmarkField +
"],timestamp");
181 cmd.
setOptions(
"select=[-/,+/" + _subIdField +
",+/" 182 + _bookmarkField +
"]");
185 _msIter = _stream.
begin();
204 return next(current_);
228 SOW_RECOVERY_HANDLE_EXCEPTION(
"SOWRecoveryPoint::next")
238 Field data = serialize(recoveryPoint_);
242 SOW_RECOVERY_HANDLE_EXCEPTION(
"SOWRecoveryPoint::update")
251 +
"='" + _trackedName +
"'");
253 SOW_RECOVERY_HANDLE_EXCEPTION(
"SOWRecoveryPoint::purge")
263 + _trackedName +
"' and /" 267 SOW_RECOVERY_HANDLE_EXCEPTION(
"SOWRecoveryPoint::purge(subId)")
285 _pExceptionListener = pListener_;
290 if (_closed || !_client.isValid())
return;
295 SOW_RECOVERY_HANDLE_EXCEPTION(
"SOWRecoveryPoint::close publishFlush")
305 SOW_RECOVERY_HANDLE_EXCEPTION(
"SOWRecoveryPoint::close disconnect")
308 void _initSerialization()
313 if (_serializeLen == 0)
315 _serializeLen = (size_t) (_nameField.length()
316 + _trackedName.length()
317 + _subIdField.length()
318 + _bookmarkField.length()
319 + (AMPS_MAX_BOOKMARK_LEN * 4UL)
320 + SUBID_LEN + JSON_EXTRA);
321 _serializeLen += (128-(_serializeLen % 128));
323 _serializeBuffer =
new char[_serializeLen];
324 AMPS_snprintf(_serializeBuffer, _serializeLen,
325 "{\"%s\":\"%s\",\"%s\":\"", _nameField.c_str()
326 , _trackedName.c_str()
327 , _subIdField.c_str());
328 _serializeStart = JSON_START + _nameField.length()
329 + _trackedName.length() + _subIdField.length();
331 SOW_RECOVERY_HANDLE_EXCEPTION(
"SOWRecoveryPoint::initSerialization")
336 virtual void initSerialization()
338 _initSerialization();
346 const Field& timestamp_)
354 const char* start = (
const char*)memchr((
const void*)data_.
data(),
355 (int)
':', data_.
len());
356 size_t remain = data_.
len() - (size_t)(start - data_.
data());
357 start = (
const char*)memchr((
const void*)start, (int)
'"', remain);
359 remain = data_.
len() - (size_t)(start - data_.
data());
360 const char* end = (
const char*)memchr((
const void*)start,
362 size_t len = (size_t)(end - start);
363 subId =
Field(start, len);
364 start = (
const char*)memchr((
const void*)start, (int)
':', data_.
len());
365 remain = data_.
len() - (size_t)(start - data_.
data());
366 start = (
const char*)memchr((
const void*)start, (int)
'"', remain);
368 remain = data_.
len() - (size_t)(start - data_.
data());
369 end = (
const char*)memchr((
const void*)start, (int)
'"', remain);
370 len = (size_t)(end - start);
371 if (_useTimestamp && !timestamp_.
empty())
373 if (_deserializeLen < len + timestamp_.
len())
375 delete[] _deserializeBuffer;
376 _deserializeBuffer = 0;
378 if (!_deserializeBuffer)
380 _deserializeLen = len + timestamp_.
len() + 1;
381 _deserializeBuffer =
new char[_deserializeLen];
383 memcpy((
void*)_deserializeBuffer, (
const void*)start, len);
384 _deserializeBuffer[len] =
',';
385 memcpy((
void*)(_deserializeBuffer+len+1),
386 (
const void*)timestamp_.
data(), timestamp_.
len());
387 bookmark =
Field(_deserializeBuffer, _deserializeLen);
391 bookmark =
Field(start, len);
394 SOW_RECOVERY_HANDLE_EXCEPTION(
"SOWRecoveryPoint::deserialize")
406 size_t fullLen = _serializeStart + subId.
len()
407 + _bookmarkField.length() + bookmark.
len() + JSON_END;
408 if (fullLen >= _serializeLen)
410 _serializeLen = fullLen + (128-(fullLen % 128));
411 delete[] _serializeBuffer;
415 AMPS_snprintf(_serializeBuffer + _serializeStart,
416 _serializeLen - _serializeStart,
417 "%.*s\",\"%s\":\"%.*s\"}", (
int)subId.
len()
419 , _bookmarkField.c_str()
420 , (int)bookmark.
len()
422 _serializeField.assign(_serializeBuffer, fullLen);
424 SOW_RECOVERY_HANDLE_EXCEPTION(
"SOWRecoveryPoint::serialize")
425 return _serializeField;
428 enum Constants :
size_t 437 size_t _serializeLen;
438 size_t _serializeStart;
439 Field _serializeField;
440 char* _serializeBuffer;
441 size_t _deserializeLen;
442 char* _deserializeBuffer;
444 std::string _trackedName;
446 std::string _nameField;
447 std::string _subIdField;
448 std::string _bookmarkField;
449 unsigned _timeoutMillis;
453 std::shared_ptr<const ExceptionListener> _pExceptionListener;
456 bool _throwNotListen;
461 #endif //_SOWRECOVERYPOINTADAPTER_H_ Command & setFilter(const std::string &v_)
Definition: ampsplusplus.hpp:540
virtual void update(RecoveryPoint &recoveryPoint_)
Update the storage information with the given recovery point.
Definition: SOWRecoveryPointAdapter.hpp:234
Message encapsulates a single message sent to or received from an AMPS server, and provides methods f...
Definition: Message.hpp:501
iterator begin(void)
Returns an iterator representing the beginning of the topic or subscription.
Definition: ampsplusplus.hpp:4173
const char * data() const
Returns the (non-null-terminated) data underlying this field.
Definition: Field.hpp:251
Field getCommand() const
Retrieves the value of the Command header of the Message as a new Field.
Definition: Message.hpp:1035
MessageStream execute(Command &command_)
Execute the provided command and return messages received in response in a MessageStream.
Definition: ampsplusplus.hpp:7613
RecoveryPointAdapter virtual base class for implementing external storage of subscription recovery po...
Definition: SOWRecoveryPointAdapter.hpp:83
Client represents a connection to an AMPS server, but does not provide failover or reconnection behav...
Definition: ampsplusplus.hpp:4249
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 & setTopic(const std::string &v_)
Definition: ampsplusplus.hpp:538
virtual void purge()
Remove all data from the storage.
Definition: SOWRecoveryPointAdapter.hpp:246
virtual bool next(RecoveryPoint ¤t_)
Recovery is done by iteration over elements in storage.
Definition: SOWRecoveryPointAdapter.hpp:163
bool empty() const
Returns 'true' if empty, 'false' otherwise.
Definition: Field.hpp:126
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:258
Command & setTimeout(unsigned v_)
Sets the client-side timeout for this command.
Definition: ampsplusplus.hpp:588
Provides AMPS::RecoveryPointAdapter, an iterface for impplementing external storage of bookmark subsc...
virtual void purge(const Field &subId_)
Remove the specified subId_ from the storage.
Definition: SOWRecoveryPointAdapter.hpp:258
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:283
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:118
Provides AMPS::RecoveryPoint, AMPS::RecoveryPointFactory, AMPS::FixedRecoveryPoint, and AMPS::DynamicRecoveryPoint.
Represents an iterator over messages in an AMPS topic.
Definition: ampsplusplus.hpp:4140
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:4792
virtual void close()
Take any necessary actions to close the associated storage.
Definition: SOWRecoveryPointAdapter.hpp:271
Field represents the value of a single field in a Message.
Definition: Field.hpp:84
An iterable object representing the results of an AMPS subscription and/or query. ...
Definition: ampsplusplus.hpp:4132
Field getTimestamp() const
Retrieves the value of the Timestamp header of the Message as a new Field.
Definition: Message.hpp:1238
Field getData() const
Returns the data from this message.
Definition: Message.hpp:1268
FixedRecoveryPoint is a RecoveryPoint implementation where subId and bookmark are set explicitly...
Definition: RecoveryPoint.hpp:133
void disconnect()
Disconnect from an AMPS server.
Definition: ampsplusplus.hpp:4428
Command & setData(const std::string &v_)
Sets the data for this command from an existing string.
Definition: ampsplusplus.hpp:574
Command & setOptions(const std::string &v_)
Sets the options string for this command: see Message.Options for a helper class for constructing the...
Definition: ampsplusplus.hpp:560
RecoveryPointAdapterImpl virtual base class for implementing external storage of subscription recover...
Definition: RecoveryPointAdapter.hpp:49
Definition: ampsplusplus.hpp:103
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:7584
Command is an encapsulation of a single AMPS command sent by the client.
Definition: ampsplusplus.hpp:407
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:5796
const Field & getBookmark() const
Get the bookmark for this recovery point.
Definition: RecoveryPoint.hpp:91