bookkeeper.hpp

Go to the documentation of this file.

 
#ifndef SNBMODULES_INCLUDE_SNBMODULES_BOOKKEEPER_HPP_
#define SNBMODULES_INCLUDE_SNBMODULES_BOOKKEEPER_HPP_
 
#include "snbmodules/common/protocols_enum.hpp"
#include "snbmodules/common/status_enum.hpp"
#include "snbmodules/group_metadata.hpp"
#include "snbmodules/ip_format.hpp"
#include "snbmodules/notification_interface.hpp"
 
// errors handling
#include "logging/Logging.hpp"
#include "snbmodules/common/errors_declaration.hpp"
 
#include <algorithm>
#include <chrono>
#include <map>
#include <set>
#include <string>
#include <utility>
#include <vector>
 
namespace dunedaq::snbmodules {
class Bookkeeper : public NotificationInterface
{
 
public:
  Bookkeeper(const IPFormat& listening_ip,
             std::string bookkeeper_id,
             std::string file_log_path = "",
             int refresh_rate = 5,
             std::string connection_prefix = "snbmodules",
             int timeout_send = 10,
             int timeout_receive = 100)
    : NotificationInterface(std::move(connection_prefix), timeout_send, timeout_receive)
  {
    set_bookkeeper_id(std::move(bookkeeper_id));
    set_ip(listening_ip);
    m_file_log_path = std::move(file_log_path);
    m_refresh_rate = refresh_rate;
    display_information();
  }
 
  ~Bookkeeper() {}
 
  bool start();
 
  void do_work(std::atomic<bool>& running_flag);
 
  bool action_on_receive_notification(NotificationData notif) override;
 
  void input_action(char input);
 
  [[deprecated("Now only the uploader can create a new transfer")]] void create_new_transfer(
    const std::string& protocol,
    const std::string& src,
    const std::set<std::string>& dests,
    const std::set<std::filesystem::path>& files,
    const nlohmann::json& protocol_options = nlohmann::json());
 
  void display_information();
 
  void request_update_metadata(bool force = false);
 
  // Setters
  inline void set_bookkeeper_id(std::string bookkeeper_id) { m_bookkeeper_id = std::move(bookkeeper_id); }
  inline void set_ip(const IPFormat& ip) { m_ip = ip; }
  void add_update_transfer(const std::string& client_id, const std::string& data);
  void add_update_grp_transfer(GroupMetadata grp_transfers);
 
  // Getters
  inline std::string get_bookkeeper_id() const { return m_bookkeeper_id; }
  inline IPFormat get_ip() const { return m_ip; }
  inline std::map<std::string, GroupMetadata>& get_grp_transfers() { return m_grp_transfers; }
  inline const std::map<std::string, GroupMetadata>& get_grp_transfers() const { return m_grp_transfers; }
  inline std::map<std::string, std::vector<std::shared_ptr<TransferMetadata>>>& get_transfers() { return m_transfers; }
  inline const std::map<std::string, std::vector<std::shared_ptr<TransferMetadata>>>& get_transfers() const
  {
    return m_transfers;
  }
 
private:
  std::string m_bookkeeper_id;
 
  IPFormat m_ip;
 
  std::map<std::string, GroupMetadata> m_grp_transfers;
 
  std::map<std::string, std::set<std::string>> m_clients_per_grp_transfer;
 
  std::map<std::string, std::vector<std::shared_ptr<TransferMetadata>>> m_transfers;
 
  std::string m_file_log_path = "";
 
  int m_refresh_rate = 5;
 
  void request_connection_and_available_files(const std::string& client);
 
  [[deprecated("Now only the uploader can start a transfer")]] void start_transfers(const std::string& transfer_id);
 
  std::string get_client_name_from_session_name(const std::string& session_name) const
  {
    return session_name.substr(0, session_name.find("_"));
  }
};
} // namespace dunedaq::snbmodules
#endif // SNBMODULES_INCLUDE_SNBMODULES_BOOKKEEPER_HPP_