Receiver.hpp

Go to the documentation of this file.

 
#ifndef IPM_INCLUDE_IPM_RECEIVER_HPP_
#define IPM_INCLUDE_IPM_RECEIVER_HPP_
 
#include "cetlib/BasicPluginFactory.h"
#include "cetlib/compiler_macros.h"
#include "ers/Issue.hpp"
#include "logging/Logging.hpp" // NOTE: if ISSUES ARE DECLARED BEFORE include logging/Logging.hpp, TLOG_DEBUG<<issue wont work.
#include "opmonlib/MonitorableObject.hpp"
 
#include <atomic>
#include <memory>
#include <string>
#include <vector>
 
namespace dunedaq {
// Disable coverage collection LCOV_EXCL_START
ERS_DECLARE_ISSUE(ipm, KnownStateForbidsReceive, "Receiver not in a state to receive data", )
ERS_DECLARE_ISSUE(ipm,
                  UnexpectedNumberOfBytes,
                  connection_name << ": Expected " << bytes1 << " bytes in message but received " << bytes2,
                  ((std::string)connection_name)((int)bytes1)((int)bytes2)) // NOLINT
ERS_DECLARE_ISSUE(ipm,
                  ReceiveTimeoutExpired,
                  connection_name << ": Unable to receive within timeout period (timeout period was " << timeout
                                  << " milliseconds)",
                  ((std::string)connection_name)((int)timeout)) // NOLINT
// Reenable coverage collection LCOV_EXCL_STOP
} // namespace dunedaq
 
#ifndef EXTERN_C_FUNC_DECLARE_START
// NOLINTNEXTLINE(build/define_used)
#define EXTERN_C_FUNC_DECLARE_START                                                                                    \
  extern "C"                                                                                                           \
  {
#endif

// NOLINTNEXTLINE
#define DEFINE_DUNE_IPM_RECEIVER(klass)                                                                                \
  EXTERN_C_FUNC_DECLARE_START                                                                                          \
  std::shared_ptr<dunedaq::ipm::Receiver> make()                                                                       \
  {                                                                                                                    \
    return std::shared_ptr<dunedaq::ipm::Receiver>(new klass());                                                       \
  }                                                                                                                    \
  }
 
namespace dunedaq::ipm {
 
class Receiver : public opmonlib::MonitorableObject
{
 
public:
  struct ConnectionInfo
  {
    std::string connection_name{ "" };
    std::string connection_string{ "" };
    std::vector<std::string> connection_strings{};
  };
  using duration_t = std::chrono::milliseconds;
  static constexpr duration_t s_block = duration_t::max();
  static constexpr duration_t s_no_block = duration_t::zero();
 
  using message_size_t = int;
  static constexpr message_size_t s_any_size =
    0; // Since "I want 0 bytes" is pointless, "0" denotes "I don't care about the size"
 
  Receiver() = default;
  virtual ~Receiver() = default;
 
  virtual std::string connect_for_receives(const ConnectionInfo& connection_info) = 0;
 
  virtual bool can_receive() const noexcept = 0;
 
  // receive() will perform some universally-desirable checks before calling user-implemented receive_:
  // -Throws KnownStateForbidsReceive if can_receive() == false
  // -Throws UnexpectedNumberOfBytes if the "nbytes" argument isn't anysize, and the
  //  received bytes inside the function aren't the same number as nbytes
 
  struct Response
  {
    std::string metadata{ "" };
    std::vector<char> data{};
  };
 
  Response receive(const duration_t& timeout, message_size_t num_bytes = s_any_size, bool no_tmoexcept_mode = false);
  virtual bool data_pending() = 0;
 
  virtual void register_callback(std::function<void(Response&)>) = 0;
  virtual void unregister_callback() = 0;
 
  Receiver(const Receiver&) = delete;
  Receiver& operator=(const Receiver&) = delete;
 
  Receiver(Receiver&&) = delete;
  Receiver& operator=(Receiver&&) = delete;
 
protected:
  ConnectionInfo m_connection_info;
  void generate_opmon_data() override;
 
  virtual Response receive_(const duration_t& timeout, bool no_tmoexcept_mode) = 0;
 
private:
  mutable std::atomic<size_t> m_bytes = { 0 };
  mutable std::atomic<size_t> m_messages = { 0 };
};
 
inline std::shared_ptr<Receiver>
make_ipm_receiver(std::string const& plugin_name)
{
  static cet::BasicPluginFactory bpf("duneIPM", "make");
  return bpf.makePlugin<std::shared_ptr<Receiver>>(plugin_name);
}
 
} // namespace dunedaq::ipm
 
#endif // IPM_INCLUDE_IPM_RECEIVER_HPP_