Merge pull request #2401 from fengzeroz/main

core: write tags resp

.github/workflows/check.yml

@@ -81,7 +81,7 @@ jobs:
     container: ghcr.io/neugates/build:x86_64-main
     strategy:
       matrix:
-        plugin: [core, ekuiper, modbus, mqtt, azure, metrics]
+        plugin: [core, ekuiper, modbus, mqtt, azure, metrics, launch]
 
     steps:
       - uses: actions/checkout@v4
@@ -109,7 +109,10 @@ jobs:
               --ignore=tests/ft/app/test_mqtt.py             \
               --ignore=tests/ft/app/test_azure.py            \
               --ignore=tests/ft/driver/test_modbus.py        \
-              --ignore=tests/ft/metrics/test_metrics.py
+              --ignore=tests/ft/metrics/test_metrics.py      \
+              --ignore=tests/ft/login/test_launch.py         \
+              --ignore=tests/ft/login/test_filter_error.py   \
+              --ignore=tests/ft/login/test_login.py
           elif [ "${{ matrix.plugin }}" = "ekuiper" ]; then
             pytest -s -v tests/ft/app/"test_ekuiper.py"
           elif [ "${{ matrix.plugin }}" = "modbus" ]; then
@@ -118,11 +121,14 @@ jobs:
             pytest -s -v tests/ft/app/"test_mqtt.py"
           elif [ "${{ matrix.plugin }}" = "azure" ]; then
             pytest -s -v tests/ft/app/"test_azure.py"
+          #elif [ "${{ matrix.plugin }}" = "launch" ]; then
+            #pytest -s -v tests/ft/login
           else
             pytest -s -v tests/ft/metrics/"test_metrics.py"          
           fi
 
       - name: create cov report
+        if: ${{ matrix.plugin != 'launch' }}
         run: |
             sudo apt-get update
             sudo apt-get -y install lcov
@@ -131,6 +137,7 @@ jobs:
 
       - name: Upload coverage to Codecov
         uses: codecov/codecov-action@v4
+        if: ${{ matrix.plugin != 'launch' }}
         with:
           verbose: true
           fail_ci_if_error: true

include/neuron/adapter.h

@@ -51,6 +51,13 @@ typedef struct {
     int64_t mtime;
 } neu_driver_file_info_t;
 
