256 lines
7.8 KiB
C++
256 lines
7.8 KiB
C++
#include <stdlib.h>
|
|
#include <string.h>
|
|
|
|
#include "log.h"
|
|
#include "utils.h"
|
|
#include "kafka.h"
|
|
#include <MESA/MESA_prof_load.h>
|
|
#include <MESA/librdkafka/rdkafka.h>
|
|
|
|
#define MAX_SYMBOL_LEN 128
|
|
|
|
struct config
|
|
{
|
|
int enable_debug;
|
|
char brokerlist[MAX_SYMBOL_LEN];
|
|
char sasl_username[MAX_SYMBOL_LEN];
|
|
char sasl_passwd[MAX_SYMBOL_LEN];
|
|
char topic_name[MAX_TOPIC_NUM][MAX_SYMBOL_LEN];
|
|
};
|
|
|
|
struct per_producer_per_topic
|
|
{
|
|
rd_kafka_t *producer;
|
|
rd_kafka_topic_t *topic;
|
|
};
|
|
|
|
struct kafka
|
|
{
|
|
struct config cfg;
|
|
struct per_producer_per_topic *pppt[MAX_TOPIC_NUM];
|
|
};
|
|
|
|
/******************************************************************************
|
|
* Private API
|
|
******************************************************************************/
|
|
|
|
static void per_producer_per_topic_free(struct per_producer_per_topic *pppt)
|
|
{
|
|
if (pppt)
|
|
{
|
|
if (pppt->topic)
|
|
{
|
|
rd_kafka_topic_destroy(pppt->topic);
|
|
pppt->topic = NULL;
|
|
}
|
|
|
|
if (pppt->producer)
|
|
{
|
|
rd_kafka_destroy(pppt->producer);
|
|
pppt->producer = NULL;
|
|
}
|
|
|
|
free(pppt);
|
|
pppt = NULL;
|
|
}
|
|
}
|
|
|
|
static struct per_producer_per_topic *per_producer_per_topic_new(const char *brokerlist, const char *sasl_username, const char *sasl_passwd, const char *topic_name)
|
|
{
|
|
char err_str[1024] = {0};
|
|
struct per_producer_per_topic *pppt = (struct per_producer_per_topic *)calloc(1, sizeof(struct per_producer_per_topic));
|
|
if (!pppt)
|
|
{
|
|
return NULL;
|
|
}
|
|
|
|
rd_kafka_conf_t *conf = rd_kafka_conf_new();
|
|
if (!conf)
|
|
{
|
|
LOG_ERROR("%s: failed to create kafka conf", LOG_TAG_KAFKA);
|
|
goto error_out;
|
|
}
|
|
if (rd_kafka_conf_set(conf, "queue.buffering.max.messages", "1000000", err_str, sizeof(err_str)) != RD_KAFKA_CONF_OK)
|
|
{
|
|
LOG_ERROR("%s: failed to set kafka queue.buffering.max.messages, %s", LOG_TAG_KAFKA, err_str);
|
|
goto error_out;
|
|
}
|
|
if (rd_kafka_conf_set(conf, "topic.metadata.refresh.interval.ms", "600000", err_str, sizeof(err_str)) != RD_KAFKA_CONF_OK)
|
|
{
|
|
LOG_ERROR("%s: failed to set kafka topic.metadata.refresh.interval.ms, %s", LOG_TAG_KAFKA, err_str);
|
|
goto error_out;
|
|
}
|
|
if (rd_kafka_conf_set(conf, "client.id", topic_name, err_str, sizeof(err_str)) != RD_KAFKA_CONF_OK)
|
|
{
|
|
LOG_ERROR("%s: failed to set kafka client.id, %s", LOG_TAG_KAFKA, err_str);
|
|
goto error_out;
|
|
}
|
|
|
|
// bootstrap.servers
|
|
if (rd_kafka_conf_set(conf, "bootstrap.servers", brokerlist, err_str, sizeof(err_str)) != RD_KAFKA_CONF_OK)
|
|
{
|
|
LOG_ERROR("%s: failed to set kafka bootstrap.servers, %s", LOG_TAG_KAFKA, err_str);
|
|
goto error_out;
|
|
}
|
|
|
|
if (strlen(sasl_username) > 0 && strlen(sasl_passwd) > 0)
|
|
{
|
|
if (rd_kafka_conf_set(conf, "security.protocol", "sasl_plaintext", err_str, sizeof(err_str)) != RD_KAFKA_CONF_OK)
|
|
{
|
|
LOG_ERROR("%s: failed to set kafka security.protocol, %s", LOG_TAG_KAFKA, err_str);
|
|
goto error_out;
|
|
}
|
|
if (rd_kafka_conf_set(conf, "sasl.mechanisms", "PLAIN", err_str, sizeof(err_str)) != RD_KAFKA_CONF_OK)
|
|
{
|
|
LOG_ERROR("%s: failed to set kafka sasl.mechanisms, %s", LOG_TAG_KAFKA, err_str);
|
|
goto error_out;
|
|
}
|
|
if (rd_kafka_conf_set(conf, "sasl.username", sasl_username, err_str, sizeof(err_str)) != RD_KAFKA_CONF_OK)
|
|
{
|
|
LOG_ERROR("%s: failed to set kafka sasl.username, %s", LOG_TAG_KAFKA, err_str);
|
|
goto error_out;
|
|
}
|
|
if (rd_kafka_conf_set(conf, "sasl.password", sasl_passwd, err_str, sizeof(err_str)) != RD_KAFKA_CONF_OK)
|
|
{
|
|
LOG_ERROR("%s: failed to set kafka sasl.password, %s", LOG_TAG_KAFKA, err_str);
|
|
goto error_out;
|
|
}
|
|
}
|
|
else
|
|
{
|
|
if (rd_kafka_conf_set(conf, "security.protocol", "plaintext", err_str, sizeof(err_str)) != RD_KAFKA_CONF_OK)
|
|
{
|
|
LOG_ERROR("%s: failed to set kafka security.protocol, %s", LOG_TAG_KAFKA, err_str);
|
|
goto error_out;
|
|
}
|
|
}
|
|
|
|
// The conf object is freed by this function and must not be used or destroyed by the application sub-sequently.
|
|
pppt->producer = rd_kafka_new(RD_KAFKA_PRODUCER, conf, err_str, sizeof(err_str));
|
|
conf = NULL;
|
|
if (pppt->producer == NULL)
|
|
{
|
|
LOG_ERROR("%s: failed to create kafka producer, %s", LOG_TAG_KAFKA, err_str);
|
|
goto error_out;
|
|
}
|
|
|
|
pppt->topic = rd_kafka_topic_new(pppt->producer, topic_name, NULL);
|
|
if (pppt->topic == NULL)
|
|
{
|
|
LOG_ERROR("%s: failed to create kafka topic: %s", LOG_TAG_KAFKA, topic_name);
|
|
goto error_out;
|
|
}
|
|
|
|
return pppt;
|
|
|
|
error_out:
|
|
if (conf)
|
|
{
|
|
rd_kafka_conf_destroy(conf);
|
|
}
|
|
|
|
per_producer_per_topic_free(pppt);
|
|
return NULL;
|
|
}
|
|
|
|
/******************************************************************************
|
|
* Public API -- Kafka
|
|
******************************************************************************/
|
|
|
|
// due to limit by client.id, need per producer per topic
|
|
struct kafka *kafka_create(const char *profile)
|
|
{
|
|
struct kafka *handle = (struct kafka *)calloc(1, sizeof(struct kafka));
|
|
if (!handle)
|
|
{
|
|
return NULL;
|
|
}
|
|
|
|
MESA_load_profile_int_def(profile, "kafka", "enable_debug", &handle->cfg.enable_debug, 0);
|
|
MESA_load_profile_string_def(profile, "kafka", "brokerlist", handle->cfg.brokerlist, sizeof(handle->cfg.brokerlist), "");
|
|
MESA_load_profile_string_def(profile, "kafka", "sasl_username", handle->cfg.sasl_username, sizeof(handle->cfg.sasl_username), "");
|
|
MESA_load_profile_string_def(profile, "kafka", "sasl_passwd", handle->cfg.sasl_passwd, sizeof(handle->cfg.sasl_passwd), "");
|
|
MESA_load_profile_string_def(profile, "kafka", "topic_name", handle->cfg.topic_name[TOPIC_RULE_HITS], sizeof(handle->cfg.topic_name[TOPIC_RULE_HITS]), "");
|
|
|
|
if (strlen(handle->cfg.brokerlist) == 0)
|
|
{
|
|
LOG_ERROR("%s: brokerlist is empty", LOG_TAG_KAFKA);
|
|
goto error_out;
|
|
}
|
|
|
|
for (int i = 0; i < MAX_TOPIC_NUM; i++)
|
|
{
|
|
if (strlen(handle->cfg.topic_name[i]) == 0)
|
|
{
|
|
LOG_ERROR("%s: topic_name[%d] is empty", LOG_TAG_KAFKA, i);
|
|
goto error_out;
|
|
}
|
|
}
|
|
|
|
for (int i = 0; i < MAX_TOPIC_NUM; i++)
|
|
{
|
|
handle->pppt[i] = per_producer_per_topic_new(handle->cfg.brokerlist, handle->cfg.sasl_username, handle->cfg.sasl_passwd, handle->cfg.topic_name[i]);
|
|
if (!handle->pppt[i])
|
|
{
|
|
goto error_out;
|
|
}
|
|
}
|
|
|
|
return handle;
|
|
|
|
error_out:
|
|
kafka_destroy(handle);
|
|
return NULL;
|
|
}
|
|
|
|
void kafka_destroy(struct kafka *handle)
|
|
{
|
|
if (handle)
|
|
{
|
|
for (int i = 0; i < MAX_TOPIC_NUM; i++)
|
|
{
|
|
per_producer_per_topic_free(handle->pppt[i]);
|
|
handle->pppt[i] = NULL;
|
|
}
|
|
|
|
free(handle);
|
|
handle = NULL;
|
|
}
|
|
}
|
|
|
|
int kafka_send(struct kafka *handle, enum topic_idx idx, const char *data, int len)
|
|
{
|
|
if (!handle)
|
|
{
|
|
LOG_ERROR("%s: handle is NULL", LOG_TAG_KAFKA);
|
|
return -1;
|
|
}
|
|
|
|
if (idx < 0 || idx >= MAX_TOPIC_NUM)
|
|
{
|
|
LOG_ERROR("%s: invalid topic index: %d", LOG_TAG_KAFKA, idx);
|
|
return -1;
|
|
}
|
|
|
|
if (handle->pppt[idx])
|
|
{
|
|
if (rd_kafka_produce(handle->pppt[idx]->topic, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_COPY, (void *)data, len, NULL, 0, NULL) == -1)
|
|
{
|
|
LOG_ERROR("%s: failed to produce message with topic [%d], %s", LOG_TAG_KAFKA, idx, rd_kafka_err2str(rd_kafka_last_error()));
|
|
return -1;
|
|
}
|
|
else
|
|
{
|
|
if (handle->cfg.enable_debug)
|
|
{
|
|
LOG_DEBUG("%s: success to produce message with topic [%d], %s", LOG_TAG_KAFKA, idx, data);
|
|
}
|
|
return 0;
|
|
}
|
|
}
|
|
else
|
|
{
|
|
LOG_ERROR("%s: topic %d not initialized", LOG_TAG_KAFKA, idx);
|
|
return -1;
|
|
}
|
|
} |