n_kafka.h

Go to the documentation of this file.

 
#ifndef __N_KAFKA
#define __N_KAFKA
 
#ifdef __cplusplus
extern "C"
{
#endif
 
#include "nilorea/n_log.h"
#include "nilorea/n_network.h"
#include "nilorea/n_network.h"
#include "cJSON.h"
 
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <locale.h>
#include <libgen.h>
#include <errno.h>
#include <unistd.h>
 
#include "rdkafka.h"
 
#define N_KAFKA_EVENT_QUEUED 0
#define N_KAFKA_EVENT_WAITING_ACK 1
#define N_KAFKA_EVENT_ERROR 2
#define N_KAFKA_EVENT_OK 4
#define N_KAFKA_EVENT_CREATED 5
 
                typedef struct N_KAFKA_EVENT                    
                {
                                N_STR *event_string ;
                                N_STR *event_files_to_delete ;
                                char *from_topic ;
                                unsigned int status ;
                                rd_kafka_headers_t *rd_kafka_headers ;
                                LIST *received_headers ;
                                int schema_id ;
                                struct N_KAFKA *parent_table ;
                } N_KAFKA_EVENT ;
 
                typedef struct N_KAFKA
                {
                                LIST *events_to_send ;
                                LIST *received_events ;
                                char *groupid;
                                /* list of topics to subscribe to */
                                char **topics;           
                                /* subscribed topics */
                                rd_kafka_topic_partition_list_t *subscription; 
                                rd_kafka_conf_t *rd_kafka_conf ;
                                rd_kafka_t* rd_kafka_handle ;
                                char *topic ;
                                char *bootstrap_servers ;
                                rd_kafka_topic_t* rd_kafka_topic ;
                                N_STR *errstr ;
                                int schema_id ;
                                pthread_rwlock_t rwlock ;
                                int mode ;
                                cJSON *configuration ;
                                int64_t poll_timeout ;
                                int64_t poll_interval ;
                                int pooling_thread_status ;
                                pthread_t pooling_thread;
                                /* number of waiting events in the producer waiting list */
                                size_t nb_queued ;
                                /* number of events waiting for an ack in the waiting list */
                                size_t nb_waiting ;
                                /* number of errored events */
                                size_t nb_error ;
                } N_KAFKA ;
 
                int32_t n_kafka_get_schema_from_char( char *string );
                int32_t n_kafka_get_schema_from_nstr( N_STR *string );
                int n_kafka_put_schema_in_char( char *string , int schema_id );
                int n_kafka_put_schema_in_nstr( N_STR *string , int schema_id );
 
                int n_kafka_get_status( N_KAFKA *kafka , int *nb_queued , int *nb_waiting , int *nb_error );
                void n_kafka_delete( N_KAFKA *kafka );
                N_KAFKA *n_kafka_new( int64_t poll_timeout , int64_t poll_interval , size_t errstr_len );
                N_KAFKA *n_kafka_load_config( char *config_file , int mode );
 
                int n_kafka_new_headers( N_KAFKA_EVENT *event , size_t count );
                int n_kafka_add_header_ex( N_KAFKA_EVENT *event , char *key , size_t key_length , char *value , size_t value_length );
                int n_kafka_add_header( N_KAFKA_EVENT *event , N_STR *key , N_STR *value );
 
                int n_kafka_produce( N_KAFKA *kafka , N_KAFKA_EVENT *event );
 
                N_KAFKA_EVENT *n_kafka_new_event(int schema_id );
                N_KAFKA_EVENT *n_kafka_new_event_from_char( char *string , size_t written , int schema_id );
                N_KAFKA_EVENT *n_kafka_new_event_from_string( N_STR *string , int schema_id );
                N_KAFKA_EVENT *n_kafka_new_event_from_file( char *filename , int schema_id );
                void n_kafka_event_destroy_ptr( void *event );
                int n_kafka_event_destroy( N_KAFKA_EVENT **event );
 
                int n_kafka_poll( N_KAFKA *kafka );
                int n_kafka_start_pooling_thread( N_KAFKA *kafka );
                int n_kafka_stop_pooling_thread( N_KAFKA *kafka );
 
                int n_kafka_dump_unprocessed( N_KAFKA *kafka , char *directory );
                int n_kafka_load_unprocessed( N_KAFKA *kafka , char *directory );
 
                N_KAFKA_EVENT *n_kafka_get_event( N_KAFKA *kafka );
 
#ifdef __cplusplus
}
#endif
 
#endif // header guard