diff --git a/.github/workflows/check.yml b/.github/workflows/check.yml index d5f81313..602144c0 100644 --- a/.github/workflows/check.yml +++ b/.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 diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index 7a1c9128..4ac562ed 100644 --- a/include/neuron/adapter.h +++ b/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); diff --git a/include/neuron/json/neu_json_rw.h b/include/neuron/json/neu_json_rw.h index 97d02438..aea758a9 100644 --- a/include/neuron/json/neu_json_rw.h +++ b/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 diff --git a/include/neuron/msg.h b/include/neuron/msg.h index a3d66d57..d70c2fc4 100644 --- a/include/neuron/msg.h +++ b/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); diff --git a/plugins/mqtt/mqtt_handle.c b/plugins/mqtt/mqtt_handle.c index 0bd4291e..629fe768 100644 --- a/plugins/mqtt/mqtt_handle.c +++ b/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) diff --git a/plugins/mqtt/mqtt_handle.h b/plugins/mqtt/mqtt_handle.h index 3c96d560..f9624df3 100644 --- a/plugins/mqtt/mqtt_handle.h +++ b/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, diff --git a/plugins/mqtt/mqtt_plugin_intf.c b/plugins/mqtt/mqtt_plugin_intf.c index 8e93491c..30e5b095 100644 --- a/plugins/mqtt/mqtt_plugin_intf.c +++ b/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; diff --git a/plugins/restful/rest.c b/plugins/restful/rest.c index 41f52cb9..6bcbdcb5 100644 --- a/plugins/restful/rest.c +++ b/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); diff --git a/plugins/restful/rw_handle.c b/plugins/restful/rw_handle.c index 99d89bb9..483e7483 100644 --- a/plugins/restful/rw_handle.c +++ b/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); } \ No newline at end of file diff --git a/plugins/restful/rw_handle.h b/plugins/restful/rw_handle.h index 9a8b5ad0..6e991771 100644 --- a/plugins/restful/rw_handle.h +++ b/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 \ No newline at end of file diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index bef42862..7005242d 100644 --- a/src/adapter/adapter.c +++ b/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); diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index efb3f9e9..7fb83562 100644 --- a/src/adapter/driver/driver.c +++ b/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; diff --git a/src/base/msg_internal.h b/src/base/msg_internal.h index eeef6da6..537010a4 100644 --- a/src/base/msg_internal.h +++ b/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) { diff --git a/src/core/manager.c b/src/core/manager.c index 7253157c..a2fdbdb6 100644 --- a/src/core/manager.c +++ b/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; diff --git a/src/parser/neu_json_rw.c b/src/parser/neu_json_rw.c index a8e71698..015be63c 100644 --- a/src/parser/neu_json_rw.c +++ b/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; +} \ No newline at end of file