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

Multiple samples can be received in a single MQTT message (Fixes #178)

Browse source
This commit is contained in:
Sonja Kolen 2018-07-17 11:20:38 +02:00
parent 2a0311f05f
commit 6c99e45264
1 changed files with 7 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
lib/nodes/mqtt.c
View file

@ -84,29 +84,29 @@ static void mqtt_message_cb(struct mosquitto *mosq, void *userdata, const struct
int ret;
struct node *n = (struct node *) userdata;
struct mqtt *m = (struct mqtt *) n->_vd;
struct sample *smp;
struct sample *smps[n->in.vectorize];
debug(5, "MQTT: Node %s received a message of %d bytes from broker %s", node_name(n), msg->payloadlen, m->host);
smp = sample_alloc(&m->pool);
if (!smp) {
ret = sample_alloc_many(&m->pool, smps, n->in.vectorize);
if (ret<0) {
warn("Pool underrun in subscriber of %s", node_name(n));
return;
}
ret = io_sscan(&m->io, msg->payload, msg->payloadlen, NULL, &smp, 1);
ret = io_sscan(&m->io, msg->payload, msg->payloadlen, NULL, smps, n->in.vectorize);
if (ret < 0) {
warn("MQTT: Node %s received an invalid message", node_name(n));
warn(" Payload: %s", (char *) msg->payload);
return;
}
if (ret != 1) {
if (ret == 0) {
debug(4, "MQTT: skip empty message for node %s", node_name(n));
sample_put(smp);
sample_put_many(smps, n->in.vectorize);
return;
}
queue_signalled_push(&m->queue, (void *) smp);
queue_signalled_push_many(&m->queue, (void *) smps, n->in.vectorize);
}
static void mqtt_subscribe_cb(struct mosquitto *mosq, void *userdata, int mid, int qos_count, const int *granted_qos)