refactor(mqtt): trans-data upload

plugins/mqtt/command/command.c

@@ -146,15 +146,11 @@ void command_response_handle(mqtt_response_t *response)
 // resp_val);
 //}
 
-// char *command_read_periodic_response(neu_plugin_t *plugin, uint64_t sender,
-// const char *         node_name,
-// neu_taggrp_config_t *config,
-// neu_data_val_t *     resp_val,
-// int                  upload_format)
-//{
-// return command_rw_read_periodic_response(plugin, sender, node_name, config,
-// resp_val, upload_format);
-//}
+char *command_read_periodic_response(neu_reqresp_trans_data_t *data,
+                                     int                       upload_format)
+{
+    return command_rw_read_periodic_response(data, upload_format);
+}
 
 // char *command_write_response(neu_json_mqtt_t *parse_header,
 // neu_data_val_t * resp_val)

plugins/mqtt/command/command.h

@@ -64,11 +64,10 @@ 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_periodic_response(neu_plugin_t *plugin, uint64_t sender,
-// const char *         node_name,
-// neu_taggrp_config_t *config,
-// neu_data_val_t *     resp_val,
-// int                  upload_format);
+
+char *command_read_periodic_response(neu_reqresp_trans_data_t *data,
+                                     int                       upload_format);
+
 // char *command_write_response(neu_json_mqtt_t *parse_header,
 // neu_data_val_t * resp_val);
 

plugins/mqtt/command/read_write.c

@@ -25,13 +25,13 @@
 #include "read_write.h"
 #include "utils/log.h"
 
-// static uint64_t current_time()
-//{
-// struct timeval tv;
-// gettimeofday(&tv, NULL);
-// uint64_t ms = tv.tv_sec;
-// return ms * 1000 + tv.tv_usec / 1000;
-//}
+static uint64_t current_time()
+{
+    struct timeval tv;
+    gettimeofday(&tv, NULL);
+    uint64_t ms = tv.tv_sec;
+    return ms * 1000 + tv.tv_usec / 1000;
+}
 
 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)
@@ -59,170 +59,97 @@ int command_rw_read_once_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
     return 0;
 }
 
-// static int wrap_read_response_json_object(neu_fixed_array_t *   array,
-// neu_json_read_resp_t *json,
-// neu_plugin_t *        plugin,
-// uint32_t              node_id)
-//{
-//(void) array;
-//(void) json;
-//(void) plugin;
-//(void) node_id;
-// neu_int_val_t *      int_val;
-// neu_data_val_t *     val;
-// neu_datatag_table_t *datatag_table =
-// neu_system_get_datatags_table(plugin, node_id);
-// json->n_tag = array->length;
+static void wrap_read_response_json(neu_reqresp_trans_data_t *data,
+                                    neu_json_read_resp_t *    json)
+{
+    json->n_tag = data->n_tag;
+    if (0 < json->n_tag) {
+        json->tags = (neu_json_read_resp_tag_t *) calloc(
+            json->n_tag, sizeof(neu_json_read_resp_tag_t));
+    }
 
-// if (0 == json->n_tag) {
-// return -1;
-//}
+    for (int i = 0; i < json->n_tag; i++) {
+        neu_resp_tag_value_t *tag  = &data->tags[i];
+        neu_type_e            type = tag->value.type;
+        json->tags[i].name         = tag->tag;
+        json->tags[i].error        = NEU_ERR_SUCCESS;
 
-// json->tags = (neu_json_read_resp_tag_t *) calloc(
-// array->length, sizeof(neu_json_read_resp_tag_t));
+        switch (type) {
+        case NEU_TYPE_ERROR:
+            json->tags[i].t             = NEU_JSON_INT;
+            json->tags[i].value.val_int = tag->value.value.i32;
+            json->tags[i].error         = tag->value.value.i32;
+            break;
+        case NEU_TYPE_UINT8:
+            json->tags[i].t             = NEU_JSON_INT;
+            json->tags[i].value.val_int = tag->value.value.u8;
+            break;
+        case NEU_TYPE_INT8:
+            json->tags[i].t             = NEU_JSON_INT;
+            json->tags[i].value.val_int = tag->value.value.i8;
+            break;
+        case NEU_TYPE_INT16:
+            json->tags[i].t             = NEU_JSON_INT;
+            json->tags[i].value.val_int = tag->value.value.i16;
+            break;
+        case NEU_TYPE_INT32:
+            json->tags[i].t             = NEU_JSON_INT;
+            json->tags[i].value.val_int = tag->value.value.i32;
+            break;
+        case NEU_TYPE_INT64:
+            json->tags[i].t             = NEU_JSON_INT;
+            json->tags[i].value.val_int = tag->value.value.i64;
+            break;
+        case NEU_TYPE_UINT16:
+            json->tags[i].t             = NEU_JSON_INT;
+            json->tags[i].value.val_int = tag->value.value.u16;
+            break;
+        case NEU_TYPE_UINT32:
+            json->tags[i].t             = NEU_JSON_INT;
+            json->tags[i].value.val_int = tag->value.value.u32;
+            break;
+        case NEU_TYPE_UINT64:
+            json->tags[i].t             = NEU_JSON_INT;
+            json->tags[i].value.val_int = tag->value.value.u64;
+            break;
+        case NEU_TYPE_FLOAT:
+            json->tags[i].t               = NEU_JSON_FLOAT;
+            json->tags[i].value.val_float = tag->value.value.f32;
+            break;
+        case NEU_TYPE_DOUBLE:
+            json->tags[i].t                = NEU_JSON_DOUBLE;
+            json->tags[i].value.val_double = tag->value.value.d64;
+            break;
+        case NEU_TYPE_BOOL:
+            json->tags[i].t              = NEU_JSON_BOOL;
+            json->tags[i].value.val_bool = tag->value.value.boolean;
+            break;
+        case NEU_TYPE_BIT:
+            json->tags[i].t             = NEU_JSON_BIT;
+            json->tags[i].value.val_bit = tag->value.value.u8;
+            break;
+        case NEU_TYPE_STRING:
+            json->tags[i].t             = NEU_JSON_STR;
+            json->tags[i].value.val_str = tag->value.value.str;
+            break;
+        default:
+            break;
+        }
+    }
+}
 
