Line data    Source code

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