Kafka class is a class that is responsible to send all generated output data, via Apache Kafka technology to the next AI Watch module. More...

#include <KafkaManager.hpp>

Public Member Functions
	KafkaManager (const char *topic)
	Construct a new Kafka Manager object. More...

	~KafkaManager (void)
	Destroy the KafkaManager object. More...

void	sendData (const char *key, Json::Value root)
	Send data contained in root via Kafka with ID equal to key. More...

Private Member Functions
void	setProducer (rd_kafka_t *producer)
	Set the Kafka producer's instance. More...

void	setConfiguration (rd_kafka_conf_t *configuration)
	Set the Kafka's configuration. More...

rd_kafka_t *	getProducer (void)
	Get the Kafka producer's instance. More...

rd_kafka_conf_t *	getConfiguration (void)
	Get the Kafka's configuration. More...

const char *	getTopic (void)
	Get the destination Kafka's topic. More...

void	loadConfigurationGroup (rd_kafka_conf_t conf, GKeyFile key_file, const char *group)
	Loads the configuration from Kafka's configuration file. More...

int	getArraySize (const char **array)
	Get the "const char *" array size. More...

void	setupProducer (void)
	Launches configuration loading, once reads all the configuration parameters and later set the producer instance. More...

Static Private Member Functions
static void	dr_msg_cb (rd_kafka_t kafka_handle, const rd_kafka_message_t rkmessage, void *opaque)
	Optional per-message delivery callback (triggered by poll() or flush()) when a message has been successfully delivered or permanently failed delivery (after retries). More...

Private Attributes
rd_kafka_t *	producer
	Kafka producer's instance. More...

rd_kafka_conf_t *	configuration
	Kafka's configuration. More...

const char *	topic
	Destination Kafka's topic. More...

Detailed Description

Kafka class is a class that is responsible to send all generated output data, via Apache Kafka technology to the next AI Watch module.

Definition at line 26 of file KafkaManager.hpp.

Constructor & Destructor Documentation

◆ KafkaManager()

KafkaManager::KafkaManager ( const char * topic )

inline

Construct a new Kafka Manager object.

Parameters

topic Destination Kafka's topic.

Definition at line 107 of file KafkaManager.hpp.

                                                 : topic(topic) {
        setupProducer();
    }

References setupProducer().

◆ ~KafkaManager()

KafkaManager::~KafkaManager ( void )

Destroy the KafkaManager object.

Definition at line 87 of file KafkaManager.cpp.

                                 {
    setConfiguration(NULL);
    rd_kafka_destroy(getProducer());
}

References getProducer(), and setConfiguration().

Member Function Documentation

◆ dr_msg_cb()

void KafkaManager::dr_msg_cb	(	rd_kafka_t *	kafka_handle,
		const rd_kafka_message_t *	rkmessage,
		void *	opaque
	)

staticprivate

Optional per-message delivery callback (triggered by poll() or flush()) when a message has been successfully delivered or permanently failed delivery (after retries).

Parameters

kafka_handle
rkmessage
opaque

Definition at line 56 of file KafkaManager.cpp.

                                                                                                            {
    if (rkmessage->err) g_error("Message delivery failed: %s", rd_kafka_err2str(rkmessage->err));
}

Referenced by setupProducer().

◆ getArraySize()

int KafkaManager::getArraySize ( const char ** array )

private

Get the "const char *" array size.

Parameters

array

Returns: int

See also: https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md

Definition at line 52 of file KafkaManager.cpp.

                                                   {
    return (array == nullptr || * array == nullptr) ? 0 : (sizeof(array) / sizeof(array[0]));
}

◆ getConfiguration()

rd_kafka_conf_t * KafkaManager::getConfiguration ( void )

private

Get the Kafka's configuration.

Returns: rd_kafka_conf_t *

Definition at line 27 of file KafkaManager.cpp.

                                                      {
    return this->configuration;
}

References configuration.

Referenced by setupProducer().

◆ getProducer()

rd_kafka_t * KafkaManager::getProducer ( void )

private

Get the Kafka producer's instance.

Returns: rd_kafka_t *

Definition at line 23 of file KafkaManager.cpp.

                                            {
    return this->producer;
}

References producer.

Referenced by sendData(), setupProducer(), and ~KafkaManager().

◆ getTopic()

const char * KafkaManager::getTopic ( void )

private

Get the destination Kafka's topic.

Returns: const char *

Definition at line 31 of file KafkaManager.cpp.

                                         {
    return this->topic;
}

References topic.

Referenced by sendData().

◆ loadConfigurationGroup()

void KafkaManager::loadConfigurationGroup	(	rd_kafka_conf_t *	conf,
		GKeyFile *	key_file,
		const char *	group
	)

private

Loads the configuration from Kafka's configuration file.

Parameters

conf
key_file
group

