GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-17 04:40:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2009 from gc87/scan-tags-v2.9

Browse Source 
[v2.9]feat: add scan tags reqest & response
This commit is contained in:
fengzeroz
2024-05-24 09:48:28 +08:00
committed by GitHub
parent 87acec1ede ba07160fc5
commit 0512edbb52
18 changed files with 431 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CMakeLists.txt
+1
View File
@@ -147,6 +147,7 @@ set(NEURON_SOURCES
plugins/restful/group_config_handle.c
plugins/restful/plugin_handle.c
plugins/restful/version_handle.c
plugins/restful/scan_handle.c
plugins/restful/rest.c
plugins/restful/user.c)
include/neuron/adapter.h
+3
View File
@@ -65,6 +65,9 @@ typedef struct adapter_callbacks {
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);
void (*scan_tags_response)(neu_adapter_t *adapter, void *r,
int error, void *tags, neu_type_e type,
bool is_array);
} driver;
};
} adapter_callbacks_t;
include/neuron/errcodes.h
+1
View File
@@ -137,6 +137,7 @@ typedef enum {
NEU_ERR_PLUGIN_NOT_SUPPORT_READ_SYNC = 3018,
NEU_ERR_PLUGIN_TYPE_NOT_SUPPORT = 3019,
NEU_ERR_PLUGIN_TAG_VALUE_OUT_OF_RANGE = 3020,
NEU_ERR_PLUGIN_NOT_SUPPORT_SCAN_TAGS = 3021,
NEU_ERR_MQTT_FAILURE = 4000,
NEU_ERR_MQTT_NO_CERTFILESET = 4001,
include/neuron/json/neu_json_scan.h
+51
View File
@@ -0,0 +1,51 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 3 of the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with this program; if not, write to the Free Software Foundation,
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
/*
* DO NOT EDIT THIS FILE MANUALLY!
* It was automatically generated by `json-autotype`.
*/
#ifndef _NEU_JSON_API_NEU_JSON_SCAN_H_
#define _NEU_JSON_API_NEU_JSON_SCAN_H_
#include "json/json.h"
#include "tag.h"
#ifdef __cplusplus
extern "C" {
#endif
typedef struct {
char *node;
char *id;
} neu_json_scan_tags_req_t;
int neu_json_decode_scan_tags_req(char *buf, neu_json_scan_tags_req_t **result);
void neu_json_decode_scan_tags_req_free(neu_json_scan_tags_req_t *req);
int neu_json_encode_scan_tags_resp(void *json_object, void *param);
#ifdef __cplusplus
}
#endif
#endif
include/neuron/msg.h
+29
View File
@@ -114,6 +114,8 @@ typedef enum neu_reqresp_type {
NEU_REQ_PRGFILE_PROCESS,
NEU_RESP_PRGFILE_PROCESS,
NEU_REQ_SCAN_TAGS,
NEU_RESP_SCAN_TAGS,
} neu_reqresp_type_e;
static const char *neu_reqresp_type_string_t[] = {
@@ -189,6 +191,9 @@ static const char *neu_reqresp_type_string_t[] = {
[NEU_REQ_PRGFILE_UPLOAD] = "NEU_REQ_PRGFILE_UPLOAD",
[NEU_REQ_PRGFILE_PROCESS] = "NEU_REQ_PRGFILE_PROCESS",
[NEU_RESP_PRGFILE_PROCESS] = "NEU_RESP_PRGFILE_PROCESS",
[NEU_REQ_SCAN_TAGS] = "NEU_REQ_SCAN_TAGS",
[NEU_RESP_SCAN_TAGS] = "NEU_RESP_SCAN_TAGS",
};
inline static const char *neu_reqresp_type_string(neu_reqresp_type_e type)
@@ -923,6 +928,30 @@ typedef struct {
char reason[256];
} neu_resp_prgfile_process_t;
typedef struct neu_req_scan_tags {
char driver[NEU_NODE_NAME_LEN];
char id[NEU_TAG_ADDRESS_LEN];
} neu_req_scan_tags_t;
typedef struct {
char name[NEU_TAG_NAME_LEN];
char id[5 + 1 + NEU_TAG_ADDRESS_LEN]; // ns + ! + address
uint8_t type;
} neu_scan_tag_t;
typedef struct {
UT_array * scan_tags;
int32_t error;
neu_type_e type;
bool is_array;
} neu_resp_scan_tags_t;
typedef struct {
char name[NEU_TAG_NAME_LEN];
char id[5 + 1 + NEU_TAG_ADDRESS_LEN];
neu_type_e type;
} neu_scan_tag_attribute_t;
#ifdef __cplusplus
}
#endif
include/neuron/plugin.h
+1
View File
@@ -108,6 +108,7 @@ typedef struct neu_plugin_intf_funs {
neu_datatag_t *tags,
int n_tag); // create tags by API
int (*del_tags)(neu_plugin_t *plugin, int n_tag);
int (*scan_tags)(neu_plugin_t *plugin, void *req, char *id);
} driver;
};
plugins/restful/handle.c
+10
View File
@@ -33,6 +33,7 @@
#include "normal_handle.h"
#include "plugin_handle.h"
#include "rw_handle.h"
#include "scan_handle.h"
#include "utils/http.h"
#include "version_handle.h"
@@ -117,6 +118,9 @@ static struct neu_http_handler cors_handler[] = {
{
.url = "/api/v2/metrics",
},
{
.url = "/api/v2/scan/tags",
},
};
static struct neu_http_handler rest_handlers[] = {
@@ -379,6 +383,12 @@ static struct neu_http_handler rest_handlers[] = {
.url = "/api/v2/metrics",
.value.handler = handle_get_metric,
},
{
.method = NEU_HTTP_METHOD_POST,
.type = NEU_HTTP_HANDLER_FUNCTION,
.url = "/api/v2/scan/tags",
.value.handler = handle_scan_tags,
},
};
void neu_rest_handler(const struct neu_http_handler **handlers, uint32_t *size)
plugins/restful/rest.c
+4
View File
@@ -34,6 +34,7 @@
#include "plugin_handle.h"
#include "rest.h"
#include "rw_handle.h"
#include "scan_handle.h"
#include "utils/http.h"
#include "utils/log.h"
#include "utils/neu_jwt.h"
@@ -223,6 +224,9 @@ static int dashb_plugin_request(neu_plugin_t * plugin,
case NEU_RESP_READ_GROUP:
handle_read_resp(header->ctx, (neu_resp_read_group_t *) data);
break;
case NEU_RESP_SCAN_TAGS:
handle_scan_tags_resp(header->ctx, (neu_resp_scan_tags_t *) data);
break;
default:
nlog_fatal("recv unhandle msg: %s",
neu_reqresp_type_string(header->type));
plugins/restful/scan_handle.c
+68
View File
@@ -0,0 +1,68 @@
#include <stdlib.h>
#include "plugin.h"
#include "utils/log.h"
#include "json/neu_json_fn.h"
#include "json/neu_json_scan.h"
#include "handle.h"
#include "utils/http.h"
#include "scan_handle.h"
void handle_scan_tags(nng_aio *aio)
{
neu_plugin_t *plugin = neu_rest_get_plugin();
NEU_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
aio, neu_json_scan_tags_req_t, neu_json_decode_scan_tags_req, {
neu_reqresp_head_t header = { 0 };
neu_req_scan_tags_t cmd = { 0 };
int err_type;
nng_http_req *nng_req = nng_aio_get_input(aio, 0);
nlog_notice("<%p> req %s %s", aio, nng_http_req_get_method(nng_req),
nng_http_req_get_uri(nng_req));
header.ctx = aio;
header.type = NEU_REQ_SCAN_TAGS;
if (NULL != req->node && NEU_NODE_NAME_LEN <= strlen(req->node)) {
err_type = NEU_ERR_NODE_NAME_TOO_LONG;
goto error;
}
if (NULL != req->id && NEU_TAG_ADDRESS_LEN <= strlen(req->id)) {
err_type = NEU_ERR_NODE_NAME_TOO_LONG;
goto error;
}
if (NULL != req->node) {
strcpy(cmd.driver, req->node);
}
if (NULL != req->id) {
strcpy(cmd.id, req->id);
}
if (0 != neu_plugin_op(plugin, header, &cmd)) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
neu_http_response(aio, NEU_ERR_IS_BUSY, result_error);
});
}
goto success;
error:
NEU_JSON_RESPONSE_ERROR(
err_type, { neu_http_response(aio, err_type, result_error); });
success:;
})
}
void handle_scan_tags_resp(nng_aio *aio, neu_resp_scan_tags_t *resp)
{
char *result = NULL;
neu_json_encode_by_fn(resp, neu_json_encode_scan_tags_resp, &result);
neu_http_ok(aio, result);
free(result);
}
plugins/restful/scan_handle.h
+29
View File
@@ -0,0 +1,29 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 3 of the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with this program; if not, write to the Free Software Foundation,
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#ifndef _NEU_SCAN_HANDLE_H_
#define _NEU_ACAN_HANDLE_H_
#include <nng/nng.h>
void handle_scan_tags(nng_aio *aio);
void handle_scan_tags_resp(nng_aio *aio, neu_resp_scan_tags_t *resp);
void handle_scan_tag(nng_aio *aio);
#endif
src/adapter/adapter.c
+15
View File
@@ -523,6 +523,11 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
strcpy(pheader->receiver, header.sender);
break;
}
case NEU_REQ_SCAN_TAGS: {
neu_req_scan_tags_t *cmd = (neu_req_scan_tags_t *) data;
strcpy(pheader->receiver, cmd->driver);
break;
}
default:
break;
}
@@ -1181,6 +1186,16 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
break;
}
case NEU_REQ_SCAN_TAGS: {
neu_adapter_driver_scan_tags((neu_adapter_driver_t *) adapter, header);
break;
}
case NEU_RESP_SCAN_TAGS: {
adapter->module->intf_funs->request(
adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
neu_msg_free(msg);
break;
}
default:
nlog_warn("adapter: %s recv msg type error, type: %s", adapter->name,
neu_reqresp_type_string(header->type));
src/adapter/driver/driver.c
+50 -7
View File
@@ -322,16 +322,33 @@ static void update(neu_adapter_t *adapter, const char *group, const char *tag,
update_with_meta(adapter, group, tag, value, NULL, 0);
}
static void scan_tags_response(neu_adapter_t *adapter, void *r, neu_error error,
void *tags, neu_type_e type, bool is_array)
{
neu_reqresp_head_t * req = (neu_reqresp_head_t *) r;
neu_resp_scan_tags_t value = {
.scan_tags = tags,
.error = error,
.type = type,
.is_array = is_array,
};
req->type = NEU_RESP_SCAN_TAGS;
nlog_notice("scan tags response <%p>", req->ctx);
adapter->cb_funs.response(adapter, req, &value);
}
neu_adapter_driver_t *neu_adapter_driver_create()
{
neu_adapter_driver_t *driver = calloc(1, sizeof(neu_adapter_driver_t));
driver->cache = neu_driver_cache_new();
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.update_im = update_im;
driver->adapter.cb_funs.driver.update_with_meta = update_with_meta;
driver->cache = neu_driver_cache_new();
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.update_im = update_im;
driver->adapter.cb_funs.driver.update_with_meta = update_with_meta;
driver->adapter.cb_funs.driver.scan_tags_response = scan_tags_response;
return driver;
}
@@ -2324,4 +2341,30 @@ void neu_adapter_driver_unsubscribe(neu_adapter_driver_t * driver,
}
}
pthread_mutex_unlock(&find->apps_mtx);
}
}
void neu_adapter_driver_scan_tags(neu_adapter_driver_t *driver,
neu_reqresp_head_t * req)
{
if (driver->adapter.state != NEU_NODE_RUNNING_STATE_RUNNING) {
driver->adapter.cb_funs.driver.scan_tags_response(
&driver->adapter, req, NEU_ERR_PLUGIN_NOT_RUNNING, NULL,
NEU_TYPE_ERROR, false);
nlog_warn("%s not running", driver->adapter.name);
return;
}
if (driver->adapter.module->intf_funs->driver.scan_tags == NULL) {
driver->adapter.cb_funs.driver.scan_tags_response(
&driver->adapter, req, NEU_ERR_PLUGIN_NOT_SUPPORT_SCAN_TAGS, NULL,
NEU_TYPE_ERROR, false);
nlog_warn("%s not support scan tags", driver->adapter.name);
return;
}
neu_req_scan_tags_t *cmd = (neu_req_scan_tags_t *) &req[1];
driver->adapter.module->intf_funs->driver.scan_tags(driver->adapter.plugin,
req, cmd->id);
}
src/adapter/driver/driver_internal.h
+3
View File
@@ -87,4 +87,7 @@ void neu_adapter_driver_subscribe(neu_adapter_driver_t *driver,
neu_req_subscribe_t * req);
void neu_adapter_driver_unsubscribe(neu_adapter_driver_t * driver,
neu_req_unsubscribe_t *req);
void neu_adapter_driver_scan_tags(neu_adapter_driver_t *driver,
neu_reqresp_head_t * req);
#endif
src/base/msg_internal.h
+3 -1
View File
@@ -93,7 +93,9 @@ extern "C" {
XX(NEU_REQ_UPDATE_LOG_LEVEL, neu_req_update_log_level_t) \
XX(NEU_REQ_PRGFILE_UPLOAD, neu_req_prgfile_upload_t) \
XX(NEU_REQ_PRGFILE_PROCESS, neu_req_prgfile_process_t) \
XX(NEU_RESP_PRGFILE_PROCESS, neu_resp_prgfile_process_t)
XX(NEU_RESP_PRGFILE_PROCESS, neu_resp_prgfile_process_t) \
XX(NEU_REQ_SCAN_TAGS, neu_req_scan_tags_t) \
XX(NEU_RESP_SCAN_TAGS, neu_resp_scan_tags_t)
static inline size_t neu_reqresp_size(neu_reqresp_type_e t)
{
src/core/manager.c
+2
View File
@@ -1036,6 +1036,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
}
case NEU_REQ_PRGFILE_PROCESS:
case NEU_REQ_PRGFILE_UPLOAD:
case NEU_REQ_SCAN_TAGS:
case NEU_REQ_GET_NODE_STATE: {
if (neu_node_manager_find(manager->node_manager, header->receiver) ==
NULL) {
@@ -1245,6 +1246,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_RESP_ERROR:
case NEU_RESP_READ_GROUP:
case NEU_RESP_PRGFILE_PROCESS:
case NEU_RESP_SCAN_TAGS:
forward_msg(manager, header, header->receiver);
break;
src/parser/neu_json_scan.c
+154
View File
@@ -0,0 +1,154 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 3 of the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with this program; if not, write to the Free Software Foundation,
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
/*
* DO NOT EDIT THIS FILE MANUALLY!
* It was automatically generated by `json-autotype`.
*/
#include <stdlib.h>
#include <string.h>
#include <jansson.h>
#include "define.h"
#include "msg.h"
#include "json/json.h"
#include "json/neu_json_scan.h"
int neu_json_decode_scan_tags_req(char *buf, neu_json_scan_tags_req_t **result)
{
void *json_obj = neu_json_decode_new(buf);
if (NULL == json_obj) {
return -1;
}
neu_json_scan_tags_req_t *req = calloc(1, sizeof(neu_json_scan_tags_req_t));
if (req == NULL) {
return -1;
}
neu_json_elem_t req_node_elems[] = { {
.name = "node",
.t = NEU_JSON_STR,
} };
neu_json_elem_t req_id_elems[] = {
{
.name = "id",
.t = NEU_JSON_STR,
},
};
if (0 !=
neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_node_elems),
req_node_elems)) {
neu_json_decode_free(json_obj);
free(req);
return -1;
}
neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_id_elems),
req_id_elems);
req->node = req_node_elems[0].v.val_str;
req->id = req_id_elems[0].v.val_str;
*result = req;
neu_json_decode_free(json_obj);
return 0;
}
void neu_json_decode_scan_tags_req_free(neu_json_scan_tags_req_t *req)
{
free(req->node);
free(req->id);
free(req);
}
int neu_json_encode_scan_tags_resp(void *json_object, void *param)
{
int ret = 0;
neu_resp_scan_tags_t *resp = (neu_resp_scan_tags_t *) param;
void * tag_array = neu_json_array();
int len = 0;
if (NULL != resp->scan_tags) {
len = utarray_len(resp->scan_tags);
utarray_foreach(resp->scan_tags, neu_scan_tag_t *, p_tag)
{
if (NULL != p_tag) {
neu_json_elem_t tag_elems[] = {
{
.name = "name",
.t = NEU_JSON_STR,
.v.val_str = p_tag->name,
},
{
.name = "id",
.t = NEU_JSON_STR,
.v.val_str = p_tag->id,
},
{
.name = "tag",
.t = NEU_JSON_INT,
.v.val_int = p_tag->type,
},
};
neu_json_encode_array(tag_array, tag_elems,
NEU_JSON_ELEM_SIZE(tag_elems));
}
}
utarray_free(resp->scan_tags);
}
neu_json_elem_t resp_elems[] = { {
.name = "error",
.t = NEU_JSON_INT,
.v.val_int = resp->error,
},
{
.name = "total",
.t = NEU_JSON_INT,
.v.val_int = len,
},
{
.name = "type",
.t = NEU_JSON_INT,
.v.val_int = resp->type,
},
{
.name = "is_array",
.t = NEU_JSON_BOOL,
.v.val_bool = resp->is_array,
},
{
.name = "tags",
.t = NEU_JSON_OBJECT,
.v.val_object = tag_array,
} };
ret = neu_json_encode_field(json_object, resp_elems,
NEU_JSON_ELEM_SIZE(resp_elems));
return ret;
}
tests/ft/neuron/api.py
+6 -1
View File
@@ -349,4 +349,9 @@ def mqtt_node_setting(node):
"offline-cache": False, "cache-sync-interval": 100, "host": "broker.emqx.io", "port": 1883, "username": "", "password": "", "ssl": False})
def get_nodes_disable_auth(type):
return requests.get(url=config.BASE_URL + '/api/v2/node', params={"type": type})
return requests.get(url=config.BASE_URL + '/api/v2/node', params={"type": type})
@gen_check
def scan_tags(node, id):
return requests.post(url=config.BASE_URL + "/api/v2/scan/tags", headers={"Authorization": config.default_jwt}, json={"node": node, "id": id})
tests/ft/neuron/error.py
+1
View File
@@ -107,5 +107,6 @@ NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS = 3017
NEU_ERR_PLUGIN_NOT_SUPPORT_READ_SYNC = 3018
NEU_ERR_PLUGIN_TYPE_NOT_SUPPORT = 3019
NEU_ERR_PLUGIN_TAG_VALUE_OUT_OF_RANGE = 3020
NEU_ERR_PLUGIN_NOT_SUPPORT_SCAN_TAGS = 3021
NEU_ERR_MQTT_SUBSCRIBE_FAILURE = 4010