CheckpointPlayer.h

Go to the documentation of this file.

#ifndef MADARA_KNOWLEDGE_CHECKPOINT_PLAYER_H
#define MADARA_KNOWLEDGE_CHECKPOINT_PLAYER_H
 
#include <fstream>
#include <thread>
#include <memory>
 
#include "madara/MadaraExport.h"
#include "madara/utility/ScopedArray.h"
#include "madara/knowledge/CheckpointSettings.h"
#include "madara/knowledge/FileHeader.h"
#include "madara/transport/MessageHeader.h"
#include "madara/utility/StlHelper.h"
 
namespace madara
{
namespace knowledge
{
class MADARA_EXPORT CheckpointReader
{
public:
  CheckpointReader(CheckpointSettings& in_checkpoint_settings)
    : checkpoint_settings(in_checkpoint_settings),
      logger_(logger::global_logger.get())
  {
  }
 
  void start();
 
  std::pair<std::string, KnowledgeRecord> next();
 
  int64_t get_total_read() const
  {
    return total_read;
  }
 
  const FileHeader* get_file_header() const
  {
    if (stage == 0)
    {
      return nullptr;
    }
 
    return &meta;
  }
 
  bool is_open() const
  {
    return file.is_open();
  }
 
  const CheckpointSettings& get_checkpoint_settings() const
  {
    return checkpoint_settings;
  }
 
private:
  CheckpointSettings& checkpoint_settings;
 
  logger::Logger* logger_;
  int stage = 0;
  std::ifstream file;
  int64_t total_read = 0;
  FileHeader meta;
  int64_t max_buffer;
  int64_t buffer_remaining;
  utility::ScopedArray<char> buffer;
  char* current;
  size_t checkpoint_start;
  uint64_t state;
  uint64_t checkpoint_size;
  transport::MessageHeader checkpoint_header;
  uint64_t update;
};
 
class MADARA_EXPORT CheckpointPlayer
{
public:
  CheckpointPlayer(ThreadSafeContext& context, CheckpointSettings settings,
      KnowledgeUpdateSettings update_settings = {})
    : context_(&context),
      settings_(std::move(settings)),
      update_settings_(update_settings)
  {
  }
 
  CheckpointPlayer(ThreadSafeContext& context, CheckpointReader reader,
      KnowledgeUpdateSettings update_settings = {})
    : context_(&context),
      settings_(reader.get_checkpoint_settings()),
      update_settings_(update_settings),
      reader_(utility::mk_unique<CheckpointReader>(std::move(reader)))
  {
  }
 
  ~CheckpointPlayer()
  {
    stop();
  }
 
  // This object spawns a thread which holds a pointer back to this object,
  // so it cannot be safely copied or moved.
  CheckpointPlayer(const CheckpointPlayer&) = delete;
  CheckpointPlayer(CheckpointPlayer&&) = delete;
  CheckpointPlayer& operator=(const CheckpointPlayer&) = delete;
  CheckpointPlayer& operator=(CheckpointPlayer&&) = delete;
 
  void start()
  {
    keep_running_.test_and_set();
    init_reader();
    thread_ = std::thread(thread_main, this);
  }
 
  void stop()
  {
    keep_running_.clear();
    if (thread_.joinable())
    {
      thread_.join();
    }
    reader_.reset();
  }
 
  bool play_until(uint64_t target_toi);
 
private:
  static void thread_main(CheckpointPlayer* self);
 
  void init_reader()
  {
    if (!reader_)
    {
      reader_ = utility::mk_unique<CheckpointReader>(settings_);
    }
  }
 
  ThreadSafeContext* context_;
  CheckpointSettings settings_;
  KnowledgeUpdateSettings update_settings_;
 
  std::atomic_flag keep_running_;
  std::thread thread_;
  std::unique_ptr<CheckpointReader> reader_;
};
}
}  // namespace madara::knowledge
 
#endif  // MADARA_KNOWLEDGE_CHECKPOINT_PLAYER_H