diff --git a/src/connection/connection.c b/src/connection/connection.c
index ef9256d7..908ec21a 100644
--- a/src/connection/connection.c
+++ b/src/connection/connection.c
@@ -237,6 +237,8 @@ int neu_conn_tcp_server_close_client(neu_conn_t *conn, int fd)
 
     conn->disconnected(conn->data, fd);
     conn_tcp_server_del_client(conn, fd);
+    conn->offset = 0;
+    memset(conn->buf, 0, conn->buf_size);
 
     pthread_mutex_unlock(&conn->mtx);
     return 0;
@@ -1100,8 +1102,7 @@ int neu_conn_stream_consume(neu_conn_t *conn, void *context,
             if (used == 0) {
                 break;
             } else if (used == -1) {
-                memset(conn->buf, 0, conn->buf_size);
-                conn->offset = 0;
+                neu_conn_disconnect(conn);
                 break;
             }
         }
@@ -1132,8 +1133,7 @@ int neu_conn_stream_tcp_server_consume(neu_conn_t *conn, int fd, void *context,
             if (used == 0) {
                 break;
             } else if (used == -1) {
-                memset(conn->buf, 0, conn->buf_size);
-                conn->offset = 0;
+                neu_conn_tcp_server_close_client(conn, fd);
                 break;
             }
         }
@@ -1171,7 +1171,10 @@ int neu_conn_wait_msg(neu_conn_t *conn, void *context, uint16_t n_byte,
                            result.need, conn->buf_size, conn->offset);
                 return -1;
             }
+        } else if (result.need == 0) {
+            return result.used;
         } else {
+            neu_conn_disconnect(conn);
             return result.used;
         }
     }
@@ -1203,7 +1206,10 @@ int neu_conn_tcp_server_wait_msg(neu_conn_t *conn, int fd, void *context,
                            result.need, conn->buf_size, conn->offset);
                 return -1;
             }
+        } else if (result.need == 0) {
+            return result.used;
         } else {
+            neu_conn_tcp_server_close_client(conn, fd);
             return result.used;
         }
     }