Function runServer

Server/src/openbmp.cpp:401–551

* Run Server loop * * \param [in] cfg Reference to the config options */

 * \param [in]  cfg    Reference to the config options
 */
void runServer(Config &cfg) {
    msgBus_kafka *kafka;
    int active_connections = 0;                 // Number of active connections/threads
    int concurrent_routers = 0;			// Number of concurrent routers
    time_t last_heartbeat_time = 0;
   
    LOG_INFO("Initializing server");

    try {
        // Define the collector hash
        MD5 hash;
        hash.update((unsigned char *)cfg.admin_id, strlen(cfg.admin_id));
        hash.finalize();

        // Save the hash
        unsigned char *hash_raw = hash.raw_digest();
        memcpy(cfg.c_hash_id, hash_raw, 16);
        delete[] hash_raw;

        // Kafka connection
        kafka = new msgBus_kafka(logger, &cfg, cfg.c_hash_id);

        // allocate and start a new bmp server
        BMPListener *bmp_svr = new BMPListener(logger, &cfg);

        collector_update_msg(kafka, cfg, MsgBusInterface::COLLECTOR_ACTION_STARTED);
        last_heartbeat_time = time(NULL);

        LOG_INFO("Ready. Waiting for connections");

        // Loop to accept new connections
        while (run) {
            /*
             * Check for any stale threads/connections
             */
             for (size_t i=0; i < thr_list.size(); i++) {

                // If thread is not running, it means it terminated, so close it out
                if (!thr_list.at(i)->running) {

                    // Join the thread to clean up
                    pthread_join(thr_list.at(i)->thr, NULL);
                    --active_connections;

                    if (!thr_list.at(i)->baselineTimeout)
                        --concurrent_routers;

                    // free the vector entry
                    delete thr_list.at(i);
                    thr_list.erase(thr_list.begin() + i);

                    collector_update_msg(kafka, cfg,
                                         MsgBusInterface::COLLECTOR_ACTION_CHANGE);

                }

		        else if (!thr_list.at(i)->baselineTimeout) {