* Collector Update Message * * \param [in] kafka Pointer to kafka instance * \param [in] cfg Reference to configuration * \param [in] code reason code for the update */
| 366 | * \param [in] code reason code for the update |
| 367 | */ |
| 368 | void collector_update_msg(msgBus_kafka *kafka, Config &cfg, |
| 369 | MsgBusInterface::collector_action_code code) { |
| 370 | |
| 371 | MsgBusInterface::obj_collector oc; |
| 372 | |
| 373 | snprintf(oc.admin_id, sizeof(oc.admin_id), "%s", cfg.admin_id); |
| 374 | |
| 375 | oc.router_count = thr_list.size(); |
| 376 | |
| 377 | string router_ips; |
| 378 | for (int i=0; i < thr_list.size(); i++) { |
| 379 | //MsgBusInterface::hash_toStr(thr_list.at(i)->client.hash_id, hash_str); |
| 380 | if (router_ips.size() > 0) |
| 381 | router_ips.append(", "); |
| 382 | |
| 383 | router_ips.append(thr_list.at(i)->client.c_ip); |
| 384 | } |
| 385 | |
| 386 | snprintf(oc.routers, sizeof(oc.routers), "%s", router_ips.c_str()); |
| 387 | |
| 388 | timeval tv; |
| 389 | gettimeofday(&tv, NULL); |
| 390 | oc.timestamp_secs = tv.tv_sec; |
| 391 | oc.timestamp_us = tv.tv_usec; |
| 392 | |
| 393 | kafka->update_Collector(oc, code); |
| 394 | } |
| 395 | |
| 396 | /** |
| 397 | * Run Server loop |
no test coverage detected