From 45b3aeda9aeee1431765560c3b7a294c629a70f5 Mon Sep 17 00:00:00 2001
From: eeff <33566201+eeff@users.noreply.github.com>
Date: Mon, 25 Mar 2024 16:22:28 +0800
Subject: [PATCH] fix(ekuiper): check received aio data race

---
 plugins/ekuiper/plugin_ekuiper.c | 9 +++++++++
 plugins/ekuiper/plugin_ekuiper.h | 1 +
 plugins/ekuiper/read_write.c     | 3 +++
 3 files changed, 13 insertions(+)

diff --git a/plugins/ekuiper/plugin_ekuiper.c b/plugins/ekuiper/plugin_ekuiper.c
index a3b68355..583a7e65 100644
--- a/plugins/ekuiper/plugin_ekuiper.c
+++ b/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;
 }
 
diff --git a/plugins/ekuiper/plugin_ekuiper.h b/plugins/ekuiper/plugin_ekuiper.h
index a84c0b51..70874ef7 100644
--- a/plugins/ekuiper/plugin_ekuiper.h
+++ b/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;
diff --git a/plugins/ekuiper/read_write.c b/plugins/ekuiper/read_write.c
index 725a767f..6609804b 100644
--- a/plugins/ekuiper/read_write.c
+++ b/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;
     }