SOWRecoveryPointAdapter.hpp

//
// Copyright (c) 2010-2025 60East Technologies Inc., All Rights Reserved.
//
// This computer software is owned by 60East Technologies Inc. and is
// protected by U.S. copyright laws and other laws and by international
// treaties.  This computer software is furnished by 60East Technologies
// Inc. pursuant to a written license agreement and may be used, copied,
// transmitted, and stored only in accordance with the terms of such
// license agreement and with the inclusion of the above copyright notice.
// This computer software or any other copies thereof may not be provided
// or otherwise made available to any other person.
//
// U.S. Government Restricted Rights.  This computer software: (a) was
// developed at private expense and is in all respects the proprietary
// information of 60East Technologies Inc.; (b) was not developed with
// government funds; (c) is a trade secret of 60East Technologies Inc.
// for all purposes of the Freedom of Information Act; and (d) is a
// commercial item and thus, pursuant to Section 12.212 of the Federal
// Acquisition Regulations (FAR) and DFAR Supplement Section 227.7202,
// Government's use, duplication or disclosure of the computer software
// is subject to the restrictions set forth by 60East Technologies Inc..
//

#ifndef _SOWRECOVERYPOINTADAPTER_H_
#define _SOWRECOVERYPOINTADAPTER_H_

#include <amps/ampsplusplus.hpp>
#include <amps/Field.hpp>
#include <amps/RecoveryPoint.hpp>
#include <amps/RecoveryPointAdapter.hpp>
#include <assert.h>
#include <memory>
#include <string>

using std::string;

#define AMPS_SOW_STORE_DEFAULT_TOPIC "/ADMIN/bookmark_store"
#define AMPS_SOW_STORE_DEFAULT_CLIENT_FIELD "clientName"
#define AMPS_SOW_STORE_DEFAULT_SUB_FIELD "subId"
#define AMPS_SOW_STORE_DEFAULT_BOOKMARK_FIELD "bookmark"

#define SOW_RECOVERY_HANDLE_EXCEPTION(x) \
  catch (const AMPSException& ex_) \
  { \
    std::ostringstream os; \
    os << x << ": AMPSException " << ex_.what(); \
    StoreException ex(os.str()); \
    if (_throwNotListen) \
    { \
      throw ex;\
    } \
    else if (_pExceptionListener) \
    { \
      _pExceptionListener->exceptionThrown(ex); \
    } \
  } \
  catch (const std::exception& ex_) \
  { \
    std::ostringstream os; \
    os << x << ": std::exception " << ex_.what(); \
    StoreException ex(os.str()); \
    if (_throwNotListen) \
    { \
      throw ex;\
    } \
    else if (_pExceptionListener) \
    { \
      _pExceptionListener->exceptionThrown(ex); \
    } \
  } \
  catch (...) \
  { \
    std::ostringstream os; \
    os << x << ": Unknown exception"; \
    StoreException ex(os.str()); \
    if (_throwNotListen) \
    { \
      throw ex;\
    } \
    else if (_pExceptionListener) \
    { \
      _pExceptionListener->exceptionThrown(ex); \
    } \
  }

namespace AMPS
{

  class SOWRecoveryPointAdapter : public RecoveryPointAdapterImpl
  {
  public:

    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
                           )
      : RecoveryPointAdapterImpl()
      , _serializeLen(0)
      , _serializeBuffer(0)
      , _deserializeLen(0)
      , _deserializeBuffer(0)
      , _client(storeClient_)
      , _trackedName(trackedClientName_)
      , _topic(topic_)
      , _nameField(clientNameField_)
      , _subIdField(subIdField_)
      , _bookmarkField(bookmarkField_)
      , _timeoutMillis(timeoutMillis_)
      , _closeClient(closeClient_)
      , _executed(false)
      , _throwNotListen(updateFailureThrows_)
      , _useTimestamp(useTimestamp_)
      , _closed(false)
    {
      if (_client.getName() == _trackedName)
      {
        throw UsageException("The SOWRecoveryPointAdapter cannot use the tracked client to update AMPS");
      }
      _initSerialization();
    }

    virtual ~SOWRecoveryPointAdapter()
    {
      _close();
      delete[] _serializeBuffer;
      delete[] _deserializeBuffer;
    }

