ex_kafka.c

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <locale.h>
#include <libgen.h>
#include <errno.h>
 
#include "nilorea/n_log.h"
#include "nilorea/n_network.h"
#include "nilorea/n_kafka.h"
#include "cJSON.h"
 
#include "rdkafka.h"
 
#define OK 0
#define ERROR -1
#define NB_TEST_EVENTS 10
 
char *config_file = NULL,
     *event_string = NULL,
     *event_file = NULL,
     *event_log_file = NULL,
     *log_prefix = NULL;
 
int log_level = LOG_ERR, /* default log level */
    getoptret = 0,       /* getopt return value */
    KAFKA_MODE = -1,
    PRODUCER_MODE = 0,
    run = 1;
 
// help func
void usage(void) {
    // TODO: add -F unprocessed event file
    fprintf(stderr,
            "Syntax is: ex_kafka -v -c config_file [-s event or -f eventfile] -o event_log_file -V LOGLEVEL\n"
            "   -v version: print version and exit\n"
            "   -c config_file: [required] Kproducer config file\n"
            "   -s : string of the event to send\n"
            "   -f : file containing the event to send\n"
            "   -C : start a consumer (default output received in terminal)"
            "   -P : start a producer and produce event"
            "   -o : optionnal, set a log file instead of default (stderr/stdout)\n"
            "   -p : optionnal, set a log prefix\n"
            "   -V verbosity: specify a log level for console output\n"
            "                 Supported: LOG_ EMERG,ALERT,CRIT,ERR,WARNING,NOTICE,INFO,DEBUG\n");
}
 
// stop handler
static void stop(int sig) {
    (void)sig;
    run = 0;
}
 
