refactor(libwaku): async (#3180)

examples/cbindings/waku_example.c

@@ -5,6 +5,7 @@
 #include <unistd.h>
 #include <fcntl.h>
 #include <stdint.h>
+#include <pthread.h>
 
 #include <sys/types.h>
 #include <unistd.h>
@@ -13,13 +14,30 @@
 #include "base64.h"
 #include "../../library/libwaku.h"
 
+
+// Shared synchronization variables
+pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
+pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
+int callback_executed = 0;
+
+void waitForCallback() {
+    pthread_mutex_lock(&mutex);
+    while (!callback_executed) {
+        pthread_cond_wait(&cond, &mutex);
+    }
+    callback_executed = 0;
+    pthread_mutex_unlock(&mutex);
+}
+
+
 #define WAKU_CALL(call)                                                        \
 do {                                                                           \
   int ret = call;                                                              \
   if (ret != 0) {                                                              \
     printf("Failed the call to: %s. Returned code: %d\n", #call, ret);         \
     exit(1);                                                                   \
   }                                                                            \
+  waitForCallback();                                                           \
 } while (0)
 
 struct ConfigNode {
@@ -99,6 +117,21 @@ void event_handler(int callerRet, const char* msg, size_t len, void* userData) {
     else if (callerRet == RET_OK) {
         printf("Receiving event: %s\n", msg);
     }
+
+    pthread_mutex_lock(&mutex);
+    callback_executed = 1;
+    pthread_cond_signal(&cond);
+    pthread_mutex_unlock(&mutex);
+}
+
+void on_event_received(int callerRet, const char* msg, size_t len, void* userData) {
+    if (callerRet == RET_ERR) {
+        printf("Error: %s\n", msg);
+        exit(1);
+    }
+    else if (callerRet == RET_OK) {
+        printf("Receiving event: %s\n", msg);
+    }
 }
 
 char* contentTopic = NULL;
@@ -161,10 +194,20 @@ void show_help_and_exit() {
 
 void print_default_pubsub_topic(int callerRet, const char* msg, size_t len, void* userData) {
     printf("Default pubsub topic: %s\n", msg);
+
+    pthread_mutex_lock(&mutex);
+    callback_executed = 1;
+    pthread_cond_signal(&cond);
+    pthread_mutex_unlock(&mutex);
 }
 
 void print_waku_version(int callerRet, const char* msg, size_t len, void* userData) {
     printf("Git Version: %s\n", msg);
+
+    pthread_mutex_lock(&mutex);
+    callback_executed = 1;
+    pthread_cond_signal(&cond);
+    pthread_mutex_unlock(&mutex);
 }
 
 // Beginning of UI program logic
@@ -247,15 +290,13 @@ void handle_user_input() {
 // End of UI program logic
 
 int main(int argc, char** argv) {
-    waku_setup();
-
     struct ConfigNode cfgNode;
     // default values
     snprintf(cfgNode.host, 128, "0.0.0.0");
     cfgNode.port = 60000;
     cfgNode.relay = 1;
 
-    cfgNode.store = 1;
+    cfgNode.store = 0;
     snprintf(cfgNode.storeNode, 2048, "");
     snprintf(cfgNode.storeRetentionPolicy, 64, "time:6000000");
     snprintf(cfgNode.storeDbUrl, 256, "postgres://postgres:test123@localhost:5432/postgres");
@@ -296,17 +337,20 @@ int main(int argc, char** argv) {
                                     cfgNode.storeMaxNumDbConnections);
 
     ctx = waku_new(jsonConfig, event_handler, userData);
+    waitForCallback();
 
     WAKU_CALL( waku_default_pubsub_topic(ctx, print_default_pubsub_topic, userData) );
     WAKU_CALL( waku_version(ctx, print_waku_version, userData) );
 
     printf("Bind addr: %s:%u\n", cfgNode.host, cfgNode.port);
     printf("Waku Relay enabled: %s\n", cfgNode.relay == 1 ? "YES": "NO");
 
-    waku_set_event_callback(ctx, event_handler, userData);
+    waku_set_event_callback(ctx, on_event_received, userData);
+
     waku_start(ctx, event_handler, userData);
+    waitForCallback();
 
-    waku_listen_addresses(ctx, event_handler, userData);
+    WAKU_CALL( waku_listen_addresses(ctx, event_handler, userData) );
 
     printf("Establishing connection with: %s\n", cfgNode.peers);
 
@@ -334,4 +378,7 @@ int main(int argc, char** argv) {
     while(1) {
         handle_user_input();
     }
+
+    pthread_mutex_destroy(&mutex);
+    pthread_cond_destroy(&cond);
 }

examples/cpp/waku.cpp

@@ -199,8 +199,6 @@ auto cify(F&& f) {
 }
 
 int main(int argc, char** argv) {
-    waku_setup();
-
     struct ConfigNode cfgNode;
     // default values
     snprintf(cfgNode.host, 128, "0.0.0.0");

examples/golang/waku.go

@@ -296,10 +296,6 @@ type WakuNode struct {
 	ctx unsafe.Pointer
 }
 
-func WakuSetup() {
-	C.waku_setup()
-}
-
 func WakuNew(config WakuConfig) (*WakuNode, error) {
 	jsonConfig, err := json.Marshal(config)
 	if err != nil {
@@ -557,8 +553,6 @@ func (self *WakuNode) WakuGetMyENR() (string, error) {
 }
 
 func main() {
-	WakuSetup()
-
 	config := WakuConfig{
 		Host:        "0.0.0.0",
 		Port:        30304,

examples/mobile/android/app/src/main/jni/waku_ffi.c

@@ -177,7 +177,6 @@ jclass loadClass(JNIEnv *env, const char *className) {
 }
 
 void Java_com_mobile_WakuModule_wakuSetup(JNIEnv *env, jobject thiz) {
-  waku_setup();
   LOGD("log example for debugging purposes...")
 }
 

examples/python/waku.py

@@ -58,9 +58,6 @@ def call_waku(func):
 
 callback_type = ctypes.CFUNCTYPE(None, ctypes.c_int, ctypes.c_char_p, ctypes.c_size_t)
 
-# libwaku setup
-libwaku.waku_setup()
-
 # Node creation
 libwaku.waku_new.restype = ctypes.c_void_p
 libwaku.waku_new.argtypes = [ctypes.c_char_p,

examples/rust/src/main.rs

@@ -13,8 +13,6 @@ pub type WakuCallback =
     );
 
 extern "C" {
-    pub fn waku_setup();
-
     pub fn waku_new(
         config_json: *const u8,
         cb: WakuCallback,
@@ -70,8 +68,6 @@ fn main() {
     }";
 
     unsafe {
-        waku_setup();
-
         // Create the waku node
         let closure = |ret: i32, data: &str| {
             println!("Ret {ret}. Error creating waku node {data}");

library/ffi_types.nim

@@ -0,0 +1,30 @@
+################################################################################
+### Exported types
+
+type WakuCallBack* = proc(
+  callerRet: cint, msg: ptr cchar, len: csize_t, userData: pointer
+) {.cdecl, gcsafe, raises: [].}
+
+const RET_OK*: cint = 0
+const RET_ERR*: cint = 1
+const RET_MISSING_CALLBACK*: cint = 2
+
+### End of exported types
+################################################################################
+
+################################################################################
+### FFI utils
+
+template foreignThreadGc*(body: untyped) =
+  when declared(setupForeignThreadGc):
+    setupForeignThreadGc()
+
+  body
+
+  when declared(tearDownForeignThreadGc):
+    tearDownForeignThreadGc()
+
+type onDone* = proc()
+
+### End of FFI utils
+################################################################################

library/libwaku.h

@@ -20,9 +20,6 @@ extern "C" {
 
 typedef void (*WakuCallBack) (int callerRet, const char* msg, size_t len, void* userData);
 
-// Initializes the library. Should be called before any other function
-void waku_setup();
-
 // Creates a new instance of the waku node.
 // Sets up the waku node from the given configuration.
 // Returns a pointer to the Context needed by the rest of the API functions.