|
|
@@ -72,52 +72,188 @@ int aclk_get_conv_log_next()
|
|
|
#define ACLK_TOPIC_PREFIX "/agent/"
|
|
|
|
|
|
struct aclk_topic {
|
|
|
- const char *topic_suffix;
|
|
|
+ enum aclk_topics topic_id;
|
|
|
+ // as received from cloud - we keep this for
|
|
|
+ // eventual topic list update when claim_id changes
|
|
|
+ char *topic_recvd;
|
|
|
+ // constructed topic
|
|
|
char *topic;
|
|
|
};
|
|
|
|
|
|
// This helps to cache finalized topics (assembled with claim_id)
|
|
|
// to not have to alloc or create buffer and construct topic every
|
|
|
// time message is sent as in old ACLK
|
|
|
-static struct aclk_topic aclk_topic_cache[] = {
|
|
|
- { .topic_suffix = "outbound/meta", .topic = NULL }, // ACLK_TOPICID_CHART
|
|
|
- { .topic_suffix = "outbound/alarms", .topic = NULL }, // ACLK_TOPICID_ALARMS
|
|
|
- { .topic_suffix = "outbound/meta", .topic = NULL }, // ACLK_TOPICID_METADATA
|
|
|
- { .topic_suffix = "inbound/cmd", .topic = NULL }, // ACLK_TOPICID_COMMAND
|
|
|
- { .topic_suffix = NULL, .topic = NULL }
|
|
|
-};
|
|
|
+static struct aclk_topic **aclk_topic_cache = NULL;
|
|
|
+static size_t aclk_topic_cache_items = 0;
|
|
|
|
|
|
void free_topic_cache(void)
|
|
|
{
|
|
|
- struct aclk_topic *tc = aclk_topic_cache;
|
|
|
- while (tc->topic_suffix) {
|
|
|
- if (tc->topic) {
|
|
|
- freez(tc->topic);
|
|
|
- tc->topic = NULL;
|
|
|
+ if (aclk_topic_cache) {
|
|
|
+ for (size_t i = 0; i < aclk_topic_cache_items; i++) {
|
|
|
+ freez(aclk_topic_cache[i]->topic);
|
|
|
+ freez(aclk_topic_cache[i]->topic_recvd);
|
|
|
+ freez(aclk_topic_cache[i]);
|
|
|
}
|
|
|
- tc++;
|
|
|
+ freez(aclk_topic_cache);
|
|
|
+ aclk_topic_cache = NULL;
|
|
|
+ aclk_topic_cache_items = 0;
|
|
|
}
|
|
|
}
|
|
|
|
|
|
-static inline void generate_topic_cache(void)
|
|
|
-{
|
|
|
- struct aclk_topic *tc = aclk_topic_cache;
|
|
|
- char *ptr;
|
|
|
- if (unlikely(!tc->topic)) {
|
|
|
- rrdhost_aclk_state_lock(localhost);
|
|
|
- while(tc->topic_suffix) {
|
|
|
- tc->topic = mallocz(strlen(ACLK_TOPIC_PREFIX) + (UUID_STR_LEN - 1) + 2 /* '/' and \0 */ + strlen(tc->topic_suffix));
|
|
|
- ptr = tc->topic;
|
|
|
- strcpy(ptr, ACLK_TOPIC_PREFIX);
|
|
|
- ptr += strlen(ACLK_TOPIC_PREFIX);
|
|
|
- strcpy(ptr, localhost->aclk_state.claimed_id);
|
|
|
- ptr += (UUID_STR_LEN - 1);
|
|
|
- *ptr++ = '/';
|
|
|
- strcpy(ptr, tc->topic_suffix);
|
|
|
- tc++;
|
|
|
+#define JSON_TOPIC_KEY_TOPIC "topic"
|
|
|
+#define JSON_TOPIC_KEY_NAME "name"
|
|
|
+
|
|
|
+struct topic_name {
|
|
|
+ enum aclk_topics id;
|
|
|
+ // cloud name - how is it called
|
|
|
+ // in answer to /password endpoint
|
|
|
+ const char *name;
|
|
|
+} topic_names[] = {
|
|
|
+ { .id = ACLK_TOPICID_CHART, .name = "chart" },
|
|
|
+ { .id = ACLK_TOPICID_ALARMS, .name = "alarms" },
|
|
|
+ { .id = ACLK_TOPICID_METADATA, .name = "meta" },
|
|
|
+ { .id = ACLK_TOPICID_COMMAND, .name = "inbox-cmd" },
|
|
|
+ { .id = ACLK_TOPICID_UNKNOWN, .name = NULL }
|
|
|
+};
|
|
|
+
|
|
|
+enum aclk_topics compulsory_topics[] = {
|
|
|
+ ACLK_TOPICID_CHART,
|
|
|
+ ACLK_TOPICID_ALARMS,
|
|
|
+ ACLK_TOPICID_METADATA,
|
|
|
+ ACLK_TOPICID_COMMAND,
|
|
|
+ ACLK_TOPICID_UNKNOWN
|
|
|
+};
|
|
|
+
|
|
|
+static enum aclk_topics topic_name_to_id(const char *name) {
|
|
|
+ struct topic_name *topic = topic_names;
|
|
|
+ while (topic->name) {
|
|
|
+ if (!strcmp(topic->name, name)) {
|
|
|
+ return topic->id;
|
|
|
}
|
|
|
+ topic++;
|
|
|
+ }
|
|
|
+ return ACLK_TOPICID_UNKNOWN;
|
|
|
+}
|
|
|
+
|
|
|
+static const char *topic_id_to_name(enum aclk_topics tid) {
|
|
|
+ struct topic_name *topic = topic_names;
|
|
|
+ while (topic->name) {
|
|
|
+ if (topic->id == tid)
|
|
|
+ return topic->name;
|
|
|
+ topic++;
|
|
|
+ }
|
|
|
+ return "unknown";
|
|
|
+}
|
|
|
+
|
|
|
+#define CLAIM_ID_REPLACE_TAG "#{claim_id}"
|
|
|
+static void topic_generate_final(struct aclk_topic *t) {
|
|
|
+ char *dest;
|
|
|
+ char *replace_tag = strstr(t->topic_recvd, CLAIM_ID_REPLACE_TAG);
|
|
|
+ if (!replace_tag)
|
|
|
+ return;
|
|
|
+
|
|
|
+ rrdhost_aclk_state_lock(localhost);
|
|
|
+ if (unlikely(!localhost->aclk_state.claimed_id)) {
|
|
|
+ error("This should never be called if agent not claimed");
|
|
|
rrdhost_aclk_state_unlock(localhost);
|
|
|
+ return;
|
|
|
}
|
|
|
+
|
|
|
+ t->topic = mallocz(strlen(t->topic_recvd) + 1 - strlen(CLAIM_ID_REPLACE_TAG) + strlen(localhost->aclk_state.claimed_id));
|
|
|
+ memcpy(t->topic, t->topic_recvd, replace_tag - t->topic_recvd);
|
|
|
+ dest = t->topic + (replace_tag - t->topic_recvd);
|
|
|
+
|
|
|
+ memcpy(dest, localhost->aclk_state.claimed_id, strlen(localhost->aclk_state.claimed_id));
|
|
|
+ dest += strlen(localhost->aclk_state.claimed_id);
|
|
|
+ rrdhost_aclk_state_unlock(localhost);
|
|
|
+ replace_tag += strlen(CLAIM_ID_REPLACE_TAG);
|
|
|
+ strcpy(dest, replace_tag);
|
|
|
+ dest += strlen(replace_tag);
|
|
|
+ *dest = 0;
|
|
|
+}
|
|
|
+
|
|
|
+static int topic_cache_add_topic(struct json_object *json, struct aclk_topic *topic)
|
|
|
+{
|
|
|
+ struct json_object_iterator it;
|
|
|
+ struct json_object_iterator itEnd;
|
|
|
+
|
|
|
+ it = json_object_iter_begin(json);
|
|
|
+ itEnd = json_object_iter_end(json);
|
|
|
+
|
|
|
+ while (!json_object_iter_equal(&it, &itEnd)) {
|
|
|
+ if (!strcmp(json_object_iter_peek_name(&it), JSON_TOPIC_KEY_NAME)) {
|
|
|
+ if (json_object_get_type(json_object_iter_peek_value(&it)) != json_type_string) {
|
|
|
+ error("topic dictionary key \"" JSON_TOPIC_KEY_NAME "\" is expected to be json_type_string");
|
|
|
+ return 1;
|
|
|
+ }
|
|
|
+ topic->topic_id = topic_name_to_id(json_object_get_string(json_object_iter_peek_value(&it)));
|
|
|
+ if (topic->topic_id == ACLK_TOPICID_UNKNOWN) {
|
|
|
+ info("topic dictionary has unkown topic name \"%s\"", json_object_get_string(json_object_iter_peek_value(&it)));
|
|
|
+ }
|
|
|
+ json_object_iter_next(&it);
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ if (!strcmp(json_object_iter_peek_name(&it), JSON_TOPIC_KEY_TOPIC)) {
|
|
|
+ if (json_object_get_type(json_object_iter_peek_value(&it)) != json_type_string) {
|
|
|
+ error("topic dictionary key \"" JSON_TOPIC_KEY_TOPIC "\" is expected to be json_type_string");
|
|
|
+ return 1;
|
|
|
+ }
|
|
|
+ topic->topic_recvd = strdupz(json_object_get_string(json_object_iter_peek_value(&it)));
|
|
|
+ json_object_iter_next(&it);
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ error("topic dictionary has Unknown/Unexpected key \"%s\" in topic description. Ignoring!", json_object_iter_peek_name(&it));
|
|
|
+ json_object_iter_next(&it);
|
|
|
+ }
|
|
|
+
|
|
|
+ if (!topic->topic_recvd) {
|
|
|
+ error("topic dictionary Missig compulsory key %s", JSON_TOPIC_KEY_TOPIC);
|
|
|
+ return 1;
|
|
|
+ }
|
|
|
+
|
|
|
+ topic_generate_final(topic);
|
|
|
+ aclk_topic_cache_items++;
|
|
|
+
|
|
|
+ return 0;
|
|
|
+}
|
|
|
+
|
|
|
+int aclk_generate_topic_cache(struct json_object *json)
|
|
|
+{
|
|
|
+ json_object *obj;
|
|
|
+
|
|
|
+ size_t array_size = json_object_array_length(json);
|
|
|
+ if (!array_size) {
|
|
|
+ error("Empty topic list!");
|
|
|
+ return 1;
|
|
|
+ }
|
|
|
+
|
|
|
+ if (aclk_topic_cache)
|
|
|
+ free_topic_cache();
|
|
|
+
|
|
|
+ aclk_topic_cache = callocz(array_size, sizeof(struct aclk_topic *));
|
|
|
+
|
|
|
+ for (size_t i = 0; i < array_size; i++) {
|
|
|
+ obj = json_object_array_get_idx(json, i);
|
|
|
+ if (json_object_get_type(obj) != json_type_object) {
|
|
|
+ error("expected json_type_object");
|
|
|
+ return 1;
|
|
|
+ }
|
|
|
+ aclk_topic_cache[i] = callocz(1, sizeof(struct aclk_topic));
|
|
|
+ if (topic_cache_add_topic(obj, aclk_topic_cache[i])) {
|
|
|
+ error("failed to parse topic @idx=%d", (int)i);
|
|
|
+ return 1;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ for (int i = 0; compulsory_topics[i] != ACLK_TOPICID_UNKNOWN; i++) {
|
|
|
+ if (!aclk_get_topic(compulsory_topics[i])) {
|
|
|
+ error("missing compulsory topic \"%s\" in password response from cloud", topic_id_to_name(compulsory_topics[i]));
|
|
|
+ return 1;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ return 0;
|
|
|
}
|
|
|
|
|
|
/*
|
|
|
@@ -127,9 +263,17 @@ static inline void generate_topic_cache(void)
|
|
|
*/
|
|
|
const char *aclk_get_topic(enum aclk_topics topic)
|
|
|
{
|
|
|
- generate_topic_cache();
|
|
|
+ if (!aclk_topic_cache) {
|
|
|
+ error("Topic cache not initialized");
|
|
|
+ return NULL;
|
|
|
+ }
|
|
|
|
|
|
- return aclk_topic_cache[topic].topic;
|
|
|
+ for (size_t i = 0; i < aclk_topic_cache_items; i++) {
|
|
|
+ if (aclk_topic_cache[i]->topic_id == topic)
|
|
|
+ return aclk_topic_cache[i]->topic;
|
|
|
+ }
|
|
|
+ error("Unknown topic");
|
|
|
+ return NULL;
|
|
|
}
|
|
|
|
|
|
/*
|
Timotej S