6 Module N_KAFKA
gullradriel edited this page 2026-10-29 20:54:40 +01:00

KAFKA: generic event producer and consumer functions

Structures

Name
struct N_KAFKA_EVENT
structure of a KAFKA message
struct N_KAFKA
structure of a KAFKA consumer or producer handle

Functions

Name
int n_kafka_stop_polling_thread(N_KAFKA * kafka)
stop the kafka polling thread
int n_kafka_start_polling_thread(N_KAFKA * kafka)
start the kafka polling thread
int n_kafka_set_sended_dir(N_KAFKA * kafka, const char * tosend_dir, const char * sended_dir)
set the 'to-send' and 'sended' base directories used to move acknowledged event files instead of deleting them
int n_kafka_set_error_timeout(N_KAFKA * kafka, int32_t error_timeout)
set the retry interval (in seconds) for errored events.
int n_kafka_put_schema_in_nstr(N_STR * string, int schema_id)
put schema id in an N_STR string
int n_kafka_put_schema_in_char(char * string, int schema_id)
put schema id in a char string
int n_kafka_produce(N_KAFKA * kafka, N_KAFKA_EVENT * event)
produce a kafka event
int n_kafka_poll(N_KAFKA * kafka)
poll kafka for events
int n_kafka_new_headers(N_KAFKA_EVENT * event, size_t count)
create new headers for an event
N_KAFKA_EVENT * n_kafka_new_event_from_string(const N_STR * string, int schema_id)
create a new kafka event from an N_STR string
N_KAFKA_EVENT * n_kafka_new_event_from_file(char * filename, int schema_id)
create a new kafka event from a file
N_KAFKA_EVENT * n_kafka_new_event_from_char(const char * string, size_t written, int schema_id)
create a new kafka event from a char string
N_KAFKA_EVENT * n_kafka_new_event(int schema_id)
create a new empty kafka event
N_KAFKA * n_kafka_new(int32_t poll_timeout, int32_t poll_interval, size_t errstr_len)
create a new kafka handle
int n_kafka_load_unprocessed(N_KAFKA * kafka, const char * directory)
load unprocessed events from a directory
N_KAFKA * n_kafka_load_config(char * config_file, int mode)
load kafka configuration from file
int n_kafka_get_status(N_KAFKA * kafka, size_t * nb_queued, size_t * nb_waiting, size_t * nb_error)
get kafka queue status counters
int32_t n_kafka_get_schema_from_nstr(const N_STR * string)
get schema id from an N_STR string
int32_t n_kafka_get_schema_from_char(const char * string)
get schema id from a char string
N_KAFKA_EVENT * n_kafka_get_event(N_KAFKA * kafka)
get the next received event
void n_kafka_event_destroy_ptr(void * event)
destroy a kafka event (void pointer version for list destructor)
int n_kafka_event_destroy(N_KAFKA_EVENT ** event)
destroy a kafka event
int n_kafka_enable_event_production(N_KAFKA * kafka)
enable event production
int n_kafka_enable_event_consumption(N_KAFKA * kafka)
enable event consumption
int n_kafka_dump_unprocessed(N_KAFKA * kafka, char * directory)
dump unprocessed events to a directory
int n_kafka_disable_event_production(N_KAFKA * kafka)
disable event production
int n_kafka_disable_event_consumption(N_KAFKA * kafka)
disable event consumption
void n_kafka_delete(N_KAFKA * kafka)
delete a kafka handle
int n_kafka_add_header_ex(N_KAFKA_EVENT * event, char * key, size_t key_length, char * value, size_t value_length)
add a header to an event (extended version)
int n_kafka_add_header(N_KAFKA_EVENT * event, N_STR * key, N_STR * value)
add a header to an event

Defines

Name
N_KAFKA_EVENT_WAITING_ACK
state of a sent event waiting for acknowledgement
N_KAFKA_EVENT_QUEUED
state of a queued event
N_KAFKA_EVENT_OK
state of an OK event
N_KAFKA_EVENT_ERROR
state of an errored event
N_KAFKA_EVENT_CREATED
state of a freshly created event

Functions Documentation

function n_kafka_stop_polling_thread

int n_kafka_stop_polling_thread(
    N_KAFKA * kafka
)

stop the kafka polling thread

Parameters:

  • kafka target kafka handle

Return: TRUE or FALSE

stop the polling thread of a kafka handle

stop the kafka polling thread

function n_kafka_start_polling_thread

int n_kafka_start_polling_thread(
    N_KAFKA * kafka
)

start the kafka polling thread

Parameters:

  • kafka kafka handle to use

Return: TRUE or FALSE

start the polling thread of a kafka handle

start the kafka polling thread

function n_kafka_set_sended_dir

int n_kafka_set_sended_dir(
    N_KAFKA * kafka,
    const char * tosend_dir,
    const char * sended_dir
)

set the 'to-send' and 'sended' base directories used to move acknowledged event files instead of deleting them

