From 90721f520df22daeb002ae2a45449c2b8268207c Mon Sep 17 00:00:00 2001 From: gc87 Date: Thu, 14 Oct 2021 01:46:34 +0800 Subject: [PATCH] Add mqtt command interface about adapter --- plugins/mqtt/CMakeLists.txt | 3 +- plugins/mqtt/command_datatag.c | 2 +- plugins/mqtt/command_group_config.c | 12 +-- plugins/mqtt/command_node.c | 102 ++++++++++++++++++ plugins/mqtt/command_node.h | 43 ++++++++ plugins/mqtt/command_rw.c | 35 +++--- plugins/mqtt/command_rw.h | 5 +- plugins/mqtt/common.c | 7 +- plugins/mqtt/common.h | 9 +- plugins/mqtt/plugin_mqtt.c | 160 ++++++++++++++++------------ 10 files changed, 272 insertions(+), 106 deletions(-) create mode 100644 plugins/mqtt/command_node.c create mode 100644 plugins/mqtt/command_node.h diff --git a/plugins/mqtt/CMakeLists.txt b/plugins/mqtt/CMakeLists.txt index b314430c..b308c0be 100644 --- a/plugins/mqtt/CMakeLists.txt +++ b/plugins/mqtt/CMakeLists.txt @@ -11,7 +11,8 @@ set(src common.c command_rw.c command_group_config.c - command_datatag.c) + command_datatag.c + command_node.c) add_library(plugin-mqtt SHARED ${src}) target_include_directories(plugin-mqtt PRIVATE diff --git a/plugins/mqtt/command_datatag.c b/plugins/mqtt/command_datatag.c index 220cf987..b8e405b4 100644 --- a/plugins/mqtt/command_datatag.c +++ b/plugins/mqtt/command_datatag.c @@ -103,7 +103,7 @@ static int get_all_tag(neu_plugin_t *plugin, const uint32_t node_id, } struct visit_context ctx = { .table = table, .tags = tags }; - common_group_configs_freach(plugin, node_id, &ctx, visit_func); + common_group_configs_foreach(plugin, node_id, &ctx, visit_func); return 0; } diff --git a/plugins/mqtt/command_group_config.c b/plugins/mqtt/command_group_config.c index 5d7172f9..208da822 100644 --- a/plugins/mqtt/command_group_config.c +++ b/plugins/mqtt/command_group_config.c @@ -109,15 +109,11 @@ char *command_get_group_configs(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, static int add_group_config(neu_plugin_t *plugin, const char *config_name, const uint32_t node_id, const int read_interval) { - if (0 == node_id) { - return -1; - } - vector_t nodes = neu_system_get_nodes(plugin, NEU_NODE_TYPE_DRIVER); int rc = common_node_id_exist(&nodes, node_id); vector_uninit(&nodes); if (0 != rc) { - return -2; + return -1; } vector_t configs = neu_system_get_group_configs(plugin, node_id); @@ -125,17 +121,17 @@ static int add_group_config(neu_plugin_t *plugin, const char *config_name, GROUP_CONFIGS_UNINIT(configs); if (0 == rc) { - return -4; + return -2; } neu_taggrp_config_t *grp_config = neu_taggrp_cfg_new((char *) config_name); if (NULL == grp_config) { - return -5; + return -3; } neu_taggrp_cfg_set_interval(grp_config, read_interval); intptr_t error = neu_system_add_group_config(plugin, node_id, grp_config); if (0 != error) { - return -6; + return -4; } return 0; } diff --git a/plugins/mqtt/command_node.c b/plugins/mqtt/command_node.c new file mode 100644 index 00000000..46301284 --- /dev/null +++ b/plugins/mqtt/command_node.c @@ -0,0 +1,102 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2021 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program; if not, write to the Free Software Foundation, + * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. + **/ + +#include + +#include "command_node.h" + +char *command_get_nodes(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_get_nodes_req_t *req) +{ + log_info("Get node list uuid:%s, node type:%d", mqtt->uuid, req->node_type); + + neu_node_type_e node_type = req->node_type; + neu_parse_get_nodes_res_t res = { 0 }; + int index = 0; + vector_t nodes = neu_system_get_nodes(plugin, node_type); + + res.n_node = nodes.size; + res.nodes = malloc(res.n_node * sizeof(neu_parse_get_nodes_res_node_t)); + memset(res.nodes, 0, res.n_node * sizeof(neu_parse_get_nodes_res_node_t)); + + VECTOR_FOR_EACH(&nodes, iter) + { + neu_node_info_t *info = (neu_node_info_t *) iterator_get(&iter); + + res.nodes[index].id = info->node_id; + res.nodes[index].name = info->node_name; + index += 1; + } + + char *json_str = NULL; + int rc = neu_json_encode_with_mqtt(&res, neu_parse_encode_get_tags, mqtt, + neu_parse_encode_mqtt_param, &json_str); + if (0 == rc) { + return json_str; + } + + return NULL; +} + +char *command_add_node(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_add_node_req_t *req) +{ + log_info("Add node uuid:%s, node type:%d", mqtt->uuid, req->node_type); + intptr_t rc = neu_system_add_node(plugin, req->node_type, req->node_name, + req->plugin_name); + char * json_str = NULL; + if (rc != 0) { + json_str = strdup("{\"error\": 1}"); + } else { + json_str = strdup("{\"error\": 0}"); + } + + return json_str; +} + +char *command_update_node(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_update_node_req_t *req) +{ + log_info("Update node uuid:%s, node type:%d", mqtt->uuid, req->node_type); + intptr_t rc = neu_system_update_node(plugin, req->node_type, req->node_name, + req->plugin_name); + char * json_str = NULL; + if (rc != 0) { + json_str = strdup("{\"error\": 1}"); + } else { + json_str = strdup("{\"error\": 0}"); + } + + return json_str; +} + +char *command_delete_node(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_del_node_req_t *req) +{ + log_info("Delete node uuid:%s, node id:%d", mqtt->uuid, req->node_id); + intptr_t rc = neu_system_del_node(plugin, req->node_id); + char * json_str = NULL; + if (rc != 0) { + json_str = strdup("{\"error\": 1}"); + } else { + json_str = strdup("{\"error\": 0}"); + } + + return json_str; +} \ No newline at end of file diff --git a/plugins/mqtt/command_node.h b/plugins/mqtt/command_node.h new file mode 100644 index 00000000..54065534 --- /dev/null +++ b/plugins/mqtt/command_node.h @@ -0,0 +1,43 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2021 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program; if not, write to the Free Software Foundation, + * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. + **/ + +#ifndef NEURON_PLUGIN_MQTT_COMMAND_NODE +#define NEURON_PLUGIN_MQTT_COMMAND_NODE + +#ifdef __cplusplus +extern "C" { +#endif + +#include + +#include "common.h" + +char *command_get_nodes(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_get_nodes_req_t *req); +char *command_add_node(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_add_node_req_t *req); +char *command_update_node(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_update_node_req_t *req); +char *command_delete_node(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_del_node_req_t *req); + +#ifdef __cplusplus +} +#endif +#endif \ No newline at end of file diff --git a/plugins/mqtt/command_rw.c b/plugins/mqtt/command_rw.c index d4576678..6d60d313 100644 --- a/plugins/mqtt/command_rw.c +++ b/plugins/mqtt/command_rw.c @@ -27,22 +27,20 @@ void command_read_once_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, neu_parse_read_req_t *req) { log_info("READ uuid:%s, node id:%u", mqtt->uuid, req->node_id); - - vector_t nodes = neu_system_get_nodes(plugin, NEU_NODE_TYPE_DRIVER); - int rc = common_node_id_exist(&nodes, req->node_id); - vector_uninit(&nodes); + int rc = common_has_node(plugin, req->node_id); if (0 != rc) { + log_debug("The requested node does not exist"); return; } - vector_t configs = neu_system_get_group_configs(plugin, req->node_id); - neu_taggrp_config_t *config; - neu_taggrp_config_t *c; - config = *(neu_taggrp_config_t **) vector_get(&configs, 0); - c = (neu_taggrp_config_t *) neu_taggrp_cfg_ref(config); - uint32_t req_id = neu_plugin_send_read_cmd(plugin, req->node_id, c); + neu_taggrp_config_t *config = neu_system_find_group_config( + plugin, req->node_id, req->group_config_name); + if (NULL == config) { + log_debug("The requested config does not exist"); + return; + } - GROUP_CONFIGS_UNINIT(configs); + uint32_t req_id = neu_plugin_send_read_cmd(plugin, req->node_id, config); UNUSED(req_id); } @@ -182,12 +180,13 @@ static void clean_read_response_json_object(neu_parse_read_res_t *json) free(json->tags); } -char *command_read_once_response(neu_plugin_t *plugin, const char *uuid, - neu_data_val_t *resp_val) +char *command_read_once_response(neu_plugin_t * plugin, + neu_parse_mqtt_t *parse_header, + neu_data_val_t * resp_val) { UNUSED(plugin); - UNUSED(uuid); UNUSED(resp_val); + UNUSED(parse_header); neu_fixed_array_t *array; int rc = neu_dvalue_get_ref_array(resp_val, &array); @@ -229,8 +228,12 @@ char *command_read_once_response(neu_plugin_t *plugin, const char *uuid, char *command_read_cycle_response(neu_plugin_t * plugin, neu_data_val_t *resp_val) { - // TODO: Generate UUID - return command_read_once_response(plugin, "", resp_val); + neu_parse_mqtt_t parse_header = { 0 }; + parse_header.function = NEU_MQTT_OP_READ; + char uuid4_str[40] = { '\0' }; + neu_uuid_v4_gen(uuid4_str); + parse_header.uuid = uuid4_str; + return command_read_once_response(plugin, &parse_header, resp_val); } static bool match_name_grp_config(const void *key, const void *item) diff --git a/plugins/mqtt/command_rw.h b/plugins/mqtt/command_rw.h index 73b3e7e4..f94b9bd3 100644 --- a/plugins/mqtt/command_rw.h +++ b/plugins/mqtt/command_rw.h @@ -30,8 +30,9 @@ extern "C" { void command_read_once_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, neu_parse_read_req_t *req); -char *command_read_once_response(neu_plugin_t *plugin, const char *uuid, - neu_data_val_t *resp_val); +char *command_read_once_response(neu_plugin_t * plugin, + neu_parse_mqtt_t *parse_header, + neu_data_val_t * resp_val); char *command_read_cycle_response(neu_plugin_t * plugin, neu_data_val_t *resp_val); void command_write_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, diff --git a/plugins/mqtt/common.c b/plugins/mqtt/common.c index d172611e..464b78ee 100644 --- a/plugins/mqtt/common.c +++ b/plugins/mqtt/common.c @@ -95,9 +95,10 @@ neu_taggrp_config_t *common_get_group_config(neu_plugin_t * plugin, return ret_config; } -void common_group_configs_freach(neu_plugin_t *plugin, const uint32_t node_id, - void *context, - void visit_func(neu_taggrp_config_t *, void *)) +void common_group_configs_foreach(neu_plugin_t *plugin, const uint32_t node_id, + void *context, + void visit_func(neu_taggrp_config_t *, + void *)) { vector_t configs = neu_system_get_group_configs(plugin, node_id); neu_taggrp_config_t *config; diff --git a/plugins/mqtt/common.h b/plugins/mqtt/common.h index f18c39a9..e25fc8f6 100644 --- a/plugins/mqtt/common.h +++ b/plugins/mqtt/common.h @@ -28,6 +28,7 @@ extern "C" { #include "parser/neu_json_fn.h" #include "parser/neu_json_group_config.h" #include "parser/neu_json_mqtt.h" +#include "parser/neu_json_node.h" #include "parser/neu_json_rw.h" #include "parser/neu_json_tag.h" @@ -54,10 +55,10 @@ int common_has_group_config(neu_plugin_t *plugin, const uint32_t node_id, neu_taggrp_config_t *common_get_group_config(neu_plugin_t * plugin, const uint32_t node_id, const char * group_config_name); -void common_group_configs_freach(neu_plugin_t *plugin, const uint32_t node_id, - void *context, - void visit_func(neu_taggrp_config_t *, - void *)); +void common_group_configs_foreach(neu_plugin_t *plugin, const uint32_t node_id, + void *context, + void visit_func(neu_taggrp_config_t *, + void *)); neu_datatag_table_t *common_get_datatags_table(neu_plugin_t * plugin, const uint32_t node_id); diff --git a/plugins/mqtt/plugin_mqtt.c b/plugins/mqtt/plugin_mqtt.c index 4cc1ba00..b3f4e865 100644 --- a/plugins/mqtt/plugin_mqtt.c +++ b/plugins/mqtt/plugin_mqtt.c @@ -26,6 +26,7 @@ #include "command_datatag.h" #include "command_group_config.h" +#include "command_node.h" #include "command_rw.h" #include "option.h" @@ -51,6 +52,7 @@ struct neu_plugin { option_t option; paho_client_t * paho; neu_list context_list; + neu_parse_mqtt_t * parse_header; }; static int plugin_subscribe(neu_plugin_t *plugin, const char *topic, @@ -83,128 +85,141 @@ static void plugin_response_handle(const char *topic_name, size_t topic_len, memset(json_str, 0x00, len + 1); memcpy(json_str, payload, len); - void * result = NULL; - neu_parse_mqtt_t *mqtt = NULL; - int rc = neu_parse_decode_mqtt_param(json_str, &mqtt); + neu_parse_mqtt_t *mqtt = NULL; + char * ret_str = NULL; + int rc = neu_parse_decode_mqtt_param(json_str, &mqtt); if (0 != rc) { log_error("JSON parsing mqtt failed"); return; } + plugin->parse_header = mqtt; + switch (mqtt->function) { case NEU_MQTT_OP_GET_GROUP_CONFIG: { - rc = neu_parse_decode_get_group_config( - json_str, (neu_parse_get_group_config_req_t **) &result); + neu_parse_get_group_config_req_t *req = NULL; + rc = neu_parse_decode_get_group_config(json_str, &req); if (0 == rc) { - char *ret_str = NULL; - ret_str = command_get_group_configs( - plugin, mqtt, (neu_parse_get_group_config_req_t *) result); - neu_parse_decode_get_group_config_free(result); - MQTT_SEND(plugin->paho, "neuronlite/response", 0, ret_str); + ret_str = command_get_group_configs(plugin, mqtt, req); + neu_parse_decode_get_group_config_free(req); } break; } case NEU_MQTT_OP_ADD_GROUP_CONFIG: { - rc = neu_parse_decode_add_group_config( - json_str, (neu_parse_add_group_config_req_t **) &result); + neu_parse_add_group_config_req_t *req = NULL; + rc = neu_parse_decode_add_group_config(json_str, &req); if (0 == rc) { - char *ret_str = NULL; - ret_str = command_add_group_config( - plugin, mqtt, (neu_parse_add_group_config_req_t *) result); - neu_parse_decode_add_group_config_free(result); - MQTT_SEND(plugin->paho, "neuronlite/response", 0, ret_str); + ret_str = command_add_group_config(plugin, mqtt, req); + neu_parse_decode_add_group_config_free(req); } break; } case NEU_MQTT_OP_UPDATE_GROUP_CONFIG: { - rc = neu_parse_decode_update_group_config( - json_str, (neu_parse_update_group_config_req_t **) &result); + neu_parse_update_group_config_req_t *req = NULL; + rc = neu_parse_decode_update_group_config(json_str, &req); if (0 == rc) { - char *ret_str = NULL; - ret_str = command_update_group_config( - plugin, mqtt, (neu_parse_update_group_config_req_t *) result); - neu_parse_decode_update_group_config_free(result); - MQTT_SEND(plugin->paho, "neuronlite/response", 0, ret_str); + ret_str = command_update_group_config(plugin, mqtt, req); + neu_parse_decode_update_group_config_free(req); } break; } case NEU_MQTT_OP_DELETE_GROUP_CONFIG: { - rc = neu_parse_decode_del_group_config( - json_str, (neu_parse_del_group_config_req_t **) &result); + neu_parse_del_group_config_req_t *req; + rc = neu_parse_decode_del_group_config(json_str, &req); if (0 == rc) { - char *ret_str = NULL; - ret_str = command_delete_group_config( - plugin, mqtt, (neu_parse_del_group_config_req_t *) result); - neu_parse_decode_del_group_config_free(result); - MQTT_SEND(plugin->paho, "neuronlite/response", 0, ret_str); + ret_str = command_delete_group_config(plugin, mqtt, req); + neu_parse_decode_del_group_config_free(req); } break; } case NEU_MQTT_OP_READ: { - rc = neu_parse_decode_read(json_str, (neu_parse_read_req_t **) &result); + neu_parse_read_req_t *req = NULL; + rc = neu_parse_decode_read(json_str, &req); if (0 == rc) { - command_read_once_request(plugin, mqtt, - (neu_parse_read_req_t *) result); - neu_parse_decode_read_free((neu_parse_read_req_t *) result); + command_read_once_request(plugin, mqtt, req); + neu_parse_decode_read_free(req); } break; } case NEU_MQTT_OP_WRITE: { - rc = neu_parse_decode_write(json_str, - (neu_parse_write_req_t **) &result); + neu_parse_write_req_t *req = NULL; + rc = neu_parse_decode_write(json_str, &req); if (0 == rc) { - command_write_request(plugin, mqtt, - (neu_parse_write_req_t *) result); - neu_parse_decode_write_free(result); + command_write_request(plugin, mqtt, req); + neu_parse_decode_write_free(req); } break; } case NEU_MQTT_OP_GET_TAGS: { - rc = neu_parse_decode_get_tags(json_str, - (neu_parse_get_tags_req_t **) &result); + neu_parse_get_tags_req_t *req; + rc = neu_parse_decode_get_tags(json_str, &req); if (0 == rc) { - char *ret_str = NULL; - ret_str = command_get_tags(plugin, mqtt, - (neu_parse_get_tags_req_t *) result); - neu_parse_decode_get_tags_free(result); - MQTT_SEND(plugin->paho, "neuronlite/response", 0, ret_str); + ret_str = command_get_tags(plugin, mqtt, req); + neu_parse_decode_get_tags_free(req); } break; } case NEU_MQTT_OP_ADD_TAGS: { - rc = neu_parse_decode_add_tags(json_str, - (neu_parse_add_tags_req_t **) &result); + neu_parse_add_tags_req_t *req = NULL; + rc = neu_parse_decode_add_tags(json_str, &req); if (0 == rc) { - char *ret_str = NULL; - ret_str = command_add_tags(plugin, mqtt, - (neu_parse_add_tags_req_t *) result); - neu_parse_decode_add_tags_free(result); - MQTT_SEND(plugin->paho, "neuronlite/response", 0, ret_str); + ret_str = command_add_tags(plugin, mqtt, req); + neu_parse_decode_add_tags_free(req); } break; } case NEU_MQTT_OP_UPDATE_TAGS: { - rc = neu_parse_decode_update_tags( - json_str, (neu_parse_update_tags_req_t **) &result); + neu_parse_update_tags_req_t *req = NULL; + rc = neu_parse_decode_update_tags(json_str, &req); if (0 == rc) { - char *ret_str = NULL; - ret_str = command_update_tags( - plugin, mqtt, (neu_parse_update_tags_req_t *) result); - neu_parse_decode_update_tags_free(result); - MQTT_SEND(plugin->paho, "neuronlite/response", 0, ret_str); + ret_str = command_update_tags(plugin, mqtt, req); + neu_parse_decode_update_tags_free(req); } break; } case NEU_MQTT_OP_DELETE_TAGS: { - rc = neu_parse_decode_del_tags(json_str, - (neu_parse_del_tags_req_t **) &result); + neu_parse_del_tags_req_t *req = NULL; + rc = neu_parse_decode_del_tags(json_str, &req); if (0 == rc) { - char *ret_str = NULL; - ret_str = command_delete_tags(plugin, mqtt, - (neu_parse_del_tags_req_t *) result); - neu_parse_decode_del_tags_free(result); - MQTT_SEND(plugin->paho, "neuronlite/response", 0, ret_str); + ret_str = command_delete_tags(plugin, mqtt, req); + neu_parse_decode_del_tags_free(req); + } + break; + } + case NEU_MQTT_OP_GET_NODES: { + neu_parse_get_nodes_req_t *req = NULL; + rc = neu_parse_decode_get_nodes(json_str, &req); + if (0 == rc) { + ret_str = command_get_nodes(plugin, mqtt, req); + neu_parse_decode_get_nodes_free(req); + } + break; + } + case NEU_MQTT_OP_ADD_NODES: { + neu_parse_add_node_req_t *req = NULL; + rc = neu_parse_decode_add_node(json_str, &req); + if (0 == rc) { + ret_str = command_add_node(plugin, mqtt, req); + neu_parse_decode_add_node_free(req); + } + break; + } + case NEU_MQTT_OP_UPDATE_NODES: { + neu_parse_update_node_req_t *req = NULL; + rc = neu_parse_decode_update_node(json_str, &req); + if (0 == rc) { + ret_str = command_update_node(plugin, mqtt, req); + neu_parse_decode_update_node_free(req); + } + break; + } + case NEU_MQTT_OP_DELETE_NODES: { + neu_parse_del_node_req_t *req = NULL; + rc = neu_parse_decode_del_node(json_str, &req); + if (0 == rc) { + ret_str = command_delete_node(plugin, mqtt, req); + neu_parse_decode_del_node_free(req); } break; } @@ -212,6 +227,7 @@ static void plugin_response_handle(const char *topic_name, size_t topic_len, break; } + MQTT_SEND(plugin->paho, "neuronlite/response", 0, ret_str); free(json_str); UNUSED(topic_len); @@ -277,6 +293,8 @@ static int mqtt_plugin_init(neu_plugin_t *plugin) plugin->option.keepalive_interval = atoi(keepalive_interval); plugin->option.clean_session = atoi(clean_session); + plugin->parse_header = NULL; + // Paho mqtt client setup client_error error = paho_client_open(&plugin->option, plugin, &plugin->paho); @@ -322,9 +340,9 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_request_t *req) switch (req->req_type) { case NEU_REQRESP_READ_RESP: { neu_reqresp_read_resp_t *read_resp; - read_resp = (neu_reqresp_read_resp_t *) req->buf; - char *json_str = - command_read_once_response(plugin, "", read_resp->data_val); + read_resp = (neu_reqresp_read_resp_t *) req->buf; + char *json_str = command_read_once_response( + plugin, plugin->parse_header, read_resp->data_val); MQTT_SEND(plugin->paho, "neuronlite/response", 0, json_str); break; }