GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-18 13:41:22 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(mqtt): new read data

Browse Source
This commit is contained in:
gc87
2022-06-13 09:54:17 +08:00
parent 36a16f1bb6
commit 61306faefd
5 changed files with 87 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/mqtt/command/command.c
+15 -21
View File
@@ -37,17 +37,13 @@ static int ping_response(char **output_str, neu_plugin_t *plugin,
static int read_response(mqtt_response_t *response, neu_plugin_t *plugin,
neu_json_mqtt_t *mqtt, char *json_str)
{
neu_json_read_req_t *req = NULL;
int rc = neu_json_decode_read_req(json_str, &req);
if (0 == rc) {
uint32_t req_id = neu_plugin_get_event_id(plugin);
if (0 < req_id) {
response->context_add(plugin, req_id, mqtt, NULL,
response->topic_pair, false);
}
UNUSED(response);
command_rw_read_once_request(plugin, mqtt, req, req_id);
neu_json_decode_read_req_free(req);
neu_json_read_req_t *json = NULL;
int rc = neu_json_decode_read_req(json_str, &json);
if (0 == rc) {
command_rw_read_once_request(plugin, mqtt, json);
neu_json_decode_read_req_free(json);
}
return rc;
@@ -138,18 +134,16 @@ void command_response_handle(mqtt_response_t *response)
neu_json_decode_mqtt_req_free(mqtt);
}
// char *command_read_once_response(neu_plugin_t *plugin, uint32_t node_id,
// neu_json_mqtt_t *parse_header,
// neu_data_val_t * resp_val)
//{
// return command_rw_read_once_response(plugin, node_id, parse_header,
// resp_val);
//}
char *command_read_periodic_response(neu_reqresp_trans_data_t *data,
int upload_format)
char *command_read_once_response(neu_reqresp_head_t * head,
neu_resp_read_group_t *data, int format)
{
return command_rw_read_periodic_response(data, upload_format);
return command_rw_read_once_response(head, data, format);
return NULL;
}
char *command_read_periodic_response(neu_reqresp_trans_data_t *data, int format)
{
return command_rw_read_periodic_response(data, format);
}
// char *command_write_response(neu_json_mqtt_t *parse_header,
plugins/mqtt/command/command.h
+4 -4
View File
@@ -61,12 +61,12 @@ typedef struct {
} mqtt_response_t;
void command_response_handle(mqtt_response_t *response);
// char *command_read_once_response(neu_plugin_t *plugin, uint32_t node_id,
// neu_json_mqtt_t *parse_header,
// neu_data_val_t * resp_val);
char *command_read_once_response(neu_reqresp_head_t * head,
neu_resp_read_group_t *data, int format);
char *command_read_periodic_response(neu_reqresp_trans_data_t *data,
int upload_format);
int format);
// char *command_write_response(neu_json_mqtt_t *parse_header,
// neu_data_val_t * resp_val);
plugins/mqtt/command/read_write.c
+35 -64
View File
@@ -34,42 +34,33 @@ static uint64_t current_time()
}
int command_rw_read_once_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
neu_json_read_req_t *req, uint32_t req_id)
neu_json_read_req_t *req)
{
(void) plugin;
(void) mqtt;
(void) req;
(void) req_id;
// neu_node_id_t node_id =
// neu_plugin_get_node_id_by_node_name(plugin, req->node_name);
// if (node_id == 0) {
// zlog_debug(neuron, "node %s does not exist", req->node_name);
// return -1;
//}
// zlog_info(neuron, "READ uuid:%s, node id:%u", mqtt->uuid, node_id);
// neu_taggrp_config_t *config =
// neu_system_find_group_config(plugin, node_id, req->group_name);
// if (NULL == config) {
// zlog_debug(neuron, "The requested config does not exist");
// return -2;
//}
neu_reqresp_head_t header = { 0 };
header.ctx = mqtt;
header.type = NEU_REQ_READ_GROUP;
neu_req_read_group_t cmd = { 0 };
strcpy(cmd.driver, req->node);
strcpy(cmd.group, req->group);
int ret = neu_plugin_op(plugin, header, &cmd);
if (ret != 0) {
return -1;
}
// neu_plugin_send_read_cmd(plugin, req_id, node_id, config);
// return 0;
return 0;
}
static void wrap_read_response_json(neu_reqresp_trans_data_t *data,
neu_json_read_resp_t * json)
static void wrap_read_response_json(neu_resp_tag_value_t *tags, uint16_t len,
neu_json_read_resp_t *json)
{
json->n_tag = data->n_tag;
json->n_tag = len;
if (0 < json->n_tag) {
json->tags = (neu_json_read_resp_tag_t *) calloc(
json->n_tag, sizeof(neu_json_read_resp_tag_t));
}
for (int i = 0; i < json->n_tag; i++) {
neu_resp_tag_value_t *tag = &data->tags[i];
neu_resp_tag_value_t *tag = &tags[i];
neu_type_e type = tag->value.type;
json->tags[i].name = tag->tag;
json->tags[i].error = NEU_ERR_SUCCESS;
@@ -151,59 +142,39 @@ static void clean_read_response_json(neu_json_read_resp_t *json)
free(json->tags);
}
// char *command_rw_read_once_response(neu_plugin_t *plugin, uint32_t node_id,
// neu_json_mqtt_t *parse_header,
// neu_data_val_t * resp_val)
//{
// UNUSED(plugin);
char *command_rw_read_once_response(neu_reqresp_head_t * head,
neu_resp_read_group_t *data, int format)
{
UNUSED(format);
// neu_fixed_array_t *array;
// int rc = neu_dvalue_get_ref_array(resp_val, &array);
// if (0 != rc) {
// zlog_info(neuron, "Get array ref error");
// return NULL;
//}
// if (0 >= array->length) {
// return NULL;
//}
// char *json_str = NULL;
// neu_json_read_resp_t json;
// memset(&json, 0, sizeof(neu_json_read_resp_t));
// rc = wrap_read_response_json_object(array, &json, plugin, node_id);
// if (0 != rc) {
// return NULL;
//}
// rc = neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp,
// parse_header, neu_json_encode_mqtt_resp,
//&json_str);
// clean_read_response_json_object(&json);
// if (0 != rc) {
// zlog_info(neuron, "Json string parse error:%d", rc);
// return NULL;
//}
// return json_str;
//}
neu_resp_tag_value_t *tags = data->tags;
uint16_t len = data->n_tag;
char * json_str = NULL;
neu_json_read_resp_t json = { 0 };
wrap_read_response_json(tags, len, &json);
neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp, head->ctx,
neu_json_encode_mqtt_resp, &json_str);
clean_read_response_json(&json);
return json_str;
}
char *command_rw_read_periodic_response(neu_reqresp_trans_data_t *data,
int upload_format)
int format)
{
neu_resp_tag_value_t * tags = data->tags;
uint16_t len = data->n_tag;
char * json_str = NULL;
neu_json_read_periodic_t header = { .group = (char *) data->group,
.node = (char *) data->driver,
.timestamp = current_time() };
neu_json_read_resp_t json = { 0 };
wrap_read_response_json(data, &json);
wrap_read_response_json(tags, len, &json);
if (0 == upload_format) { // values
if (0 == format) { // values
neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp1, &header,
neu_json_encode_read_periodic_resp,
&json_str);
} else if (1 == upload_format) { // tags
} else if (1 == format) { // tags
neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp, &header,
neu_json_encode_read_periodic_resp,
&json_str);
plugins/mqtt/command/read_write.h
+5 -5
View File
@@ -29,13 +29,13 @@ extern "C" {
#include "common.h"
int command_rw_read_once_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
neu_json_read_req_t *req, uint32_t req_id);
// char *command_rw_read_once_response(neu_plugin_t *plugin, uint32_t node_id,
// neu_json_mqtt_t *parse_header,
// neu_data_val_t * resp_val);
neu_json_read_req_t *req);
char *command_rw_read_once_response(neu_reqresp_head_t * head,
neu_resp_read_group_t *data, int format);
char *command_rw_read_periodic_response(neu_reqresp_trans_data_t *data,
int upload_format);
int format);
int command_rw_write_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
neu_json_write_req_t *write_req, uint32_t req_id);
plugins/mqtt/mqtt.c
+28 -33
View File
@@ -709,38 +709,33 @@ static int mqtt_plugin_stop(neu_plugin_t *plugin)
// return NEU_ERR_SUCCESS;
//}
// static neu_err_code_e trans_data(neu_plugin_t *plugin, neu_request_t *req)
// {
// (void) plugin;
// (void) req;
// return 0;
// pthread_mutex_lock(&plugin->mutex);
static neu_err_code_e read_response(neu_plugin_t * plugin,
neu_reqresp_head_t *head, void *data)
{
neu_resp_read_group_t *read_data = data;
// if (!plugin->routine_running) {
// pthread_mutex_unlock(&plugin->mutex);
// return NEU_ERR_FAILURE;
// }
pthread_mutex_lock(&plugin->mutex);
// mqtt_routine_t * routine = plugin->routine;
// neu_reqresp_data_t * neu_data = (neu_reqresp_data_t *) req->buf;
// struct topic_pair * pair = NULL;
// int type = TOPIC_TYPE_UPLOAD;
// uint64_t sender = req->sender_id;
// const char * node_name = req->node_name;
// neu_taggrp_config_t *config = neu_data->grp_config;
// pair = vector_find_item(&routine->topics, (void *) &type,
// topic_type_match); char *json = command_read_periodic_response(plugin,
// sender, node_name,
// config, neu_data->data_val,
// plugin->routine->option.format);
// struct context *c = mqtt_routine_context_create(0, NULL, json, pair,
// true); if (NULL != c) {
// mqtt_routine_send(routine, c);
// }
if (!plugin->routine_running) {
pthread_mutex_unlock(&plugin->mutex);
return NEU_ERR_FAILURE;
}
// pthread_mutex_unlock(&plugin->mutex);
// return NEU_ERR_SUCCESS;
// }
pthread_mutex_lock(&plugin->routine->contexts_mtx);
mqtt_routine_t * routine = plugin->routine;
int type = TOPIC_TYPE_READ;
struct topic_pair *pair = topics_find_type(routine->topics, type);
char * json =
command_read_once_response(head, read_data, routine->option.format);
struct context *c = mqtt_routine_context_create(0, NULL, json, pair, true);
if (NULL != c) {
mqtt_routine_send(routine, c);
}
pthread_mutex_unlock(&plugin->routine->contexts_mtx);
pthread_mutex_unlock(&plugin->mutex);
return NEU_ERR_SUCCESS;
}
static neu_err_code_e trans_data(neu_plugin_t *plugin, void *data)
{
@@ -782,10 +777,10 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
neu_err_code_e error = NEU_ERR_SUCCESS;
switch (head->type) {
// case NEU_REQRESP_READ_RESP:
// error = read_response(plugin, req);
// break;
// case NEU_REQRESP_WRITE_RESP:
case NEU_RESP_READ_GROUP:
error = read_response(plugin, head, data);
break;
// case NEU_RESP_WRITE_TAG:
// error = write_response(plugin, req);
// break;
case NEU_REQRESP_TRANS_DATA: