GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-19 14:23:05 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix mqtt command

Browse Source
This commit is contained in:
gc87
2021-10-18 19:11:28 +08:00
committed by Parker Liu
parent 5ba8e31494
commit 25bbff646e
4 changed files with 101 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/mqtt/command_rw.c
+17 -51
View File
@@ -23,26 +23,26 @@
#include "command_rw.h"
void command_read_once_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_read_req_t *req)
int 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);
int rc = common_has_node(plugin, req->node_id);
if (0 != rc) {
log_debug("The requested node does not exist");
return;
return -1;
}
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;
return -2;
}
uint32_t req_id = neu_plugin_get_event_id(plugin);
neu_plugin_send_read_cmd(plugin, req_id, req->node_id, config);
UNUSED(req_id);
return req_id;
}
static int wrap_read_response_json_object(neu_fixed_array_t * array,
@@ -208,7 +208,8 @@ char *command_read_once_response(neu_plugin_t * plugin,
}
char *json_str = NULL;
rc = neu_json_encode_by_fn(&json, neu_parse_encode_read, &json_str);
rc = neu_json_encode_with_mqtt(&json, neu_parse_encode_read, parse_header,
neu_parse_encode_mqtt_param, &json_str);
clean_read_response_json_object(&json);
if (0 != rc) {
log_info("Json string parse error:%d", rc);
@@ -229,54 +230,18 @@ char *command_read_cycle_response(neu_plugin_t * plugin,
return command_read_once_response(plugin, &parse_header, resp_val);
}
static bool match_name_grp_config(const void *key, const void *item)
{
if (NULL == key || NULL == item) {
return false;
}
const char * name = (const char *) key;
neu_taggrp_config_t *config = *(neu_taggrp_config_t **) item;
if (0 == strcmp(name, neu_taggrp_cfg_get_name(config))) {
return true;
}
return false;
}
static neu_taggrp_config_t *
get_group_config_by_name(neu_plugin_t *plugin, uint32_t dest_node_id,
const char *group_config_name)
{
vector_t configs = neu_system_get_group_configs(plugin, dest_node_id);
neu_taggrp_config_t *config;
neu_taggrp_config_t *c;
// config1 = *(neu_taggrp_config_t **) vector_get(&configs, 0);
config = *(neu_taggrp_config_t **) vector_find_item(
&configs, group_config_name, match_name_grp_config);
if (NULL == config) {
GROUP_CONFIGS_UNINIT(configs);
return NULL;
}
c = (neu_taggrp_config_t *) neu_taggrp_cfg_ref(config);
GROUP_CONFIGS_UNINIT(configs);
return c;
}
static int write_command(neu_plugin_t *plugin, uint32_t dest_node_id,
const char * group_config_name,
neu_data_val_t *write_val)
{
vector_t nodes = neu_system_get_nodes(plugin, NEU_NODE_TYPE_DRIVER);
int rc = common_node_id_exist(&nodes, dest_node_id);
vector_uninit(&nodes);
int rc = common_has_node(plugin, dest_node_id);
if (0 != rc) {
return -1;
}
neu_taggrp_config_t *config;
config = get_group_config_by_name(plugin, dest_node_id, group_config_name);
config =
neu_system_find_group_config(plugin, dest_node_id, group_config_name);
if (NULL == config) {
return -2;
}
@@ -287,8 +252,8 @@ static int write_command(neu_plugin_t *plugin, uint32_t dest_node_id,
return req_id;
}
void command_write_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_write_req_t *write_req)
int command_write_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_write_req_t *write_req)
{
log_info("WRITE uuid:%s, group config name:%s", mqtt->uuid,
write_req->group_config_name);
@@ -298,14 +263,14 @@ void command_write_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
write_val = neu_dvalue_unit_new();
if (write_val == NULL) {
log_error("Failed to allocate data value for write data");
return;
return -1;
}
array = neu_fixed_array_new(write_req->n_tag, sizeof(neu_int_val_t));
if (array == NULL) {
log_error("Failed to allocate array for write data");
neu_dvalue_free(write_val);
return;
return -2;
}
neu_int_val_t int_val;
@@ -350,9 +315,10 @@ void command_write_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
}
neu_dvalue_init_move_array(write_val, NEU_DTYPE_INT_VAL, array);
write_command(plugin, write_req->node_id, write_req->group_config_name,
write_val);
int req_id = write_command(plugin, write_req->node_id,
write_req->group_config_name, write_val);
neu_dvalue_free(write_val);
return req_id;
}
static int wrap_write_response_json_object(neu_fixed_array_t * array,
plugins/mqtt/command_rw.h
+2 -2
View File
@@ -28,14 +28,14 @@ extern "C" {
#include "common.h"
void command_read_once_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
int 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,
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,
int command_write_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_write_req_t *write_req);
char *command_write_response(neu_plugin_t *plugin, const char *uuid,
neu_data_val_t *resp_val);
plugins/mqtt/plugin_mqtt.c
+81 -3
View File
@@ -47,6 +47,12 @@
const neu_plugin_module_t neu_plugin_module;
struct context {
neu_list_node node;
int req_id;
neu_parse_mqtt_t parse_header;
};
struct neu_plugin {
neu_plugin_common_t common;
option_t option;
@@ -55,6 +61,64 @@ struct neu_plugin {
neu_parse_mqtt_t * parse_header;
};
struct context *context_create()
{
struct context *ctx = NULL;
ctx = malloc(sizeof(struct context));
if (NULL != ctx) {
memset(ctx, 0, sizeof(struct context));
}
return ctx;
}
void context_list_add(neu_list *list, struct context *ctx, int req_id,
neu_parse_mqtt_t *parse_header)
{
ctx->req_id = req_id;
ctx->parse_header.function = parse_header->function;
ctx->parse_header.uuid = strdup(parse_header->uuid);
NEU_LIST_NODE_INIT(&ctx->node);
neu_list_append(list, ctx);
}
struct context *context_list_find(neu_list *list, const int id)
{
struct context *item;
NEU_LIST_FOREACH(list, item)
{
if (id == item->req_id) {
return item;
}
}
return NULL;
}
void context_destroy(struct context *ctx)
{
if (NULL != ctx) {
if (NULL != ctx->parse_header.uuid) {
free(ctx->parse_header.uuid);
}
free(ctx);
}
}
void context_list_remove(neu_list *list, struct context *ctx)
{
UNUSED(list);
UNUSED(ctx);
}
void context_list_destroy(neu_list *list)
{
while (!neu_list_empty(list)) {
struct context *ctx = neu_list_first(list);
neu_list_remove(list, ctx);
context_destroy(ctx);
}
}
static int plugin_subscribe(neu_plugin_t *plugin, const char *topic,
const int qos, subscribe_handle handle)
{
@@ -137,7 +201,8 @@ static void plugin_response_handle(const char *topic_name, size_t topic_len,
neu_parse_read_req_t *req = NULL;
rc = neu_parse_decode_read(json_str, &req);
if (0 == rc) {
command_read_once_request(plugin, mqtt, req);
int req_id = command_read_once_request(plugin, mqtt, req);
UNUSED(req_id);
neu_parse_decode_read_free(req);
}
break;
@@ -146,7 +211,8 @@ static void plugin_response_handle(const char *topic_name, size_t topic_len,
neu_parse_write_req_t *req = NULL;
rc = neu_parse_decode_write(json_str, &req);
if (0 == rc) {
command_write_request(plugin, mqtt, req);
int req_id = command_write_request(plugin, mqtt, req);
UNUSED(req_id);
neu_parse_decode_write_free(req);
}
break;
@@ -228,7 +294,17 @@ static void plugin_response_handle(const char *topic_name, size_t topic_len,
}
MQTT_SEND(plugin->paho, "neuronlite/response", 0, ret_str);
free(json_str);
// if (NULL != mqtt) {
// if (NULL != mqtt->uuid) {
// free(mqtt->uuid);
// }
// free(mqtt);
// }
if (NULL != json_str) {
free(json_str);
}
UNUSED(topic_len);
}
@@ -295,6 +371,8 @@ static int mqtt_plugin_init(neu_plugin_t *plugin)
plugin->parse_header = NULL;
NEU_LIST_INIT(&plugin->context_list, struct context, node);
// Paho mqtt client setup
client_error error =
paho_client_open(&plugin->option, plugin, &plugin->paho);
plugins/opcua/plugin_handle.c
+1 -1
View File
@@ -548,6 +548,7 @@ static void periodic_read(nng_aio *aio, void *arg, int code)
stop_periodic_read(tuple);
neu_taggrp_cfg_anchor(new_config);
if (new_config != NULL) {
neu_taggrp_cfg_free(tuple->config);
tuple->config = new_config;
@@ -636,7 +637,6 @@ static opc_subscribe_tuple_t *find_subscribe(opc_handle_context_t *context,
static void stop_periodic_read(opc_subscribe_tuple_t *tuple)
{
nng_aio_cancel(tuple->aio);
nng_aio_free(tuple->aio);
}
static void remove_subscribe(opc_subscribe_tuple_t *tuple)