diff --git a/include/neuron/connection/mqtt_client_intf.h b/include/neuron/connection/mqtt_client_intf.h index 427c28a2..6801c140 100644 --- a/include/neuron/connection/mqtt_client_intf.h +++ b/include/neuron/connection/mqtt_client_intf.h @@ -58,10 +58,11 @@ typedef struct { char *keypass; /*application*/ - int format; - char * upload_topic; - char * group_id; - state_update state_update_func; + int format; + char * upload_topic; + char * group_id; + state_update state_update_func; + zlog_category_t *log; } neu_mqtt_option_t; typedef void *neu_mqtt_client_t; diff --git a/plugins/mqtt/CMakeLists.txt b/plugins/mqtt/CMakeLists.txt index b64b9fed..8549d8c1 100644 --- a/plugins/mqtt/CMakeLists.txt +++ b/plugins/mqtt/CMakeLists.txt @@ -1,4 +1,8 @@ # MQTT Plugin +# SparkplugB Plugin +cmake_minimum_required(VERSION 3.12) +project(plugin-mqtt) + set(LIBRARY_OUTPUT_PATH "${CMAKE_BINARY_DIR}/plugins") file(COPY ${CMAKE_SOURCE_DIR}/plugins/mqtt/mqtt.json DESTINATION ${CMAKE_BINARY_DIR}/plugins/schema/) @@ -8,10 +12,8 @@ set(src command/read_write.c command/command.c) -add_library(plugin-mqtt SHARED ${src}) - -target_include_directories(plugin-mqtt PRIVATE +add_library(${PROJECT_NAME} SHARED ${src}) +target_include_directories(${PROJECT_NAME} PRIVATE ${CMAKE_SOURCE_DIR}/include/neuron) - -target_link_libraries(plugin-mqtt neuron-base) -target_link_libraries(plugin-mqtt ${CMAKE_THREAD_LIBS_INIT}) +target_link_libraries(${PROJECT_NAME} neuron-base) +target_link_libraries(${PROJECT_NAME} ${CMAKE_THREAD_LIBS_INIT}) diff --git a/plugins/mqtt/mqtt.c b/plugins/mqtt/mqtt.c index eb9c6765..847b7862 100644 --- a/plugins/mqtt/mqtt.c +++ b/plugins/mqtt/mqtt.c @@ -218,6 +218,7 @@ static mqtt_routine_t *mqtt_routine_start(neu_plugin_t *plugin, } routine->option.state_update_func = mqtt_routine_state; + routine->option.log = plugin->common.log; neu_mqtt_client_t *client = (neu_mqtt_client_t *) &routine->client; neu_err_code_e error = NEU_ERR_SUCCESS; error = neu_mqtt_client_open(client, &routine->option, plugin); diff --git a/src/connection/mqtt_c_client.c b/src/connection/mqtt_c_client.c index 9ddc26a9..db762c4d 100644 --- a/src/connection/mqtt_c_client.c +++ b/src/connection/mqtt_c_client.c @@ -83,6 +83,7 @@ struct mqtt_c_client { pthread_t daemon; void * user_data; state_update state_update_func; + zlog_category_t * log; UT_array * array; }; @@ -195,15 +196,21 @@ static void ssl_ctx_init(struct reconnect_state *state) SSL_library_init(); state->ssl_ctx = SSL_CTX_new(SSLv23_client_method()); + mqtt_c_client_t *client = state->client; + // setup cert if (NULL != cert_file && 0 < strlen(cert_file)) { state->cert = ssl_ctx_load_cert(cert_file, NULL); if (NULL != state->cert) { if (0 >= SSL_CTX_use_certificate(state->ssl_ctx, state->cert)) { - nlog_error("failed to use certificate"); + if (NULL != client->log) { + zlog_error(client->log, "failed to use certificate"); + } } } else { - nlog_error("failed to load certificate"); + if (NULL != client->log) { + zlog_error(client->log, "failed to load certificate"); + } } } @@ -213,22 +220,30 @@ static void ssl_ctx_init(struct reconnect_state *state) state->key = ssl_ctx_load_key(key_file, keypass); if (NULL != state->key) { if (0 >= SSL_CTX_use_PrivateKey(state->ssl_ctx, state->key)) { - nlog_error("failed to use privatekey"); + if (NULL != client->log) { + zlog_error(client->log, "failed to use privatekey"); + } } } else { - nlog_error("failed to load privatekey"); + if (NULL != client->log) { + zlog_error(client->log, "failed to load privatekey"); + } } } if (NULL != state->cert && NULL != state->key) { if (!SSL_CTX_check_private_key(state->ssl_ctx)) { - nlog_error("failed to check privatekey"); + if (NULL != client->log) { + zlog_error(client->log, "failed to check privatekey"); + } } } // setup ca from base64 if (!ssl_ctx_load_ca(state->ssl_ctx, ca_file)) { - nlog_error("failed to load ca"); + if (NULL != client->log) { + zlog_error(client->log, "failed to load ca"); + } } } @@ -407,7 +422,15 @@ static void subscribe_on_reconnect(mqtt_c_client_t *client) p = (struct subscribe_tuple **) utarray_next(client->array, p)) { neu_err_code_e error = client_subscribe_send(client, (*p)); if (NEU_ERR_SUCCESS != error) { - nlog_error("send subscribe %s error:%d", (*p)->topic, error); + if (NULL != client->log) { + zlog_error(client->log, "send subscribe %s error:%d", + (*p)->topic, error); + } + } + + if (NULL != client->log) { + zlog_error(client->log, "send subscribe %s success:%d", (*p)->topic, + error); } } } @@ -481,7 +504,9 @@ static void publish_callback(void ** p_reconnect_state, char *topic_name = (char *) malloc(published->topic_name_size + 1); if (NULL == topic_name) { - nlog_error("topic name alloc error"); + if (NULL != client->log) { + zlog_error(client->log, "topic name alloc error"); + } return; } @@ -537,8 +562,10 @@ static void *client_refresher(void *context) client->state_update_func(client->user_data, rc); if (0 != rc) { - nlog_error("connect to %s:%s erorr, retry...", - client->state.hostname, client->state.port); + if (NULL != client->log) { + zlog_error(client->log, "connect to %s:%s erorr, retry...", + client->state.hostname, client->state.port); + } usleep(INTERVAL * 10); } } @@ -554,13 +581,14 @@ static mqtt_c_client_t *client_create(const neu_mqtt_option_t *option, { mqtt_c_client_t *client = calloc(1, sizeof(mqtt_c_client_t)); if (NULL == client) { - nlog_error("mqtt c client alloc error"); + zlog_error(option->log, "mqtt c client alloc error"); return NULL; } client->option = option; client->user_data = context; client->state_update_func = client->option->state_update_func; + client->log = client->option->log; UT_icd tuple_icd = { sizeof(struct subscribe_tuple *), NULL, NULL, NULL }; utarray_new(client->array, &tuple_icd); @@ -632,7 +660,10 @@ static neu_err_code_e client_connect(mqtt_c_client_t *client) publish_callback); if (0 != pthread_create(&client->daemon, NULL, client_refresher, client)) { - nlog_error("pthread create error"); + if (NULL != client->log) { + zlog_error(client->log, "pthread create error"); + } + return NEU_ERR_MQTT_CONNECT_FAILURE; } @@ -733,7 +764,10 @@ client_subscribe_create(mqtt_c_client_t *client, const char *topic, { struct subscribe_tuple *tuple = calloc(1, sizeof(struct subscribe_tuple)); if (NULL == tuple) { - nlog_error("tuple alloc error"); + if (NULL != client->log) { + nlog_error("tuple alloc error"); + } + return NULL; } @@ -896,6 +930,7 @@ neu_err_code_e mqtt_c_client_close(mqtt_c_client_t *client) client->state_update_func = NULL; client->user_data = NULL; + client->log = NULL; pthread_mutex_lock(&client->state.running_mutex); client->state.running = false; pthread_mutex_unlock(&client->state.running_mutex);