diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index df9bc53d..35e8a5f2 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -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; diff --git a/include/neuron/plugin.h b/include/neuron/plugin.h index 3bfd6778..ebe66791 100644 --- a/include/neuron/plugin.h +++ b/include/neuron/plugin.h @@ -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; }; diff --git a/plugins/mqtt/mqtt.json b/plugins/mqtt/mqtt.json index ae7bb187..2beca2b5 100644 --- a/plugins/mqtt/mqtt.json +++ b/plugins/mqtt/mqtt.json @@ -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": { diff --git a/plugins/mqtt/mqtt_config.c b/plugins/mqtt/mqtt_config.c index dfad485d..4ba09aa4 100644 --- a/plugins/mqtt/mqtt_config.c +++ b/plugins/mqtt/mqtt_config.c @@ -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); } diff --git a/plugins/mqtt/mqtt_config.h b/plugins/mqtt/mqtt_config.h index b4c1342a..69a17a14 100644 --- a/plugins/mqtt/mqtt_config.h +++ b/plugins/mqtt/mqtt_config.h @@ -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; diff --git a/plugins/mqtt/mqtt_handle.c b/plugins/mqtt/mqtt_handle.c index d2fee82f..250c0fbe 100644 --- a/plugins/mqtt/mqtt_handle.c +++ b/plugins/mqtt/mqtt_handle.c @@ -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; diff --git a/plugins/mqtt/mqtt_plugin_intf.c b/plugins/mqtt/mqtt_plugin_intf.c index f480301f..c9ca86f9 100644 --- a/plugins/mqtt/mqtt_plugin_intf.c +++ b/plugins/mqtt/mqtt_plugin_intf.c @@ -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; } diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 72616dbc..79f8ddbd 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -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;