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

mqtt driver topic prefix

Browse Source
This commit is contained in:
fengzero
2025-02-06 06:24:51 +00:00
parent 2a7c7edee8
commit bdf142ca22
8 changed files with 181 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/adapter.h
+9
View File
@@ -44,6 +44,12 @@ typedef int (*neu_adapter_register_metric_cb_t)(neu_adapter_t * adapter,
neu_metric_type_e type,
uint64_t init);
typedef struct {
char * path;
uint64_t size;
uint64_t mtime;
} neu_driver_file_info_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,
@@ -75,6 +81,9 @@ typedef struct adapter_callbacks {
neu_json_type_e t, neu_type_e type,
neu_json_value_u value,
int64_t error);
void (*directory_response)(neu_adapter_t *adapter, void *req,
int error, neu_driver_file_info_t *infos,
int n_info);
} driver;
};
} adapter_callbacks_t;
include/neuron/plugin.h
+2
View File
@@ -115,6 +115,8 @@ typedef struct neu_plugin_intf_funs {
int (*test_read_tag)(neu_plugin_t *plugin, void *req,
neu_datatag_t tag);
int (*action)(neu_plugin_t *plugin, const char *action);
int (*directory)(neu_plugin_t *plugin, void *req, const char *path);
} driver;
};
plugins/mqtt/mqtt.json
+6 -19
View File
@@ -129,28 +129,15 @@
"length": 255
}
},
"driver-action-req-topic": {
"name": "Driver Action Request Topic",
"name_zh": "驱动执行动作请求主题",
"description": "MQTT topic to receive driver action request messages.",
"description_zh": "驱动执行动作请求的 MQTT 主题。",
"driver-topic-prefix": {
"name": "Driver Topic Prefix",
"name_zh": "驱动主题前缀",
"description": "MQTT topic prefix for driver function.",
"attribute": "required",
"type": "string",
"default": "/neuron/${random_str}/action/req",
"default": "/neuron/${random_str}",
"valid": {
"length": 255
}
},
"driver-action-resp-topic": {
"name": "Driver Action Response Topic",
"name_zh": "驱动执行动作响应主题",
"description": "MQTT topic to send driver action response messages.",
"description_zh": "驱动执行动作响应的 MQTT 主题。",
"attribute": "required",
"type": "string",
"default": "/neuron/${random_str}/action/resp",
"valid": {
"length": 255
"length": 128
}
},
"upload_drv_state": {
plugins/mqtt/mqtt_config.c
+81 -56
View File
@@ -275,14 +275,8 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
.v.val_str = NULL,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL, // for backward compatibility
};
neu_json_elem_t driver_action_req_topic = {
.name = "driver-action-req-topic",
.t = NEU_JSON_STR,
.v.val_str = NULL,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
};
neu_json_elem_t driver_action_resp_topic = {
.name = "driver-action-resp-topic",
neu_json_elem_t driver_topic_prefix = {
.name = "driver-topic-prefix",
.t = NEU_JSON_STR,
.v.val_str = NULL,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
@@ -379,10 +373,10 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
free(schema.v.val_str);
}
ret = neu_parse_param(setting, &err_param, 2, &driver_action_req_topic,
&driver_action_resp_topic);
ret = neu_parse_param(setting, &err_param, 1, &driver_topic_prefix);
if (0 != ret) {
plog_warn(plugin, "parsing action topic fail, key: `%s`", err_param);
plog_warn(plugin, "parsing driver topic prefix fail, key: `%s`",
err_param);
}
// write request topic
@@ -402,18 +396,10 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
}
// driver action request topic
if (NULL == driver_action_req_topic.v.val_str &&
0 > neu_asprintf(&driver_action_req_topic.v.val_str,
"/neuron/%s/action/req", plugin->common.name)) {
plog_error(plugin, "setting driver action request topic error");
goto error;
}
// driver action response topic
if (NULL == driver_action_resp_topic.v.val_str &&
0 > neu_asprintf(&driver_action_resp_topic.v.val_str,
"/neuron/%s/action/resp", plugin->common.name)) {
plog_error(plugin, "setting driver action response topic error");
if (NULL == driver_topic_prefix.v.val_str &&
0 > neu_asprintf(&driver_topic_prefix.v.val_str, "/neuron/%s",
plugin->common.name)) {
plog_error(plugin, "setting driver topic prefix error");
goto error;
}
@@ -465,31 +451,31 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
plog_notice(plugin, "setting upload_err failed");
}
config->version = version.v.val_int;
config->client_id = client_id.v.val_str;
config->qos = qos.v.val_int;
config->format = format.v.val_int;
config->write_req_topic = write_req_topic.v.val_str;
config->write_resp_topic = write_resp_topic.v.val_str;
config->driver_action_req_topic = driver_action_req_topic.v.val_str;
config->driver_action_resp_topic = driver_action_resp_topic.v.val_str;
config->cache = offline_cache.v.val_bool;
config->cache_mem_size = cache_mem_size.v.val_int * MB;
config->cache_disk_size = cache_disk_size.v.val_int * MB;
config->cache_sync_interval = cache_sync_interval.v.val_int;
config->host = host.v.val_str;
config->port = port.v.val_int;
config->username = username.v.val_str;
config->password = password.v.val_str;
config->ssl = ssl.v.val_bool;
config->ca = ca.v.val_str;
config->cert = cert.v.val_str;
config->key = key.v.val_str;
config->keypass = keypass.v.val_str;
config->upload_drv_state = upload_drv_state.v.val_bool;
config->heartbeat_topic = upload_drv_state_topic.v.val_str;
config->heartbeat_interval = upload_drv_state_interval.v.val_int;
config->upload_err = upload_err.v.val_bool;
config->version = version.v.val_int;
config->client_id = client_id.v.val_str;
config->qos = qos.v.val_int;
config->format = format.v.val_int;
config->write_req_topic = write_req_topic.v.val_str;
config->write_resp_topic = write_resp_topic.v.val_str;
config->cache = offline_cache.v.val_bool;
config->cache_mem_size = cache_mem_size.v.val_int * MB;
config->cache_disk_size = cache_disk_size.v.val_int * MB;
config->cache_sync_interval = cache_sync_interval.v.val_int;
config->host = host.v.val_str;
config->port = port.v.val_int;
config->username = username.v.val_str;
config->password = password.v.val_str;
config->ssl = ssl.v.val_bool;
config->ca = ca.v.val_str;
config->cert = cert.v.val_str;
config->key = key.v.val_str;
config->keypass = keypass.v.val_str;
config->upload_drv_state = upload_drv_state.v.val_bool;
config->heartbeat_topic = upload_drv_state_topic.v.val_str;
config->heartbeat_interval = upload_drv_state_interval.v.val_int;
config->upload_err = upload_err.v.val_bool;
config->driver_topic_prefix = driver_topic_prefix.v.val_str;
plog_notice(plugin, "config MQTT version : %d", config->version);
plog_notice(plugin, "config client-id : %s", config->client_id);
@@ -499,10 +485,8 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
plog_notice(plugin, "config write-req-topic : %s", config->write_req_topic);
plog_notice(plugin, "config write-resp-topic: %s",
config->write_resp_topic);
plog_notice(plugin, "config driver-action-req-topic : %s",
config->driver_action_req_topic);
plog_notice(plugin, "config driver-action-resp-topic: %s",
config->driver_action_resp_topic);
plog_notice(plugin, "config driver-topic-prefix : %s",
config->driver_topic_prefix);
plog_notice(plugin, "config upload-drv-state: %d",
config->upload_drv_state);
plog_notice(plugin, "config upload-err: %d", config->upload_err);
@@ -545,6 +529,48 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
plog_notice(plugin, "config keypass : %s", placeholder);
}
snprintf(config->driver_topic.action_req,
sizeof(config->driver_topic.action_req), "%s/%s",
config->driver_topic_prefix, ACTION_REQ_TOPIC);
snprintf(config->driver_topic.action_resp,
sizeof(config->driver_topic.action_resp), "%s/%s",
config->driver_topic_prefix, ACTION_RESP_TOPIC);
snprintf(config->driver_topic.files_req,
sizeof(config->driver_topic.files_req), "%s/%s",
config->driver_topic_prefix, FILES_REQ_TOPIC);
snprintf(config->driver_topic.files_resp,
sizeof(config->driver_topic.files_resp), "%s/%s",
config->driver_topic_prefix, FILES_RESP_TOPIC);
snprintf(config->driver_topic.file_up_req,
sizeof(config->driver_topic.file_up_req), "%s/%s",
config->driver_topic_prefix, FILE_UP_REQ_TOPIC);
snprintf(config->driver_topic.file_up_resp,
sizeof(config->driver_topic.file_up_resp), "%s/%s",
config->driver_topic_prefix, FILE_UP_RESP_TOPIC);
snprintf(config->driver_topic.file_up_data_req,
sizeof(config->driver_topic.file_up_data_req), "%s/%s",
config->driver_topic_prefix, FILE_UP_DATA_REQ_TOPIC);
snprintf(config->driver_topic.file_up_data_resp,
sizeof(config->driver_topic.file_up_data_resp), "%s/%s",
config->driver_topic_prefix, FILE_UP_DATA_RESP_TOPIC);
snprintf(config->driver_topic.file_down_req,
sizeof(config->driver_topic.file_down_req), "%s/%s",
config->driver_topic_prefix, FILE_DOWN_REQ_TOPIC);
snprintf(config->driver_topic.file_down_resp,
sizeof(config->driver_topic.file_down_resp), "%s/%s",
config->driver_topic_prefix, FILE_DOWN_RESP_TOPIC);
snprintf(config->driver_topic.file_down_data_req,
sizeof(config->driver_topic.file_down_data_req), "%s/%s",
config->driver_topic_prefix, FILE_DOWN_DATA_REQ_TOPIC);
snprintf(config->driver_topic.file_down_data_resp,
sizeof(config->driver_topic.file_down_data_resp), "%s/%s",
config->driver_topic_prefix, FILE_DOWN_DATA_RESP_TOPIC);
return 0;
error:
@@ -560,8 +586,7 @@ error:
free(key.v.val_str);
free(keypass.v.val_str);
free(upload_drv_state_topic.v.val_str);
free(driver_action_req_topic.v.val_str);
free(driver_action_resp_topic.v.val_str);
free(driver_topic_prefix.v.val_str);
return -1;
}
@@ -570,8 +595,6 @@ void mqtt_config_fini(mqtt_config_t *config)
free(config->client_id);
free(config->write_req_topic);
free(config->write_resp_topic);
free(config->driver_action_req_topic);
free(config->driver_action_resp_topic);
free(config->host);
free(config->username);
free(config->password);
@@ -581,6 +604,8 @@ void mqtt_config_fini(mqtt_config_t *config)
free(config->keypass);
free(config->heartbeat_topic);
free(config->driver_topic_prefix);
if (config->schema_vts) {
free(config->schema_vts);
}
plugins/mqtt/mqtt_config.h
+64 -27
View File
@@ -55,34 +55,71 @@ static inline const char *mqtt_upload_format_str(mqtt_upload_format_e f)
}
}
#define ACTION_REQ_TOPIC "action/req"
#define ACTION_RESP_TOPIC "action/resp"
#define FILES_REQ_TOPIC "flist/req"
#define FILES_RESP_TOPIC "flist/resp"
#define FILE_UP_REQ_TOPIC "fup/req"
#define FILE_UP_RESP_TOPIC "fup/resp"
#define FILE_UP_DATA_REQ_TOPIC "fupdata/req"
#define FILE_UP_DATA_RESP_TOPIC "fupdata/resp"
#define FILE_DOWN_REQ_TOPIC "fdown/req"
#define FILE_DOWN_RESP_TOPIC "fdown/resp"
#define FILE_DOWN_DATA_REQ_TOPIC "fdowndata/req"
#define FILE_DOWN_DATA_RESP_TOPIC "fdowndata/resp"
typedef struct {
neu_mqtt_version_e version; // mqtt version
char * client_id; // client id
neu_mqtt_qos_e qos; // message QoS
mqtt_upload_format_e format; // upload format
char * write_req_topic; // write request topic
char * write_resp_topic; // write response topic
char * driver_action_req_topic; // driver action request topic
char * driver_action_resp_topic; // driver action response topic
bool upload_err; // Upload tag error code flag
bool upload_drv_state; // upload driver state flag
char * heartbeat_topic; // upload driver state topic
uint16_t heartbeat_interval; // upload driver state interval
size_t cache; // cache enable flag
size_t cache_mem_size; // cache memory size in bytes
size_t cache_disk_size; // cache disk size in bytes
size_t cache_sync_interval; // cache sync interval
char * host; // broker host
uint16_t port; // broker port
char * username; // user name
char * password; // user password
bool ssl; // ssl flag
char * ca; // CA
char * cert; // client cert
char * key; // client key
char * keypass; // client key password
// remove in 2.6, keep it here
// for backward compatibility
char action_req[256];
char action_resp[256];
char files_req[256];
char files_resp[256];
char file_up_req[256];
char file_up_resp[256];
char file_up_data_req[256];
char file_up_data_resp[256];
char file_down_req[256];
char file_down_resp[256];
char file_down_data_req[256];
char file_down_data_resp[256];
} mqtt_driver_topic_t;
typedef struct {
neu_mqtt_version_e version; // mqtt version
char * client_id; // client id
neu_mqtt_qos_e qos; // message QoS
mqtt_upload_format_e format; // upload format
char * write_req_topic; // write request topic
char * write_resp_topic; // write response topic
char * driver_topic_prefix; // driver topic prefix
mqtt_driver_topic_t driver_topic;
bool upload_err; // Upload tag error code flag
bool upload_drv_state; // upload driver state flag
char * heartbeat_topic; // upload driver state topic
uint16_t heartbeat_interval; // upload driver state interval
size_t cache; // cache enable flag
size_t cache_mem_size; // cache memory size in bytes
size_t cache_disk_size; // cache disk size in bytes
size_t cache_sync_interval; // cache sync interval
char * host; // broker host
uint16_t port; // broker port
char * username; // user name
char * password; // user password
bool ssl; // ssl flag
char * ca; // CA
char * cert; // client cert
char * key; // client key
char * keypass; // client key password
// remove in 2.6, keep it here
// for backward compatibility
size_t n_schema_vt;
mqtt_schema_vt_t *schema_vts;
plugins/mqtt/mqtt_handle.c
+1 -1
View File
@@ -679,7 +679,7 @@ int handle_driver_action_response(neu_plugin_t * plugin,
goto end;
}
char * topic = plugin->config.driver_action_resp_topic;
char * topic = plugin->config.driver_topic.action_resp;
neu_mqtt_qos_e qos = plugin->config.qos;
rv = publish(plugin, qos, topic, json_str, strlen(json_str));
json_str = NULL;
plugins/mqtt/mqtt_plugin_intf.c
+3 -3
View File
@@ -308,10 +308,10 @@ static int subscribe(neu_plugin_t *plugin, const mqtt_config_t *config)
if (0 !=
neu_mqtt_client_subscribe(plugin->client, config->qos,
config->driver_action_req_topic, plugin,
config->driver_topic.action_req, plugin,
handle_driver_action_req)) {
plog_error(plugin, "subscribe [%s] fail",
plugin->config.driver_action_req_topic);
plugin->config.driver_topic.action_req);
return NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
}
@@ -323,7 +323,7 @@ static int unsubscribe(neu_plugin_t *plugin, const mqtt_config_t *config)
neu_mqtt_client_unsubscribe(plugin->client, plugin->read_req_topic);
neu_mqtt_client_unsubscribe(plugin->client, config->write_req_topic);
neu_mqtt_client_unsubscribe(plugin->client,
config->driver_action_req_topic);
config->driver_topic.action_req);
neu_msleep(100); // wait for message completion
return 0;
}
src/adapter/driver/driver.c
+15
View File
@@ -116,6 +116,8 @@ 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 directory_response(neu_adapter_t *adapter, void *req, int error,
neu_driver_file_info_t *infos, int n_info);
static group_t * find_group(neu_adapter_driver_t *driver, const char *name);
static void store_write_tag(group_t *group, to_be_write_tag_t *tag);
static inline void start_group_timer(neu_adapter_driver_t *driver,
@@ -218,6 +220,18 @@ static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
}
}
static void directory_response(neu_adapter_t *adapter, void *r, int error,
neu_driver_file_info_t *infos, int n_info)
{
(void) infos;
(void) n_info;
(void) error;
(void) adapter;
(void) r;
// nlog_notice("directory response start <%p>", req->ctx);
}
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)
@@ -425,6 +439,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.directory_response = directory_response;
driver->adapter.cb_funs.driver.update_im = update_im;
driver->adapter.cb_funs.driver.update_with_trace = update_with_trace;
driver->adapter.cb_funs.driver.update_with_meta = update_with_meta;