fix(ekuiper): check received aio data race

plugins/ekuiper/plugin_ekuiper.c

@@ -142,7 +142,16 @@ static inline int start(neu_plugin_t *plugin, const char *url)
         return rv;
     }
 
+    nng_mtx_lock(plugin->mtx);
+    while (plugin->receiving) {
+        nng_mtx_unlock(plugin->mtx);
+        nng_msleep(10);
+        nng_mtx_lock(plugin->mtx);
+    }
+    plugin->receiving = true;
     nng_recv_aio(plugin->sock, plugin->recv_aio);
+    nng_mtx_unlock(plugin->mtx);
+
     return NEU_ERR_SUCCESS;
 }
 

plugins/ekuiper/plugin_ekuiper.h

@@ -37,6 +37,7 @@ struct neu_plugin {
     nng_socket          sock;
     nng_mtx *           mtx;
     bool                started;
+    bool                receiving;
     nng_aio *           recv_aio;
     char *              host;
     uint16_t            port;

plugins/ekuiper/read_write.c

@@ -86,6 +86,9 @@ void recv_data_callback(void *arg)
     rv = nng_aio_result(plugin->recv_aio);
     if (0 != rv) {
         plog_error(plugin, "nng_recv error: %s", nng_strerror(rv));
+        nng_mtx_lock(plugin->mtx);
+        plugin->receiving = false;
+        nng_mtx_unlock(plugin->mtx);
         return;
     }