Parameters:

  • kafka handler to configure
  • tosend_dir base directory the event files are read from; used to compute the subpath to preserve. Can be NULL (then only the basename is preserved)
  • sended_dir base directory acknowledged event files are moved into. NULL restores the legacy delete-on-ack behaviour

Return: TRUE or FALSE

set the 'to-send' and 'sended' base directories so acknowledged event files are moved instead of deleted

set the 'to-send' and 'sended' base directories used to move acknowledged event files instead of deleting them

function n_kafka_set_error_timeout

int n_kafka_set_error_timeout(
    N_KAFKA * kafka,
    int32_t error_timeout
)

set the retry interval (in seconds) for errored events.

Parameters:

  • kafka handler to configure
  • error_timeout seconds to wait before re-queuing an errored event; 0 disables retries (default)

Return: TRUE or FALSE

set the retry interval (in seconds) for events stuck in N_KAFKA_EVENT_ERROR

0 disables retries.

set the retry interval (in seconds) for errored events.

function n_kafka_put_schema_in_nstr

int n_kafka_put_schema_in_nstr(
    N_STR * string,
    int schema_id
)

put schema id in an N_STR string

Parameters:

  • string N_STR *destination where to write schema id
  • schema_id the schema_id to write

Return: zero or positive number on success, -1 on error

put a htonl schema id into the first 4 bytes of a N_STR *string

put schema id in an N_STR string

function n_kafka_put_schema_in_char

int n_kafka_put_schema_in_char(
    char * string,
    int schema_id
)

put schema id in a char string

Parameters:

  • string char *destination where to write schema id
  • schema_id the schema_id to write

Return: zero or positive number on success, -1 on error

put a htonl schema id into the first 4 bytes of a char *string

put schema id in a char string

function n_kafka_produce

int n_kafka_produce(
    N_KAFKA * kafka,
    N_KAFKA_EVENT * event
)

produce a kafka event

Parameters:

  • kafka the producer N_KAFKA handle to use
  • event event to send. Will be owned by N_KAFKA *kafka handle once done

Return: TRUE or FALSE

put an event in the events_to_send list

produce a kafka event

function n_kafka_poll

int n_kafka_poll(
    N_KAFKA * kafka
)

poll kafka for events

Parameters:

  • kafka kafka handle to use

Return: TRUE or FALSE

Poll kafka handle in producer or consumer mode.

poll kafka for events

function n_kafka_new_headers

int n_kafka_new_headers(
    N_KAFKA_EVENT * event,
    size_t count
)

create new headers for an event

Parameters:

  • event target event
  • count size in elements of the created headers array

Return: TRUE or FALSE

allocate a headers array for the event

create new headers for an event

function n_kafka_new_event_from_string

N_KAFKA_EVENT * n_kafka_new_event_from_string(
    const N_STR * string,
    int schema_id
)

create a new kafka event from an N_STR string

Parameters:

  • string source string
  • schema_id the schema id to use. Pass -1 for non avro formatted events

Return: a new N_KAFKA *event or NULL

make a new event from a N_STR *string

create a new kafka event from an N_STR string

function n_kafka_new_event_from_file

N_KAFKA_EVENT * n_kafka_new_event_from_file(
    char * filename,
    int schema_id
)

create a new kafka event from a file

Parameters:

  • filename source file path and filename
  • schema_id the schema id to use. Pass -1 for non avro formatted events

Return: a new N_KAFKA *event or NULL

make a new event from a N_STR *string

create a new kafka event from a file

function n_kafka_new_event_from_char

N_KAFKA_EVENT * n_kafka_new_event_from_char(
    const char * string,
    size_t written,
    int schema_id
)

create a new kafka event from a char string

Parameters:

  • string source string
  • written lenght of the string
  • schema_id the schema id to use. Pass -1 for non avro formatted events

Return: a new N_KAFKA *event or NULL

make a new event from a char *string

create a new kafka event from a char string

function n_kafka_new_event

N_KAFKA_EVENT * n_kafka_new_event(
    int schema_id
)

create a new empty kafka event

Parameters:

  • schema_id schema id that the event is using, -1 if not using avro schema_id formatted events

Return: a new empty N_KAFKA_EVENT *event

create a new empty event

create a new empty kafka event

function n_kafka_new

N_KAFKA * n_kafka_new(
    int32_t poll_timeout,
    int32_t poll_interval,
    size_t errstr_len
)

create a new kafka handle

Parameters:

  • poll_interval set polling interval
  • poll_timeout set polling
  • errstr_len set the size of the error string buffer

Return: a new empty N_KAFKA *kafka handle or NULL

allocate a new kafka handle

create a new kafka handle

function n_kafka_load_unprocessed

int n_kafka_load_unprocessed(
    N_KAFKA * kafka,
    const char * directory
)

load unprocessed events from a directory

Parameters:

  • kafka kafka handle to use
  • directory the directory from which to load the events

Return: TRUE or FALSE

load unprocessed/unset events

load unprocessed events from a directory

function n_kafka_load_config

N_KAFKA * n_kafka_load_config(
    char * config_file,
    int mode
)