-// for (uint32_t i = 0; i < array->length; i++) {
-// int_val          = neu_fixed_array_get(array, i);
-// val              = int_val->val;
-// neu_dtype_e type = neu_dvalue_get_value_type(val);
+static void clean_read_response_json(neu_json_read_resp_t *json)
+{
+    if (NULL == json) {
+        return;
+    }
 
-// neu_datatag_t *tag = neu_datatag_tbl_get(datatag_table, int_val->key);
+    if (NULL == json->tags) {
+        return;
+    }
 
-// if (tag == NULL) {
-// continue;
-//}
-
-// json->tags[i].name  = tag->name;
-// json->tags[i].error = NEU_ERR_SUCCESS;
-
-// switch (type) {
-// case NEU_DTYPE_ERRORCODE: {
-// int32_t value;
-// neu_dvalue_get_errorcode(val, &value);
-// json->tags[i].t             = NEU_JSON_INT;
-// json->tags[i].value.val_int = value;
-// json->tags[i].error         = value;
-// break;
-//}
-// case NEU_DTYPE_BYTE: {
-// uint8_t value;
-// neu_dvalue_get_uint8(val, &value);
-// json->tags[i].t             = NEU_JSON_INT;
-// json->tags[i].value.val_int = value;
-// break;
-//}
-// case NEU_DTYPE_INT8: {
-// int8_t value;
-// neu_dvalue_get_int8(val, &value);
-// json->tags[i].t             = NEU_JSON_INT;
-// json->tags[i].value.val_int = value;
-// break;
-//}
-// case NEU_DTYPE_INT16: {
-// int16_t value;
-// neu_dvalue_get_int16(val, &value);
-// json->tags[i].t             = NEU_JSON_INT;
-// json->tags[i].value.val_int = value;
-// break;
-//}
-// case NEU_DTYPE_INT32: {
-// int32_t value;
-// neu_dvalue_get_int32(val, &value);
-// json->tags[i].t             = NEU_JSON_INT;
-// json->tags[i].value.val_int = value;
-// break;
-//}
-// case NEU_DTYPE_INT64: {
-// int64_t value;
-// neu_dvalue_get_int64(val, &value);
-// json->tags[i].t             = NEU_JSON_INT;
-// json->tags[i].value.val_int = value;
-// break;
-//}
-// case NEU_DTYPE_UINT8: {
-// uint8_t value;
-// neu_dvalue_get_uint8(val, &value);
-// json->tags[i].t             = NEU_JSON_INT;
-// json->tags[i].value.val_int = value;
-// break;
-//}
-// case NEU_DTYPE_UINT16: {
-// uint16_t value;
-// neu_dvalue_get_uint16(val, &value);
-// json->tags[i].t             = NEU_JSON_INT;
-// json->tags[i].value.val_int = value;
-// break;
-//}
-// case NEU_DTYPE_UINT32: {
-// uint32_t value;
-// neu_dvalue_get_uint32(val, &value);
-// json->tags[i].t             = NEU_JSON_INT;
-// json->tags[i].value.val_int = value;
-// break;
-//}
-// case NEU_DTYPE_UINT64: {
-// uint64_t value;
-// neu_dvalue_get_uint64(val, &value);
-// json->tags[i].t             = NEU_JSON_INT;
-// json->tags[i].value.val_int = value;
-// break;
-//}
-// case NEU_DTYPE_FLOAT: {
-// float value;
-// neu_dvalue_get_float(val, &value);
-// json->tags[i].t               = NEU_JSON_FLOAT;
-// json->tags[i].value.val_float = value;
-// break;
-//}
-// case NEU_DTYPE_DOUBLE: {
-// double value;
-// neu_dvalue_get_double(val, &value);
-// json->tags[i].t                = NEU_JSON_DOUBLE;
-// json->tags[i].value.val_double = value;
-// break;
-//}
-// case NEU_DTYPE_BOOL: {
-// bool value;
-// neu_dvalue_get_bool(val, &value);
-// json->tags[i].t              = NEU_JSON_BOOL;
-// json->tags[i].value.val_bool = value;
-// break;
-//}
-// case NEU_DTYPE_BIT: {
-// uint8_t value;
-// neu_dvalue_get_bit(val, &value);
-// json->tags[i].t             = NEU_JSON_BIT;
-// json->tags[i].value.val_bit = value;
-// break;
-//}
-// case NEU_DTYPE_CSTR: {
-// char *value;
-// neu_dvalue_get_ref_cstr(val, &value);
-// json->tags[i].t             = NEU_JSON_STR;
-// json->tags[i].value.val_str = value;
-// break;
-//}
-// default:
-// break;
-//}
-//}
-
-// return 0;
-// return 0;
-//}
-
-// static void clean_read_response_json_object(neu_json_read_resp_t *json)
-//{
-// if (NULL == json) {
-// return;
-//}
-
-// if (NULL == json->tags) {
-// return;
-//}
-
-// free(json->tags);
-//}
+    free(json->tags);
+}
 
 // char *command_rw_read_once_response(neu_plugin_t *plugin, uint32_t node_id,
 // neu_json_mqtt_t *parse_header,
