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

feat(plugin): add ThingsNeural north plugin

Browse Source
This commit is contained in:
hxy7yx
2026-10-25 19:04:51 -07:00
parent 6750b7f828
commit b7211ebe9d
15 changed files with 1515 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CMakeLists.txt
+1
View File
@@ -197,6 +197,7 @@ file(COPY ${CMAKE_SOURCE_DIR}/default_plugins.json DESTINATION ${CMAKE_BINARY_DI
add_subdirectory(plugins/modbus)
add_subdirectory(plugins/mqtt)
add_subdirectory(plugins/thingsneural)
add_subdirectory(plugins/ekuiper)
add_subdirectory(plugins/file)
add_subdirectory(plugins/monitor)
default_plugins.json
+1
View File
@@ -1,6 +1,7 @@
{
"plugins": [
"libplugin-mqtt.so",
"libplugin-thingsneural.so",
"libplugin-aws-iot.so",
"libplugin-azure-iot.so",
"libplugin-ekuiper.so",
plugins/restful/adapter_handle.c
+3 -3
View File
@@ -64,9 +64,9 @@ static bool tags_check_symbol(const char *tags)
{
for (size_t i = 0; i < strlen(tags); i++) {
char c = tags[i];
if (c == '[' || c == ']' || c == '{' || c == '}' || c == ';' ||
c == ':' || c == '\'' || c == '"' || c == '<' || c == '>' ||
c == '?' || c == '\\' || c == '|' || c == '`' || c == '\t' ||
if (c == '[' || c == ']' || c == '{' || c == '}' || c == ';' ||
c == ':' || c == '\'' || c == '"' || c == '<' || c == '>' ||
c == '?' || c == '\\' || c == '|' || c == '`' || c == '\t' ||
c == '\n') {
return false;
}
plugins/restful/normal_handle.c
+4
View File
@@ -479,6 +479,10 @@ static struct {
.plugin = "MQTT",
.schema = "mqtt",
},
{
.plugin = "ThingsNeural",
.schema = "thingsneural",
},
{
.plugin = "NON A11",
.schema = "nona11",
plugins/thingsneural/CMakeLists.txt
+23
View File
@@ -0,0 +1,23 @@
cmake_minimum_required(VERSION 3.12)
project(plugin-thingsneural)
set(LIBRARY_OUTPUT_PATH "${CMAKE_BINARY_DIR}/plugins")
file(COPY ${CMAKE_SOURCE_DIR}/plugins/thingsneural/thingsneural.json
DESTINATION ${CMAKE_BINARY_DIR}/plugins/schema/)
add_library(${PROJECT_NAME} SHARED
thingsneural_config.c
thingsneural_handle.c
thingsneural_plugin.c
thingsneural_plugin_intf.c
)
target_include_directories(${PROJECT_NAME} PRIVATE
${CMAKE_SOURCE_DIR}/include/neuron
${CMAKE_SOURCE_DIR}/plugins/thingsneural
)
target_link_libraries(${PROJECT_NAME} neuron-base)
target_link_libraries(${PROJECT_NAME} ${CMAKE_THREAD_LIBS_INIT})
plugins/thingsneural/thingsneural.json
+137
View File
@@ -0,0 +1,137 @@
{
"version": {
"name": "MQTT Version",
"name_zh": "MQTT 版本",
"type": "map",
"attribute": "optional",
"default": 4,
"valid": {
"map": [
{
"key": "3.1.1",
"value": 4
},
{
"key": "5.0",
"value": 5
}
]
}
},
"client-id": {
"name": "Client ID",
"name_zh": "客户端 ID",
"type": "string",
"attribute": "required",
"default": "neuron_${random_str}",
"valid": {
"length": 255
}
},
"qos": {
"name": "QoS Level",
"name_zh": "QoS 等级",
"type": "map",
"attribute": "optional",
"default": 0,
"valid": {
"map": [
{
"key": "QoS 0",
"value": 0
},
{
"key": "QoS 1",
"value": 1
},
{
"key": "QoS 2",
"value": 2
}
]
}
},
"host": {
"name": "Broker Host",
"name_zh": "服务器地址",
"attribute": "required",
"type": "string",
"default": "127.0.0.1",
"valid": {
"length": 255
}
},
"port": {
"name": "Broker Port",
"name_zh": "服务器端口",
"attribute": "required",
"type": "int",
"default": 1883,
"valid": {
"min": 1,
"max": 65535
}
},
"username": {
"name": "Username",
"name_zh": "用户名",
"attribute": "optional",
"type": "string",
"default": "",
"valid": {
"length": 255
}
},
"password": {
"name": "Password",
"name_zh": "密码",
"attribute": "optional",
"type": "string",
"default": "",
"valid": {
"length": 255
}
},
"ssl": {
"name": "SSL",
"name_zh": "SSL",
"attribute": "optional",
"type": "bool",
"default": false,
"valid": {}
},
"product-key": {
"name": "Product Key",
"name_zh": "产品标识",
"description": "ThingsNeural productKey used in NLINK topics.",
"description_zh": "拼 NLINK 主题使用的 productKey。",
"attribute": "required",
"type": "string",
"valid": {
"length": 128
}
},
"device-key": {
"name": "Device Key",
"name_zh": "设备标识",
"description": "ThingsNeural deviceKey used in NLINK topics.",
"description_zh": "拼 NLINK 主题使用的 deviceKey。",
"attribute": "required",
"type": "string",
"valid": {
"length": 128
}
},
"points": {
"name": "Point Map",
"name_zh": "点位映射",
"description": "JSON array string. Each item maps a thing-model property to a south node, group and tag.",
"description_zh": "JSON 数组字符串。每一项把物模型属性映射到南向 node、group 和 tag。",
"attribute": "required",
"type": "string",
"default": "[]",
"valid": {
"length": 81960
}
}
}
plugins/thingsneural/thingsneural_config.c
+279
View File
@@ -0,0 +1,279 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2026 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.
**/
#include <jansson.h>
#include <string.h>
#include "utils/asprintf.h"
#include "thingsneural_plugin.h"
static int copy_name(char *dst, size_t dst_len, const char *src)
{
size_t len = 0;
if (NULL == src || '\0' == src[0]) {
return -1;
}
len = strlen(src);
if (len >= dst_len) {
return -1;
}
memcpy(dst, src, len + 1);
return 0;
}
static char *build_topic(const char *suffix, const char *product_key,
const char *device_key)
{
char *topic = NULL;
if (neu_asprintf(&topic, "/iot/thing/%s/%s/property/%s", product_key,
device_key, suffix) <= 0) {
return NULL;
}
return topic;
}
static void points_free(tn_point_t **points)
{
tn_point_t *point = NULL;
tn_point_t *tmp = NULL;
HASH_ITER(hh, *points, point, tmp)
{
HASH_DEL(*points, point);
free(point);
}
}
void tn_points_free(tn_point_t **points)
{
points_free(points);
}
void tn_points_rename_node(tn_point_t *points, const char *node,
const char *new_name)
{
tn_point_t *point = NULL;
tn_point_t *tmp = NULL;
HASH_ITER(hh, points, point, tmp)
{
if (0 == strcmp(point->node, node)) {
snprintf(point->node, sizeof(point->node), "%s", new_name);
}
}
}
void tn_points_rename_group(tn_point_t *points, const char *node,
const char *group, const char *new_name)
{
tn_point_t *point = NULL;
tn_point_t *tmp = NULL;
HASH_ITER(hh, points, point, tmp)
{
if (0 == strcmp(point->node, node) &&
0 == strcmp(point->group, group)) {
snprintf(point->group, sizeof(point->group), "%s", new_name);
}
}
}
static int parse_points(neu_plugin_t *plugin, const char *text,
tn_point_t **points)
{
json_error_t error;
json_t * root = NULL;
size_t index = 0;
json_t * item = NULL;
int rv = 0;
*points = NULL;
if (NULL == text || '\0' == text[0]) {
return 0;
}
root = json_loads(text, 0, &error);
if (NULL == root || !json_is_array(root)) {
plog_error(plugin, "points is not a JSON array");
json_decref(root);
return -1;
}
json_array_foreach(root, index, item)
{
tn_point_t *point = NULL;
tn_point_t *existing = NULL;
const char *property = NULL;
const char *node = NULL;
const char *group = NULL;
const char *tag = NULL;
if (!json_is_object(item)) {
plog_error(plugin, "points[%zu] is not an object", index);
rv = -1;
break;
}
property = json_string_value(json_object_get(item, "property"));
node = json_string_value(json_object_get(item, "node"));
group = json_string_value(json_object_get(item, "group"));
tag = json_string_value(json_object_get(item, "tag"));
point = calloc(1, sizeof(*point));
if (NULL == point ||
0 !=
copy_name(point->property, sizeof(point->property), property) ||
0 != copy_name(point->node, sizeof(point->node), node) ||
0 != copy_name(point->group, sizeof(point->group), group) ||
0 != copy_name(point->tag, sizeof(point->tag), tag)) {
plog_error(plugin, "points[%zu] missing property/node/group/tag",
index);
free(point);
rv = -1;
break;
}
HASH_FIND_STR(*points, point->property, existing);
if (NULL != existing) {
plog_error(plugin, "duplicate property:%s", point->property);
free(point);
rv = -1;
break;
}
HASH_ADD_STR(*points, property, point);
}
json_decref(root);
if (0 != rv) {
points_free(points);
}
return rv;
}
static void take_optional_string(neu_json_elem_t *elem, char **out)
{
if (NULL == elem->v.val_str || '\0' == elem->v.val_str[0]) {
free(elem->v.val_str);
elem->v.val_str = NULL;
return;
}
*out = elem->v.val_str;
elem->v.val_str = NULL;
}
int tn_config_parse(neu_plugin_t *plugin, const char *setting,
tn_config_t *config, tn_point_t **points)
{
int rv = 0;
neu_json_elem_t version = { .name = "version",
.t = NEU_JSON_INT,
.v.val_int = 4,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL };
neu_json_elem_t qos = { .name = "qos",
.t = NEU_JSON_INT,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL };
neu_json_elem_t client_id = { .name = "client-id", .t = NEU_JSON_STR };
neu_json_elem_t host = { .name = "host", .t = NEU_JSON_STR };
neu_json_elem_t port = { .name = "port", .t = NEU_JSON_INT };
neu_json_elem_t username = { .name = "username",
.t = NEU_JSON_STR,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL };
neu_json_elem_t password = { .name = "password",
.t = NEU_JSON_STR,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL };
neu_json_elem_t ssl = { .name = "ssl",
.t = NEU_JSON_BOOL,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL };
neu_json_elem_t product = { .name = "product-key", .t = NEU_JSON_STR };
neu_json_elem_t device = { .name = "device-key", .t = NEU_JSON_STR };
neu_json_elem_t points_el = { .name = "points", .t = NEU_JSON_STR };
memset(config, 0, sizeof(*config));
*points = NULL;
if (0 !=
neu_parse_param(setting, NULL, 9, &version, &qos, &client_id, &host,
&port, &product, &device, &points_el, &ssl)) {
plog_error(plugin, "parse required setting fail");
rv = -1;
goto end;
}
neu_parse_param(setting, NULL, 2, &username, &password);
if (port.v.val_int <= 0 || port.v.val_int > 65535 || qos.v.val_int < 0 ||
qos.v.val_int > 2 ||
(version.v.val_int != NEU_MQTT_VERSION_V311 &&
version.v.val_int != NEU_MQTT_VERSION_V5)) {
plog_error(plugin, "invalid port, qos or mqtt version");
rv = -1;
goto end;
}
config->version = version.v.val_int;
config->qos = qos.v.val_int;
config->port = (uint16_t) port.v.val_int;
config->ssl = ssl.v.val_bool;
config->client_id = client_id.v.val_str;
config->host = host.v.val_str;
config->product_key = product.v.val_str;
config->device_key = device.v.val_str;
client_id.v.val_str = NULL;
host.v.val_str = NULL;
product.v.val_str = NULL;
device.v.val_str = NULL;
take_optional_string(&username, &config->username);
take_optional_string(&password, &config->password);
config->report_topic =
build_topic("report/request", config->product_key, config->device_key);
config->set_req_topic =
build_topic("set/request", config->product_key, config->device_key);
config->set_resp_topic =
build_topic("set/response", config->product_key, config->device_key);
if (NULL == config->report_topic || NULL == config->set_req_topic ||
NULL == config->set_resp_topic) {
rv = -1;
goto end;
}
if (0 != parse_points(plugin, points_el.v.val_str, points)) {
rv = -1;
goto end;
}
end:
free(client_id.v.val_str);
free(host.v.val_str);
free(username.v.val_str);
free(password.v.val_str);
free(product.v.val_str);
free(device.v.val_str);
free(points_el.v.val_str);
if (0 != rv) {
tn_config_fini(config);
points_free(points);
}
return rv;
}
void tn_config_fini(tn_config_t *config)
{
free(config->client_id);
free(config->host);
free(config->username);
free(config->password);
free(config->product_key);
free(config->device_key);
free(config->report_topic);
free(config->set_req_topic);
free(config->set_resp_topic);
memset(config, 0, sizeof(*config));
}
plugins/thingsneural/thingsneural_config.h
+47
View File
@@ -0,0 +1,47 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2026 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.
**/
#ifndef NEURON_PLUGIN_THINGSNEURAL_CONFIG_H
#define NEURON_PLUGIN_THINGSNEURAL_CONFIG_H
#ifdef __cplusplus
extern "C" {
#endif
#include "connection/mqtt_client.h"
#include "neuron.h"
struct tn_point;
typedef struct {
neu_mqtt_version_e version;
neu_mqtt_qos_e qos;
uint16_t port;
bool ssl;
char * client_id;
char * host;
char * username;
char * password;
char * product_key;
char * device_key;
char * report_topic;
char * set_req_topic;
char * set_resp_topic;
} tn_config_t;
int tn_config_parse(neu_plugin_t *plugin, const char *setting,
tn_config_t *config, struct tn_point **points);
void tn_config_fini(tn_config_t *config);
#ifdef __cplusplus
}
#endif
#endif
plugins/thingsneural/thingsneural_handle.c
+436
View File
@@ -0,0 +1,436 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2026 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.
**/
#include <inttypes.h>
#include <jansson.h>
#include <math.h>
#include <string.h>
#include "connection/mqtt_client.h"
#include "utils/asprintf.h"
#include "utils/time.h"
#include "thingsneural_handle.h"
#include "thingsneural_plugin.h"
#define TN_METHOD_REPORT "nlink.thing.property.report"
#define TN_METHOD_SET "nlink.thing.property.set"
#define TN_CODE_OK 200
#define TN_CODE_BAD 400
#define TN_CODE_FAIL 500
static void publish_cb(int errcode, neu_mqtt_qos_e qos, char *topic,
uint8_t *payload, uint32_t len, void *data)
{
neu_plugin_t *plugin = data;
(void) topic;
(void) len;
if (0 == errcode) {
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSGS_TOTAL, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_5S, len, NULL);
} else {
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, 1,
NULL);
plog_error(plugin, "publish qos:%d fail:%d", qos, errcode);
}
free(payload);
}
static int publish_json(neu_plugin_t *plugin, char *topic, char *payload)
{
size_t len = strlen(payload);
int rv = neu_mqtt_client_publish(plugin->client, plugin->config.qos, topic,
(uint8_t *) payload, (uint32_t) len,
plugin, publish_cb);
if (0 != rv) {
plog_error(plugin, "publish %s fail", topic);
free(payload);
return -1;
}
return 0;
}
static char *next_message_id(neu_plugin_t *plugin)
{
char *id = NULL;
neu_asprintf(&id, "%" PRIu64 "-%" PRIu64, (uint64_t) neu_time_ms(),
++plugin->msg_seq);
return id;
}
static json_t *dvalue_to_json(neu_dvalue_t *value)
{
switch (value->type) {
case NEU_TYPE_INT8:
return json_integer(value->value.i8);
case NEU_TYPE_UINT8:
case NEU_TYPE_BIT:
return json_integer(value->value.u8);
case NEU_TYPE_INT16:
return json_integer(value->value.i16);
case NEU_TYPE_UINT16:
case NEU_TYPE_WORD:
return json_integer(value->value.u16);
case NEU_TYPE_INT32:
return json_integer(value->value.i32);
case NEU_TYPE_UINT32:
case NEU_TYPE_DWORD:
return json_integer(value->value.u32);
case NEU_TYPE_INT64:
return json_integer(value->value.i64);
case NEU_TYPE_UINT64:
case NEU_TYPE_LWORD:
return json_integer((json_int_t) value->value.u64);
case NEU_TYPE_FLOAT:
if (isnan(value->value.f32)) {
return NULL;
}
return json_real(value->value.f32);
case NEU_TYPE_DOUBLE:
if (isnan(value->value.d64)) {
return NULL;
}
return json_real(value->value.d64);
case NEU_TYPE_BOOL:
return json_boolean(value->value.boolean);
case NEU_TYPE_STRING:
return json_string(value->value.str);
default:
return NULL;
}
}
static const tn_point_t *find_uplink(tn_point_t *points, const char *node,
const char *group, const char *tag)
{
tn_point_t *point = NULL;
tn_point_t *tmp = NULL;
HASH_ITER(hh, points, point, tmp)
{
if (0 == strcmp(point->node, node) &&
0 == strcmp(point->group, group) && 0 == strcmp(point->tag, tag)) {
return point;
}
}
return NULL;
}
static int json_to_dvalue(json_t *item, neu_dvalue_t *value)
{
memset(value, 0, sizeof(*value));
if (json_is_integer(item)) {
value->type = NEU_TYPE_INT64;
value->value.i64 = (int64_t) json_integer_value(item);
return 0;
}
if (json_is_real(item)) {
value->type = NEU_TYPE_DOUBLE;
value->value.d64 = json_real_value(item);
return 0;
}
if (json_is_boolean(item)) {
value->type = NEU_TYPE_BOOL;
value->value.boolean = json_boolean_value(item);
return 0;
}
if (json_is_string(item)) {
value->type = NEU_TYPE_STRING;
snprintf(value->value.str, sizeof(value->value.str), "%s",
json_string_value(item));
return 0;
}
return -1;
}
static int write_point(neu_plugin_t *plugin, const tn_point_t *point,
json_t *item)
{
neu_reqresp_head_t header = { 0 };
neu_req_write_tag_t cmd = { 0 };
if (0 != json_to_dvalue(item, &cmd.value)) {
return -1;
}
header.type = NEU_REQ_WRITE_TAG;
cmd.driver = strdup(point->node);
cmd.group = strdup(point->group);
cmd.tag = strdup(point->tag);
if (NULL == cmd.driver || NULL == cmd.group || NULL == cmd.tag) {
free(cmd.driver);
free(cmd.group);
free(cmd.tag);
return -1;
}
if (0 != neu_plugin_op(plugin, header, &cmd)) {
plog_error(plugin, "write %s/%s/%s fail", point->node, point->group,
point->tag);
free(cmd.driver);
free(cmd.group);
free(cmd.tag);
return -1;
}
return 0;
}
static void publish_set_response(neu_plugin_t *plugin, const char *id, int code)
{
json_t *root = NULL;
char * text = NULL;
if (NULL == plugin->client || NULL == plugin->config.set_resp_topic) {
return;
}
root = json_pack("{s:s, s:i, s:s, s:s}", "id", id ? id : "", "code", code,
"method", TN_METHOD_SET, "version", "1.0");
if (NULL == root) {
return;
}
text = json_dumps(root, JSON_COMPACT);
json_decref(root);
if (NULL != text) {
publish_json(plugin, plugin->config.set_resp_topic, text);
}
}
static void on_set_message(neu_mqtt_qos_e qos, const char *topic,
const uint8_t *payload, uint32_t len, void *data,
trace_w3c_t *trace)
{
neu_plugin_t *plugin = data;
char * text = NULL;
(void) qos;
(void) topic;
(void) trace;
text = calloc(1, (size_t) len + 1);
if (NULL == text) {
return;
}
memcpy(text, payload, len);
tn_handle_set_payload(plugin, text);
free(text);
}
int tn_subscribe_downlink(neu_plugin_t *plugin)
{
if (NULL == plugin->client || NULL == plugin->config.set_req_topic) {
return -1;
}
return neu_mqtt_client_subscribe(plugin->client, plugin->config.qos,
plugin->config.set_req_topic, plugin,
on_set_message);
}
int tn_handle_trans_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *data)
{
json_t *values = NULL;
json_t *params = NULL;
json_t *root = NULL;
char * text = NULL;
char * id = NULL;
char time_text[32] = { 0 };
int n_value = 0;
if (NULL == plugin->client ||
!neu_mqtt_client_is_connected(plugin->client)) {
return NEU_ERR_PLUGIN_NOT_RUNNING;
}
values = json_object();
if (NULL == values) {
return NEU_ERR_EINTERNAL;
}
pthread_mutex_lock(&plugin->lock);
utarray_foreach(data->tags, neu_resp_tag_value_meta_t *, tv)
{
const tn_point_t *point = NULL;
json_t * item = NULL;
if (tv->value.type == NEU_TYPE_ERROR) {
continue;
}
point = find_uplink(plugin->points, data->driver, data->group, tv->tag);
if (NULL == point) {
continue;
}
item = dvalue_to_json(&tv->value);
if (NULL == item) {
continue;
}
json_object_set_new(values, point->property, item);
n_value += 1;
}
pthread_mutex_unlock(&plugin->lock);
if (0 == n_value) {
json_decref(values);
return NEU_ERR_SUCCESS;
}
snprintf(time_text, sizeof(time_text), "%" PRId64, neu_time_ms());
id = next_message_id(plugin);
params = json_object();
root = json_object();
if (NULL == params || NULL == root) {
json_decref(values);
json_decref(params);
json_decref(root);
free(id);
return NEU_ERR_EINTERNAL;
}
json_object_set_new(params, "time", json_string(time_text));
json_object_set_new(params, "values", values);
values = NULL;
json_object_set_new(root, "id", json_string(id ? id : ""));
json_object_set_new(root, "version", json_string("1.0"));
json_object_set_new(root, "ack", json_integer(1));
json_object_set_new(root, "method", json_string(TN_METHOD_REPORT));
json_object_set_new(root, "params", params);
params = NULL;
free(id);
text = json_dumps(root, JSON_COMPACT);
json_decref(root);
if (NULL == text) {
return NEU_ERR_EINTERNAL;
}
if (0 != publish_json(plugin, plugin->config.report_topic, text)) {
return NEU_ERR_MQTT_PUBLISH_FAILURE;
}
return NEU_ERR_SUCCESS;
}
int tn_handle_set_payload(neu_plugin_t *plugin, const char *payload)
{
json_error_t error;
json_t * root = NULL;
json_t * params = NULL;
json_t * values = NULL;
const char * id = NULL;
const char * method = NULL;
const char * key = NULL;
json_t * item = NULL;
int code = TN_CODE_OK;
int n_ok = 0;
root = json_loads(payload, 0, &error);
if (NULL == root || !json_is_object(root)) {
json_decref(root);
return NEU_ERR_BODY_IS_WRONG;
}
id = json_string_value(json_object_get(root, "id"));
method = json_string_value(json_object_get(root, "method"));
params = json_object_get(root, "params");
values = json_is_object(params) ? json_object_get(params, "values") : NULL;
if ((NULL != method && 0 != strcmp(method, TN_METHOD_SET)) ||
!json_is_object(values)) {
publish_set_response(plugin, id, TN_CODE_BAD);
json_decref(root);
return NEU_ERR_BODY_IS_WRONG;
}
json_object_foreach(values, key, item)
{
tn_point_t point = { 0 };
tn_point_t *found = NULL;
int rv = 0;
pthread_mutex_lock(&plugin->lock);
HASH_FIND_STR(plugin->points, key, found);
if (NULL != found) {
point = *found;
}
pthread_mutex_unlock(&plugin->lock);
if (NULL == found) {
plog_notice(plugin, "property %s has no south mapping", key);
code = TN_CODE_BAD;
continue;
}
rv = write_point(plugin, &point, item);
if (0 != rv) {
code = TN_CODE_FAIL;
continue;
}
n_ok += 1;
}
if (0 == n_ok && code == TN_CODE_OK) {
code = TN_CODE_BAD;
}
publish_set_response(plugin, id, code);
json_decref(root);
return code == TN_CODE_OK ? NEU_ERR_SUCCESS : NEU_ERR_BODY_IS_WRONG;
}
int tn_handle_subscribe_group(neu_plugin_t *plugin, neu_req_subscribe_t *sub)
{
plog_notice(plugin, "subscribe driver:%s group:%s", sub->driver,
sub->group);
free(sub->params);
free(sub->static_tags);
return NEU_ERR_SUCCESS;
}
int tn_handle_update_subscribe(neu_plugin_t *plugin, neu_req_subscribe_t *sub)
{
plog_notice(plugin, "update subscribe driver:%s group:%s", sub->driver,
sub->group);
free(sub->params);
free(sub->static_tags);
return NEU_ERR_SUCCESS;
}
int tn_handle_unsubscribe_group(neu_plugin_t * plugin,
neu_req_unsubscribe_t *unsub)
{
plog_notice(plugin, "unsubscribe driver:%s group:%s", unsub->driver,
unsub->group);
return NEU_ERR_SUCCESS;
}
int tn_handle_update_group(neu_plugin_t *plugin, neu_req_update_group_t *req)
{
pthread_mutex_lock(&plugin->lock);
tn_points_rename_group(plugin->points, req->driver, req->group,
req->new_name);
pthread_mutex_unlock(&plugin->lock);
return NEU_ERR_SUCCESS;
}
int tn_handle_del_group(neu_plugin_t *plugin, neu_req_del_group_t *req)
{
(void) plugin;
(void) req;
return NEU_ERR_SUCCESS;
}
int tn_handle_update_driver(neu_plugin_t *plugin, neu_req_update_node_t *req)
{
pthread_mutex_lock(&plugin->lock);
tn_points_rename_node(plugin->points, req->node, req->new_name);
pthread_mutex_unlock(&plugin->lock);
return NEU_ERR_SUCCESS;
}
int tn_handle_del_driver(neu_plugin_t *plugin, neu_reqresp_node_deleted_t *req)
{
(void) plugin;
(void) req;
return NEU_ERR_SUCCESS;
}
plugins/thingsneural/thingsneural_handle.h
+36
View File
@@ -0,0 +1,36 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2026 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.
**/
#ifndef NEURON_PLUGIN_THINGSNEURAL_HANDLE_H
#define NEURON_PLUGIN_THINGSNEURAL_HANDLE_H
#ifdef __cplusplus
extern "C" {
#endif
#include "neuron.h"
int tn_subscribe_downlink(neu_plugin_t *plugin);
int tn_handle_trans_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *data);
int tn_handle_set_payload(neu_plugin_t *plugin, const char *payload);
int tn_handle_subscribe_group(neu_plugin_t *plugin, neu_req_subscribe_t *sub);
int tn_handle_update_subscribe(neu_plugin_t *plugin, neu_req_subscribe_t *sub);
int tn_handle_unsubscribe_group(neu_plugin_t * plugin,
neu_req_unsubscribe_t *unsub);
int tn_handle_update_group(neu_plugin_t *plugin, neu_req_update_group_t *req);
int tn_handle_del_group(neu_plugin_t *plugin, neu_req_del_group_t *req);
int tn_handle_update_driver(neu_plugin_t *plugin, neu_req_update_node_t *req);
int tn_handle_del_driver(neu_plugin_t *plugin, neu_reqresp_node_deleted_t *req);
#ifdef __cplusplus
}
#endif
#endif
plugins/thingsneural/thingsneural_plugin.c
+28
View File
@@ -0,0 +1,28 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2026 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.
**/
#include "thingsneural_plugin_intf.h"
#define DESCRIPTION \
"Northbound ThingsNeural plugin using NLINK property messages."
#define DESCRIPTION_ZH "按 NLINK 物模型属性报文收发的北向应用插件"
const neu_plugin_module_t neu_plugin_module = {
.version = NEURON_PLUGIN_VER_1_0,
.schema = "thingsneural",
.module_name = "ThingsNeural",
.module_descr = DESCRIPTION,
.module_descr_zh = DESCRIPTION_ZH,
.intf_funs = &tn_plugin_intf_funs,
.kind = NEU_PLUGIN_KIND_SYSTEM,
.type = NEU_NA_TYPE_APP,
.display = true,
.single = false,
};
plugins/thingsneural/thingsneural_plugin.h
+51
View File
@@ -0,0 +1,51 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2026 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.
**/
#ifndef NEURON_PLUGIN_THINGSNEURAL_H
#define NEURON_PLUGIN_THINGSNEURAL_H
#ifdef __cplusplus
extern "C" {
#endif
#include <pthread.h>
#include "neuron.h"
#include "thingsneural_config.h"
typedef struct tn_point {
char property[NEU_TAG_NAME_LEN];
char node[NEU_NODE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
char tag[NEU_TAG_NAME_LEN];
UT_hash_handle hh;
} tn_point_t;
struct neu_plugin {
neu_plugin_common_t common;
tn_config_t config;
neu_mqtt_client_t * client;
tn_point_t * points;
pthread_mutex_t lock;
uint64_t msg_seq;
};
void tn_points_free(tn_point_t **points);
void tn_points_rename_node(tn_point_t *points, const char *node,
const char *new_name);
void tn_points_rename_group(tn_point_t *points, const char *node,
const char *group, const char *new_name);
#ifdef __cplusplus
}
#endif
#endif
plugins/thingsneural/thingsneural_plugin_intf.c
+247
View File
@@ -0,0 +1,247 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2026 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.
**/
#include <stdlib.h>
#include <string.h>
#include "thingsneural_config.h"
#include "thingsneural_handle.h"
#include "thingsneural_plugin.h"
#include "thingsneural_plugin_intf.h"
extern const neu_plugin_module_t neu_plugin_module;
static void connect_cb(void *data)
{
neu_plugin_t *plugin = data;
plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
plog_notice(plugin, "plugin `%s` connected", neu_plugin_module.module_name);
if (0 != tn_subscribe_downlink(plugin)) {
plog_error(plugin, "subscribe property set topic fail");
}
}
static void disconnect_cb(void *data)
{
neu_plugin_t *plugin = data;
plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
plog_notice(plugin, "plugin `%s` disconnected",
neu_plugin_module.module_name);
}
static int open_client(neu_plugin_t *plugin, const tn_config_t *config)
{
int rv = 0;
neu_mqtt_client_t *client = neu_mqtt_client_new(config->version);
if (NULL == client) {
return -1;
}
rv = neu_mqtt_client_set_zlog_category(client, plugin->common.log);
if (0 != rv) {
goto error;
}
rv = neu_mqtt_client_set_addr(client, config->host, config->port);
if (0 != rv) {
goto error;
}
rv = neu_mqtt_client_set_id(client, config->client_id);
if (0 != rv) {
goto error;
}
rv = neu_mqtt_client_set_connect_cb(client, connect_cb, plugin);
if (0 != rv) {
goto error;
}
rv = neu_mqtt_client_set_disconnect_cb(client, disconnect_cb, plugin);
if (0 != rv) {
goto error;
}
if (NULL != config->username) {
rv = neu_mqtt_client_set_user(client, config->username,
config->password);
if (0 != rv) {
goto error;
}
}
rv = neu_mqtt_client_set_tls(client, config->ssl, NULL, NULL, NULL, NULL);
if (0 != rv) {
goto error;
}
if (0 != neu_mqtt_client_open(client)) {
goto error;
}
plugin->client = client;
return 0;
error:
neu_mqtt_client_free(client);
return -1;
}
static void close_client(neu_plugin_t *plugin)
{
if (NULL == plugin->client) {
return;
}
if (NULL != plugin->config.set_req_topic) {
neu_mqtt_client_unsubscribe(plugin->client,
plugin->config.set_req_topic);
}
neu_mqtt_client_close(plugin->client);
neu_mqtt_client_free(plugin->client);
plugin->client = NULL;
plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
}
neu_plugin_t *tn_plugin_open(void)
{
neu_plugin_t *plugin = calloc(1, sizeof(neu_plugin_t));
neu_plugin_common_init(&plugin->common);
pthread_mutex_init(&plugin->lock, NULL);
return plugin;
}
int tn_plugin_close(neu_plugin_t *plugin)
{
pthread_mutex_destroy(&plugin->lock);
free(plugin);
return NEU_ERR_SUCCESS;
}
int tn_plugin_init(neu_plugin_t *plugin, bool load)
{
(void) load;
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_TRANS_DATA_5S, 5000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_BYTES_5S, 5000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_MSGS_TOTAL, 0);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, 0);
plog_notice(plugin, "initialize plugin `%s` success",
neu_plugin_module.module_name);
return NEU_ERR_SUCCESS;
}
int tn_plugin_uninit(neu_plugin_t *plugin)
{
close_client(plugin);
tn_config_fini(&plugin->config);
pthread_mutex_lock(&plugin->lock);
tn_points_free(&plugin->points);
pthread_mutex_unlock(&plugin->lock);
plog_notice(plugin, "uninitialize plugin `%s` success",
neu_plugin_module.module_name);
return NEU_ERR_SUCCESS;
}
int tn_plugin_config(neu_plugin_t *plugin, const char *setting)
{
tn_config_t config = { 0 };
tn_point_t *points = NULL;
if (0 != tn_config_parse(plugin, setting, &config, &points)) {
return NEU_ERR_NODE_SETTING_INVALID;
}
close_client(plugin);
if (0 != open_client(plugin, &config)) {
tn_config_fini(&config);
tn_points_free(&points);
return NEU_ERR_MQTT_CONNECT_FAILURE;
}
tn_config_fini(&plugin->config);
pthread_mutex_lock(&plugin->lock);
tn_points_free(&plugin->points);
plugin->points = points;
pthread_mutex_unlock(&plugin->lock);
plugin->config = config;
if (0 != tn_subscribe_downlink(plugin)) {
plog_error(plugin, "subscribe property set topic fail");
}
plog_notice(plugin, "config plugin `%s` success, report:%s",
neu_plugin_module.module_name, plugin->config.report_topic);
return NEU_ERR_SUCCESS;
}
int tn_plugin_start(neu_plugin_t *plugin)
{
if (NULL == plugin->client) {
return NEU_ERR_MQTT_IS_NULL;
}
if (0 != neu_mqtt_client_open(plugin->client)) {
return NEU_ERR_MQTT_CONNECT_FAILURE;
}
if (0 != tn_subscribe_downlink(plugin)) {
plog_error(plugin, "subscribe property set topic fail");
}
plog_notice(plugin, "start plugin `%s` success",
neu_plugin_module.module_name);
return NEU_ERR_SUCCESS;
}
int tn_plugin_stop(neu_plugin_t *plugin)
{
if (NULL != plugin->client && NULL != plugin->config.set_req_topic) {
neu_mqtt_client_unsubscribe(plugin->client,
plugin->config.set_req_topic);
neu_mqtt_client_close(plugin->client);
}
plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
plog_notice(plugin, "stop plugin `%s` success",
neu_plugin_module.module_name);
return NEU_ERR_SUCCESS;
}
static int tn_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
void *data)
{
switch (head->type) {
case NEU_REQRESP_TRANS_DATA:
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_5S, 1, NULL);
return tn_handle_trans_data(plugin, data);
case NEU_REQ_SUBSCRIBE_GROUP:
return tn_handle_subscribe_group(plugin, data);
case NEU_REQ_UPDATE_SUBSCRIBE_GROUP:
return tn_handle_update_subscribe(plugin, data);
case NEU_REQ_UNSUBSCRIBE_GROUP:
return tn_handle_unsubscribe_group(plugin, data);
case NEU_REQ_UPDATE_GROUP:
return tn_handle_update_group(plugin, data);
case NEU_REQ_DEL_GROUP:
return tn_handle_del_group(plugin, data);
case NEU_REQ_UPDATE_NODE:
return tn_handle_update_driver(plugin, data);
case NEU_REQRESP_NODE_DELETED: {
neu_reqresp_node_deleted_t *req = data;
if (0 != strcmp(plugin->common.name, req->node)) {
return tn_handle_del_driver(plugin, data);
}
return NEU_ERR_SUCCESS;
}
default:
return NEU_ERR_PLUGIN_NOT_RUNNING;
}
}
const neu_plugin_intf_funs_t tn_plugin_intf_funs = {
.open = tn_plugin_open,
.close = tn_plugin_close,
.init = tn_plugin_init,
.uninit = tn_plugin_uninit,
.start = tn_plugin_start,
.stop = tn_plugin_stop,
.setting = tn_plugin_config,
.request = tn_plugin_request,
.try_connect = NULL,
};
plugins/thingsneural/thingsneural_plugin_intf.h
+18
View File
@@ -0,0 +1,18 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2026 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.
**/
#ifndef NEURON_PLUGIN_THINGSNEURAL_INTF_H
#define NEURON_PLUGIN_THINGSNEURAL_INTF_H
#include "plugin.h"
extern const neu_plugin_intf_funs_t tn_plugin_intf_funs;
#endif
tests/ft/app/test_thingsneural.py
+204
View File
@@ -0,0 +1,204 @@
import json
import time
import pytest
import neuron.api as api
import neuron.config as config
from neuron.common import description, process, random_port
from neuron.mqtt import Mock
APP = "tn-device"
DRIVER_A = "modbus-a"
DRIVER_B = "modbus-b"
GROUP_A = "ga"
GROUP_B = "gb"
INTERVAL = 100
PRODUCT_KEY = "pkdemo"
DEVICE_KEY = "dkdemo"
REPORT_TOPIC = f"/iot/thing/{PRODUCT_KEY}/{DEVICE_KEY}/property/report/request"
SET_REQ_TOPIC = f"/iot/thing/{PRODUCT_KEY}/{DEVICE_KEY}/property/set/request"
SET_RESP_TOPIC = f"/iot/thing/{PRODUCT_KEY}/{DEVICE_KEY}/property/set/response"
def tag(address):
return {
"name": "tag0",
"address": address,
"attribute": config.NEU_TAG_ATTRIBUTE_RW,
"type": config.NEU_TYPE_INT16,
}
POINTS = json.dumps(
[
{
"property": "temperature",
"node": DRIVER_A,
"group": GROUP_A,
"tag": "tag0",
},
{
"property": "humidity",
"node": DRIVER_B,
"group": GROUP_B,
"tag": "tag0",
},
]
)
@pytest.fixture(autouse=True, scope="class")
def neuron_process():
process.remove_persistence()
neuron = process.NeuronProcess()
neuron.start()
deadline = time.time() + 10
while time.time() < deadline:
try:
if api.ping().status_code == 200:
break
except Exception:
time.sleep(0.2)
else:
neuron.stop()
raise AssertionError("neuron did not listen on port 7000")
yield neuron
neuron.stop()
@pytest.fixture(autouse=True, scope="class")
def thingsneural_node(neuron_process):
api.add_node_check(APP, "ThingsNeural")
yield
api.del_node(APP)
@pytest.fixture(autouse=True, scope="class")
def modbus_nodes(thingsneural_node):
port = random_port()
simulator = process.start_simulator(
["./modbus_simulator", "tcp", f"{port}", "ip_v4"]
)
try:
for node, group, address in (
(DRIVER_A, GROUP_A, "1!400001"),
(DRIVER_B, GROUP_B, "1!400002"),
):
api.add_node_check(node, "Modbus TCP")
api.add_group_check(node, group, INTERVAL)
api.add_tags_check(node, group, tags=[tag(address)])
api.subscribe_group_check(APP, node, group, params={})
response = api.modbus_tcp_node_setting(node, port)
assert 200 == response.status_code
yield
finally:
api.del_node(DRIVER_A)
api.del_node(DRIVER_B)
process.stop_simulator(simulator)
@pytest.fixture(autouse=True, scope="class")
def mocker():
mock = Mock()
yield mock
mock.done()
def setting(mocker):
return {
"client-id": mocker.rand_id(),
"qos": 0,
"host": "127.0.0.1",
"port": mocker.port,
"ssl": False,
"product-key": PRODUCT_KEY,
"device-key": DEVICE_KEY,
"points": POINTS,
}
def collect_reports(mocker):
messages = mocker.get(REPORT_TOPIC, num=6, timeout=8)
assert messages, "no NLINK property reports"
if isinstance(messages, dict):
messages = [messages]
return messages
class TestThingsNeural:
@description(
given="ThingsNeural plugin is loaded",
when="list plugins",
then="ThingsNeural is visible",
)
def test_plugin_listed(self):
response = api.get_plugins()
assert 200 == response.status_code
names = [item["name"] for item in response.json().get("plugins", [])]
assert "ThingsNeural" in names
@description(
given="two south groups mapped to thing-model properties",
when="tags are reported",
then="payload is NLINK and keys are property identifiers",
)
def test_uplink_nlink(self, mocker):
api.node_setting_check(APP, setting(mocker))
api.node_ctl(APP, config.NEU_CTL_START)
mocker.assert_client_connected()
properties = set()
for message in collect_reports(mocker):
assert "node" not in message
assert "group" not in message
assert message["version"] == "1.0"
assert message["method"] == "nlink.thing.property.report"
assert message["ack"] == 1
values = message["params"]["values"]
assert "tag0" not in values
properties.update(values.keys())
assert "temperature" in properties
assert "humidity" in properties
@description(
given="a property set without node or group",
when="temperature and an unknown property are written",
then="only the mapped south tag changes",
)
def test_downlink_reverse_map(self, mocker):
api.node_setting_check(APP, setting(mocker))
api.node_ctl(APP, config.NEU_CTL_START)
mocker.assert_client_connected()
response = mocker.reqresp(
SET_RESP_TOPIC,
SET_REQ_TOPIC,
{
"id": "set-1",
"version": "1.0",
"method": "nlink.thing.property.set",
"params": {"values": {"temperature": 1234, "humidity": 56}},
},
timeout=3,
)
assert response is not None
assert response["id"] == "set-1"
assert response["code"] == 200
assert response["method"] == "nlink.thing.property.set"
time.sleep(0.5)
assert 1234 == api.read_tag(DRIVER_A, GROUP_A, "tag0")
assert 56 == api.read_tag(DRIVER_B, GROUP_B, "tag0")
unknown = mocker.reqresp(
SET_RESP_TOPIC,
SET_REQ_TOPIC,
{
"id": "set-unknown",
"version": "1.0",
"method": "nlink.thing.property.set",
"params": {"values": {"pressure": 1}},
},
timeout=3,
)
assert unknown is not None
assert unknown["code"] == 400
assert 1234 == api.read_tag(DRIVER_A, GROUP_A, "tag0")