Function ClientThread

Server/src/client_thread.cpp:74–289

* Client thread function * * Thread function that is called when starting a new thread. * The DB/mysql is initialized for each thread. * * @param [in] arg Pointer to the BMPServer ClientInfo */

 * @param [in]  arg     Pointer to the BMPServer ClientInfo
 */
void *ClientThread(void *arg) {
    // Setup the args
    ThreadMgmt *thr = static_cast<ThreadMgmt *>(arg);
    Logger *logger = thr->log;

    // Setup the client thread info struct
    ClientThreadInfo cInfo;
    cInfo.mbus = NULL;
    cInfo.client = &thr->client;
    cInfo.log = thr->log;
    cInfo.closing = false;

    int sock_fds[2];
    pollfd pfd;
    unsigned char *sock_buf = NULL;

    /*
     * Setup the cleanup routine for when the thread is canceled.
     *  A thread is only canceled if openbmpd is terminated.
     */
    pthread_cleanup_push(ClientThread_cancel, &cInfo);

    try {
        // connect to message bus
        cInfo.mbus = new msgBus_kafka(logger, thr->cfg, thr->cfg->c_hash_id);

        if (thr->cfg->debug_msgbus)
            cInfo.mbus->enableDebug();

        BMPReader rBMP(logger, thr->cfg);
        LOG_INFO("Thread started to monitor BMP from router %s using socket %d buffer in bytes = %u",
                cInfo.client->c_ip, cInfo.client->c_sock, thr->cfg->bmp_buffer_size);

        // Buffer client socket using pipe
        socketpair(PF_LOCAL, SOCK_STREAM, 0, sock_fds);
        cInfo.bmp_write_end_sock = sock_fds[1];
        cInfo.client->pipe_sock = sock_fds[0];

        /*
         * Create and start the reader thread to monitor the pipe fd (read end)
         */
        bool bmp_run = true;
        //cInfo.bmp_reader_thread = new std::thread([&] {rBMP.readerThreadLoop(bmp_run,cInfo.client,
        cInfo.bmp_reader_thread = new std::thread(&BMPReader::readerThreadLoop, &rBMP, std::ref(bmp_run), cInfo.client,
                                                                             (MsgBusInterface *)cInfo.mbus );

        // Variables to handle circular buffer
        sock_buf = new unsigned char[thr->cfg->bmp_buffer_size];
        int bytes_read = 0;
        int write_buf_pos = 0;
        int read_buf_pos = 0;
        bool wrap_state = false;
        unsigned char *sock_buf_read_ptr = sock_buf;
        unsigned char *sock_buf_write_ptr = sock_buf;

        /*
         * monitor and buffer the client socket
         */