Line data    Source code

       1              : /**
       2              :  * @file ProtoBufStream.cpp ProtoBufStream class implementation
       3              :  *
       4              :  * This is part of the DUNE DAQ software, copyright 2020.
       5              :  * Licensing/copyright details are in the COPYING file that you should have
       6              :  * received with this code.
       7              :  */
       8              : 
       9              : #include "ProtoBufStream.hpp"
      10              : #include <boost/crc.hpp>
      11              : #include <chrono>
      12              : #include <ers/StreamFactory.hpp>
      13              : #include <iostream>
      14              : #include <string>
      15              : #include <vector>
      16              : 
      17            0 : ERS_REGISTER_OUTPUT_STREAM(erskafka::ProtoBufStream, "protobufstream", param)
      18              : 
      19              : /** Constructor that creates a new instance of the ersstream stream with the given configuration.
      20              :  * \param format elastic search connection string.
      21              :  */
      22              : namespace erskafka {
      23            0 : erskafka::ProtoBufStream::ProtoBufStream(const std::string& param)
      24            0 :   : m_session("Uknown")
      25            0 :   , m_application("Uknown")
      26              : {
      27              : 
      28            0 :   nlohmann::json conf;
      29            0 :   conf["bootstrap"] = param;
      30              : 
      31            0 :   m_publisher = std::make_unique<dunedaq::erskafka::ERSPublisher>(conf);
      32              : 
      33            0 :   if (auto env_p = std::getenv("DUNEDAQ_SESSION"))
      34            0 :     m_session = env_p;
      35              : 
      36            0 :   if (auto app_p = std::getenv("DUNEDAQ_APPLICATION_NAME"))
      37            0 :     m_application = app_p;
      38            0 : }
      39              : 
      40              : /** Write method
      41              :  * \param issue issue to be sent.
      42              :  */
      43              : void
      44            0 : erskafka::ProtoBufStream::write(const ers::Issue& issue)
      45              : {
      46            0 :   try {
      47              : 
      48            0 :     auto schema = to_schema_chain(issue);
      49            0 :     schema.set_session(m_session);
      50            0 :     schema.set_application(m_application);
      51            0 :     m_publisher->publish(std::move(schema));
      52              : 
      53            0 :   } catch (const std::exception& e) {
      54            0 :     std::cerr << "Producer error : " << e.what() << '\n';
      55            0 :   }
      56            0 : }
      57              : } // namespace erskafka