ProtoBufStream.cpp

Go to the documentation of this file.

 
#include "ProtoBufStream.hpp"
#include <boost/crc.hpp>
#include <chrono>
#include <ers/StreamFactory.hpp>
#include <iostream>
#include <string>
#include <vector>
 
ERS_REGISTER_OUTPUT_STREAM(erskafka::ProtoBufStream, "protobufstream", param)
 

namespace erskafka {
erskafka::ProtoBufStream::ProtoBufStream(const std::string& param)
  : m_session("Uknown")
  , m_application("Uknown")
{
 
  nlohmann::json conf;
  conf["bootstrap"] = param;
 
  m_publisher = std::make_unique<dunedaq::erskafka::ERSPublisher>(conf);
 
  if (auto env_p = std::getenv("DUNEDAQ_SESSION"))
    m_session = env_p;
 
  if (auto app_p = std::getenv("DUNEDAQ_APPLICATION_NAME"))
    m_application = app_p;
}

void
erskafka::ProtoBufStream::write(const ers::Issue& issue)
{
  try {
 
    auto schema = to_schema_chain(issue);
    schema.set_session(m_session);
    schema.set_application(m_application);
    m_publisher->publish(std::move(schema));
 
  } catch (const std::exception& e) {
    std::cerr << "Producer error : " << e.what() << '\n';
  }
}
} // namespace erskafka