Generic kafka consume and produce event functions. More...

#include "nilorea/n_kafka.h"
#include "nilorea/n_common.h"
#include "nilorea/n_base64.h"
#include <limits.h>
#include <sys/stat.h>
#include <sys/types.h>

Include dependency graph for n_kafka.c:

Go to the source code of this file.

Functions
int	n_kafka_add_header (N_KAFKA_EVENT event, N_STR key, N_STR *value)
	add a header entry to an event.

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 entry to an event.

void	n_kafka_delete (N_KAFKA *kafka)
	delete a N_KAFKA handle

static void	n_kafka_delivery_message_callback (rd_kafka_t rk, const rd_kafka_message_t rkmessage, void *opaque)
	Message delivery report callback.

int	n_kafka_disable_event_consumption (N_KAFKA *kafka)
	disable event consumption

int	n_kafka_disable_event_production (N_KAFKA *kafka)
	disable event production

int	n_kafka_dump_unprocessed (N_KAFKA kafka, char directory)
	dump unprocessed/unset events

int	n_kafka_enable_event_consumption (N_KAFKA *kafka)
	enable event consumption

int	n_kafka_enable_event_production (N_KAFKA *kafka)
	enable event production

int	n_kafka_event_destroy (N_KAFKA_EVENT **event)
	destroy a kafka event and set it's pointer to NULL

void	n_kafka_event_destroy_ptr (void *event_ptr)
	festroy a kafka event

static const char *	n_kafka_event_fdesc (const N_KAFKA_EVENT *event)
	return a " (file: ...)" log fragment naming the source file(s) of an event

N_KAFKA_EVENT *	n_kafka_get_event (N_KAFKA *kafka)
	get a received event from the N_KAFKA kafka handle

int32_t	n_kafka_get_schema_from_char (const char *string)
	get a schema from the first 4 bytes of a char *string, returning ntohl of the value

int32_t	n_kafka_get_schema_from_nstr (const N_STR *string)
	get a schema from the first 4 bytes of a N_STR *string, returning ntohl of the value

int	n_kafka_get_status (N_KAFKA kafka, size_t nb_queued, size_t nb_waiting, size_t nb_error)
	return the queues status

N_KAFKA *	n_kafka_load_config (char *config_file, int mode)
	load a kafka configuration from a file

int	n_kafka_load_unprocessed (N_KAFKA kafka, const char directory)
	load unprocessed/unset events

static int	n_kafka_mkdir_p (const char *path)
	create a directory and all its missing parents (mkdir -p)

static int	n_kafka_move_file_to_sended (const N_KAFKA kafka, const char filepath)
	move a single acknowledged event file to the 'sended' directory, preserving its subpath relative to 'to-send'

N_KAFKA *	n_kafka_new (int32_t poll_timeout, int32_t poll_interval, size_t errstr_len)
	allocate a new kafka handle

N_KAFKA_EVENT *	n_kafka_new_event (int schema_id)
	create a new empty event

N_KAFKA_EVENT *	n_kafka_new_event_from_char (const char *string, size_t written, int schema_id)
	make a new event from a char *string

N_KAFKA_EVENT *	n_kafka_new_event_from_file (char *filename, int schema_id)
	make a new event from a N_STR *string

N_KAFKA_EVENT *	n_kafka_new_event_from_string (const N_STR *string, int schema_id)
	make a new event from a N_STR *string

int	n_kafka_new_headers (N_KAFKA_EVENT *event, size_t count)
	allocate a headers array for the event

int	n_kafka_poll (N_KAFKA *kafka)
	Poll kafka handle in producer or consumer mode.

void *	n_kafka_polling_thread (void *ptr)
	kafka produce or consume polling thread function

int	n_kafka_produce (N_KAFKA kafka, N_KAFKA_EVENT event)
	put an event in the events_to_send list

int	n_kafka_produce_ex (N_KAFKA kafka, N_KAFKA_EVENT event)
	produce an event on a N_KAFKA *kafka handle

int	n_kafka_put_schema_in_char (char *string, int schema_id)
	put a htonl schema id into the first 4 bytes of a char *string

int	n_kafka_put_schema_in_nstr (N_STR *string, int schema_id)
	put a htonl schema id into the first 4 bytes of a N_STR *string

int	n_kafka_set_error_timeout (N_KAFKA *kafka, int32_t error_timeout)
	set the retry interval (in seconds) for events stuck in N_KAFKA_EVENT_ERROR

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 so acknowledged event files are moved instead of deleted

int	n_kafka_start_polling_thread (N_KAFKA *kafka)
	start the polling thread of a kafka handle

int	n_kafka_stop_polling_thread (N_KAFKA *kafka)
	stop the polling thread of a kafka handle

Detailed Description

Generic kafka consume and produce event functions.

Author: Castagnier Mickael

Version: 1.0

Date: 23/11/2022

Definition in file n_kafka.c.

Function Documentation

◆ n_kafka_delivery_message_callback()

static void n_kafka_delivery_message_callback	(	rd_kafka_t *	rk,
		const rd_kafka_message_t *	rkmessage,
		void *	opaque
	)

static

Message delivery report callback.

This callback is called exactly once per message, indicating if the message was succesfully delivered The callback is triggered from rd_kafka_poll() and executes on the application's thread

Parameters

rk	kafka rd_kafka_t kafka handle
rkmessage	pointer to the received event in kafka form
opaque	opaque structure holding the pointer of a sent event

Definition at line 267 of file n_kafka.c.

References __n_assert, _nstr, free_split_result(), LOG_DEBUG, LOG_ERR, LOG_INFO, N_KAFKA_EVENT_ERROR, n_kafka_event_fdesc(), N_KAFKA_EVENT_OK, n_kafka_move_file_to_sended(), n_log, split(), split_count(), unlock, and write_lock.

Referenced by n_kafka_load_config().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ n_kafka_event_fdesc()

static const char * n_kafka_event_fdesc ( const N_KAFKA_EVENT * event )

static

return a " (file: ...)" log fragment naming the source file(s) of an event

Parameters

event the event to describe

Returns: a never-NULL const C string to append after the opaque pointer in a log line. When the event carries its source file(s) (event_files_to_delete, set when the event is produced from a monitored directory) the fragment names them; for events with no source file (string events, single-file sends, received events) it is empty so no misleading "(file: ...)" is printed.

Definition at line 44 of file n_kafka.c.

References _Thread_local, N_STR::data, and N_KAFKA_EVENT::event_files_to_delete.

Referenced by n_kafka_add_header_ex(), n_kafka_delivery_message_callback(), n_kafka_dump_unprocessed(), n_kafka_new_headers(), n_kafka_poll(), n_kafka_produce(), and n_kafka_produce_ex().

Here is the caller graph for this function: