Line data    Source code

       1              : /**
       2              :  * @file ERSPublisher.cpp ERSPublusher Class Implementation
       3              :  *
       4              :  * This is part of the DUNE DAQ Software Suite, copyright 2020.
       5              :  * Licensing/copyright details are in the COPYING file that you should have
       6              :  * received with this code.
       7              :  */
       8              : 
       9              : #include "erskafka/ERSPublisher.hpp"
      10              : 
      11              : #include <iostream>
      12              : 
      13              : #include <stdexcept>
      14              : 
      15              : using namespace dunedaq::erskafka;
      16              : 
      17            0 : ERSPublisher::ERSPublisher(const nlohmann::json& conf)
      18              : {
      19              : 
      20            0 :   RdKafka::Conf* k_conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
      21            0 :   std::string errstr;
      22              : 
      23            0 :   auto it = conf.find("bootstrap");
      24            0 :   if (it == conf.end()) {
      25            0 :     std::cerr << "Missing bootstrap from json file";
      26            0 :     throw std::runtime_error("Missing bootstrap from json file");
      27              :   }
      28              : 
      29            0 :   k_conf->set("bootstrap.servers", *it, errstr);
      30            0 :   if (errstr != "") {
      31            0 :     throw std::runtime_error(errstr);
      32              :   }
      33              : 
      34            0 :   std::string client_id;
      35            0 :   it = conf.find("cliend_id");
      36            0 :   if (it != conf.end())
      37            0 :     client_id = *it;
      38            0 :   else if (const char* env_p = std::getenv("DUNEDAQ_APPLICATION_NAME"))
      39            0 :     client_id = env_p;
      40              :   else
      41            0 :     client_id = "erskafkaproducerdefault";
      42              : 
      43            0 :   k_conf->set("client.id", client_id, errstr);
      44            0 :   if (errstr != "") {
      45            0 :     throw std::runtime_error(errstr);
      46              :   }
      47              : 
      48              :   // Create producer instance
      49            0 :   m_producer.reset(RdKafka::Producer::create(k_conf, errstr));
      50              : 
      51            0 :   if (errstr != "") {
      52            0 :     throw std::runtime_error(errstr);
      53              :   }
      54              : 
      55            0 :   it = conf.find("default_topic");
      56            0 :   if (it != conf.end())
      57            0 :     m_default_topic = *it;
      58            0 : }
      59              : 
      60              : bool
      61            0 : ERSPublisher::publish(ersschema::IssueChain&& issue) const
      62              : {
      63              : 
      64            0 :   std::string binary;
      65            0 :   issue.SerializeToString(&binary);
      66              : 
      67              :   // get the topic
      68            0 :   auto topic = ERSPublisher::topic(issue);
      69              : 
      70            0 :   auto key = ERSPublisher::key(issue);
      71              : 
      72              :   //      RdKafka::Producer::RK_MSG_COPY to be investigated
      73            0 :   RdKafka::ErrorCode err = m_producer->produce(topic,
      74              :                                                RdKafka::Topic::PARTITION_UA,
      75              :                                                RdKafka::Producer::RK_MSG_COPY,
      76            0 :                                                const_cast<char*>(binary.c_str()),
      77              :                                                binary.size(),
      78            0 :                                                key.c_str(),
      79              :                                                key.size(),
      80              :                                                0,
      81              :                                                nullptr);
      82            0 :   if (err != RdKafka::ERR_NO_ERROR) {
      83            0 :     return false;
      84              :   }
      85              : 
      86              :   return true;
      87            0 : }