Mirrors/VILLASnode
1
0
Fork 0
mirror of https://git.rwth-aachen.de/acs/public/villas/node/ synced 2025-03-16 00:00:02 +01:00
Code Issues Releases Wiki Activity

kafka: minor tweaks in coding style

Browse source
This commit is contained in:
Steffen Vogel 2021-06-19 13:10:35 -04:00
parent 6a3e5d59af
commit cbcca7c8d9
1 changed files with 6 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
lib/nodes/kafka.cpp
View file

@ -67,20 +67,20 @@ static void kafka_logger_cb(const rd_kafka_t *rk, int level, const char *fac, co
static void kafka_message_cb(void *ctx, const rd_kafka_message_t *msg)
{
int ret;
int ret, cnt = n->in.vectorize;
struct vnode *n = (struct vnode *) ctx;
struct kafka *k = (struct kafka *) n->_vd;
struct sample *smps[n->in.vectorize];
struct sample *smps[cnt];
n->logger->debug("Received a message of {} bytes from broker {}", msg->len, k->server);
ret = sample_alloc_many(&k->pool, smps, n->in.vectorize);
ret = sample_alloc_many(&k->pool, smps, cnt);
if (ret <= 0) {
n->logger->warn("Pool underrun in consumer");
return;
}
ret = k->formatter->sscan((char *) msg->payload, msg->len, nullptr, smps, n->in.vectorize);
ret = k->formatter->sscan((char *) msg->payload, msg->len, nullptr, smps, cnt);
if (ret < 0) {
n->logger->warn("Received an invalid message");
n->logger->warn(" Payload: {}", (char *) msg->payload);
@ -89,11 +89,11 @@ static void kafka_message_cb(void *ctx, const rd_kafka_message_t *msg)
if (ret == 0) {
n->logger->debug("Skip empty message");
sample_decref_many(smps, n->in.vectorize);
sample_decref_many(smps, cnt);
return;
}
ret = queue_signalled_push_many(&k->queue, (void **) smps, n->in.vectorize);
ret = queue_signalled_push_many(&k->queue, (void **) smps, cnt);
if (ret < (int) n->in.vectorize)
n->logger->warn("Failed to enqueue samples");
}