notification_interface.hpp

Go to the documentation of this file.

 
#ifndef SNBMODULES_INCLUDE_SNBMODULES_NOTIFICATION_INTERFACE_HPP_
#define SNBMODULES_INCLUDE_SNBMODULES_NOTIFICATION_INTERFACE_HPP_
 
#include "iomanager/IOManager.hpp"
#include "iomanager/network/ConfigClient.hpp"
#include "iomanager/network/NetworkIssues.hpp"
 
#include "snbmodules/common/notification_enum.hpp"
#include "snbmodules/iomanager_wrapper.hpp"
#include "snbmodules/ip_format.hpp"
 
#include <set>
#include <string>
#include <utility>
#include <vector>
 
namespace dunedaq {
namespace snbmodules {
struct NotificationData
{
  std::string m_source_id;
  std::string m_target_id;
  std::string m_notification;
  std::string m_data;
 
  NotificationData() = default;
  NotificationData(std::string source_id, std::string target_id, std::string notification, std::string data = "")
    : m_source_id(std::move(source_id))
    , m_target_id(std::move(target_id))
    , m_notification(std::move(notification))
    , m_data(std::move(data))
  {
  }
  NotificationData(std::string source_id,
                   std::string target_id,
                   notification_type::e_notification_type notification,
                   std::string data = "")
    : m_source_id(std::move(source_id))
    , m_target_id(std::move(target_id))
    , m_notification(notification_type::notification_to_string(notification))
    , m_data(std::move(data))
  {
  }
  virtual ~NotificationData() = default;
  NotificationData(NotificationData const&) = default;
  NotificationData& operator=(NotificationData const&) = default;
  NotificationData(NotificationData&&) = default;
  NotificationData& operator=(NotificationData&&) = default;
 
  DUNE_DAQ_SERIALIZE(NotificationData, m_source_id, m_target_id, m_notification, m_data);
};
 
class NotificationInterface
{
 
public:
  explicit NotificationInterface(std::string connection_prefix = "snbmodules",
                                 int timeout_send = 10,
                                 int timeout_receive = 100)
    : m_timeout_send(timeout_send)
    , m_timeout_receive(timeout_receive)
    , m_connection_prefix(std::move(connection_prefix))
  {
  }
 
  NotificationInterface(std::vector<std::string> bk_conn,
                        std::set<std::string> client_conn,
                        const std::string& connection_prefix = "snbmodules",
                        int timeout_send = 10,
                        int timeout_receive = 100)
    : m_timeout_send(timeout_send)
    , m_timeout_receive(timeout_receive)
    , m_connection_prefix(connection_prefix)
  {
    m_bookkeepers_conn = std::move(bk_conn);
    TLOG() << "CCC " << __LINE__ << " client_conn size=" << client_conn.size();
    m_clients_conn = std::move(client_conn);
  }
  virtual ~NotificationInterface() = default;
 
  std::optional<NotificationData> listen_for_notification(const std::string& id,
                                                          const std::string& expected_from = "",
                                                          int timeout = -1,
                                                          int tries = -1);
 
  bool send_notification(const notification_type::e_notification_type& notif,
                         const std::string& src,
                         const std::string& dst,
                         const std::string& id_conn,
                         const std::string& data = "",
                         int tries = -1);
 
  virtual bool action_on_receive_notification(NotificationData notif) = 0;
 
  void lookups_connections()
  {
 
    iomanager::ConnectionId id = { m_connection_prefix + ".*bookkeeper.*", "notification_t", "" };
    TLOG() << "LAB " << __LINE__ << " " << id.uid << " " << id.data_type << " " << id.session;
 
    try {
      iomanager::ConnectionResponse result = IOManagerWrapper::get().lookups_connection(id, false);
      for (const auto& conn : result.connections) {
        TLOG() << "CCC " << __LINE__ << " adding bookkeeper conn.uid=" << conn.uid;
        m_bookkeepers_conn.push_back(conn.uid);
      }
    } catch (...) // Ignore if no connection found
    {
      TLOG() << "LAB " << __LINE__;
    }
    try {
      id = { m_connection_prefix + ".*transfer.*", "notification_t", "" };
      TLOG() << "LAB " << __LINE__ << " " << id.uid << " " << id.data_type << " " << id.session;
      iomanager::ConnectionResponse result = IOManagerWrapper::get().lookups_connection(id, false);
      for (const auto& conn : result.connections) {
        TLOG() << "CCC " << __LINE__ << " adding transfer conn.uid=" << conn.uid;
        m_clients_conn.insert(conn.uid);
      }
    } catch (...) // Ignore if no connection found
    {
      TLOG() << "LAB " << __LINE__;
    }
  }
 
#if 0
            virtual inline void add_connection(const IPFormat &ip, const std::string &connection_name, const std::string &data_type, bool is_client)
            {
                IOManagerWrapper::get().add_connection(ip, connection_name, data_type);
 
                if (is_client)
        {
          TLOG() << "CCC " << __LINE__ << " adding connection_name=" << connection_name;
          m_clients_conn.insert(connection_name);
                }
                else
                {
                    m_bookkeepers_conn.push_back(connection_name);
                }
            }
#endif
 
  // TEMP_REMOVED//void init_connection_interface(const std::string &session_name = "SNBMODULES", bool
  // use_connectivity_service = false, const IPFormat &ip = IPFormat("localhost", 5000)) TEMP_REMOVED//{
  // TEMP_REMOVED//IOManagerWrapper::get().init_connection_interface(session_name, use_connectivity_service, ip);
  // TEMP_REMOVED//}
 
  // Getters
  inline const std::vector<std::string>& get_bookkeepers_conn() const { return m_bookkeepers_conn; }
  inline const std::set<std::string>& get_clients_conn() const { return m_clients_conn; }
 
private:
  std::vector<std::string> m_bookkeepers_conn;
  std::set<std::string> m_clients_conn;
 
  int m_timeout_send = 10;
  int m_timeout_receive = 100;
  std::string m_connection_prefix = "snbmodules";
 
  int m_max_tries = 3;
};
 
} // namespace snbmodules
 
// Must be in dunedaq namespace only
DUNE_DAQ_SERIALIZABLE(snbmodules::NotificationData, "notification_t");
 
} // namespace dunedaq
 
#endif // SNBMODULES_INCLUDE_SNBMODULES_NOTIFICATION_INTERFACE_HPP_