+typedef struct {
+    char group[NEU_GROUP_NAME_LEN];
+    char name[NEU_TAG_NAME_LEN];
+
+    int error;
+} neu_driver_write_responses_t;
+
 typedef struct adapter_callbacks {
     int (*command)(neu_adapter_t *adapter, neu_reqresp_head_t head, void *data);
     int (*response)(neu_adapter_t *adapter, neu_reqresp_head_t *head,
@@ -73,6 +80,9 @@ typedef struct adapter_callbacks {
                                      neu_tag_meta_t *metas, int n_meta);
             void (*write_response)(neu_adapter_t *adapter, void *req,
                                    int error);
+            void (*write_responses)(neu_adapter_t *adapter, void *req,
+                                    neu_driver_write_responses_t *responses,
+                                    int                           n_response);
             void (*update_im)(neu_adapter_t *adapter, const char *group,
                               const char *tag, neu_dvalue_t value,
                               neu_tag_meta_t *metas, int n_meta);

include/neuron/json/neu_json_rw.h

@@ -206,6 +206,12 @@ void neu_json_decode_test_read_tag_req_free(neu_json_test_read_tag_req_t *req);
 
 int neu_json_encode_test_read_tag_resp(void *json_object, void *param);
 
+typedef struct {
+    UT_array *tags;
+} neu_json_write_tags_resp_t;
+
+int neu_json_encode_write_tags_resp(void *json_object, void *param);
+
 #ifdef __cplusplus
 }
 #endif

include/neuron/msg.h

@@ -52,6 +52,7 @@ typedef enum neu_reqresp_type {
     NEU_RESP_TEST_READ_TAG,
     NEU_REQ_WRITE_TAG,
     NEU_REQ_WRITE_TAGS,
+    NEU_RESP_WRITE_TAGS,
     NEU_REQ_WRITE_GTAGS,
 
     NEU_REQ_SUBSCRIBE_GROUP,
@@ -177,6 +178,7 @@ static const char *neu_reqresp_type_string_t[] = {
     [NEU_RESP_TEST_READ_TAG]       = "NEU_RESP_TEST_READ_TAG",
     [NEU_REQ_WRITE_TAG]            = "NEU_REQ_WRITE_TAG",
     [NEU_REQ_WRITE_TAGS]           = "NEU_REQ_WRITE_TAGS",
+    [NEU_RESP_WRITE_TAGS]          = "NEU_RESP_WRITE_TAGS",
     [NEU_REQ_WRITE_GTAGS]          = "NEU_REQ_WRITE_GTAGS",
 
     [NEU_REQ_SUBSCRIBE_GROUP]        = "NEU_REQ_SUBSCRIBE_GROUP",
@@ -878,6 +880,16 @@ typedef struct neu_req_write_tags {
     neu_resp_tag_value_t *tags;
 } neu_req_write_tags_t;
 
+typedef struct neu_resp_write_tags_ele {
+    char group[NEU_GROUP_NAME_LEN];
+    char tag[NEU_TAG_NAME_LEN];
+    int  error;
+} neu_resp_write_tags_ele_t;
+
+typedef struct neu_resp_write_tags {
+    UT_array *tags; // array neu_resp_write_tags_ele_t
+} neu_resp_write_tags_t;
+
 static inline void neu_req_write_tags_fini(neu_req_write_tags_t *req)
 {
     free(req->driver);

plugins/mqtt/mqtt_handle.c

@@ -743,6 +743,81 @@ end:
     return rv;
 }
 
+int handle_write_tags_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
+                               neu_resp_write_tags_t *data)
+{
+    int    rv       = 0;
+    char * json_str = NULL;
+    size_t size     = 0;
+
+    if (NULL == plugin->client) {
+        rv = NEU_ERR_MQTT_IS_NULL;
+        goto end;
+    }
+
+    if (0 == plugin->config.cache &&
+        !neu_mqtt_client_is_connected(plugin->client)) {
+        // cache disable and we are disconnected
+        rv = NEU_ERR_MQTT_FAILURE;
+        goto end;
+    }
+
+    if (plugin->config.format == MQTT_UPLOAD_FORMAT_PROTOBUF) {
+        Model__WriteResponse resp = MODEL__WRITE_RESPONSE__INIT;
+
+        resp.uuid     = mqtt_json->uuid;
+        resp.n_errors = utarray_len(data->tags);
+        resp.errors = calloc(resp.n_errors, sizeof(Model__WriteResponseItem *));
+        int index   = 0;
+
+        utarray_foreach(data->tags, neu_resp_write_tags_ele_t *, tag)
+        {
+            resp.errors[index] = calloc(1, sizeof(Model__WriteResponseItem));
+            model__write_response_item__init(resp.errors[index]);
+            resp.errors[index]->name  = strdup(tag->tag);
+            resp.errors[index]->error = tag->error;
+
+            index++;
+        }
+
+        size     = model__write_response__get_packed_size(&resp);
+        json_str = malloc(size);
+        model__write_response__pack(&resp, (uint8_t *) json_str);
+
+        for (size_t i = 0; i < resp.n_errors; i++) {
+            free(resp.errors[i]->name);
+            free(resp.errors[i]);
+        }
+        free(resp.errors);
+    } else {
+        neu_json_write_tags_resp_t write_resp = { 0 };
+        write_resp.tags                       = data->tags;
+
+        neu_json_encode_with_mqtt(&write_resp, neu_json_encode_write_tags_resp,
+                                  mqtt_json, neu_json_encode_mqtt_resp,
+                                  &json_str);
+        if (NULL == json_str) {
+            plog_error(plugin,
+                       "generate driver write tags resp json fail, uuid:%s",
+                       mqtt_json->uuid);
+            rv = NEU_ERR_EINTERNAL;
+            goto end;
+        }
+
+        size = strlen(json_str);
+    }
+
+    char *         topic = plugin->config.write_resp_topic;
+    neu_mqtt_qos_e qos   = plugin->config.qos;
+    rv                   = publish(plugin, qos, topic, json_str, size);
+    json_str             = NULL;
+
+end:
+    neu_json_decode_mqtt_req_free(mqtt_json);
+    utarray_free(data->tags);
+    return rv;
+}
+
 int handle_driver_action_response(neu_plugin_t *            plugin,
                                   neu_json_mqtt_t *         mqtt_json,
                                   neu_resp_driver_action_t *data)

plugins/mqtt/mqtt_handle.h

@@ -42,6 +42,8 @@ void handle_write_req(neu_mqtt_qos_e qos, const char *topic,
 int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
                           neu_resp_error_t *data, void *trace_scope,
                           void *trace_ctx, char *span_id);
+int handle_write_tags_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
+                               neu_resp_write_tags_t *data);
 
 void handle_read_req(neu_mqtt_qos_e qos, const char *topic,
                      const uint8_t *payload, uint32_t len, void *data,

plugins/mqtt/mqtt_plugin_intf.c

@@ -550,6 +550,9 @@ int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
         error = handle_write_response(plugin, head->ctx, data, scope, trace,
                                       new_span_id);
         break;
+    case NEU_RESP_WRITE_TAGS:
+        error = handle_write_tags_response(plugin, head->ctx, data);
+        break;
     case NEU_RESP_DRIVER_ACTION:
         error = handle_driver_action_response(plugin, head->ctx, data);
         break;

plugins/restful/rest.c

@@ -212,6 +212,10 @@ static int dashb_plugin_request(neu_plugin_t *      plugin,
         }
         break;
     }
+    case NEU_RESP_WRITE_TAGS: {
+        handle_write_tags_resp(header->ctx, (neu_resp_write_tags_t *) data);
+        break;
+    }
     case NEU_RESP_CHECK_SCHEMA: {
         handle_get_plugin_schema_resp(header->ctx,
                                       (neu_resp_check_schema_t *) data);

plugins/restful/rw_handle.c

@@ -986,4 +986,13 @@ void handle_test_read_tag_resp(nng_aio *aio, neu_resp_test_read_tag_t *resp)
     neu_json_encode_by_fn(resp, neu_json_encode_test_read_tag_resp, &result);
     neu_http_ok(aio, result);
     free(result);
+}
+
+void handle_write_tags_resp(nng_aio *aio, neu_resp_write_tags_t *resp)
+{
+    char *result = NULL;
+    neu_json_encode_by_fn(resp, neu_json_encode_write_tags_resp, &result);
+    neu_http_ok(aio, result);
+    free(result);
+    utarray_free(resp->tags);
 }

plugins/restful/rw_handle.h

@@ -36,5 +36,6 @@ void handle_read_resp(nng_aio *aio, neu_resp_read_group_t *resp);
 void handle_read_paginate_resp(nng_aio *                       aio,
                                neu_resp_read_group_paginate_t *resp);
 void handle_test_read_tag_resp(nng_aio *aio, neu_resp_test_read_tag_t *resp);
+void handle_write_tags_resp(nng_aio *aio, neu_resp_write_tags_t *resp);
 
 #endif

src/adapter/adapter.c

@@ -790,6 +790,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_REQ_UNSUBSCRIBE_GROUP_EVENT:
     case NEU_REQ_UPDATE_SUBSCRIBE_GROUP_EVENT:
     case NEU_REQ_SUBSCRIBE_GROUPS_EVENT:
+    case NEU_RESP_WRITE_TAGS:
         adapter->module->intf_funs->request(
             adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
         neu_msg_free(msg);

src/adapter/driver/driver.c

@@ -116,6 +116,9 @@ static void update_with_meta(neu_adapter_t *adapter, const char *group,
                              const char *tag, neu_dvalue_t value,
                              neu_tag_meta_t *metas, int n_meta);
 static void write_response(neu_adapter_t *adapter, void *r, neu_error error);
+static void write_responses(neu_adapter_t *adapter, void *r,
+                            neu_driver_write_responses_t *response,
+                            int                           n_response);
 static void directory_response(neu_adapter_t *adapter, void *req, int error,
                                neu_driver_file_info_t *infos, int n_info);
 static void fup_open_response(neu_adapter_t *adapter, void *req, int error,
@@ -167,6 +170,32 @@ static void format_tag_value(neu_dvalue_t *value)
     value->value.d64 *= negative;
 }
 
+static void write_responses(neu_adapter_t *adapter, void *r,
+                            neu_driver_write_responses_t *response,
+                            int                           n_response)
+{
+    neu_reqresp_head_t *  req   = (neu_reqresp_head_t *) r;
+    neu_resp_write_tags_t nresp = { 0 };
+    req->type                   = NEU_RESP_WRITE_TAGS;
+    UT_icd    icd   = { sizeof(neu_resp_write_tags_ele_t), NULL, NULL, NULL };
+    UT_array *array = NULL;
+
+    utarray_new(array, &icd);
+
+    for (int i = 0; i < n_response; i++) {
+        neu_driver_write_responses_t *resp = &response[i];
+        neu_resp_write_tags_ele_t     ele  = { 0 };
+
+        strcpy(ele.group, resp->group);
+        strcpy(ele.tag, resp->name);
+        ele.error = resp->error;
+        utarray_push_back(array, &ele);
+    }
+
+    nresp.tags = array;
+    adapter->cb_funs.response(adapter, req, &nresp);
+}
+
 static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
 {
     neu_reqresp_head_t *req    = (neu_reqresp_head_t *) r;
@@ -507,6 +536,7 @@ neu_adapter_driver_t *neu_adapter_driver_create()
     driver->driver_events                              = neu_event_new();
     driver->adapter.cb_funs.driver.update              = update;
     driver->adapter.cb_funs.driver.write_response      = write_response;
+    driver->adapter.cb_funs.driver.write_responses     = write_responses;
     driver->adapter.cb_funs.driver.directory_response  = directory_response;
     driver->adapter.cb_funs.driver.fup_open_response   = fup_open_response;
     driver->adapter.cb_funs.driver.fdown_open_response = fdown_open_response;

src/base/msg_internal.h

@@ -130,7 +130,8 @@ extern "C" {
     XX(NEU_REQ_SUBSCRIBE_GROUP_EVENT, neu_req_subscribe_t)           \
     XX(NEU_REQ_UNSUBSCRIBE_GROUP_EVENT, neu_req_unsubscribe_t)       \
     XX(NEU_REQ_UPDATE_SUBSCRIBE_GROUP_EVENT, neu_req_subscribe_t)    \
-    XX(NEU_REQ_SUBSCRIBE_GROUPS_EVENT, neu_req_subscribe_groups_t)
+    XX(NEU_REQ_SUBSCRIBE_GROUPS_EVENT, neu_req_subscribe_groups_t)   \
+    XX(NEU_RESP_WRITE_TAGS, neu_resp_write_tags_t)
 
 static inline size_t neu_reqresp_size(neu_reqresp_type_e t)
 {

src/core/manager.c

@@ -1374,6 +1374,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_RESP_TEST_READ_TAG:
     case NEU_RESP_PRGFILE_PROCESS:
     case NEU_RESP_SCAN_TAGS:
+    case NEU_RESP_WRITE_TAGS:
         forward_msg(manager, header, header->receiver);
         break;
 

src/parser/neu_json_rw.c

@@ -1436,3 +1436,43 @@ void neu_json_decode_write_gtags_req_free(neu_json_write_gtags_req_t *req)
     free(req->groups);
     free(req);
 }
+
+int neu_json_encode_write_tags_resp(void *json_object, void *param)
+{
+    int                         ret   = 0;
+    neu_json_write_tags_resp_t *resp  = (neu_json_write_tags_resp_t *) param;
+    void *                      array = json_array();
+
+    utarray_foreach(resp->tags, neu_resp_write_tags_ele_t *, ele)
+    {
+        neu_json_elem_t node_elems[] = {
+            {
+                .name      = "group",
+                .t         = NEU_JSON_STR,
+                .v.val_str = ele->group,
+            },
+            {
+                .name      = "name",
+                .t         = NEU_JSON_STR,
+                .v.val_str = ele->tag,
+            },
+            {
+                .name      = "error",
+                .t         = NEU_JSON_INT,
+                .v.val_int = ele->error,
+            },
+        };
+        array = neu_json_encode_array(array, node_elems,
+                                      NEU_JSON_ELEM_SIZE(node_elems));
+    }
+
+    neu_json_elem_t resp_elems[] = { {
+        .name         = "tags",
+        .t            = NEU_JSON_OBJECT,
+        .v.val_object = array,
+    } };
+    ret = neu_json_encode_field(json_object, resp_elems,
+                                NEU_JSON_ELEM_SIZE(resp_elems));
+
+    return ret;
+}