Sender.hpp

Go to the documentation of this file.

 
#ifndef IPM_INCLUDE_IPM_SENDER_HPP_
#define IPM_INCLUDE_IPM_SENDER_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, KnownStateForbidsSend, "Sender not in a state to send data", )
ERS_DECLARE_ISSUE(ipm,
                  NullPointerPassedToSend,
                  connection_name << ": An null pointer to memory was passed to Sender::send",
                  ((std::string)connection_name))
ERS_DECLARE_ISSUE(ipm,
                  SendTimeoutExpired,
                  connection_name << ": Unable to send 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_SENDER(klass)                                                                                  \
  EXTERN_C_FUNC_DECLARE_START                                                                                          \
  std::shared_ptr<dunedaq::ipm::Sender> make()                                                                         \
  {                                                                                                                    \
    return std::shared_ptr<dunedaq::ipm::Sender>(new klass());                                                         \
  }                                                                                                                    \
  }
 
namespace dunedaq::ipm {
 
class Sender : public opmonlib::MonitorableObject
{
 
public:
  struct ConnectionInfo
  {
    std::string connection_name{ "" };
    std::string connection_string{ "inproc://default" };
    int capacity{ 0 };
  };
  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;
 
  Sender() = default;
  virtual ~Sender() = default;
 
  virtual std::string connect_for_sends(const ConnectionInfo& connection_info) = 0;
 
  virtual bool can_send() const noexcept = 0;
 
  // send() will perform some universally-desirable checks before calling user-implemented send_()
  // -Throws KnownStateForbidsSend if can_send() == false
  // -Throws NullPointerPassedToSend if message is a null pointer
  // -If message_size == 0, function is a no-op
 
  bool send(const void* message,
            message_size_t message_size,
            const duration_t& timeout,
            std::string const& metadata = "",
            bool no_tmoexcept_mode = false);
 
  Sender(const Sender&) = delete;
  Sender& operator=(const Sender&) = delete;
 
  Sender(Sender&&) = delete;
  Sender& operator=(Sender&&) = delete;
 
protected:
  ConnectionInfo m_connection_info;
  void generate_opmon_data() override;
 
  virtual bool send_(const void* message,
                     message_size_t N,
                     const duration_t& timeout,
                     std::string const& metadata,
                     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<Sender>
make_ipm_sender(std::string const& plugin_name)
{
  static cet::BasicPluginFactory bpf("duneIPM", "make");
  return bpf.makePlugin<std::shared_ptr<Sender>>(plugin_name);
}
 
} // namespace dunedaq::ipm
 
#endif // IPM_INCLUDE_IPM_SENDER_HPP_