@@ -262,47 +189,32 @@ int command_rw_read_once_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
 // return json_str;
 //}
 
-// char *command_rw_read_periodic_response(neu_plugin_t *plugin, uint64_t
-// sender, const char *         node_name, neu_taggrp_config_t *config,
-// neu_data_val_t *     resp_val,
-// int                  upload_format)
-//{
-// UNUSED(plugin);
+char *command_rw_read_periodic_response(neu_reqresp_trans_data_t *data,
+                                        int                       upload_format)
+{
+    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);
 
-// neu_fixed_array_t *array = NULL;
-// neu_dvalue_get_ref_array(resp_val, &array);
-// assert(NULL != array);
+    if (0 == upload_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
+        neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp, &header,
+                                  neu_json_encode_read_periodic_resp,
+                                  &json_str);
+    } else {
+        clean_read_response_json(&json);
+        return NULL;
+    }
 
-// if (0 >= array->length) {
-// return NULL;
-//}
-
-// char *                   json_str = NULL;
-// neu_json_read_periodic_t header   = {
-//.group_name = (char *) neu_taggrp_cfg_get_name(config),
-//.node_name  = (char *) node_name,
-//.timestamp  = current_time()
-//};
-
-// neu_json_read_resp_t json = { 0 };
-// wrap_read_response_json_object(array, &json, plugin, (uint32_t) sender);
-
-// if (0 == upload_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
-// neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp, &header,
-// neu_json_encode_read_periodic_resp,
-//&json_str);
-//} else {
-// clean_read_response_json_object(&json);
-// return NULL;
-//}
-
-// clean_read_response_json_object(&json);
-// return json_str;
-//}
+    clean_read_response_json(&json);
+    return json_str;
+}
 
 // static int write_command(neu_plugin_t *plugin, uint32_t dest_node_id,
 // const char *    group_config_name,

plugins/mqtt/command/read_write.h

@@ -33,10 +33,10 @@ int command_rw_read_once_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
 // 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);