int main(int argc, char* argv[]) {
    /* Signal handler for clean shutdown */
    signal(SIGINT, stop);
 
    /* temporary header structure */
    rd_kafka_headers_t* headers = NULL;
 
    /* Analysing arguments */
    while ((getoptret = getopt(argc, argv, "vhCPH:c:s:f:V:o:p:")) != -1) {
        switch (getoptret) {
            case 'v':
                fprintf(stderr, "     Version compiled on %s at %s\n", __DATE__, __TIME__);
                exit(TRUE);
            case 'h':
                usage();
                exit(0);
                break;
            case 'C':
                if (KAFKA_MODE == -1) {
                    KAFKA_MODE = RD_KAFKA_CONSUMER;
                } else {
                    fprintf(stderr, "-C and -P can not be used at the ame time!");
                    exit(TRUE);
                }
                break;
            case 'P':
                if (KAFKA_MODE == -1) {
                    KAFKA_MODE = RD_KAFKA_PRODUCER;
                } else {
                    fprintf(stderr, "-C and -P can not be used at the ame time!");
                    exit(TRUE);
                }
                break;
            case 'H': {
                char *name = NULL, *val = NULL;
                size_t name_sz = -1;
 
                name = optarg;
                val = strchr(name, '=');
                if (val) {
                    name_sz = (size_t)(val - name);
                    val++; /* past the '=' */
                }
 
                if (!headers)
                    headers = rd_kafka_headers_new(16);
 
                int err = rd_kafka_header_add(headers, name, name_sz, val, -1);
                if (err) {
                    fprintf(stderr,
                            "%% Failed to add header %s: %s\n",
                            name, rd_kafka_err2str(err));
                    exit(1);
                }
            } break;
            case 'c':
                config_file = local_strdup(optarg);
                break;
            case 's':
                Malloc(event_string, char, strlen(optarg) + 1);
                strcpy(event_string, optarg);
                break;
            case 'f':
                event_file = local_strdup(optarg);
                break;
            case 'o':
                event_log_file = local_strdup(optarg);
                break;
            case 'p':
                log_prefix = local_strdup(optarg);
                break;
            case 'V':
                if (!strncmp("LOG_NULL", optarg, 8)) {
                    log_level = LOG_NULL;
                } else if (!strncmp("LOG_NOTICE", optarg, 10)) {
                    log_level = LOG_NOTICE;
                } else if (!strncmp("LOG_INFO", optarg, 8)) {
                    log_level = LOG_INFO;
                } else if (!strncmp("LOG_ERR", optarg, 7)) {
                    log_level = LOG_ERR;
                } else if (!strncmp("LOG_DEBUG", optarg, 9)) {
                    log_level = LOG_DEBUG;
                } else {
                    fprintf(stderr, "%s n'est pas un niveau de log valide.\n", optarg);
                    exit(-1);
                }
                break;
            case '?':
                if (optopt == 'c' || optopt == 's' || optopt == 'f') {
                    fprintf(stderr, "Option -%c need a parameter\n", optopt);
                    exit(FALSE);
                }
                FALL_THROUGH;
            default:
                usage();
                exit(-1);
                break;
        }
    }
 
    if (KAFKA_MODE == -1) {
        n_log(LOG_ERR, "consumer (-C) or producer (-P) mode is not defined !", log_prefix);
        exit(1);
    }
 
    if (!log_prefix) {
        log_prefix = strdup("");
    }
 
    set_log_level(log_level);
    if (event_log_file) {
        int log_file_ret = set_log_file(event_log_file);
        n_log(LOG_DEBUG, "%s log to file: %s , %d  , %p", log_prefix, event_log_file, log_file_ret, get_log_file());
    }
 
    /* testing parameters */
    if (!config_file) {
        n_log(LOG_ERR, "%s parameter config_file needs to be set !", log_prefix);
        exit(1);
    }
 
    if (KAFKA_MODE == RD_KAFKA_PRODUCER) {
        if (!event_string && !event_file) {
            n_log(LOG_ERR, "%s one of (event_string|event_file) needs to be set !", log_prefix);
            exit(1);
        }
    }
 
    if (event_string && event_file) {
        n_log(LOG_ERR, "%s do not define event_string AND event_file, only one needs to be set !", log_prefix);
        exit(1);
    }
 
    // load kafka config file
    unsigned int exit_code = 0;
    N_KAFKA* kafka_handle = n_kafka_load_config(config_file, KAFKA_MODE);
    __n_assert(kafka_handle, n_log(LOG_ERR, "kafka handke is NULL !!"); exit(1));
 
    n_kafka_start_pooling_thread(kafka_handle);
 
    size_t nb_queued = 0;
    size_t nb_waiting = 0;
    size_t nb_error = 0;
    int poll_status = 1;
    N_KAFKA_EVENT* event = NULL;
 
    if (KAFKA_MODE == RD_KAFKA_PRODUCER) {
        // create a new kafka event from a string or from a file
        if (event_string) {
            event = n_kafka_new_event_from_char(event_string, strlen(event_string), kafka_handle->schema_id);
        }
        if (event_file) {
            event = n_kafka_new_event_from_file(event_file, kafka_handle->schema_id);
        }
        // set headers if any
        if (headers) {
            event->rd_kafka_headers = rd_kafka_headers_copy(headers);
            // clean them if no more needed
            rd_kafka_headers_destroy(headers);
        }
        // produce the event, API is charging itself of destroying it
        if (n_kafka_produce(kafka_handle, event) == FALSE) {
            n_log(LOG_ERR, "n_kafka_produce returned an error for event %p", event);
        } else {
            n_log(LOG_INFO, "n_kafka_produce returned OK for event %p", event);
        }
    }
 
    // loop on pool
    do {
        poll_status = n_kafka_get_status(kafka_handle, &nb_queued, &nb_waiting, &nb_error);
        n_log(LOG_DEBUG, "polling kafka handle, status: %d, %d in queue, %d waiting for ack, %d on error", poll_status, nb_queued, nb_waiting, nb_error);
 
        // if we were waiting only for producing elemens we could use a test like this to break out of the loop
        if (KAFKA_MODE == RD_KAFKA_PRODUCER) {
            usleep(30000);
            if (nb_queued == 0 && nb_waiting == 0 && nb_error == 0)
                break;
        } else {
            event = n_kafka_get_event(kafka_handle);
            if (event) {
                if (kafka_handle->schema_id != -1)
                    n_log(LOG_INFO, "received event schema id %d string:\n%s", event->schema_id, _str(event->event_string->data + 4));
                else
                    n_log(LOG_INFO, "received event string:\n%s", event->event_string->data);
 
                list_foreach(node, event->received_headers) {
                    N_STR* header = (N_STR*)node->ptr;
                    n_log(LOG_INFO, "headers: %s", _nstr(header));
                }
                n_kafka_event_destroy(&event);
            } else {
                usleep(30000);
            }
        }
    } while (run && poll_status > 0);
 
    n_log(LOG_INFO, "kafka_handle: %d queued, %d waiting ack, %d on error", nb_queued, nb_waiting, nb_error);
    if (nb_error > 0 || nb_waiting > 0) {
        n_log(LOG_ERR, "kafka_handle: %d events are still waiting for ack, and %d are on error !", nb_waiting, nb_error);
        n_kafka_dump_unprocessed(kafka_handle, "DATAS/kafka/unprocessed");
    }
 
    // log unprocessed events
    list_foreach(node, kafka_handle->received_events) {
        if (node) {
            if (kafka_handle->schema_id != -1)
                n_log(LOG_INFO, "[unprocessed]received event schema id %d string:\n%s", event->schema_id, event->event_string->data + 4);
            else
                n_log(LOG_INFO, "[unprocessed]received event string:\n%s", event->event_string->data);
        }
    }
 
    // closing kafka handle
    n_kafka_delete(kafka_handle);
 
    exit(exit_code);
}