    virtual bool next(RecoveryPoint& current_)
    {
      static Field emptyField;
      try
      {
        if (!_executed)
        {
          Command cmd("sow");
          cmd.setTopic(_topic)
          .setFilter("/" + _nameField + "='" + _trackedName + "'")
          .setTimeout(_timeoutMillis);
          if (_useTimestamp)
          {
            cmd.setOptions("select=[-/,+/" + _subIdField + ",+/"
                           + _bookmarkField + "],timestamp");
          }
          else
          {
            cmd.setOptions("select=[-/,+/" + _subIdField + ",+/"
                           + _bookmarkField + "]");
          }
          _stream = _client.execute(cmd).timeout(_timeoutMillis);
          _msIter = _stream.begin();
          _executed = true;
        }
        else
        {
          ++_msIter;
        }
        if (_msIter == MessageStream::iterator())
        {
          return false;
        }
        Message m = *_msIter;
        if (!m.isValid())
        {
          current_ = RecoveryPoint(NULL);
          return false;
        }
        if (m.getCommand() == "group_begin")
        {
          return next(current_);
        }
        else if (m.getCommand() == "sow")
        {
          if (_useTimestamp)
          {
            current_ = RecoveryPoint(deserialize(m.getData(),
                                                 m.getTimestamp()));
          }
          else
          {
            current_ = RecoveryPoint(deserialize(m.getData(),
                                                 emptyField));
          }
          return true;
        }
        else if (m.getCommand() == "group_end" || m.getCommand() == "ack")
        {
          current_ = RecoveryPoint(NULL);
          _msIter = MessageStream::iterator();
          _stream = MessageStream();
          return false;
        }
      }
      SOW_RECOVERY_HANDLE_EXCEPTION("SOWRecoveryPoint::next")
      return false;
    }

    virtual void update(RecoveryPoint& recoveryPoint_)
    {
      try
      {
        Field data = serialize(recoveryPoint_);
        _client.publish(_topic.data(), _topic.length(), data.data(), data.len());
      }
      SOW_RECOVERY_HANDLE_EXCEPTION("SOWRecoveryPoint::update")
    }

    virtual void purge()
    {
      try
      {
        Message m = _client.sowDelete(_topic, "/" + _nameField
                                      + "='" + _trackedName + "'");
      }
      SOW_RECOVERY_HANDLE_EXCEPTION("SOWRecoveryPoint::purge")
    }

    virtual void purge(const Field& subId_)
    {
      try
      {
        Message m = _client.sowDelete(_topic, "/" + _nameField + "='"
                                      + _trackedName + "' and /"
                                      + _subIdField + "='"
                                      + subId_ + "'");
      }
      SOW_RECOVERY_HANDLE_EXCEPTION("SOWRecoveryPoint::purge(subId)")
    }

    virtual void close()
    {
      _close();
    }

    void setExceptionListener(const std::shared_ptr<const ExceptionListener>& pListener_)
    {
      _pExceptionListener = pListener_;
    }
  protected:
    void _close()
    {
      if (_closed || !_client.isValid())
      {
        return;
      }
      try
      {
        _client.publishFlush();
      }
      SOW_RECOVERY_HANDLE_EXCEPTION("SOWRecoveryPoint::close publishFlush")
      try
      {
        if (_closeClient)
        {
          _closed = true;
          _client.disconnect();
          _client = Client();
        }
      }
      SOW_RECOVERY_HANDLE_EXCEPTION("SOWRecoveryPoint::close disconnect")
    }

    void _initSerialization()
    {
      try
      {
        // Set up json serialization
        if (_serializeLen == 0)
        {
          _serializeLen = (size_t) (_nameField.length()
                                    + _trackedName.length()
                                    + _subIdField.length()
                                    + _bookmarkField.length()
                                    + (AMPS_MAX_BOOKMARK_LEN * 4UL)
                                    + SUBID_LEN + JSON_EXTRA);
          _serializeLen += (128 - (_serializeLen % 128));
        }
        _serializeBuffer = new char[_serializeLen];
        AMPS_snprintf(_serializeBuffer, _serializeLen,
                      "{\"%s\":\"%s\",\"%s\":\"", _nameField.c_str()
                      , _trackedName.c_str()
                      , _subIdField.c_str());
        _serializeStart = JSON_START + _nameField.length()
                          + _trackedName.length() + _subIdField.length();
      }
      SOW_RECOVERY_HANDLE_EXCEPTION("SOWRecoveryPoint::initSerialization")
    }

    // Subclasses can override this to set up for something other than json
    // serialization if not using json.
    virtual void initSerialization()
    {
      _initSerialization();
    }