load kafka configuration from file

Parameters:

  • config_file path and filename of the config file
  • mode RD_KAFKA_CONSUMER or RD_KAFKA_PRODUCER

Return: an allocated and configured N_KAFKA *kafka handle or NULL

load a kafka configuration from a file

load kafka configuration from file

function n_kafka_get_status

int n_kafka_get_status(
    N_KAFKA * kafka,
    size_t * nb_queued,
    size_t * nb_waiting,
    size_t * nb_error
)

get kafka queue status counters

Parameters:

  • kafka handler to use
  • nb_queued pointer to queue number holder
  • nb_waiting pointer to waiting number holder
  • nb_error pointer to error number holder

Return: TRUE or FALSE

return the queues status

get kafka queue status counters

function n_kafka_get_schema_from_nstr

int32_t n_kafka_get_schema_from_nstr(
    const N_STR * string
)

get schema id from an N_STR string

Parameters:

  • string N_STR *source from where to read schema id

Return: zero or positive number on success, -1 on error

get a schema from the first 4 bytes of a N_STR *string, returning ntohl of the value

get schema id from an N_STR string

function n_kafka_get_schema_from_char

int32_t n_kafka_get_schema_from_char(
    const char * string
)

get schema id from a char string

Parameters:

  • string char *source from where to read schema id

Return: zero or positive number on success, -1 on error

get a schema from the first 4 bytes of a char *string, returning ntohl of the value

get schema id from a char string

function n_kafka_get_event

N_KAFKA_EVENT * n_kafka_get_event(
    N_KAFKA * kafka
)

get the next received event

Parameters:

  • kafka kafka handle to use

Return: a received event ror NULL

get a received event from the N_KAFKA kafka handle

get the next received event

function n_kafka_event_destroy_ptr

void n_kafka_event_destroy_ptr(
    void * event
)

destroy a kafka event (void pointer version for list destructor)

Parameters:

  • event_ptr void pointer to target

festroy a kafka event

destroy a kafka event (void pointer version for list destructor)

function n_kafka_event_destroy

int n_kafka_event_destroy(
    N_KAFKA_EVENT ** event
)

destroy a kafka event

Parameters:

  • event event to delete

Return: TRUE or FALSE

destroy a kafka event and set it's pointer to NULL

destroy a kafka event

function n_kafka_enable_event_production

int n_kafka_enable_event_production(
    N_KAFKA * kafka
)

enable event production

Parameters:

Return: TRUE or FALSE

function n_kafka_enable_event_consumption

int n_kafka_enable_event_consumption(
    N_KAFKA * kafka
)

enable event consumption

Parameters:

Return: TRUE or FALSE

function n_kafka_dump_unprocessed

int n_kafka_dump_unprocessed(
    N_KAFKA * kafka,
    char * directory
)

dump unprocessed events to a directory

Parameters:

  • kafka kafka handle to use
  • directory the directory in which to dump the events

Return: TRUE or FALSE

dump unprocessed/unset events

dump unprocessed events to a directory

function n_kafka_disable_event_production

int n_kafka_disable_event_production(
    N_KAFKA * kafka
)

disable event production

Parameters:

Return: TRUE or FALSE

function n_kafka_disable_event_consumption

int n_kafka_disable_event_consumption(
    N_KAFKA * kafka
)

disable event consumption

Parameters:

Return: TRUE or FALSE

function n_kafka_delete

void n_kafka_delete(
    N_KAFKA * kafka
)

delete a kafka handle

Parameters:

delete a N_KAFKA handle

delete a kafka handle

function n_kafka_add_header_ex

int n_kafka_add_header_ex(
    N_KAFKA_EVENT * event,
    char * key,
    size_t key_length,
    char * value,
    size_t value_length
)

add a header to an event (extended version)

Parameters:

  • event target event
  • key string of the key
  • key_length size of the key string
  • value string of the value
  • value_length size of the value string

Return: TRUE or FALSE

add a header entry to an event.

add a header to an event (extended version)

headers array must have been allocated before

function n_kafka_add_header

int n_kafka_add_header(
    N_KAFKA_EVENT * event,
    N_STR * key,
    N_STR * value
)

add a header to an event

Parameters:

  • event target event
  • key N_STR *string of the key
  • value N_STR *string of the value

Return: TRUE or FALSE

add a header entry to an event.

add a header to an event

headers array must have been allocated before

Macros Documentation

define N_KAFKA_EVENT_WAITING_ACK

#define N_KAFKA_EVENT_WAITING_ACK 1

state of a sent event waiting for acknowledgement

define N_KAFKA_EVENT_QUEUED

#define N_KAFKA_EVENT_QUEUED 0

state of a queued event

define N_KAFKA_EVENT_OK

#define N_KAFKA_EVENT_OK 4

state of an OK event

define N_KAFKA_EVENT_ERROR

#define N_KAFKA_EVENT_ERROR 2

state of an errored event

define N_KAFKA_EVENT_CREATED

#define N_KAFKA_EVENT_CREATED 5

state of a freshly created event