Merge pull request #722 from fengzeroz/v2.1.0

version: 2.1.4

include/neuron/connection/neu_connection.h

@@ -117,6 +117,8 @@ neu_conn_t *neu_conn_new(neu_conn_param_t *param, void *data,
 neu_conn_t *neu_conn_reconfig(neu_conn_t *conn, neu_conn_param_t *param);
 void        neu_conn_destory(neu_conn_t *conn);
 void        neu_conn_disconnect(neu_conn_t *conn);
+void        neu_conn_stop(neu_conn_t *conn);
+void        neu_conn_start(neu_conn_t *conn);
 int         neu_conn_tcp_server_accept(neu_conn_t *conn);
 int         neu_conn_tcp_server_close_client(neu_conn_t *conn, int fd);
 

plugins/modbus/modbus_tcp.c

@@ -107,14 +107,13 @@ static int driver_uninit(neu_plugin_t *plugin)
 
 static int driver_start(neu_plugin_t *plugin)
 {
-    (void) plugin;
-
+    neu_conn_start(plugin->conn);
     return 0;
 }
 
 static int driver_stop(neu_plugin_t *plugin)
 {
-    neu_conn_disconnect(plugin->conn);
+    neu_conn_stop(plugin->conn);
     return 0;
 }
 

src/connection/connection.c

@@ -42,6 +42,7 @@ struct neu_conn {
     neu_conn_param_t param;
     void *           data;
     bool             is_connected;
+    bool             stop;
 
     neu_conn_callback connected;
     neu_conn_callback disconnected;
@@ -84,6 +85,7 @@ neu_conn_t *neu_conn_new(neu_conn_param_t *param, void *data,
     conn->data             = data;
     conn->disconnected     = disconnected;
     conn->connected        = connected;
+    conn->stop             = false;
 
     conn_tcp_server_listen(conn);
 
@@ -92,6 +94,21 @@ neu_conn_t *neu_conn_new(neu_conn_param_t *param, void *data,
     return conn;
 }
 
+void neu_conn_stop(neu_conn_t *conn)
+{
+    pthread_mutex_lock(&conn->mtx);
+    conn->stop = true;
+    conn_disconnect(conn);
+    pthread_mutex_unlock(&conn->mtx);
+}
+
+void neu_conn_start(neu_conn_t *conn)
+{
+    pthread_mutex_lock(&conn->mtx);
+    conn->stop = false;
+    pthread_mutex_unlock(&conn->mtx);
+}
+
 neu_conn_t *neu_conn_reconfig(neu_conn_t *conn, neu_conn_param_t *param)
 {
     pthread_mutex_lock(&conn->mtx);
@@ -198,8 +215,20 @@ int neu_conn_tcp_server_close_client(neu_conn_t *conn, int fd)
 ssize_t neu_conn_tcp_server_send(neu_conn_t *conn, int fd, uint8_t *buf,
                                  ssize_t len)
 {
+    ssize_t ret = 0;
+
+    pthread_mutex_lock(&conn->mtx);
+    if (conn->stop) {
+        pthread_mutex_unlock(&conn->mtx);
+        return ret;
+    }
+
     conn_tcp_server_listen(conn);
-    return send(fd, buf, len, MSG_NOSIGNAL | MSG_DONTWAIT);
+    ret = send(fd, buf, len, MSG_NOSIGNAL | MSG_DONTWAIT);
+
+    pthread_mutex_unlock(&conn->mtx);
+
+    return ret;
 }
 
 ssize_t neu_conn_tcp_server_recv(neu_conn_t *conn, int fd, uint8_t *buf,
@@ -207,11 +236,19 @@ ssize_t neu_conn_tcp_server_recv(neu_conn_t *conn, int fd, uint8_t *buf,
 {
     ssize_t ret = 0;
 
+    pthread_mutex_lock(&conn->mtx);
+    if (conn->stop) {
+        pthread_mutex_unlock(&conn->mtx);
+        return ret;
+    }
+
     if (conn->block) {
         ret = recv(fd, buf, len, MSG_WAITALL);
     } else {
         ret = recv(fd, buf, len, 0);
     }
+    pthread_mutex_unlock(&conn->mtx);
+
     return ret;
 }
 
@@ -220,6 +257,10 @@ ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len)
     ssize_t ret = 0;
 
     pthread_mutex_lock(&conn->mtx);
+    if (conn->stop) {
+        pthread_mutex_unlock(&conn->mtx);
+        return ret;
+    }
 
     if (conn->is_connected) {
         switch (conn->param.type) {
@@ -268,6 +309,10 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
     ssize_t ret = 0;
 
     pthread_mutex_lock(&conn->mtx);
+    if (conn->stop) {
+        pthread_mutex_unlock(&conn->mtx);
+        return ret;
+    }
 
     switch (conn->param.type) {
     case NEU_CONN_TCP_SERVER:

src/connection/mqtt_c_client.c

@@ -61,9 +61,10 @@ struct reconnect_state {
     EVP_PKEY *       key;
     SSL_CTX *        ssl_ctx;
     BIO *            sock_fd;
-    int              sock_state; // sock state, value same as mqtt.error
-    bool             running;    // refresher stop flag
-    bool             working;    // publish & publish callback working flag
+    int              sock_state;        // sock state, value same as mqtt.error
+    int              sock_do_connected; // sock do connect state
+    bool             running;           // refresher stop flag
+    bool             working; // publish & publish callback working flag
     pthread_mutex_t  sock_state_mutex;
     pthread_mutex_t  running_mutex;
     pthread_mutex_t  working_mutex;
@@ -341,11 +342,16 @@ static int bio_connect(struct reconnect_state *state)
     return 0;
 }
 
-static void bio_update_connection(struct reconnect_state *state)
+static int bio_update_connection(struct reconnect_state *state)
 {
     bio_uninit(state);
     bio_init(state);
-    bio_connect(state);
+
+    if (NULL == state->sock_fd) {
+        return -1;
+    }
+
+    return bio_connect(state);
 }
 
 static void reconnect_state_uninit(struct reconnect_state *state)
@@ -427,22 +433,44 @@ static void subscribe_on_reconnect(mqtt_c_client_t *client)
                            (*p)->topic, error);
             }
         }
-
-        if (NULL != client->log) {
-            zlog_error(client->log, "send subscribe %s success:%d", (*p)->topic,
-                       error);
-        }
     }
 }
 
 enum MQTTErrors inspector_callback(struct mqtt_client *mqtt)
 {
     struct reconnect_state *state = mqtt->reconnect_state;
+
+    if (NULL == state->sock_fd) {
+        mqtt->error = MQTT_ERROR_SOCKET_ERROR;
+
+        if (NULL != state->client->log) {
+            zlog_error(state->client->log, "bio is null");
+        }
+
+        return mqtt->error;
+    }
+
+    if (0 != state->sock_do_connected) {
+        mqtt->error = MQTT_ERROR_SOCKET_ERROR;
+
+        if (NULL != state->client->log) {
+            zlog_error(state->client->log, "bio do connect failed");
+        }
+
+        return mqtt->error;
+    }
+
     if (0 == strcmp(state->connection, "ssl://")) {
         SSL *ssl = NULL;
         BIO_get_ssl(state->sock_fd, &ssl);
         if (X509_V_OK != SSL_get_verify_result(ssl)) {
             mqtt->error = MQTT_ERROR_SOCKET_ERROR;
+
+            if (NULL != state->client->log) {
+                zlog_error(state->client->log, "bio ssl verify failed");
+            }
+
+            return mqtt->error;
         }
     }
 
@@ -458,8 +486,7 @@ static void client_reconnect(struct mqtt_client *mqtt, void **p_reconnect_state)
         *((struct reconnect_state **) p_reconnect_state);
     mqtt_c_client_t *client  = state->client;
     mqtt->inspector_callback = inspector_callback;
-
-    bio_update_connection(state);
+    state->sock_do_connected = bio_update_connection(state);
 
     mqtt_reinit(mqtt, state->sock_fd, state->send_buf, state->send_buf_sz,
                 state->recv_buf, state->recv_buf_sz);
@@ -513,8 +540,10 @@ static void publish_callback(void **                       p_reconnect_state,
     memcpy(topic_name, published->topic_name, published->topic_name_size);
     topic_name[published->topic_name_size] = '\0';
 
-    nlog_debug("received publish('%s'): %s", topic_name,
-               (const char *) published->application_message);
+    if (NULL != client->log) {
+        zlog_debug(client->log, "received publish('%s'): %s", topic_name,
+                   (const char *) published->application_message);
+    }
 
     for (struct subscribe_tuple **p =
              (struct subscribe_tuple **) utarray_front(client->array);
@@ -563,8 +592,10 @@ static void *client_refresher(void *context)
             client->state_update_func(client->user_data, rc);
 
             if (0 != rc && error_count % 100 == 0) {
-                zlog_error(client->log, "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);
+                }
             }
         }
         error_count++;
@@ -580,7 +611,10 @@ 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) {
-        zlog_error(option->log, "mqtt c client alloc error");
+        if (NULL != option->log) {
+            zlog_error(option->log, "mqtt c client alloc error");
+        }
+
         return NULL;
     }
 

version

@@ -1 +1 @@
-2.1.3
+2.1.4

zlog.conf

@@ -9,4 +9,4 @@ simple = "%d [%V] %f:%L %m%n"
 
 [rules]
 
-*.INFO     "./logs/%c.log", 10MB * 3 ~ "./logs/%c-%d(%Y%m%d).#2s.log"; simple
+*.INFO     "./logs/%c.log", 10MB * 3 ~ "./logs/%c.#2s.log"; simple