Add tokio to node tests

Cargo.lock

@@ -439,6 +439,17 @@ version = "0.3.25"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "00f5fb52a06bdcadeb54e8d3671f8888a39697dcb0b81b23b55174030427f4eb"
 
+[[package]]
+name = "futures-macro"
+version = "0.3.25"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bdfb8ce053d86b91919aad980c220b1fb8401a9394410e1c289ed7e66b61835d"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "futures-sink"
 version = "0.3.25"
@@ -460,6 +471,7 @@ dependencies = [
  "futures-channel",
  "futures-core",
  "futures-io",
+ "futures-macro",
  "futures-sink",
  "futures-task",
  "memchr",
@@ -1232,6 +1244,29 @@ version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cda74da7e1a664f795bb1f8a87ec406fb89a02522cf6e50620d016add6dbbf5c"
 
+[[package]]
+name = "tokio"
+version = "1.24.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "597a12a59981d9e3c38d216785b0c37399f6e415e8d0712047620f189371b0bb"
+dependencies = [
+ "autocfg",
+ "pin-project-lite",
+ "tokio-macros",
+ "windows-sys 0.42.0",
+]
+
+[[package]]
+name = "tokio-macros"
+version = "1.8.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d266c00fde287f55d3f1c3e96c500c362a2b8c695076ec180f27918820bc6df8"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "toml"
 version = "0.5.9"
@@ -1350,6 +1385,7 @@ version = "0.1.0-beta3"
 dependencies = [
  "aes-gcm",
  "base64",
+ "futures",
  "hex",
  "multiaddr",
  "once_cell",
@@ -1360,6 +1396,7 @@ dependencies = [
  "serial_test",
  "smart-default",
  "sscanf",
+ "tokio",
  "url",
  "waku-sys",
 ]

waku-bindings/Cargo.toml

@@ -30,4 +30,4 @@ waku-sys = { version = "0.1.0-beta3", path = "../waku-sys" }
 [dev-dependencies]
 futures = "0.3.25"
 serial_test = "0.10.0"
-tokio = { version = "1.24.2", features = ["sync"] }
+tokio = { version = "1.24.2", features = ["macros", "rt", "sync", "time"] }

waku-bindings/tests/node.rs

@@ -5,14 +5,18 @@ use secp256k1::{PublicKey, Secp256k1, SecretKey};
 use serial_test::serial;
 use std::net::IpAddr;
 use std::str::FromStr;
-use std::time::SystemTime;
+use std::time::{Duration, SystemTime};
 use std::{collections::HashSet, str::from_utf8};
 use tokio::sync::mpsc::{self, Sender};
+use tokio::time;
 use waku_bindings::{
     waku_new, waku_set_event_callback, Encoding, Event, Key, MessageId, ProtocolId, Running,
     WakuContentTopic, WakuLogLevel, WakuMessage, WakuNodeConfig, WakuNodeHandle,
 };
 
+const ECHO_TIMEOUT: u64 = 10;
+const ECHO_MESSAGE: &str = "Hi from 🦀!";
+
 const NODES: &[&str] = &[
     "/dns4/node-01.ac-cn-hongkong-c.wakuv2.test.statusim.net/tcp/30303/p2p/16Uiu2HAkvWiyFsgRhuJEb9JfjYxEkoHLgnUQmr1N5mKWnYjxYRVm",
     "/dns4/node-01.do-ams3.wakuv2.test.statusim.net/tcp/30303/p2p/16Uiu2HAmPLe7Mzm8TsYUubgCAW1aJoeFScxrLj8ppHFivPo97bUZ",
@@ -97,7 +101,7 @@ fn set_callback(tx: Sender<Response>, sk: SecretKey, ssk: Key<Aes256Gcm>) {
     });
 }
 
-fn test_echo_messages(
+async fn test_echo_messages(
     node: &WakuNodeHandle<Running>,
     content: &'static str,
     content_topic: WakuContentTopic,
@@ -116,7 +120,7 @@ fn test_echo_messages(
             .unwrap(),
     );
 
-    let (tx, mut rx) = mpsc::channel(100);
+    let (tx, mut rx) = mpsc::channel(1);
     set_callback(tx, sk, ssk);
 
     let mut ids =
@@ -126,7 +130,7 @@ fn test_echo_messages(
         try_publish_lightpush_messages(node, &message, &sk, &ssk).expect("send lightpush messages"),
     );
 
-    while let Some(res) = futures::executor::block_on(rx.recv()) {
+    while let Some(res) = rx.recv().await {
         if ids.take(&res.id).is_some() {
             let msg = from_utf8(&res.payload).expect("should be valid message");
             assert_eq!(content, msg);
@@ -139,23 +143,16 @@ fn test_echo_messages(
 }
 
 #[ignore]
-#[test]
+#[tokio::test]
 #[serial]
-fn discv5_echo() -> Result<(), String> {
+async fn discv5_echo() -> Result<(), String> {
     let config = WakuNodeConfig {
         host: IpAddr::from_str("0.0.0.0").ok(),
-        port: None,
-        advertise_addr: None,
-        node_key: None,
-        keep_alive_interval: None,
-        relay: None,
-        relay_topics: vec![],
-        min_peers_to_publish: None,
-        filter: None,
         log_level: Some(WakuLogLevel::Error),
         discv5: Some(true),
         discv5_udp_port: Some(9000),
         discv5_bootstrap_nodes: Vec::new(),
+        ..Default::default()
     };
 
     let node = waku_new(Some(config))?;
@@ -175,14 +172,20 @@ fn discv5_echo() -> Result<(), String> {
     let sk = SecretKey::new(&mut thread_rng());
     let ssk = Aes256Gcm::generate_key(&mut thread_rng());
 
-    let content = "Hi from 🦀!";
-
     // Subscribe to default channel.
     node.relay_subscribe(None)?;
     let content_topic = WakuContentTopic::new("toychat", 2, "huilong", Encoding::Proto);
 
+    let sleep = time::sleep(Duration::from_secs(ECHO_TIMEOUT));
+    tokio::pin!(sleep);
+
     // Send and receive messages. Waits until all messages received.
-    test_echo_messages(&node, content, content_topic, sk, ssk);
+    let got_all = tokio::select! {
+        _ = sleep => false,
+        _ = test_echo_messages(&node, ECHO_MESSAGE, content_topic, sk, ssk) => true,
+    };
+
+    assert!(got_all);
 
     for node_data in node.peers()? {
         if node_data.peer_id() != &node.peer_id()? {
@@ -195,10 +198,14 @@ fn discv5_echo() -> Result<(), String> {
 }
 
 #[ignore]
-#[test]
+#[tokio::test]
 #[serial]
-fn default_echo() -> Result<(), String> {
+async fn default_echo() -> Result<(), String> {
-    let config = Default::default();
+    let config = WakuNodeConfig {
+        log_level: Some(WakuLogLevel::Error),
+        ..Default::default()
+    };
+
     let node = waku_new(Some(config))?;
     let node = node.start()?;
     println!("Node peer id: {}", node.peer_id()?);
@@ -216,14 +223,20 @@ fn default_echo() -> Result<(), String> {
     let sk = SecretKey::new(&mut thread_rng());
     let ssk = Aes256Gcm::generate_key(&mut thread_rng());
 
-    let content = "Hi from 🦀!";
-
     // subscribe to default channel
     node.relay_subscribe(None)?;
     let content_topic = WakuContentTopic::new("toychat", 2, "huilong", Encoding::Proto);
 
+    let sleep = time::sleep(Duration::from_secs(ECHO_TIMEOUT));
+    tokio::pin!(sleep);
+
     // Send and receive messages. Waits until all messages received.
-    test_echo_messages(&node, content, content_topic, sk, ssk);
+    let got_all = tokio::select! {
+        _ = sleep => false,
+        _ = test_echo_messages(&node, ECHO_MESSAGE, content_topic, sk, ssk) => true,
+    };
+
+    assert!(got_all);
 
     for node_data in node.peers()? {
         if node_data.peer_id() != &node.peer_id()? {