Line data    Source code

       1              : /**
       2              :  * @file streamOpMonFacility.cpp kafkaopmon 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 "streamOpMonFacility.hpp"
      10              : 
      11              : using json = nlohmann::json;
      12              : 
      13              : namespace dunedaq::kafkaopmon { // namespace dunedaq
      14              : 
      15            2 : streamOpMonFacility::streamOpMonFacility(std::string uri, dunedaq::opmonlib::OptionalOrigin o)
      16            2 :   : dunedaq::opmonlib::OpMonFacility(uri)
      17              : {
      18              : 
      19            2 :   std::regex uri_re(R"(([a-zA-Z]+):\/\/([^:\/?#\s]+):(\d+)\/([^:\/?#\s]+))");
      20              :   //* 1st Capturing Group `([a-zA-Z])`: Matches protocol
      21              :   //* 2nd Capturing Group `([^:\/?#\s])+`: Matches hostname
      22              :   //* 3rd Capturing Group `(\d)`: Matches port
      23              :   //* 4th Capturing Group `([^\/?#]+)?`: Matches kafka topic
      24              : 
      25            2 :   std::smatch uri_match;
      26            2 :   if (!std::regex_match(uri, uri_match, uri_re)) {
      27            1 :     throw WrongURI(ERS_HERE, uri);
      28              :   }
      29              : 
      30            1 :   json config;
      31            1 :   std::string bootstrap = uri_match[2];
      32            1 :   bootstrap += ':';
      33            1 :   bootstrap += uri_match[3];
      34            1 :   config["bootstrap"] = bootstrap;
      35              : 
      36              :   // optionally set the ID of the application
      37              :   // But really this is temporary, and in the future we should avoid env variables
      38            1 :   if (o) {
      39            0 :     config["cliend_id"] = dunedaq::opmonlib::to_string(o.value());
      40              :   }
      41              : 
      42            1 :   config["default_topic"] = uri_match[4];
      43              : 
      44            1 :   m_publisher.reset(new OpMonPublisher(config));
      45            5 : }
      46              : 
      47              : } // namespace dunedaq::kafkaopmon
      48              : 
      49            4 : DEFINE_DUNE_OPMON_FACILITY(dunedaq::kafkaopmon::streamOpMonFacility)