DataSubscriberModel.hpp

Go to the documentation of this file.

#ifndef DATAHANDLINGLIBS_SRC_SOURCEMODEL_HPP_
#define DATAHANDLINGLIBS_SRC_SOURCEMODEL_HPP_
 
#include "datahandlinglibs/concepts/SourceConcept.hpp"
#include "datahandlinglibs/opmon/datahandling_info.pb.h"
 
#include "iomanager/IOManager.hpp"
#include "iomanager/Receiver.hpp"
#include "iomanager/Sender.hpp"
#include "logging/Logging.hpp"
// #include "utilities/ReusableThread.hpp"
 
#include "confmodel/Connection.hpp"
 
// #include <folly/ProducerConsumerQueue.h>
// #include <nlohmann/json.hpp>
 
// #include <atomic>
// #include <memory>
// #include <mutex>
#include <functional>
 
namespace dunedaq::datahandlinglibs {
 
template<class PayloadType>
class DataSubscriberModel : public SourceConcept
{
public:
  using inherited = SourceConcept;

  DataSubscriberModel()
    : SourceConcept()
  {
  }
  ~DataSubscriberModel() {}
 
  void init(const confmodel::DaqModule* cfg) override
  {
    if (cfg->get_outputs().size() != 1) {
      throw datahandlinglibs::InitializationError(ERS_HERE, "Only 1 output supported for subscribers");
    }
    m_data_sender = get_iom_sender<PayloadType>(cfg->get_outputs()[0]->UID());
 
    if (cfg->get_inputs().size() != 1) {
      throw datahandlinglibs::InitializationError(ERS_HERE, "Only 1 input supported for subscribers");
    }
    m_data_receiver = get_iom_receiver<PayloadType>(cfg->get_inputs()[0]->UID());
  }
 
  void start()
  {
    m_packets = 0;
    m_sum_packets = 0;
    m_dropped_packets = 0;
    m_data_receiver->add_callback(std::bind(&DataSubscriberModel::handle_payload, this, std::placeholders::_1));
  }
 
  void stop() { m_data_receiver->remove_callback(); }
 
  bool handle_payload(PayloadType& message) // NOLINT(build/unsigned)
  {
    ++m_packets;
    ++m_sum_packets;
    if (!m_data_sender->try_send(std::move(message), iomanager::Sender::s_no_block)) {
      ++m_dropped_packets;
    }
    return true;
  }
 
protected:
  virtual void generate_opmon_data() override
  {
    opmon::DataSourceInfo info;
    info.set_num_packets(m_packets.exchange(0));
    info.set_sum_packets(m_sum_packets);
    info.set_num_dropped_packets(m_dropped_packets.exchange(0));
 
    this->publish(std::move(info));
  }
 
private:
  using source_t = dunedaq::iomanager::ReceiverConcept<PayloadType>;
  std::shared_ptr<source_t> m_data_receiver;
 
  using sink_t = dunedaq::iomanager::SenderConcept<PayloadType>;
  std::shared_ptr<sink_t> m_data_sender;
 
  std::atomic<uint64_t> m_packets{ 0 };
  std::atomic<uint64_t> m_sum_packets{ 0 };
  std::atomic<uint64_t> m_dropped_packets{ 0 };
};
 
} // namespace dunedaq::datahandlinglibs
 
#endif // DATAHANDLINGLIBS_SRC_SOURCEMODEL_HPP_