-// char *command_rw_read_periodic_response(neu_plugin_t *plugin, uint64_t
-// sender, const char *         node_name, neu_taggrp_config_t *grp_config,
-// neu_data_val_t *     resp_val,
-// int                  upload_format);
+
+char *command_rw_read_periodic_response(neu_reqresp_trans_data_t *data,
+                                        int upload_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);
 // char *command_rw_write_response(neu_json_mqtt_t *parse_header,

plugins/mqtt/mqtt.c

@@ -167,19 +167,19 @@ static struct topic_pair *topics_find_topic(UT_array *topics, const char *topic)
     return NULL;
 }
 
-// static struct topic_pair *topics_find_type(UT_array *topics, int type)
-// {
-//     for (struct topic_pair **pair =
-//              (struct topic_pair **) utarray_front(topics);
-//          NULL != pair;
-//          pair = (struct topic_pair **) utarray_next(topics, pair)) {
-//         if (type == (*pair)->type) {
-//             return *pair;
-//         }
-//     }
+static struct topic_pair *topics_find_type(UT_array *topics, int type)
+{
+    for (struct topic_pair **pair =
+             (struct topic_pair **) utarray_front(topics);
+         NULL != pair;
+         pair = (struct topic_pair **) utarray_next(topics, pair)) {
+        if (type == (*pair)->type) {
+            return *pair;
+        }
+    }
 
-//     return NULL;
-// }
+    return NULL;
+}
 
 static int mqtt_routine_send(mqtt_routine_t *routine, struct context *context);
 
@@ -710,35 +710,67 @@ static int mqtt_plugin_stop(neu_plugin_t *plugin)
 //}
 
 // 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);
+// {
+//     (void) plugin;
+//     (void) req;
+//     return 0;
+//     pthread_mutex_lock(&plugin->mutex);
 
-// if (!plugin->routine_running) {
-// pthread_mutex_unlock(&plugin->mutex);
-// return NEU_ERR_FAILURE;
-//}
+//     if (!plugin->routine_running) {
+//         pthread_mutex_unlock(&plugin->mutex);
+//         return NEU_ERR_FAILURE;
+//     }
 
-// 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);
-//}
+//     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);
+//     }
 
-// pthread_mutex_unlock(&plugin->mutex);
-// return NEU_ERR_SUCCESS;
-//}
+//     pthread_mutex_unlock(&plugin->mutex);
+//     return NEU_ERR_SUCCESS;
+// }
+
+static neu_err_code_e trans_data(neu_plugin_t *plugin, void *data)
+{
+    neu_reqresp_trans_data_t *trans_data = data;
+
+    pthread_mutex_lock(&plugin->mutex);
+
+    if (!plugin->routine_running) {
+        pthread_mutex_unlock(&plugin->mutex);
+        return NEU_ERR_FAILURE;
+    }
+
+    mqtt_routine_t *   routine = plugin->routine;
+    int                type    = TOPIC_TYPE_UPLOAD;
+    struct topic_pair *pair    = topics_find_type(routine->topics, type);
+    char *             json =
+        command_read_periodic_response(trans_data, routine->option.format);
+    if (NULL == json) {
+        pthread_mutex_unlock(&plugin->mutex);
+        return NEU_ERR_FAILURE;
+    }
+
+    struct context *c = mqtt_routine_context_create(0, NULL, json, pair, true);
+    if (NULL != c) {
+        mqtt_routine_send(routine, c);
+    }
+
+    pthread_mutex_unlock(&plugin->mutex);
+    return NEU_ERR_SUCCESS;
+}
 
 static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
                                void *data)
@@ -757,7 +789,7 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
     // error = write_response(plugin, req);
     // break;
     case NEU_REQRESP_TRANS_DATA:
-        // error = trans_data(plugin, req);
+        error = trans_data(plugin, data);
         break;
     default:
         error = NEU_ERR_FAILURE;

src/connection/mqtt_c_client.c

@@ -828,9 +828,9 @@ neu_err_code_e mqtt_c_client_publish(mqtt_c_client_t *client, const char *topic,
     assert(NULL != client);
     assert(NULL != topic && 0 < strlen(topic));
 
-    if (SEND_BUF_SIZE < len) {
-        return NEU_ERR_MQTT_PUBLISH_OVER_LENGTH;
-    }
+    // if (SEND_BUF_SIZE < len) {
+    //     return NEU_ERR_MQTT_PUBLISH_OVER_LENGTH;
+    // }
 
     neu_err_code_e error = mqtt_c_client_is_connected(client);
     if (0 != error) {