ERSPublisher.cpp

Go to the documentation of this file.

 
#include "erskafka/ERSPublisher.hpp"
 
#include <iostream>
 
#include <stdexcept>
 
using namespace dunedaq::erskafka;
 
ERSPublisher::ERSPublisher(const nlohmann::json& conf)
{
 
  RdKafka::Conf* k_conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
  std::string errstr;
 
  auto it = conf.find("bootstrap");
  if (it == conf.end()) {
    std::cerr << "Missing bootstrap from json file";
    throw std::runtime_error("Missing bootstrap from json file");
  }
 
  k_conf->set("bootstrap.servers", *it, errstr);
  if (errstr != "") {
    throw std::runtime_error(errstr);
  }
 
  std::string client_id;
  it = conf.find("cliend_id");
  if (it != conf.end())
    client_id = *it;
  else if (const char* env_p = std::getenv("DUNEDAQ_APPLICATION_NAME"))
    client_id = env_p;
  else
    client_id = "erskafkaproducerdefault";
 
  k_conf->set("client.id", client_id, errstr);
  if (errstr != "") {
    throw std::runtime_error(errstr);
  }
 
  // Create producer instance
  m_producer.reset(RdKafka::Producer::create(k_conf, errstr));
 
  if (errstr != "") {
    throw std::runtime_error(errstr);
  }
 
  it = conf.find("default_topic");
  if (it != conf.end())
    m_default_topic = *it;
}
 
bool
ERSPublisher::publish(ersschema::IssueChain&& issue) const
{
 
  std::string binary;
  issue.SerializeToString(&binary);
 
  // get the topic
  auto topic = ERSPublisher::topic(issue);
 
  auto key = ERSPublisher::key(issue);
 
  //      RdKafka::Producer::RK_MSG_COPY to be investigated
  RdKafka::ErrorCode err = m_producer->produce(topic,
                                               RdKafka::Topic::PARTITION_UA,
                                               RdKafka::Producer::RK_MSG_COPY,
                                               const_cast<char*>(binary.c_str()),
                                               binary.size(),
                                               key.c_str(),
                                               key.size(),
                                               0,
                                               nullptr);
  if (err != RdKafka::ERR_NO_ERROR) {
    return false;
  }
 
  return true;
}