OpMonPublisher.hpp

Go to the documentation of this file.

 
#ifndef KAFKAOPMON_INCLUDE_KAFKAOPMON_OPMONPUBLISHER_HPP_
#define KAFKAOPMON_INCLUDE_KAFKAOPMON_OPMONPUBLISHER_HPP_
 
#include <ers/ers.hpp>
#include <librdkafka/rdkafkacpp.h>
#include <nlohmann/json.hpp>
 
#include <memory>
#include <string>
 
#include "opmonlib/Utils.hpp"
#include "opmonlib/opmon_entry.pb.h"
 
namespace dunedaq {
 
ERS_DECLARE_ISSUE(kafkaopmon,
                  MissingParameter,
                  "No " << parameter << " in " << conf,
                  ((std::string)parameter)((std::string)conf))
 
ERS_DECLARE_ISSUE(kafkaopmon,
                  FailedConfiguration,
                  "Invalid " << parameter << ", cause: " << reason,
                  ((std::string)parameter)((std::string)reason))
 
ERS_DECLARE_ISSUE(kafkaopmon,
                  FailedProducerCreation,
                  "Failed creation of a Kafka producer, cause: " << reason,
                  ((std::string)reason))
 
ERS_DECLARE_ISSUE(kafkaopmon,
                  FailedProduce,
                  "Failed produce of message with key " << key << ", cause: " << reason,
                  ((std::string)key)((std::string)reason))
 
ERS_DECLARE_ISSUE(kafkaopmon,
                  TimeoutReachedWhileFlushing,
                  "Publisher destroyed before all messages were completed, timeout: " << timeout << " ms",
                  ((int)timeout))
 
} // dunedaq namespace
 
namespace dunedaq::kafkaopmon {
 
class OpMonPublisher
{
 
public:
  OpMonPublisher(const nlohmann::json& conf);
 
  OpMonPublisher() = delete;
  OpMonPublisher(const OpMonPublisher&) = delete;
  OpMonPublisher& operator=(const OpMonPublisher&) = delete;
  OpMonPublisher(OpMonPublisher&&) = delete;
  OpMonPublisher& operator=(OpMonPublisher&&) = delete;
 
  ~OpMonPublisher();
 
  void publish(dunedaq::opmon::OpMonEntry&&) const;
 
protected:
  std::string extract_topic(const dunedaq::opmon::OpMonEntry&) const noexcept { return m_default_topic; }
  std::string extract_key(const dunedaq::opmon::OpMonEntry& e) const noexcept
  {
    return dunedaq::opmonlib::to_string(e.origin()) + '/' + e.measurement();
  }
 
private:
  std::unique_ptr<RdKafka::Producer> m_producer;
  std::string m_default_topic = "monitoring.opmon_stream";
};
 
} // namespace dunedaq::kafkaopmon
 
#endif // KAFKAOPMON_INCLUDE_KAFKAOPMON_OPMONPUBLISHER_HPP_