Line data    Source code

       1              : /**
       2              :  * @brief test application for the opmon publisher
       3              :  */
       4              : 
       5              : #include <chrono>
       6              : #include <future>
       7              : #include <thread>
       8              : 
       9              : #include "opmonlib/Utils.hpp"
      10              : #include "opmonlib/opmon/test.pb.h"
      11              : #include <kafkaopmon/OpMonPublisher.hpp>
      12              : 
      13              : #include <boost/program_options.hpp>
      14              : 
      15              : using nlohmann::json;
      16              : namespace po = boost::program_options;
      17              : 
      18              : using namespace dunedaq::kafkaopmon;
      19              : using namespace dunedaq::opmonlib;
      20              : 
      21              : int
      22            0 : main(int argc, char const* argv[])
      23              : {
      24            0 :   po::options_description desc("Allowed options");
      25            0 :   desc.add_options()("help,h", "produce help message")(
      26            0 :     "bootstrap", po::value<std::string>()->default_value("monkafka.cern.ch:30092"), "kafka bootstrap server")(
      27            0 :     "topic,t", po::value<std::string>(), "Optional specification of a topic")(
      28            0 :     "n_threads,n", po::value<unsigned int>()->default_value(10), "Number of threads used for test");
      29              : 
      30            0 :   po::variables_map input_map;
      31            0 :   po::store(po::parse_command_line(argc, argv, desc), input_map);
      32            0 :   po::notify(input_map);
      33              : 
      34            0 :   if (input_map.count("help")) {
      35            0 :     std::cout << desc << std::endl;
      36              :     return 0;
      37              :   }
      38              : 
      39            0 :   json conf;
      40            0 :   conf["bootstrap"] = input_map["bootstrap"].as<std::string>();
      41            0 :   conf["client_id"] = "opmon_publisher_test";
      42            0 :   if (input_map.count("topic")) {
      43            0 :     conf["default_topic"] = input_map["topic"].as<std::string>();
      44              :   }
      45              : 
      46            0 :   OpMonPublisher p(conf);
      47              : 
      48            0 :   auto pub_func = [&](int i) {
      49            0 :     dunedaq::opmon::OpMonId id;
      50            0 :     id += "test";
      51            0 :     id += "app";
      52            0 :     id += "thread_" + std::to_string(i);
      53            0 :     for (auto j = 0; j < 20; ++j) {
      54            0 :       dunedaq::opmon::TestInfo ti;
      55            0 :       ti.set_int_example(j * 1000 + i);
      56            0 :       ti.set_string_example("test");
      57            0 :       auto e = to_entry(ti, {});
      58            0 :       *e.mutable_origin() = id;
      59            0 :       p.publish(std::move(e));
      60            0 :     }
      61            0 :   };
      62              : 
      63            0 :   auto n = input_map["n_threads"].as<unsigned int>();
      64            0 :   std::vector<std::future<void>> threads(n);
      65              : 
      66            0 :   for (size_t i = 0; i < n; ++i) {
      67            0 :     threads[i] = async(std::launch::async, pub_func, i);
      68              :   }
      69              : 
      70            0 :   for (auto& t : threads) {
      71            0 :     t.get();
      72              :   }
      73              : 
      74            0 :   std::this_thread::sleep_for(std::chrono::seconds(2));
      75              : 
      76            0 :   return 0;
      77            0 : }