diff --git a/plugins/mqtt/CMakeLists.txt b/plugins/mqtt/CMakeLists.txt index 6a5e2f8d..a4bffe4a 100644 --- a/plugins/mqtt/CMakeLists.txt +++ b/plugins/mqtt/CMakeLists.txt @@ -2,8 +2,8 @@ set(LIBRARY_OUTPUT_PATH "${CMAKE_BINARY_DIR}/plugins") set(src - plugin_mqtt.c - mqtt_util.c + mqtt.c + util.c command/read_write.c command/group_config.c command/datatag.c diff --git a/plugins/mqtt/plugin_mqtt.c b/plugins/mqtt/mqtt.c similarity index 92% rename from plugins/mqtt/plugin_mqtt.c rename to plugins/mqtt/mqtt.c index e6c01f8c..a2347400 100644 --- a/plugins/mqtt/plugin_mqtt.c +++ b/plugins/mqtt/mqtt.c @@ -24,7 +24,7 @@ #include "command/command.h" #include "connection/mqtt_client_intf.h" -#include "mqtt_util.h" +#include "util.h" #include #define UNUSED(x) (void) (x) @@ -350,6 +350,7 @@ static void mqtt_context_publish(neu_plugin_t *plugin) struct context *ctx = NULL; char * topic = NULL; char * result = NULL; + pthread_mutex_lock(&plugin->list_mutex); ctx = neu_list_first(&plugin->context_list); @@ -374,8 +375,8 @@ static void mqtt_context_publish(neu_plugin_t *plugin) if (NULL != result && NULL != topic) { neu_err_code_e error = neu_mqtt_client_publish( plugin->client, topic, 0, (unsigned char *) result, strlen(result)); - log_debug("Publish error code:%d, topic('%s'): %s", error, topic, - result); + log_debug_node(plugin, "Publish error code:%d, topic('%s'): %s", error, + topic, result); free(result); } } @@ -453,37 +454,37 @@ static void *mqtt_send_loop(void *argument) static neu_plugin_t *mqtt_plugin_open(neu_adapter_t * adapter, const adapter_callbacks_t *callbacks) { - if (NULL == adapter || NULL == callbacks) { - log_error("Open plugin with NULL adapter or callbacks"); - return NULL; - } + assert(NULL != adapter); + assert(NULL != callbacks); neu_plugin_t *plugin = NULL; - plugin = (neu_plugin_t *) malloc(sizeof(neu_plugin_t)); - if (NULL == plugin) { - log_error("Failed to allocate plugin %s", - neu_plugin_module.module_name); - return NULL; - } - memset(plugin, 0, sizeof(neu_plugin_t)); + plugin = (neu_plugin_t *) calloc(1, sizeof(neu_plugin_t)); + assert(NULL != plugin); neu_plugin_common_init(&plugin->common); plugin->common.adapter = adapter; plugin->common.adapter_callbacks = callbacks; - log_info("Success to create plugin: %s", neu_plugin_module.module_name); + log_info_node(plugin, "Success to create plugin: %s", + neu_plugin_module.module_name); return plugin; } static int mqtt_plugin_close(neu_plugin_t *plugin) { + assert(NULL != plugin); + + log_info_node(plugin, "Success to free plugin:%s", + neu_plugin_module.module_name); + free(plugin); - log_info("Success to free plugin:%s", neu_plugin_module.module_name); - return 0; + return NEU_ERR_SUCCESS; } static int mqtt_plugin_init(neu_plugin_t *plugin) { + assert(NULL != plugin); + NEU_LIST_INIT(&plugin->context_list, struct context, node); vector_init(&plugin->topics, 1, sizeof(struct topic_pair)); @@ -495,18 +496,19 @@ static int mqtt_plugin_init(neu_plugin_t *plugin) pthread_mutex_unlock(&plugin->running_mutex); if (0 != pthread_create(&plugin->daemon, NULL, mqtt_send_loop, plugin)) { - log_error("Failed to start send thread daemon."); + log_error_node(plugin, "Failed to start send thread daemon."); return NEU_ERR_FAILURE; } - log_info("Initialize plugin: %s", neu_plugin_module.module_name); + log_info_node(plugin, "Initialize plugin: %s", + neu_plugin_module.module_name); return NEU_ERR_SUCCESS; } -static int mqtt_plugin_stop(neu_plugin_t *plugin); - static int mqtt_plugin_uninit(neu_plugin_t *plugin) { + assert(NULL != plugin); + neu_mqtt_client_close(plugin->client); plugin->client = NULL; @@ -523,12 +525,16 @@ static int mqtt_plugin_uninit(neu_plugin_t *plugin) topics_cleanup(&plugin->topics); vector_uninit(&plugin->topics); - log_info("Uninitialize plugin: %s", neu_plugin_module.module_name); - return 0; + log_info_node(plugin, "Uninitialize plugin: %s", + neu_plugin_module.module_name); + return NEU_ERR_SUCCESS; } static int mqtt_plugin_config(neu_plugin_t *plugin, neu_config_t *configs) { + assert(NULL != plugin); + assert(NULL != configs); + neu_mqtt_client_close(plugin->client); plugin->client = NULL; @@ -539,47 +545,46 @@ static int mqtt_plugin_config(neu_plugin_t *plugin, neu_config_t *configs) int rc = mqtt_option_init(configs, &plugin->option); if (0 != rc) { const char *name = neu_plugin_module.module_name; - log_error("MQTT option init fail:%d, %s", rc, name); + log_error_node(plugin, "MQTT option init fail:%d, %s", rc, name); mqtt_option_uninit(&plugin->option); return NEU_ERR_FAILURE; } - neu_err_code_e error = neu_mqtt_client_open( - (neu_mqtt_client_t *) &plugin->client, &plugin->option, plugin); + neu_mqtt_client_t *client = (neu_mqtt_client_t *) &plugin->client; + neu_err_code_e error = NEU_ERR_SUCCESS; + error = neu_mqtt_client_open(client, &plugin->option, plugin); neu_mqtt_client_continue(plugin->client); - log_info("Open mqtt client: %s:%s, code:%d", plugin->option.host, - plugin->option.port, error); + + log_info_node(plugin, "Open mqtt client: %s:%s, code:%d", + plugin->option.host, plugin->option.port, error); topics_generate(&plugin->topics, plugin->option.clientid); topics_subscribe(&plugin->topics, plugin->client); - log_info("Config plugin: %s", neu_plugin_module.module_name); - return 0; + log_info_node(plugin, "Config plugin: %s", neu_plugin_module.module_name); + return NEU_ERR_SUCCESS; } static int mqtt_plugin_start(neu_plugin_t *plugin) { + assert(NULL != plugin); + neu_mqtt_client_continue(plugin->client); return NEU_ERR_SUCCESS; } static int mqtt_plugin_stop(neu_plugin_t *plugin) { + assert(NULL != plugin); + neu_mqtt_client_suspend(plugin->client); return NEU_ERR_SUCCESS; } static int mqtt_plugin_request(neu_plugin_t *plugin, neu_request_t *req) { - int rv = 0; - - if (plugin == NULL || req == NULL) { - log_warn("The plugin pointer or request is NULL"); - return (-1); - } - - log_info_node(plugin, "send request to plugin: %s, type:%d", - neu_plugin_module.module_name, req->req_type); + assert(NULL != plugin); + assert(NULL != req); switch (req->req_type) { case NEU_REQRESP_READ_RESP: { @@ -638,7 +643,7 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_request_t *req) break; } - return rv; + return NEU_ERR_SUCCESS; } static int mqtt_plugin_event_reply(neu_plugin_t * plugin, @@ -646,15 +651,14 @@ static int mqtt_plugin_event_reply(neu_plugin_t * plugin, { UNUSED(plugin); UNUSED(reply); - return 0; + return NEU_ERR_SUCCESS; } static int mqtt_plugin_validate_tag(neu_plugin_t *plugin, neu_datatag_t *tag) { (void) plugin; (void) tag; - - return 0; + return NEU_ERR_SUCCESS; } static const neu_plugin_intf_funs_t plugin_intf_funs = { diff --git a/plugins/mqtt/mqtt_util.c b/plugins/mqtt/util.c similarity index 99% rename from plugins/mqtt/mqtt_util.c rename to plugins/mqtt/util.c index dc1768d7..dbd4a16b 100644 --- a/plugins/mqtt/mqtt_util.c +++ b/plugins/mqtt/util.c @@ -22,7 +22,7 @@ #include #include -#include "mqtt_util.h" +#include "util.h" #include "json/json.h" #include #include diff --git a/plugins/mqtt/mqtt_util.h b/plugins/mqtt/util.h similarity index 100% rename from plugins/mqtt/mqtt_util.h rename to plugins/mqtt/util.h diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index ae0b3d65..c2d38969 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -122,7 +122,7 @@ target_link_libraries(http_test neuron-base gtest_main gtest yaml jansson nng) #file(COPY ${CMAKE_SOURCE_DIR}/plugin_param_schema.json DESTINATION ${CMAKE_BINARY_DIR}/tests) add_executable(mqtt_test mqtt_test.cc ${CMAKE_SOURCE_DIR}/src/config.c - ${CMAKE_SOURCE_DIR}/plugins/mqtt/mqtt_util.c + ${CMAKE_SOURCE_DIR}/plugins/mqtt/util.c ${CMAKE_SOURCE_DIR}/plugins/mqtt/command/node.c) target_include_directories(mqtt_test PRIVATE ${CMAKE_SOURCE_DIR}/src diff --git a/tests/mqtt_test.cc b/tests/mqtt_test.cc index 7da5e62b..b54dd774 100644 --- a/tests/mqtt_test.cc +++ b/tests/mqtt_test.cc @@ -1,7 +1,7 @@ #include #include "command/node.h" -#include "mqtt_util.h" +#include "util.h" #include TEST(MQTTTest, mqtt_option_init_by_config)