    // Subclasses can override this function if not using json data type.
    // It needs to return an allocated RecoveryPointImpl based on the data
    // field from a sow message that contains only 2 fields: _subIdField and
    // _bookmarkField. If you'd like more, override begin()
    virtual RecoveryPointImpl* deserialize(const Field& data_,
                                           const Field& timestamp_)
    {
      Field subId;
      Field bookmark;
      try
      {
        // We have 2 fields subId and bookmark and we only need the
        // values. Find : then start ", then end ".
        const char* start = (const char*)memchr((const void*)data_.data(),
                                                (int)':', data_.len());
        if (!start)
        {
          throw StoreException("Failure parsing json RecoveryPoint subId, no :");
        }
        size_t remain = data_.len() - (size_t)(start - data_.data());
        start = (const char*)memchr((const void*)start, (int)'"', remain);
        if (!start)
        {
          throw StoreException("Failure parsing json RecoveryPoint subId, no start \"");
        }
        ++start;
        remain = data_.len() - (size_t)(start - data_.data());
        const char* end = (const char*)memchr((const void*)start,
                                              (int)'"', remain);
        if (!end)
        {
          throw StoreException("Failure parsing json RecoveryPoint subId, no end \"");
        }
        size_t len = (size_t)(end - start);
        subId = Field(start, len);
        start = (const char*)memchr((const void*)start, (int)':', data_.len());
        if (!start)
        {
          throw StoreException("Failure parsing json RecoveryPoint bookmark, no :");
        }
        remain = data_.len() - (size_t)(start - data_.data());
        start = (const char*)memchr((const void*)start, (int)'"', remain);
        if (!start)
        {
          throw StoreException("Failure parsing json RecoveryPoint bookmark, no start \"");
        }
        ++start;
        remain = data_.len() - (size_t)(start - data_.data());
        end = (const char*)memchr((const void*)start, (int)'"', remain);
        if (!end)
        {
          throw StoreException("Failure parsing json RecoveryPoint bookmark, no end \"");
        }
        len = (size_t)(end - start);
        if (_useTimestamp && !timestamp_.empty())
        {
          if (_deserializeLen < len + timestamp_.len())
          {
            delete[] _deserializeBuffer;
            _deserializeBuffer = 0;
          }
          if (!_deserializeBuffer)
          {
            _deserializeLen = len + timestamp_.len() + 1;
            _deserializeBuffer = new char[_deserializeLen];
          }
          memcpy((void*)_deserializeBuffer, (const void*)start, len);
          _deserializeBuffer[len] = ',';
          memcpy((void*)(_deserializeBuffer + len + 1),
                 (const void*)timestamp_.data(), timestamp_.len());
          bookmark = Field(_deserializeBuffer, _deserializeLen);
        }
        else
        {
          bookmark = Field(start, len);
        }
      }
      SOW_RECOVERY_HANDLE_EXCEPTION("SOWRecoveryPoint::deserialize")
      // Return a recovery point that will copy current field values and
      // clear them when destructed.
      return new FixedRecoveryPoint(subId, bookmark, true);
    }

    virtual Field& serialize(const RecoveryPoint& recoveryPoint_)
    {
      try
      {
        Field subId = recoveryPoint_.getSubId();
        Field bookmark = recoveryPoint_.getBookmark();
        size_t fullLen = _serializeStart + subId.len()
                         + _bookmarkField.length() + bookmark.len() + JSON_END;
        if (fullLen >= _serializeLen)
        {
          _serializeLen = fullLen + (128 - (fullLen % 128));
          delete[] _serializeBuffer;
          // This will reallocate the buffer and fill with predicate
          initSerialization();
        }
        AMPS_snprintf(_serializeBuffer + _serializeStart,
                      _serializeLen - _serializeStart,
                      "%.*s\",\"%s\":\"%.*s\"}", (int)subId.len()
                      , subId.data()
                      , _bookmarkField.c_str()
                      , (int)bookmark.len()
                      , bookmark.data());
        _serializeField.assign(_serializeBuffer, fullLen);
      }
      SOW_RECOVERY_HANDLE_EXCEPTION("SOWRecoveryPoint::serialize")
      return _serializeField;
    }

    enum Constants : size_t
    {
      JSON_START = 11, // '{', 7 '"', 2 ':', 1 ','
      JSON_END   =  8, // '}', 5 '"', 1 ':', 1 ','
      JSON_EXTRA = 19, // '{', '}', 3 ':', 12 '"', 2 ','
      SUBID_LEN  = 64  // rough guess on typical max len
    };

  private:
    size_t                                      _serializeLen;
    size_t                                      _serializeStart;
    Field                                       _serializeField;
    char*                                       _serializeBuffer;
    size_t                                      _deserializeLen;
    char*                                       _deserializeBuffer;
    Client                                      _client;
    std::string                                 _trackedName;
    std::string                                 _topic;
    std::string                                 _nameField;
    std::string                                 _subIdField;
    std::string                                 _bookmarkField;
    unsigned                                    _timeoutMillis;
    MessageStream                               _stream;
    MessageStream::iterator                     _msIter;
    std::shared_ptr<const ExceptionListener>    _pExceptionListener;
    bool                                        _closeClient;
    bool                                        _executed;
    bool                                        _throwNotListen;
    bool                                        _useTimestamp;
    bool                                        _closed;
  };
} // namespace AMPS
#endif //_SOWRECOVERYPOINTADAPTER_H_