1
0
Fork 0
mirror of https://git.rwth-aachen.de/acs/public/villas/node/ synced 2025-03-09 00:00:00 +01:00
VILLASnode/lib/formats/json_kafka.cpp
Steffen Vogel aa16979fdd Apply clang-format changes
Signed-off-by: Steffen Vogel <steffen.vogel@opal-rt.com>
2023-09-07 13:00:16 +02:00

169 lines
4.8 KiB
C++

/* JSON serializtion for Kafka schema/payloads.
*
* Author: Steffen Vogel <post@steffenvogel.de>
* SPDX-FileCopyrightText: 2014-2023 Institute for Automation of Complex Power Systems, RWTH Aachen University
* SPDX-License-Identifier: Apache-2.0
*/
#include <villas/exceptions.hpp>
#include <villas/formats/json_kafka.hpp>
#include <villas/timing.hpp>
#include <villas/utils.hpp>
using namespace villas;
using namespace villas::node;
const char *JsonKafkaFormat::villasToKafkaType(enum SignalType vt) {
switch (vt) {
case SignalType::FLOAT:
return "double";
case SignalType::INTEGER:
return "int64";
case SignalType::BOOLEAN:
return "boolean";
default:
case SignalType::COMPLEX:
case SignalType::INVALID:
return "unknown";
}
}
int JsonKafkaFormat::packSample(json_t **json_smp, const struct Sample *smp) {
json_t *json_payload, *json_fields, *json_field, *json_value;
json_fields = json_array();
json_payload = json_object();
// Include sample timestamp
if (smp->flags & (int)SampleFlags::HAS_TS_ORIGIN) {
json_field = json_pack("{ s: s, s: b, s: s }", "type", "int64", "optional",
false, "field", "timestamp");
uint64_t ts_origin_ms =
smp->ts.origin.tv_sec * 1e3 + smp->ts.origin.tv_nsec / 1e6;
json_array_append_new(json_fields, json_field);
json_object_set_new(json_payload, "timestamp", json_integer(ts_origin_ms));
}
// Include sample sequence no
if (smp->flags & (int)SampleFlags::HAS_SEQUENCE) {
json_field = json_pack("{ s: s, s: b, s: s }", "type", "int64", "optional",
false, "field", "sequence");
json_array_append_new(json_fields, json_field);
json_object_set_new(json_payload, "sequence", json_integer(smp->sequence));
}
// Include sample data
for (size_t i = 0; i < MIN(smp->length, smp->signals->size()); i++) {
const auto sig = smp->signals->getByIndex(i);
const auto *data = &smp->data[i];
json_field =
json_pack("{ s: s, s: b, s: s }", "type", villasToKafkaType(sig->type),
"optional", false, "field", sig->name);
json_value = data->toJson(sig->type);
json_array_append_new(json_fields, json_field);
json_object_set_new(json_payload, sig->name.c_str(), json_value);
}
json_object_set_new(json_schema, "fields", json_fields);
json_incref(json_schema);
*json_smp = json_pack("{ s: o, s: o }", "schema", json_schema, "payload",
json_payload);
if (*json_smp == nullptr)
return -1;
return 0;
}
int JsonKafkaFormat::unpackSample(json_t *json_smp, struct Sample *smp) {
json_t *json_payload, *json_value;
json_payload = json_object_get(json_smp, "payload");
if (!json_payload)
return -1;
smp->length = 0;
smp->flags = 0;
smp->signals = signals;
// Unpack timestamp
json_value = json_object_get(json_payload, "timestamp");
if (json_value) {
uint64_t ts_origin_ms = json_integer_value(json_value);
smp->ts.origin = time_from_double(ts_origin_ms / 1e3);
smp->flags |= (int)SampleFlags::HAS_TS_ORIGIN;
}
// 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;
}
// Unpack signal data
for (size_t i = 0; i < signals->size(); i++) {
auto sig = signals->getByIndex(i);
json_value = json_object_get(json_payload, sig->name.c_str());
if (!json_value)
continue;
smp->data[i].parseJson(sig->type, json_value);
smp->length++;
}
if (smp->length > 0)
smp->flags |= (int)SampleFlags::HAS_DATA;
return 0;
}
void JsonKafkaFormat::parse(json_t *json) {
int ret;
json_error_t err;
json_t *json_schema_tmp = nullptr;
ret = json_unpack_ex(json, &err, 0, "{ s?: o }", "schema", &json_schema_tmp);
if (ret)
throw ConfigError(json, err, "node-config-format-json-kafka",
"Failed to parse format configuration");
if (json_schema_tmp) {
if (!json_is_object(json_schema_tmp))
throw ConfigError(json, "node-config-format-json-kafka-schema",
"Kafka schema must be configured as a dictionary");
if (json_schema)
json_decref(json_schema);
json_schema = json_schema_tmp;
}
JsonFormat::parse(json);
}
JsonKafkaFormat::JsonKafkaFormat(int fl) : JsonFormat(fl) {
json_schema = json_pack("{ s: s, s: s }", "type", "struct", "name",
"villas-node.Value");
}
// Register format
static char n[] = "json.kafka";
static char d[] = "JSON Kafka schema/payload messages";
static FormatPlugin<JsonKafkaFormat, n, d,
(int)SampleFlags::HAS_TS_ORIGIN |
(int)SampleFlags::HAS_SEQUENCE |
(int)SampleFlags::HAS_DATA>
p;