Definition at line 35 of file KafkaManager.cpp.

                                                                                                         {
    char errstr[512];
    g_autoptr(GError) error = NULL;
    
    gchar ** ptr = g_key_file_get_keys(key_file, group, NULL, & error);
    if (error) CV_Error(RDKAFKA_GET_KEYS_FROM_JSON_ERROR, RDKAFKA_GET_KEYS_FROM_JSON_SCOPE);
    
    while (* ptr) {
        const char * key = * ptr;
        g_autofree gchar * value = g_key_file_get_string(key_file, group, key, & error);
        if (error) CV_Error(RDKAFKA_GET_KEY_FROM_JSON_ERROR, RDKAFKA_GET_KEY_FROM_JSON_SCOPE);
        
        if (rd_kafka_conf_set(conf, key, value, errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) CV_Error(RDKAFKA_CONF_SET_ERROR, RDKAFKA_CONF_SET_SCOPE);
        ptr++;
    }
}

References RDKAFKA_CONF_SET_ERROR, RDKAFKA_CONF_SET_SCOPE, RDKAFKA_GET_KEY_FROM_JSON_ERROR, RDKAFKA_GET_KEY_FROM_JSON_SCOPE, RDKAFKA_GET_KEYS_FROM_JSON_ERROR, and RDKAFKA_GET_KEYS_FROM_JSON_SCOPE.

Referenced by setupProducer().

◆ sendData()

void KafkaManager::sendData	(	const char *	key,
		Json::Value	root
	)

Send data contained in root via Kafka with ID equal to key.

Parameters

key
root

Definition at line 92 of file KafkaManager.cpp.

                                                             {
    // Produce data
    std::string keyString = root.toStyledString();
    const char * value = keyString.c_str();
    size_t messageSize = (size_t) keyString.size();
    rd_kafka_resp_err_t err = rd_kafka_producev(getProducer(),
                                                RD_KAFKA_V_TOPIC(getTopic()),
                                                RD_KAFKA_V_MSGFLAGS(RD_KAFKA_MSG_F_COPY),
                                                RD_KAFKA_V_KEY((void *) key, strlen(key) * sizeof(char)),
                                                RD_KAFKA_V_VALUE((void *) value, messageSize),
                                                RD_KAFKA_V_OPAQUE(NULL), RD_KAFKA_V_END);
                      
    if (err) g_error("Failed to produce to topic %s: %s", topic, rd_kafka_err2str(err));
    /* } else {
           g_message("Produced event to topic %s: key = %12s value = %12s", topic, key, value);
       }
    */
    
    rd_kafka_poll(producer, 0);
    // Block until the messages are all sent.
    // g_message("Flushing final messages..");
    rd_kafka_flush(producer, 10 * 1000);
    
    if (rd_kafka_outq_len(producer) > 0) g_error("%d message(s) were not delivered", rd_kafka_outq_len(producer));
//    g_message("%d events were produced to topic %s.", message_count, topic);
}

References getProducer(), getTopic(), producer, and topic.

Referenced by FacadeSingleton::sendData().

◆ setConfiguration()

void KafkaManager::setConfiguration ( rd_kafka_conf_t * configuration )

private

Set the Kafka's configuration.

Parameters

configuration Kafka's configuration.

Definition at line 19 of file KafkaManager.cpp.

                                                                    {
    this->configuration = configuration;
}

References configuration.

Referenced by setupProducer(), and ~KafkaManager().

◆ setProducer()

void KafkaManager::setProducer ( rd_kafka_t * producer )

private

Set the Kafka producer's instance.

Parameters

producer Kafka producer's instance.

Definition at line 15 of file KafkaManager.cpp.

                                                     {
    this->producer = producer;
}

References producer.

Referenced by setupProducer().

◆ setupProducer()

void KafkaManager::setupProducer ( void )

private

Launches configuration loading, once reads all the configuration parameters and later set the producer instance.

Definition at line 60 of file KafkaManager.cpp.

                                      {
    char errorString[512];
    const char * configurationFile = "../configuration_file.ini";
    
    g_autoptr(GError) error = NULL;
    g_autoptr(GKeyFile) key_file = g_key_file_new();
    if (!g_key_file_load_from_file (key_file, configurationFile, G_KEY_FILE_NONE, & error)) {
        g_error ("Error loading config file: %s", error->message);
    }
    
    // Load the relevant configuration sections.
    setConfiguration(rd_kafka_conf_new());
    loadConfigurationGroup(getConfiguration(), key_file, "default");
    
    // Install a delivery-error callback.
    rd_kafka_conf_set_dr_msg_cb(getConfiguration(), dr_msg_cb);
    
    // Create the Producer instance.
    setProducer(rd_kafka_new(RD_KAFKA_PRODUCER, getConfiguration(), errorString, sizeof(errorString)));
    if (!getProducer()) g_error("Failed to create new producer: %s", errorString);
    
    // Configuration object is now owned, and freed, by the rd_kafka_t instance.
    setConfiguration(NULL);
}

References dr_msg_cb(), getConfiguration(), getProducer(), loadConfigurationGroup(), setConfiguration(), and setProducer().

Referenced by KafkaManager().

Member Data Documentation

◆ configuration

rd_kafka_conf_t* KafkaManager::configuration

private

Kafka's configuration.

Definition at line 35 of file KafkaManager.hpp.

Referenced by getConfiguration(), and setConfiguration().

◆ producer

rd_kafka_t* KafkaManager::producer

private

Kafka producer's instance.

Definition at line 31 of file KafkaManager.hpp.

Referenced by getProducer(), sendData(), and setProducer().

◆ topic

const char* KafkaManager::topic

private

Destination Kafka's topic.

Definition at line 39 of file KafkaManager.hpp.

Referenced by getTopic(), and sendData().

The documentation for this class was generated from the following files:

Managers/KafkaManager.hpp
Managers/KafkaManager.cpp

Public Member Functions

Private Member Functions

Static Private Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

◆ KafkaManager()

◆ ~KafkaManager()

Member Function Documentation

◆ dr_msg_cb()

◆ getArraySize()

◆ getConfiguration()

◆ getProducer()

◆ getTopic()

◆ loadConfigurationGroup()

◆ sendData()

◆ setConfiguration()

◆ setProducer()

◆ setupProducer()

Member Data Documentation

◆ configuration

◆ producer

◆ topic