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

json.kafka: minor tweaks

Browse source
This commit is contained in:
Steffen Vogel 2021-06-03 03:34:55 -04:00
parent d1a9d1e089
commit b79436432a
1 changed files with 7 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
lib/formats/json_kafka.cpp
View file

@ -41,12 +41,10 @@ static const char * json_kafka_type_villas_to_kafka(enum SignalType vt)
return "int64";
case SignalType::BOOLEAN:
return "?";
case SignalType::COMPLEX:
return "?";
return "boolean";
default:
case SignalType::COMPLEX:
case SignalType::INVALID:
return "unknown";
}
@ -129,7 +127,7 @@ static int json_kafka_unpack_sample(struct io *io, json_t *json_smp, struct samp
smp->flags = 0;
smp->signals = io->signals;
/* Get timestamp */
/* Unpack timestamp */
json_value = json_object_get(json_payload, "timestamp");
if (json_value) {
uint64_t ts_origin_ms = json_integer_value(json_value);
@ -138,18 +136,18 @@ static int json_kafka_unpack_sample(struct io *io, json_t *json_smp, struct samp
smp->flags |= (int) SampleFlags::HAS_TS_ORIGIN;
}
/* Get sequence no */
/* Unpack sequence no */
json_value = json_object_get(json_payload, "sequence");
if (json_value) {
smp->sequence = json_integer_value(json_value);
smp->flags |= (int) SampleFlags::HAS_SEQUENCE;
}
/* Get sample signals */
/* Unpack signal data */
for (size_t i = 0; i < vlist_length(io->signals); i++) {
struct signal *sig = (struct signal *) vlist_at(io->signals, i);
json_value = json_object_get(json_payload, sig->name);
if (!json_value)
continue;