diff --git a/Cargo.lock b/Cargo.lock index 80d5ee9..12ea66c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1,6 +1,6 @@ # This file is automatically @generated by Cargo. # It is not intended for manual editing. -version = 3 +version = 4 [[package]] name = "addr2line" @@ -72,7 +72,7 @@ dependencies = [ "cfg-if", "once_cell", "version_check", - "zerocopy", + "zerocopy 0.7.32", ] [[package]] @@ -379,6 +379,7 @@ checksum = "b645a089122eccb6111b4f81cbc1a49f5900ac4666bb93ac027feaecf15607bf" name = "basic" version = "0.1.0" dependencies = [ + "base64 0.21.0", "futures", "serde_json", "tokio", @@ -386,28 +387,6 @@ dependencies = [ "waku-bindings", ] -[[package]] -name = "bindgen" -version = "0.64.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4243e6031260db77ede97ad86c27e501d646a27ab57b59a574f725d98ab1fb4" -dependencies = [ - "bitflags 1.3.2", - "cexpr", - "clang-sys", - "lazy_static", - "lazycell", - "log", - "peeking_take_while", - "proc-macro2", - "quote", - "regex", - "rustc-hash", - "shlex", - "syn 1.0.100", - "which", -] - [[package]] name = "bitflags" version = "1.3.2" @@ -508,15 +487,6 @@ version = "1.0.73" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2fff2a6927b3bb87f9595d67196a70493f627687a71d87a0d692242c33f58c11" -[[package]] -name = "cexpr" -version = "0.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6fac387a98bb7c37292057cffc56d62ecb629900026402633ae9160df93a8766" -dependencies = [ - "nom", -] - [[package]] name = "cfg-if" version = "1.0.0" @@ -538,6 +508,33 @@ dependencies = [ "winapi", ] +[[package]] +name = "ciborium" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42e69ffd6f0917f5c029256a24d0161db17cea3997d185db0d35926308770f0e" +dependencies = [ + "ciborium-io", + "ciborium-ll", + "serde", +] + +[[package]] +name = "ciborium-io" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05afea1e0a06c9be33d539b876f1ce3692f4afea2cb41f740e7743225ed1c757" + +[[package]] +name = "ciborium-ll" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57663b653d948a338bfb3eeba9bb2fd5fcfaecb9e199e87e1eda4d9e8b240fd9" +dependencies = [ + "ciborium-io", + "half 2.7.1", +] + [[package]] name = "cipher" version = "0.4.3" @@ -548,17 +545,6 @@ dependencies = [ "inout", ] -[[package]] -name = "clang-sys" -version = "1.4.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fa2e27ae6ab525c3d369ded447057bca5438d86dc3a68f6faafb8269ba82ebf3" -dependencies = [ - "glob", - "libc", - "libloading", -] - [[package]] name = "clap" version = "2.34.0" @@ -1148,6 +1134,17 @@ dependencies = [ "subtle", ] +[[package]] +name = "flume" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095" +dependencies = [ + "futures-core", + "futures-sink", + "spin", +] + [[package]] name = "fnv" version = "1.0.7" @@ -1325,12 +1322,6 @@ version = "0.28.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4271d37baee1b8c7e4b708028c57d816cf9d2434acb33a549475f78c181f6253" -[[package]] -name = "glob" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9b919933a397b79c37e33b77bb2aa3dc8eb6e165ad809e58ff75bc7db2e34574" - [[package]] name = "group" version = "0.12.1" @@ -1348,6 +1339,17 @@ version = "1.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "eabb4a44450da02c90444cf74558da904edde8fb4e9035a9a6a4e15445af0bd7" +[[package]] +name = "half" +version = "2.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ea2d84b969582b4b1864a92dc5d27cd2b77b622a8d79306834f1be5ba20d84b" +dependencies = [ + "cfg-if", + "crunchy", + "zerocopy 0.8.54", +] + [[package]] name = "hashbrown" version = "0.11.2" @@ -1560,12 +1562,6 @@ version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646" -[[package]] -name = "lazycell" -version = "1.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55" - [[package]] name = "leb128" version = "0.2.5" @@ -1676,12 +1672,6 @@ dependencies = [ "autocfg", ] -[[package]] -name = "minimal-lexical" -version = "0.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" - [[package]] name = "miniz_oxide" version = "0.7.2" @@ -1762,16 +1752,6 @@ dependencies = [ "synstructure", ] -[[package]] -name = "nom" -version = "7.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a8903e5a29a317527874d0402f867152a3d21c908bb0b933e416c65e301d4c36" -dependencies = [ - "memchr", - "minimal-lexical", -] - [[package]] name = "num" version = "0.4.0" @@ -1958,12 +1938,6 @@ version = "1.0.14" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "de3145af08024dea9fa9914f381a17b8fc6034dfb00f3a84013f7ff43f29ed4c" -[[package]] -name = "peeking_take_while" -version = "0.1.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "19b17cddbe7ec3f8bc800887bab5e717348c95ea2ca0b1bf0837fb964dc67099" - [[package]] name = "percent-encoding" version = "2.2.0" @@ -2084,9 +2058,9 @@ dependencies = [ [[package]] name = "proc-macro2" -version = "1.0.78" +version = "1.0.106" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e2422ad645d89c99f8f3e6b88a9fdeca7fabeac836b1002371c4367c8f984aae" +checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934" dependencies = [ "unicode-ident", ] @@ -2113,9 +2087,9 @@ dependencies = [ [[package]] name = "quote" -version = "1.0.35" +version = "1.0.46" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "291ec9ab5efd934aaf503a6466c5d5251535d108ee747472c3977cc5acc868ef" +checksum = "dfbc457d0c7a0759a614551b11a6409e5951f6c7537be1f1b7682b9ae9230368" dependencies = [ "proc-macro2", ] @@ -2485,7 +2459,7 @@ version = "0.11.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2bef2ebfde456fb76bbcf9f59315333decc4fda0b2b44b420243c11e0f5ec1f5" dependencies = [ - "half", + "half 1.8.2", "serde", ] @@ -2566,12 +2540,6 @@ dependencies = [ "lazy_static", ] -[[package]] -name = "shlex" -version = "1.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "43b2853a4d09f215c24cc5489c992ce46052d359b5109343cbafbf26bc62f8a3" - [[package]] name = "signal-hook-registry" version = "1.4.1" @@ -2649,6 +2617,15 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "spin" +version = "0.9.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3763264f6b73151db08c50ff20d7d8a0b8796e021cdea7ceedad07b80155fa0e" +dependencies = [ + "lock_api", +] + [[package]] name = "spki" version = "0.6.0" @@ -3097,8 +3074,11 @@ dependencies = [ name = "waku-sys" version = "1.0.0" dependencies = [ - "bindgen", "cc", + "ciborium", + "flume", + "serde", + "tokio", ] [[package]] @@ -3743,7 +3723,16 @@ version = "0.7.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "74d4d3961e53fa4c9a25a8637fc2bfaf2595b3d3ae34875568a5cf64787716be" dependencies = [ - "zerocopy-derive", + "zerocopy-derive 0.7.32", +] + +[[package]] +name = "zerocopy" +version = "0.8.54" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7cbbc0a705a0fd05cc3676525980d2bf5a9bc4adac6d6475209a7887cf59d19" +dependencies = [ + "zerocopy-derive 0.8.54", ] [[package]] @@ -3757,6 +3746,17 @@ dependencies = [ "syn 2.0.48", ] +[[package]] +name = "zerocopy-derive" +version = "0.8.54" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e2e817b7b52d0c7358d3246da9d69935ebb18116b2b102b4230dac079b4862f5" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.48", +] + [[package]] name = "zeroize" version = "1.5.7" diff --git a/examples/basic/Cargo.toml b/examples/basic/Cargo.toml index 29ddd5e..a539876 100644 --- a/examples/basic/Cargo.toml +++ b/examples/basic/Cargo.toml @@ -11,3 +11,4 @@ tokio = { version = "1.36.0", features = ["full"] } tokio-util = { version = "0.7.10", features = ["rt"] } waku = { path = "../../waku-bindings", package = "waku-bindings" } serde_json = "1.0" +base64 = "0.21" diff --git a/examples/basic/src/main.rs b/examples/basic/src/main.rs index d0da1c3..df2411f 100644 --- a/examples/basic/src/main.rs +++ b/examples/basic/src/main.rs @@ -1,101 +1,60 @@ +use base64::Engine; use std::io::Error; use std::str::from_utf8; use tokio::time::{sleep, Duration}; use waku::{ - general::pubsubtopic::PubsubTopic, waku_new, Encoding, LibwakuResponse, WakuContentTopic, - WakuEvent, WakuMessage, WakuNodeConfig, + Encoding, LogosDeliveryCtx, ReceivedMessagePayload, WakuContentTopic, WakuMessage, + WakuNodeConfig, }; +const TOPIC: &str = "test"; +const TIMEOUT: Duration = Duration::from_secs(30); + +fn new_node(tcp_port: usize) -> LogosDeliveryCtx { + let config = serde_json::to_string(&WakuNodeConfig { + tcp_port: Some(tcp_port), + ..Default::default() + }) + .expect("config should serialise"); + + LogosDeliveryCtx::create(config, TIMEOUT).expect("should instantiate") +} + +fn print_received(node_name: &'static str) -> impl Fn(&ReceivedMessagePayload) { + move |event| { + let payload = base64::engine::general_purpose::STANDARD + .decode(&event.waku_message.payload) + .expect("payload should be base64"); + let msg = from_utf8(&payload).expect("should be valid message"); + println!("::::::::::::::::::::::::::::::::::::::::::::::::::::"); + println!("Message Received in {node_name}: {msg}"); + println!("::::::::::::::::::::::::::::::::::::::::::::::::::::"); + } +} + #[tokio::main] async fn main() -> Result<(), Error> { - let node1 = waku_new(Some(WakuNodeConfig { - tcp_port: Some(60010), // TODO: use any available port. - ..Default::default() - })) - .await - .expect("should instantiate"); - - let node2 = waku_new(Some(WakuNodeConfig { - tcp_port: Some(60020), // TODO: use any available port. - ..Default::default() - })) - .await - .expect("should instantiate"); + let node1 = new_node(60010); // TODO: use any available port. + let node2 = new_node(60020); // TODO: use any available port. // ======================================================================== - // Setting an event callback to be executed each time a message is received - node2 - .set_event_callback(|response| { - if let LibwakuResponse::Success(v) = response { - let event: WakuEvent = - serde_json::from_str(v.unwrap().as_str()).expect("Parsing event to succeed"); + // Registering a listener to be executed each time a message is received + node2.add_on_received_message_listener(print_received("NODE 2")); + node1.add_on_received_message_listener(print_received("NODE 1")); - match event { - WakuEvent::WakuMessage(evt) => { - // println!("WakuMessage event received: {:?}", evt.waku_message); - let message = evt.waku_message; - let payload = message.payload.to_vec(); - let msg = from_utf8(&payload).expect("should be valid message"); - println!("::::::::::::::::::::::::::::::::::::::::::::::::::::"); - println!("Message Received in NODE 2: {}", msg); - println!("::::::::::::::::::::::::::::::::::::::::::::::::::::"); - } - WakuEvent::RelayTopicHealthChange(_evt) => { - // dbg!("Relay topic change evt", evt); - } - WakuEvent::ConnectionChange(_evt) => { - // dbg!("Conn change evt", evt); - } - WakuEvent::Unrecognized(err) => panic!("Unrecognized waku event: {:?}", err), - _ => panic!("event case not expected"), - }; - } - }) - .expect("set event call back working"); - - node1 - .set_event_callback(|response| { - if let LibwakuResponse::Success(v) = response { - let event: WakuEvent = - serde_json::from_str(v.unwrap().as_str()).expect("Parsing event to succeed"); - - match event { - WakuEvent::WakuMessage(evt) => { - // println!("WakuMessage event received: {:?}", evt.waku_message); - let message = evt.waku_message; - let payload = message.payload.to_vec(); - let msg = from_utf8(&payload).expect("should be valid message"); - println!("::::::::::::::::::::::::::::::::::::::::::::::::::::"); - println!("Message Received in NODE 1: {}", msg); - println!("::::::::::::::::::::::::::::::::::::::::::::::::::::"); - } - WakuEvent::RelayTopicHealthChange(_evt) => { - // dbg!("Relay topic change evt", evt); - } - WakuEvent::ConnectionChange(_evt) => { - // dbg!("Conn change evt", evt); - } - WakuEvent::Unrecognized(err) => panic!("Unrecognized waku event: {:?}", err), - _ => panic!("event case not expected"), - }; - } - }) - .expect("set event call back working"); - - let node1 = node1.start().await.expect("node1 should start"); - let node2 = node2.start().await.expect("node2 should start"); + node1.start_node_async().await.expect("node1 should start"); + node2.start_node_async().await.expect("node2 should start"); // ======================================================================== // Subscribe to pubsub topic - let topic = PubsubTopic::new("test"); node1 - .relay_subscribe(&topic) + .waku_relay_subscribe_async(TOPIC.to_string()) .await .expect("node1 should subscribe"); node2 - .relay_subscribe(&topic) + .waku_relay_subscribe_async(TOPIC.to_string()) .await .expect("node2 should subscribe"); @@ -103,12 +62,16 @@ async fn main() -> Result<(), Error> { // Connect nodes with each other let addresses2 = node2 - .listen_addresses() + .waku_listen_addresses_async() .await .expect("should obtain the addresses"); + let address2 = addresses2 + .split(',') + .next() + .expect("node2 should report a listen address"); node1 - .connect(&addresses2[0], None) + .waku_connect_async(address2.to_string(), 10_000) .await .expect("node1 should connect to node2"); @@ -122,8 +85,9 @@ async fn main() -> Result<(), Error> { let content_topic = WakuContentTopic::new("waku", "2", "test", Encoding::Proto); let message = WakuMessage::new("Hello world", content_topic, 0, Vec::new(), false); + let message = serde_json::to_string(&message).expect("message should serialise"); node1 - .relay_publish_message(&message, &topic, None) + .waku_relay_publish_async(TOPIC.to_string(), message, 10_000) .await .expect("should have sent the message"); @@ -135,13 +99,10 @@ async fn main() -> Result<(), Error> { // ======================================================================== // Stop both instances - let node1 = node1.stop().await.expect("should stop"); - let node2 = node2.stop().await.expect("should stop"); + node1.stop_node_async().await.expect("should stop"); + node2.stop_node_async().await.expect("should stop"); - // ======================================================================== - // Free resources - node1.waku_destroy().await.expect("should deallocate"); - node2.waku_destroy().await.expect("should deallocate"); + // Resources are freed by LogosDeliveryCtx's Drop impl. Ok(()) } diff --git a/waku-bindings/src/general/libwaku_response.rs b/waku-bindings/src/general/libwaku_response.rs deleted file mode 100644 index cd66bd8..0000000 --- a/waku-bindings/src/general/libwaku_response.rs +++ /dev/null @@ -1,64 +0,0 @@ -use crate::general::waku_decode::WakuDecode; -use crate::general::Result; -use std::convert::TryFrom; -use std::str; -use waku_sys::{RET_ERR, RET_MISSING_CALLBACK, RET_OK}; - -#[derive(Debug, Clone, Default, PartialEq)] -pub enum LibwakuResponse { - Success(Option), - Failure(String), - MissingCallback, - #[default] - Undefined, -} - -impl TryFrom<(u32, &str)> for LibwakuResponse { - type Error = String; - - fn try_from((ret_code, response): (u32, &str)) -> std::result::Result { - let opt_value = Some(response.to_string()).filter(|s| !s.is_empty()); - match ret_code { - RET_OK => Ok(LibwakuResponse::Success(opt_value)), - RET_ERR => Ok(LibwakuResponse::Failure(format!( - "waku error: {}", - response - ))), - RET_MISSING_CALLBACK => Ok(LibwakuResponse::MissingCallback), - _ => Err(format!("undefined return code {}", ret_code)), - } - } -} - -/// Used in cases where the FFI call doesn't return additional information in the -/// callback. Instead, it returns RET_OK, RET_ERR, etc. -pub fn handle_no_response(code: i32, result: LibwakuResponse) -> Result<()> { - if result == LibwakuResponse::Undefined && code as u32 == RET_OK { - // Some functions will only execute the callback on error - return Ok(()); - } - - match result { - LibwakuResponse::Success(_) => Ok(()), - LibwakuResponse::Failure(v) => Err(v), - LibwakuResponse::MissingCallback => panic!("callback is required"), - LibwakuResponse::Undefined => panic!( - "undefined ffi state: code({}) was returned but callback was not executed", - code - ), - } -} - -/// Used in cases where the FFI function returns a code (RET_OK, RET_ERR, etc) plus additional -/// information, i.e. LibwakuResponse -pub fn handle_response(code: i32, result: LibwakuResponse) -> Result { - match result { - LibwakuResponse::Success(v) => WakuDecode::decode(&v.unwrap_or_default()), - LibwakuResponse::Failure(v) => Err(v), - LibwakuResponse::MissingCallback => panic!("callback is required"), - LibwakuResponse::Undefined => panic!( - "undefined ffi state: code({}) was returned but callback was not executed", - code - ), - } -} diff --git a/waku-bindings/src/general/mod.rs b/waku-bindings/src/general/mod.rs index 25e74f8..d078eb4 100644 --- a/waku-bindings/src/general/mod.rs +++ b/waku-bindings/src/general/mod.rs @@ -1,7 +1,6 @@ //! Waku [general](https://rfc.vac.dev/spec/36/#general) types pub mod contenttopic; -pub mod libwaku_response; pub mod messagehash; pub mod pubsubtopic; pub mod time; diff --git a/waku-bindings/src/lib.rs b/waku-bindings/src/lib.rs index 7576a56..2014f0f 100644 --- a/waku-bindings/src/lib.rs +++ b/waku-bindings/src/lib.rs @@ -1,12 +1,17 @@ //! # Waku //! -//! Implementation on top of [`waku-bindings`](https://rfc.vac.dev/spec/36/) +//! The FFI surface is generated from the Nim source by nim-ffi and re-exported +//! here wholesale: drive a node through `LogosDeliveryCtx`, and observe it +//! through its typed `add_on_*_listener` methods. What this crate adds on top is +//! the domain layer — [`WakuMessage`], [`WakuContentTopic`], [`MessageHash`] and +//! friends — plus [`WakuNodeConfig`], which serialises to the JSON that +//! `LogosDeliveryCtx::create` expects. pub mod general; -mod macros; pub mod node; -// Re-export the LibwakuResponse type to make it accessible outside this module -pub use general::libwaku_response::LibwakuResponse; +// The generated bindings: LogosDeliveryCtx, ListenerHandle, and one payload type +// per event. +pub use waku_sys::*; // Required so functions inside libwaku can call RLN functions even if we // use it within the bindings functions @@ -14,12 +19,8 @@ pub use general::libwaku_response::LibwakuResponse; #[allow(unused)] use rln; -pub use node::{ - waku_create_content_topic, waku_new, ChannelMessageErrorEvent, ChannelMessageReceivedEvent, - ChannelMessageSentEvent, ConnectionStatusChangeEvent, Initialized, Key, MessageErrorEvent, - MessagePropagatedEvent, MessageReceivedEvent, MessageSentEvent, Multiaddr, PublicKey, - RLNConfig, Running, SecretKey, WakuEvent, WakuMessageEvent, WakuNodeConfig, WakuNodeHandle, -}; - pub use general::contenttopic::{Encoding, WakuContentTopic}; pub use general::{messagehash::MessageHash, Result, WakuMessage, WakuMessageVersion}; +pub use node::{ + Key, Multiaddr, PublicKey, PubsubTopic, RLNConfig, SecretKey, WakuNodeConfig, +}; diff --git a/waku-bindings/src/macros.rs b/waku-bindings/src/macros.rs deleted file mode 100644 index 665a1b0..0000000 --- a/waku-bindings/src/macros.rs +++ /dev/null @@ -1,73 +0,0 @@ -use crate::general::libwaku_response::LibwakuResponse; - -use std::{slice, str}; -use waku_sys::FFICallBack; - -unsafe extern "C" fn trampoline( - ret_code: ::std::os::raw::c_int, - data: *const ::std::os::raw::c_char, - data_len: usize, - user_data: *mut ::std::os::raw::c_void, -) where - F: FnMut(LibwakuResponse), -{ - let closure = &mut *(user_data as *mut F); - - let response = if data.is_null() { - "" - } else { - str::from_utf8(slice::from_raw_parts(data as *mut u8, data_len)) - .expect("could not retrieve response") - }; - - let result = LibwakuResponse::try_from((ret_code as u32, response)) - .expect("invalid response obtained from libwaku"); - - closure(result); -} - -pub fn get_trampoline(_closure: &F) -> FFICallBack -where - F: FnMut(LibwakuResponse), -{ - Some(trampoline::) -} - -#[macro_export] -macro_rules! handle_ffi_call { - // Case: With or without additional arguments - ($waku_fn:expr, $resp_hndlr:expr, $ctx:expr $(, $($arg:expr),*)?) => {{ - use $crate::macros::get_trampoline; - use std::sync::Arc; - use tokio::sync::Notify; - use libc::*; - - let mut result = LibwakuResponse::default(); - let notify = Arc::new(Notify::new()); - let notify_clone = notify.clone(); - - // Callback to update the result and notify the waiter - let result_cb = |r: LibwakuResponse| { - result = r; - notify_clone.notify_one(); - }; - - let mut closure = result_cb; - // Create trampoline and invoke the `waku_sys` function - let code = unsafe { - let cb = get_trampoline(&closure); - $waku_fn( - $ctx, // Pass the context - cb, // Pass the callback trampoline - &mut closure as *mut _ as *mut c_void, - $($($arg),*)? // Expand the variadic arguments if provided - ) - }; - - // Wait for the callback to notify us - notify.notified().await; - - // Handle the response - $resp_hndlr(code, result) - }}; -} diff --git a/waku-bindings/src/node/channels.rs b/waku-bindings/src/node/channels.rs deleted file mode 100644 index b43e145..0000000 --- a/waku-bindings/src/node/channels.rs +++ /dev/null @@ -1,93 +0,0 @@ -//! Reliable channels: an ordered, gap-detecting message stream over a content topic. - -// std -use std::ffi::CString; -// crates -use base64::Engine; -use serde::Serialize; -// internal -use crate::general::libwaku_response::{handle_no_response, handle_response, LibwakuResponse}; -use crate::general::Result; -use crate::handle_ffi_call; -use crate::node::context::WakuNodeContext; - -/// Body of `logosdelivery_channel_send`, whose payload travels base64-encoded. -#[derive(Serialize)] -struct ChannelMessage { - payload: String, - ephemeral: bool, -} - -/// Create (or restore) a reliable channel, returning its id. -/// -/// Channel state is persisted, so re-creating a previously closed channel -/// resumes it rather than starting a new one. -pub async fn logosdelivery_channel_create( - ctx: &WakuNodeContext, - channel_id: &str, - content_topic: &str, - sender_id: &str, -) -> Result { - let channel_id = - CString::new(channel_id).expect("CString should build properly from the channel id"); - let content_topic = - CString::new(content_topic).expect("CString should build properly from the content topic"); - let sender_id = - CString::new(sender_id).expect("CString should build properly from the sender id"); - - handle_ffi_call!( - waku_sys::logosdelivery_channel_create, - handle_response, - ctx.get_ptr(), - channel_id.as_ptr(), - content_topic.as_ptr(), - sender_id.as_ptr() - ) -} - -/// Send a message over a reliable channel, returning the id of the send request. -/// -/// That id is what correlates this call with the `onChannelMessageSent` / -/// `onChannelMessageError` events reporting the message's fate. -pub async fn logosdelivery_channel_send( - ctx: &WakuNodeContext, - channel_id: &str, - payload: &[u8], - ephemeral: bool, -) -> Result { - let channel_id = - CString::new(channel_id).expect("CString should build properly from the channel id"); - - let message = ChannelMessage { - payload: base64::engine::general_purpose::STANDARD.encode(payload), - ephemeral, - }; - let message = CString::new( - serde_json::to_string(&message) - .expect("ChannelMessage should always be able to success serializing"), - ) - .expect("CString should build properly from the serialized channel message"); - - handle_ffi_call!( - waku_sys::logosdelivery_channel_send, - handle_response, - ctx.get_ptr(), - channel_id.as_ptr(), - message.as_ptr() - ) -} - -/// Close a reliable channel, stopping its loops. -/// -/// Persisted state survives, so the channel can be restored by re-creating it. -pub async fn logosdelivery_channel_close(ctx: &WakuNodeContext, channel_id: &str) -> Result<()> { - let channel_id = - CString::new(channel_id).expect("CString should build properly from the channel id"); - - handle_ffi_call!( - waku_sys::logosdelivery_channel_close, - handle_no_response, - ctx.get_ptr(), - channel_id.as_ptr() - ) -} diff --git a/waku-bindings/src/node/context.rs b/waku-bindings/src/node/context.rs deleted file mode 100644 index 30b3c14..0000000 --- a/waku-bindings/src/node/context.rs +++ /dev/null @@ -1,63 +0,0 @@ -use std::ffi::c_void; -use std::ptr::null_mut; -use std::sync::{Arc, Mutex}; - -use crate::general::libwaku_response::LibwakuResponse; -use crate::macros::get_trampoline; - -type LibwakuResponseClosure = dyn FnMut(LibwakuResponse) + Send + Sync; - -pub struct WakuNodeContext { - obj_ptr: *mut c_void, - msg_observer: Arc>>, -} - -impl WakuNodeContext { - pub fn new(obj_ptr: *mut c_void) -> Self { - let me = Self { - obj_ptr, - msg_observer: Arc::new(Mutex::new(Box::new(|_| {}))), - }; - - // By default we set a callback that will panic if the user didn't specify a valid callback. - // And by valid callback we mean a callback that can properly handle the waku events. - me.waku_set_event_callback(WakuNodeContext::panic_callback) - .expect("correctly set default callback"); - me - } - - // default callback that does nothing. A valid callback should be set - fn panic_callback(_response: LibwakuResponse) { - panic!("callback not set. Please use waku_set_event_callback to set a valid callback") - } - - pub fn get_ptr(&self) -> *mut c_void { - self.obj_ptr - } - - pub fn reset_ptr(mut self) { - self.obj_ptr = null_mut(); - } - - /// Register callback to act as event handler and receive application events, - /// which are used to react to asynchronous events in Waku - pub fn waku_set_event_callback( - &self, - closure: F, - ) -> Result<(), String> { - if let Ok(mut boxed_closure) = self.msg_observer.lock() { - *boxed_closure = Box::new(closure); - unsafe { - let cb = get_trampoline(&(*boxed_closure)); - waku_sys::logosdelivery_set_event_callback( - self.obj_ptr, - cb, - &mut (*boxed_closure) as *mut _ as *mut c_void, - ) - }; - Ok(()) - } else { - Err("Failed to acquire lock in waku_set_event_callback!".to_string()) - } - } -} diff --git a/waku-bindings/src/node/events.rs b/waku-bindings/src/node/events.rs deleted file mode 100644 index 028b9e9..0000000 --- a/waku-bindings/src/node/events.rs +++ /dev/null @@ -1,279 +0,0 @@ -//! Waku message [event](https://rfc.vac.dev/spec/36/#events) related items -//! -//! Asynchronous events require a callback to be registered. -//! An example of an asynchronous event that might be emitted is receiving a message. -//! When an event is emitted, this callback will be triggered receiving an [`WakuEvent`] - -// crates -use serde::{Deserialize, Serialize}; -// internal -use crate::general::WakuMessage; -use std::str; - -use crate::MessageHash; - -/// Waku event -/// For now just WakuMessage is supported -#[non_exhaustive] -#[derive(Serialize, Deserialize, Debug)] -#[serde(tag = "eventType", rename_all = "camelCase")] -pub enum WakuEvent { - #[serde(rename = "message")] - WakuMessage(WakuMessageEvent), - - #[serde(rename = "relay_topic_health_change")] - RelayTopicHealthChange(TopicHealthEvent), - - #[serde(rename = "connection_change")] - ConnectionChange(ConnectionChangeEvent), - - #[serde(rename = "connection_status_change")] - ConnectionStatusChange(ConnectionStatusChangeEvent), - - #[serde(rename = "message_sent")] - MessageSent(MessageSentEvent), - - #[serde(rename = "message_error")] - MessageError(MessageErrorEvent), - - #[serde(rename = "message_propagated")] - MessagePropagated(MessagePropagatedEvent), - - #[serde(rename = "message_received")] - MessageReceived(MessageReceivedEvent), - - #[serde(rename = "channel_message_received")] - ChannelMessageReceived(ChannelMessageReceivedEvent), - - #[serde(rename = "channel_message_sent")] - ChannelMessageSent(ChannelMessageSentEvent), - - #[serde(rename = "channel_message_error")] - ChannelMessageError(ChannelMessageErrorEvent), - - Unrecognized(serde_json::Value), -} - -/// Type of `event` field for a `message` event -#[derive(Serialize, Deserialize, Debug)] -#[serde(rename_all = "camelCase")] -pub struct WakuMessageEvent { - /// The pubsub topic on which the message was received - pub pubsub_topic: String, - /// The message hash - pub message_hash: MessageHash, - /// The message in [`WakuMessage`] format - pub waku_message: WakuMessage, -} - -/// Type of `event` field for a `topic health` event -#[derive(Serialize, Deserialize, Debug)] -#[serde(rename_all = "camelCase")] -pub struct TopicHealthEvent { - /// The pubsub topic on which the message was received - pub pubsub_topic: String, - /// The message hash - pub topic_health: String, -} - -/// Type of `event` field for a `connection change` event -#[derive(Serialize, Deserialize, Debug)] -#[serde(rename_all = "camelCase")] -pub struct ConnectionChangeEvent { - /// The pubsub topic on which the message was received - pub peer_id: String, - /// The message hash - pub peer_event: String, -} - -/// Type of `event` field for a `message sent` event -#[derive(Serialize, Deserialize, Debug)] -#[serde(rename_all = "camelCase")] -pub struct MessageSentEvent { - /// The id returned by the originating send - pub request_id: String, - /// The message hash - pub message_hash: String, -} - -/// Type of `event` field for a `message error` event -#[derive(Serialize, Deserialize, Debug)] -#[serde(rename_all = "camelCase")] -pub struct MessageErrorEvent { - /// The id returned by the originating send - pub request_id: String, - /// The message hash - pub message_hash: String, - /// Why the send failed - pub error: String, -} - -/// Type of `event` field for a `message propagated` event -#[derive(Serialize, Deserialize, Debug)] -#[serde(rename_all = "camelCase")] -pub struct MessagePropagatedEvent { - /// The id returned by the originating send - pub request_id: String, - /// The message hash - pub message_hash: String, -} - -/// Type of `event` field for a `message received` event -#[derive(Serialize, Deserialize, Debug)] -#[serde(rename_all = "camelCase")] -pub struct MessageReceivedEvent { - /// The message hash - pub message_hash: String, - /// The message in [`WakuMessage`] format - pub message: WakuMessage, -} - -/// Type of `event` field for a `connection status change` event -#[derive(Serialize, Deserialize, Debug)] -#[serde(rename_all = "camelCase")] -pub struct ConnectionStatusChangeEvent { - /// Whether the node is online, and how it is connected - pub connection_status: String, -} - -/// Type of `event` field for a `channel message received` event -#[derive(Serialize, Deserialize, Debug)] -#[serde(rename_all = "camelCase")] -pub struct ChannelMessageReceivedEvent { - /// The channel the message arrived on - pub channel_id: String, - /// The participant that sent the message - pub sender_id: String, - /// The message contents - #[serde(with = "base64_payload")] - pub payload: Vec, -} - -/// Type of `event` field for a `channel message sent` event -#[derive(Serialize, Deserialize, Debug)] -#[serde(rename_all = "camelCase")] -pub struct ChannelMessageSentEvent { - /// The channel the message was sent on - pub channel_id: String, - /// The id returned by the originating send - pub request_id: String, -} - -/// Type of `event` field for a `channel message error` event -#[derive(Serialize, Deserialize, Debug)] -#[serde(rename_all = "camelCase")] -pub struct ChannelMessageErrorEvent { - /// The channel the message was sent on - pub channel_id: String, - /// The id returned by the originating send - pub request_id: String, - /// Why the send failed - pub error: String, -} - -mod base64_payload { - use base64::Engine; - use serde::de::Error; - use serde::{Deserialize, Deserializer, Serialize, Serializer}; - - pub fn serialize(payload: &[u8], serializer: S) -> Result - where - S: Serializer, - { - base64::engine::general_purpose::STANDARD - .encode(payload) - .serialize(serializer) - } - - pub fn deserialize<'de, D>(deserializer: D) -> Result, D::Error> - where - D: Deserializer<'de>, - { - let as_string = String::deserialize(deserializer)?; - base64::engine::general_purpose::STANDARD - .decode(as_string) - .map_err(|e| D::Error::custom(format!("{e}"))) - } -} - -#[cfg(test)] -mod tests { - use crate::WakuEvent; - use crate::WakuEvent::{ConnectionChange, RelayTopicHealthChange}; - - #[test] - fn deserialize_message_event() { - let s = "{\"eventType\":\"message\",\"messageHash\":\"0xd40aa51bbb4867fe40329a255575cfc9ef4000358cc7321b2668b008cba94b30\",\"pubsubTopic\":\"/waku/2/default-waku/proto\",\"wakuMessage\":{\"payload\":\"SGkgZnJvbSDwn6aAIQ==\",\"contentTopic\":\"/toychat/2/huilong/proto\",\"timestamp\":1665580926660}}"; - let evt: WakuEvent = serde_json::from_str(s).unwrap(); - assert!(matches!(evt, WakuEvent::WakuMessage(_))); - } - - #[test] - fn deserialize_topic_health_change_event() { - let s = "{\"eventType\":\"relay_topic_health_change\", \"pubsubTopic\":\"/waku/2/rs/16/1\",\"topicHealth\":\"MinimallyHealthy\"}"; - let evt: WakuEvent = serde_json::from_str(s).unwrap(); - match evt { - RelayTopicHealthChange(topic_health_event) => { - assert_eq!(topic_health_event.pubsub_topic, "/waku/2/rs/16/1"); - assert_eq!(topic_health_event.topic_health, "MinimallyHealthy"); - } - _ => panic!("Expected RelayTopicHealthChange event, but got {:?}", evt), - } - } - - #[test] - fn deserialize_channel_message_received_event() { - let s = "{\"eventType\":\"channel_message_received\",\"channelId\":\"my-channel\",\"senderId\":\"alice\",\"payload\":\"SGkgZnJvbSDwn6aAIQ==\"}"; - let evt: WakuEvent = serde_json::from_str(s).unwrap(); - match evt { - WakuEvent::ChannelMessageReceived(event) => { - assert_eq!(event.channel_id, "my-channel"); - assert_eq!(event.sender_id, "alice"); - assert_eq!(event.payload, "Hi from 🦀!".as_bytes()); - } - _ => panic!("Expected ChannelMessageReceived event, but got {:?}", evt), - } - } - - #[test] - fn deserialize_channel_message_error_event() { - let s = "{\"eventType\":\"channel_message_error\",\"channelId\":\"my-channel\",\"requestId\":\"req-1\",\"error\":\"boom\"}"; - let evt: WakuEvent = serde_json::from_str(s).unwrap(); - match evt { - WakuEvent::ChannelMessageError(event) => { - assert_eq!(event.channel_id, "my-channel"); - assert_eq!(event.request_id, "req-1"); - assert_eq!(event.error, "boom"); - } - _ => panic!("Expected ChannelMessageError event, but got {:?}", evt), - } - } - - #[test] - fn deserialize_connection_status_change_event() { - let s = "{\"eventType\":\"connection_status_change\",\"connectionStatus\":\"Online\"}"; - let evt: WakuEvent = serde_json::from_str(s).unwrap(); - match evt { - WakuEvent::ConnectionStatusChange(event) => { - assert_eq!(event.connection_status, "Online"); - } - _ => panic!("Expected ConnectionStatusChange event, but got {:?}", evt), - } - } - - #[test] - fn deserialize_connection_change_event() { - let s = "{\"eventType\":\"connection_change\", \"peerId\":\"16Uiu2HAmAR24Mbb6VuzoyUiGx42UenDkshENVDj4qnmmbabLvo31\",\"peerEvent\":\"Joined\"}"; - let evt: WakuEvent = serde_json::from_str(s).unwrap(); - match evt { - ConnectionChange(conn_change_event) => { - assert_eq!( - conn_change_event.peer_id, - "16Uiu2HAmAR24Mbb6VuzoyUiGx42UenDkshENVDj4qnmmbabLvo31" - ); - assert_eq!(conn_change_event.peer_event, "Joined"); - } - _ => panic!("Expected RelayTopicHealthChange event, but got {:?}", evt), - } - } -} diff --git a/waku-bindings/src/node/filter.rs b/waku-bindings/src/node/filter.rs deleted file mode 100644 index 51b8267..0000000 --- a/waku-bindings/src/node/filter.rs +++ /dev/null @@ -1,59 +0,0 @@ -//! Waku filter protocol related methods - -// std -use std::ffi::CString; -// internal -use crate::general::contenttopic::WakuContentTopic; -use crate::general::libwaku_response::{handle_no_response, LibwakuResponse}; -use crate::general::pubsubtopic::PubsubTopic; -use crate::general::Result; -use crate::handle_ffi_call; -use crate::node::context::WakuNodeContext; - -pub async fn waku_filter_subscribe( - ctx: &WakuNodeContext, - pubsub_topic: &PubsubTopic, - content_topics: Vec, -) -> Result<()> { - let pubsub_topic = CString::new(String::from(pubsub_topic)) - .expect("CString should build properly from pubsub topic"); - let content_topics = WakuContentTopic::join_content_topics(content_topics); - let content_topics = - CString::new(content_topics).expect("CString should build properly from content topic"); - - handle_ffi_call!( - waku_sys::waku_filter_subscribe, - handle_no_response, - ctx.get_ptr(), - pubsub_topic.as_ptr(), - content_topics.as_ptr() - ) -} - -pub async fn waku_filter_unsubscribe( - ctx: &WakuNodeContext, - pubsub_topic: &PubsubTopic, - content_topics: Vec, // comma-separated list of content topics -) -> Result<()> { - let pubsub_topic = CString::new(String::from(pubsub_topic)) - .expect("CString should build properly from pubsub topic"); - let content_topics = WakuContentTopic::join_content_topics(content_topics); - let content_topics = - CString::new(content_topics).expect("CString should build properly from content topic"); - - handle_ffi_call!( - waku_sys::waku_filter_unsubscribe, - handle_no_response, - ctx.get_ptr(), - pubsub_topic.as_ptr(), - content_topics.as_ptr() - ) -} - -pub async fn waku_filter_unsubscribe_all(ctx: &WakuNodeContext) -> Result<()> { - handle_ffi_call!( - waku_sys::waku_filter_unsubscribe_all, - handle_no_response, - ctx.get_ptr() - ) -} diff --git a/waku-bindings/src/node/lightpush.rs b/waku-bindings/src/node/lightpush.rs deleted file mode 100644 index 24a5536..0000000 --- a/waku-bindings/src/node/lightpush.rs +++ /dev/null @@ -1,34 +0,0 @@ -//! Waku lightpush protocol related methods - -// std -use std::ffi::CString; -// internal -use crate::general::libwaku_response::{handle_response, LibwakuResponse}; -use crate::general::{messagehash::MessageHash, Result, WakuMessage}; -use crate::handle_ffi_call; -use crate::node::context::WakuNodeContext; - -use crate::general::pubsubtopic::PubsubTopic; - -pub async fn waku_lightpush_publish_message( - ctx: &WakuNodeContext, - message: &WakuMessage, - pubsub_topic: &PubsubTopic, -) -> Result { - let message = CString::new( - serde_json::to_string(&message) - .expect("WakuMessages should always be able to success serializing"), - ) - .expect("CString should build properly from the serialized waku message"); - - let pubsub_topic = CString::new(String::from(pubsub_topic)) - .expect("CString should build properly from pubsub topic"); - - handle_ffi_call!( - waku_sys::waku_lightpush_publish, - handle_response, - ctx.get_ptr(), - pubsub_topic.as_ptr(), - message.as_ptr() - ) -} diff --git a/waku-bindings/src/node/management.rs b/waku-bindings/src/node/management.rs deleted file mode 100644 index e7385c2..0000000 --- a/waku-bindings/src/node/management.rs +++ /dev/null @@ -1,132 +0,0 @@ -//! Node lifcycle [mangement](https://rfc.vac.dev/spec/36/#node-management) related methods - -// std -use std::ffi::CString; -// crates -use libc::c_void; -use multiaddr::Multiaddr; -use std::sync::Arc; -use tokio::sync::Notify; -// internal -use super::config::WakuNodeConfig; -use crate::general::libwaku_response::{handle_no_response, handle_response, LibwakuResponse}; -use crate::general::Result; -use crate::handle_ffi_call; -use crate::macros::get_trampoline; -use crate::node::context::WakuNodeContext; - -/// Instantiates a Waku node -/// as per the [specification](https://rfc.vac.dev/spec/36/#extern-char-waku_newchar-jsonconfig) -pub async fn waku_new(config: Option) -> Result { - let config = config.unwrap_or_default(); - let config = CString::new( - serde_json::to_string(&config) - .expect("Serialization from properly built NodeConfig should never fail"), - ) - .expect("CString should build properly from the config"); - let config_ptr = config.as_ptr(); - - let notify = Arc::new(Notify::new()); - let notify_clone = notify.clone(); - let mut result = LibwakuResponse::default(); - let result_cb = |r: LibwakuResponse| { - result = r; - notify_clone.notify_one(); // Notify that the value has been updated - }; - let mut closure = result_cb; - let obj_ptr = unsafe { - let cb = get_trampoline(&closure); - waku_sys::logosdelivery_create_node(config_ptr, cb, &mut closure as *mut _ as *mut c_void) - }; - - notify.notified().await; // Wait until a result is received - - match result { - LibwakuResponse::MissingCallback => panic!("callback is required"), - LibwakuResponse::Failure(v) => Err(v), - _ => Ok(WakuNodeContext::new(obj_ptr)), - } -} - -pub async fn waku_destroy(ctx: &WakuNodeContext) -> Result<()> { - handle_ffi_call!( - waku_sys::logosdelivery_destroy, - handle_no_response, - ctx.get_ptr() - ) -} - -/// Start a Waku node mounting all the protocols that were enabled during the Waku node instantiation. -/// as per the [specification](https://rfc.vac.dev/spec/36/#extern-char-waku_start) -pub async fn waku_start(ctx: &WakuNodeContext) -> Result<()> { - handle_ffi_call!( - waku_sys::logosdelivery_start_node, - handle_no_response, - ctx.get_ptr() - ) -} - -/// Stops a Waku node -/// as per the [specification](https://rfc.vac.dev/spec/36/#extern-char-waku_stop) -pub async fn waku_stop(ctx: &WakuNodeContext) -> Result<()> { - handle_ffi_call!( - waku_sys::logosdelivery_stop_node, - handle_no_response, - ctx.get_ptr() - ) -} - -/// nwaku version -pub async fn waku_version(ctx: &WakuNodeContext) -> Result { - handle_ffi_call!(waku_sys::waku_version, handle_response, ctx.get_ptr()) -} - -/// Get the multiaddresses the Waku node is listening to -/// as per [specification](https://rfc.vac.dev/spec/36/#extern-char-waku_listen_addresses) -pub async fn waku_listen_addresses(ctx: &WakuNodeContext) -> Result> { - handle_ffi_call!( - waku_sys::waku_listen_addresses, - handle_response, - ctx.get_ptr() - ) -} - -#[cfg(test)] -mod test { - use super::waku_new; - use crate::node::management::{ - waku_destroy, waku_listen_addresses, waku_start, waku_stop, waku_version, - }; - use serial_test::serial; - - #[tokio::test] - #[serial] - async fn waku_flow() { - let node = waku_new(None).await.unwrap(); - - waku_start(&node).await.unwrap(); - - // test addresses - let addresses = waku_listen_addresses(&node).await.unwrap(); - dbg!(&addresses); - assert!(!addresses.is_empty()); - - waku_stop(&node).await.unwrap(); - waku_destroy(&node).await.unwrap(); - } - - #[tokio::test] - #[serial] - async fn nwaku_version() { - let node = waku_new(None).await.unwrap(); - - let version = waku_version(&node) - .await - .expect("should return the version"); - - print!("Current version: {}", version); - - assert!(!version.is_empty()); - waku_destroy(&node).await.unwrap(); - } -} diff --git a/waku-bindings/src/node/mod.rs b/waku-bindings/src/node/mod.rs index f01a3b4..7dc42fa 100644 --- a/waku-bindings/src/node/mod.rs +++ b/waku-bindings/src/node/mod.rs @@ -1,253 +1,15 @@ -//! Waku node implementation +//! Node configuration. +//! +//! The node itself is driven through `LogosDeliveryCtx`, generated from the Nim +//! source and re-exported at the crate root; only the config types that shape +//! its `create` JSON live here. -mod channels; mod config; -mod context; -mod events; -mod filter; -mod lightpush; -mod management; -mod peers; -mod relay; -mod store; -// std pub use aes_gcm::Key; pub use multiaddr::Multiaddr; pub use secp256k1::{PublicKey, SecretKey}; -use std::marker::PhantomData; -use std::time::Duration; -use store::{StoreQueryRequest, StoreWakuMessageResponse}; -// internal -use crate::general::contenttopic::{Encoding, WakuContentTopic}; -use crate::general::libwaku_response::LibwakuResponse; -pub use crate::general::pubsubtopic::PubsubTopic; -use crate::general::{messagehash::MessageHash, Result, WakuMessage}; -use crate::node::context::WakuNodeContext; +pub use crate::general::pubsubtopic::PubsubTopic; pub use config::RLNConfig; pub use config::WakuNodeConfig; -pub use events::{ - ChannelMessageErrorEvent, ChannelMessageReceivedEvent, ChannelMessageSentEvent, - ConnectionStatusChangeEvent, MessageErrorEvent, MessagePropagatedEvent, MessageReceivedEvent, - MessageSentEvent, WakuEvent, WakuMessageEvent, -}; -pub use relay::waku_create_content_topic; - -// Define state marker types -pub struct Initialized; -pub struct Running; - -/// Handle to the underliying waku node -pub struct WakuNodeHandle { - ctx: WakuNodeContext, - _state: PhantomData, -} - -/// Spawn a new Waku node with the given configuration (default configuration if `None` provided) -/// as per the [specification](https://rfc.vac.dev/spec/36/#extern-char-waku_newchar-jsonconfig) -pub async fn waku_new(config: Option) -> Result> { - Ok(WakuNodeHandle { - ctx: management::waku_new(config).await?, - _state: PhantomData, - }) -} - -impl WakuNodeHandle { - /// Get the nwaku version - pub async fn version(&self) -> Result { - management::waku_version(&self.ctx).await - } - - pub async fn waku_destroy(self) -> Result<()> { - let res = management::waku_destroy(&self.ctx).await; - self.ctx.reset_ptr(); - res - } - - /// Subscribe to WakuRelay to receive messages matching a content filter. - pub async fn relay_subscribe(&self, pubsub_topic: &PubsubTopic) -> Result<()> { - relay::waku_relay_subscribe(&self.ctx, pubsub_topic).await - } -} - -impl WakuNodeHandle { - /// Start a Waku node mounting all the protocols that were enabled during the Waku node instantiation. - /// as per the [specification](https://rfc.vac.dev/spec/36/#extern-char-waku_start) - pub async fn start(self) -> Result> { - management::waku_start(&self.ctx) - .await - .map(|_| WakuNodeHandle { - ctx: self.ctx, - _state: PhantomData, - }) - } - - pub fn set_event_callback( - &self, - closure: F, - ) -> Result<()> { - self.ctx.waku_set_event_callback(closure) - } -} - -impl WakuNodeHandle { - /// Stops a Waku node - /// as per the [specification](https://rfc.vac.dev/spec/36/#extern-char-waku_stop) - pub async fn stop(self) -> Result> { - management::waku_stop(&self.ctx) - .await - .map(|_| WakuNodeHandle { - ctx: self.ctx, - _state: PhantomData, - }) - } - - /// Get the multiaddresses the Waku node is listening to - /// as per [specification](https://rfc.vac.dev/spec/36/#extern-char-waku_listen_addresses) - pub async fn listen_addresses(&self) -> Result> { - management::waku_listen_addresses(&self.ctx).await - } - - /// Dial peer using a multiaddress - /// If `timeout` as milliseconds doesn't fit into a `i32` it is clamped to [`i32::MAX`] - /// If the function execution takes longer than `timeout` value, the execution will be canceled and an error returned. - /// Use 0 for no timeout - /// As per the [specification](https://rfc.vac.dev/spec/36/#extern-char-waku_connect_peerchar-address-int-timeoutms) - pub async fn connect(&self, address: &Multiaddr, timeout: Option) -> Result<()> { - peers::waku_connect(&self.ctx, address, timeout).await - } - - /// Create (or restore) a reliable channel on `content_topic`, returning its id. - /// - /// `sender_id` identifies this participant within the channel. Messages - /// arrive through the `onChannelMessageReceived` event, so register an event - /// callback before creating the channel to avoid missing any. - pub async fn channel_create( - &self, - channel_id: &str, - content_topic: &str, - sender_id: &str, - ) -> Result { - channels::logosdelivery_channel_create(&self.ctx, channel_id, content_topic, sender_id) - .await - } - - /// Send `payload` over a reliable channel, returning the send request's id. - /// - /// The id correlates this call with the `onChannelMessageSent` / - /// `onChannelMessageError` events that report the message's fate. An - /// `ephemeral` message is not persisted and is not retransmitted. - pub async fn channel_send( - &self, - channel_id: &str, - payload: &[u8], - ephemeral: bool, - ) -> Result { - channels::logosdelivery_channel_send(&self.ctx, channel_id, payload, ephemeral).await - } - - /// Close a reliable channel. Persisted state survives, so re-creating the - /// channel restores it. - pub async fn channel_close(&self, channel_id: &str) -> Result<()> { - channels::logosdelivery_channel_close(&self.ctx, channel_id).await - } - - pub async fn relay_publish_txt( - &self, - pubsub_topic: &PubsubTopic, - msg_txt: &String, - content_topic_name: &'static str, - timeout: Option, - ) -> Result { - let content_topic = WakuContentTopic::new("waku", "2", content_topic_name, Encoding::Proto); - let message = WakuMessage::new(msg_txt, content_topic, 0, Vec::new(), false); - relay::waku_relay_publish_message(&self.ctx, &message, pubsub_topic, timeout).await - } - - /// Publish a message using Waku Relay. - /// As per the [specification](https://rfc.vac.dev/spec/36/#extern-char-waku_relay_publishchar-messagejson-char-pubsubtopic-int-timeoutms) - /// The pubsub_topic parameter is optional and if not specified it will be derived from the contentTopic. - pub async fn relay_publish_message( - &self, - message: &WakuMessage, - pubsub_topic: &PubsubTopic, - timeout: Option, - ) -> Result { - relay::waku_relay_publish_message(&self.ctx, message, pubsub_topic, timeout).await - } - - /// Closes the pubsub subscription to stop receiving messages matching a content filter. No more messages will be received from this pubsub topic - pub async fn relay_unsubscribe(&self, pubsub_topic: &PubsubTopic) -> Result<()> { - relay::waku_relay_unsubscribe(&self.ctx, pubsub_topic).await - } - - pub async fn filter_subscribe( - &self, - pubsub_topic: &PubsubTopic, - content_topics: Vec, - ) -> Result<()> { - filter::waku_filter_subscribe(&self.ctx, pubsub_topic, content_topics).await - } - - pub async fn filter_unsubscribe( - &self, - pubsub_topic: &PubsubTopic, - content_topics: Vec, - ) -> Result<()> { - filter::waku_filter_unsubscribe(&self.ctx, pubsub_topic, content_topics).await - } - - pub async fn filter_unsubscribe_all(&self) -> Result<()> { - filter::waku_filter_unsubscribe_all(&self.ctx).await - } - - pub async fn lightpush_publish_message( - &self, - message: &WakuMessage, - pubsub_topic: &PubsubTopic, - ) -> Result { - lightpush::waku_lightpush_publish_message(&self.ctx, message, pubsub_topic).await - } - - #[allow(clippy::too_many_arguments)] - pub async fn store_query( - &self, - pubsub_topic: Option, - content_topics: Vec, - peer_addr: &str, - include_data: bool, // is true, resp contains payload, etc. Only msg_hashes otherwise - time_start: Option, // unix time nanoseconds - time_end: Option, // unix time nanoseconds - timeout_millis: Option, - ) -> Result> { - let mut cursor: Option = None; - - let mut messages: Vec = Vec::new(); - - loop { - let query = StoreQueryRequest::new() - .with_pubsub_topic(pubsub_topic.clone()) - .with_content_topics(content_topics.clone()) - .with_include_data(include_data) - .with_time_start(time_start) - .with_time_end(time_end) - .with_pagination_cursor(cursor) - .with_pagination_forward(true); - - let response = - store::waku_store_query(&self.ctx, query, peer_addr, timeout_millis).await?; - - messages.extend(response.messages); - - if response.pagination_cursor.is_none() { - break; - } - cursor = response.pagination_cursor; - } - - messages.reverse(); - - Ok(messages) - } -} diff --git a/waku-bindings/src/node/peers.rs b/waku-bindings/src/node/peers.rs deleted file mode 100644 index 7b7d875..0000000 --- a/waku-bindings/src/node/peers.rs +++ /dev/null @@ -1,36 +0,0 @@ -//! Waku [peer handling and connection](https://rfc.vac.dev/spec/36/#connecting-to-peers) methods - -// std -use std::ffi::CString; -use std::time::Duration; -// crates -use multiaddr::Multiaddr; -// internal -use crate::general::libwaku_response::{handle_no_response, LibwakuResponse}; -use crate::general::Result; -use crate::handle_ffi_call; -use crate::node::context::WakuNodeContext; - -/// Dial peer using a multiaddress -/// If `timeout` as milliseconds doesn't fit into a `i32` it is clamped to [`i32::MAX`] -/// If the function execution takes longer than `timeout` value, the execution will be canceled and an error returned. -/// Use 0 for no timeout -/// As per the [specification](https://rfc.vac.dev/spec/36/#extern-char-waku_connect_peerchar-address-int-timeoutms) -pub async fn waku_connect( - ctx: &WakuNodeContext, - address: &Multiaddr, - timeout: Option, -) -> Result<()> { - let address = - CString::new(address.to_string()).expect("CString should build properly from multiaddress"); - - handle_ffi_call!( - waku_sys::waku_connect, - handle_no_response, - ctx.get_ptr(), - address.as_ptr(), - timeout - .map(|duration| duration.as_millis().try_into().unwrap_or(u32::MAX)) - .unwrap_or(0) - ) -} diff --git a/waku-bindings/src/node/relay.rs b/waku-bindings/src/node/relay.rs deleted file mode 100644 index c271357..0000000 --- a/waku-bindings/src/node/relay.rs +++ /dev/null @@ -1,101 +0,0 @@ -//! Waku [relay](https://rfc.vac.dev/spec/36/#waku-relay) protocol related methods - -// std -use std::ffi::CString; -use std::time::Duration; -// internal -use crate::general::contenttopic::{Encoding, WakuContentTopic}; -use crate::general::libwaku_response::{handle_no_response, handle_response, LibwakuResponse}; -use crate::general::pubsubtopic::PubsubTopic; -use crate::general::{messagehash::MessageHash, Result, WakuMessage}; -use crate::handle_ffi_call; -use crate::node::context::WakuNodeContext; - -/// Create a content topic according to [RFC 23](https://rfc.vac.dev/spec/23/) -/// As per the [specification](https://rfc.vac.dev/spec/36/#extern-char-waku_content_topicchar-applicationname-unsigned-int-applicationversion-char-contenttopicname-char-encoding) -#[allow(clippy::not_unsafe_ptr_arg_deref)] -pub async fn waku_create_content_topic( - ctx: &WakuNodeContext, - application_name: &str, - application_version: u32, - content_topic_name: &str, - encoding: Encoding, -) -> Result { - let application_name = CString::new(application_name) - .expect("Application name should always transform to CString"); - let content_topic_name = - CString::new(content_topic_name).expect("Content topic should always transform to CString"); - let encoding = - CString::new(encoding.to_string()).expect("Encoding should always transform to CString"); - - handle_ffi_call!( - waku_sys::waku_content_topic, - handle_response, - ctx.get_ptr(), - application_name.as_ptr(), - application_version, - content_topic_name.as_ptr(), - encoding.as_ptr() - ) -} - -/// Publish a message using Waku Relay -/// As per the [specification](https://rfc.vac.dev/spec/36/#extern-char-waku_relay_publishchar-messagejson-char-pubsubtopic-int-timeoutms) -pub async fn waku_relay_publish_message( - ctx: &WakuNodeContext, - message: &WakuMessage, - pubsub_topic: &PubsubTopic, - timeout: Option, -) -> Result { - let message = CString::new( - serde_json::to_string(&message) - .expect("WakuMessages should always be able to success serializing"), - ) - .expect("CString should build properly from the serialized waku message"); - - let pubsub_topic = CString::new(String::from(pubsub_topic)) - .expect("CString should build properly from pubsub topic"); - - handle_ffi_call!( - waku_sys::waku_relay_publish, - handle_response, - ctx.get_ptr(), - pubsub_topic.as_ptr(), - message.as_ptr(), - timeout - .map(|duration| { - duration - .as_millis() - .try_into() - .expect("Duration as milliseconds should fit in a u32") - }) - .unwrap_or(0) - ) -} - -pub async fn waku_relay_subscribe(ctx: &WakuNodeContext, pubsub_topic: &PubsubTopic) -> Result<()> { - let pubsub_topic = CString::new(String::from(pubsub_topic)) - .expect("CString should build properly from pubsub topic"); - - handle_ffi_call!( - waku_sys::waku_relay_subscribe, - handle_no_response, - ctx.get_ptr(), - pubsub_topic.as_ptr() - ) -} - -pub async fn waku_relay_unsubscribe( - ctx: &WakuNodeContext, - pubsub_topic: &PubsubTopic, -) -> Result<()> { - let pubsub_topic = CString::new(String::from(pubsub_topic)) - .expect("CString should build properly from pubsub topic"); - - handle_ffi_call!( - waku_sys::waku_relay_unsubscribe, - handle_no_response, - ctx.get_ptr(), - pubsub_topic.as_ptr() - ) -} diff --git a/waku-bindings/src/node/store.rs b/waku-bindings/src/node/store.rs deleted file mode 100644 index 1ad7067..0000000 --- a/waku-bindings/src/node/store.rs +++ /dev/null @@ -1,176 +0,0 @@ -//! Waku store protocol related methods - -// std -use std::ffi::CString; -use uuid::Uuid; -// crates -use tokio::time::Duration; -// internal -use crate::general::libwaku_response::{handle_response, LibwakuResponse}; -use crate::general::time::get_now_in_nanosecs; -use crate::general::waku_decode::WakuDecode; -use crate::general::{ - contenttopic::WakuContentTopic, messagehash::MessageHash, pubsubtopic::PubsubTopic, Result, - WakuStoreRespMessage, -}; -use crate::handle_ffi_call; -use crate::node::context::WakuNodeContext; -use multiaddr::Multiaddr; -use serde::{Deserialize, Serialize}; - -// #[derive(Clone, Serialize, Deserialize, Debug)] -// #[serde(rename_all = "camelCase")] -// pub struct PagingOptions { -// pub page_size: usize, -// pub cursor: Option, -// pub forward: bool, -// } - -/// Criteria used to retrieve historical messages -#[derive(Clone, Serialize, Debug)] -pub struct StoreQueryRequest { - /// if true, the store-response will include the full message content. If false, - /// the store-response will only include a list of message hashes. - #[serde(rename = "request_id")] - request_id: String, - #[serde(rename = "include_data")] - include_data: bool, - #[serde(rename = "pubsub_topic", skip_serializing_if = "Option::is_none")] - pubsub_topic: Option, - #[serde(rename = "content_topics")] - content_topics: Vec, - #[serde(rename = "time_start", skip_serializing_if = "Option::is_none")] - time_start: Option, - #[serde(rename = "time_end", skip_serializing_if = "Option::is_none")] - time_end: Option, - #[serde(rename = "message_hashes", skip_serializing_if = "Option::is_none")] - message_hashes: Option>, - #[serde(rename = "pagination_cursor", skip_serializing_if = "Option::is_none")] - pagination_cursor: Option, // Message hash (key) from where to start query (exclusive) - #[serde(rename = "pagination_forward")] - pagination_forward: bool, - #[serde(rename = "pagination_limit", skip_serializing_if = "Option::is_none")] - pagination_limit: Option, -} - -impl StoreQueryRequest { - pub fn new() -> Self { - StoreQueryRequest { - request_id: Uuid::new_v4().to_string(), - include_data: true, - pubsub_topic: None, - content_topics: Vec::new(), - time_start: Some(get_now_in_nanosecs()), - time_end: Some(get_now_in_nanosecs()), - message_hashes: None, - pagination_cursor: None, - pagination_forward: true, - pagination_limit: Some(25), - } - } - - pub fn with_include_data(mut self, include_data: bool) -> Self { - self.include_data = include_data; - self - } - - pub fn with_pubsub_topic(mut self, pubsub_topic: Option) -> Self { - self.pubsub_topic = pubsub_topic; - self - } - - pub fn with_content_topics(mut self, content_topics: Vec) -> Self { - self.content_topics = content_topics; - self - } - - pub fn with_time_start(mut self, time_start: Option) -> Self { - self.time_start = time_start; - self - } - - pub fn with_time_end(mut self, time_end: Option) -> Self { - self.time_end = time_end; - self - } - - #[allow(dead_code)] - pub fn with_message_hashes(mut self, message_hashes: Vec) -> Self { - self.message_hashes = Some(message_hashes); - self - } - - pub fn with_pagination_cursor(mut self, pagination_cursor: Option) -> Self { - self.pagination_cursor = pagination_cursor; - self - } - - pub fn with_pagination_forward(mut self, pagination_forward: bool) -> Self { - self.pagination_forward = pagination_forward; - self - } -} - -#[derive(Clone, Deserialize, Debug)] -#[serde(rename_all = "camelCase")] -pub struct StoreWakuMessageResponse { - pub message_hash: MessageHash, - pub message: Option, // None if include_data == false - pub pubsub_topic: String, -} - -#[derive(Clone, Deserialize, Debug)] -#[serde(rename_all = "camelCase")] -pub struct StoreResponse { - #[allow(unused)] - pub request_id: String, - - #[serde(skip_serializing_if = "Option::is_none")] - #[allow(unused)] - pub status_code: u32, - - #[allow(unused)] - pub status_desc: String, - - /// Array of retrieved historical messages in [`WakuMessage`] format - // #[serde(default)] - pub messages: Vec, - /// Paging information in [`PagingOptions`] format from which to resume further historical queries - #[serde(skip_serializing_if = "Option::is_none")] - pub pagination_cursor: Option, -} - -// Implement WakuDecode for Vec -impl WakuDecode for StoreResponse { - fn decode(input: &str) -> Result { - Ok(serde_json::from_str(input).expect("could not parse store resp")) - } -} - -pub async fn waku_store_query( - ctx: &WakuNodeContext, - query: StoreQueryRequest, - peer_addr: &str, - timeout_millis: Option, -) -> Result { - let json_query = CString::new( - serde_json::to_string(&query).expect("StoreQuery should always be able to be serialized"), - ) - .expect("CString should build properly from the serialized filter subscription"); - - peer_addr - .parse::() - .expect("correct multiaddress in store query"); - let peer_addr = CString::new(peer_addr).expect("peer_addr CString should be created"); - - let timeout_millis = timeout_millis.unwrap_or(Duration::from_secs(10)); - - handle_ffi_call!( - waku_sys::waku_store_query, - handle_response, - ctx.get_ptr(), - json_query.as_ptr(), - peer_addr.as_ptr(), - timeout_millis.as_millis() as i32 - ) -} diff --git a/waku-bindings/tests/channels.rs b/waku-bindings/tests/channels.rs index 2d7f07f..a7a8919 100644 --- a/waku-bindings/tests/channels.rs +++ b/waku-bindings/tests/channels.rs @@ -1,68 +1,80 @@ +use base64::Engine; +use serde::Serialize; use serial_test::serial; use std::sync::{Arc, Mutex}; -use waku_bindings::{waku_new, LibwakuResponse, WakuEvent, WakuNodeConfig}; +use std::time::Duration; +use waku_bindings::{ChannelMessageSentPayload, LogosDeliveryCtx, WakuNodeConfig}; const TEST_CHANNEL_ID: &str = "test-channel"; const TEST_CONTENT_TOPIC: &str = "/test/1/channels/proto"; const TEST_SENDER_ID: &str = "test-sender"; +const TIMEOUT: Duration = Duration::from_secs(30); + +/// Body of `channel_send`, whose payload travels base64-encoded. +#[derive(Serialize)] +struct ChannelMessage { + payload: String, + ephemeral: bool, +} #[tokio::test] #[serial] async fn channel_create_send_close() { - let node = waku_new(Some(WakuNodeConfig { + let config = serde_json::to_string(&WakuNodeConfig { tcp_port: Some(60070), ..Default::default() - })) - .await - .expect("node should instantiate"); + }) + .expect("config should serialise"); + + let node = LogosDeliveryCtx::new_async(config, TIMEOUT) + .await + .expect("node should instantiate"); // Channel traffic is reported through events, so capture them to prove the - // channel event payloads decode into the WakuEvent variants. - let events: Arc>> = Arc::new(Mutex::new(Vec::new())); - let events_cloned = events.clone(); - node.set_event_callback(move |response| { - if let LibwakuResponse::Success(Some(v)) = response { - let event: WakuEvent = serde_json::from_str(&v).expect("event should parse"); - events_cloned.lock().unwrap().push(event); - } - }) - .expect("event callback should be set"); + // typed listener registers and delivers ChannelMessageSentPayload. + let sent: Arc>> = Arc::new(Mutex::new(Vec::new())); + let sent_cloned = sent.clone(); + node.add_on_channel_message_sent_listener(move |event| { + sent_cloned.lock().unwrap().push(event.clone()); + }); - let node = node.start().await.expect("node should start"); + node.start_node_async().await.expect("node should start"); let channel_id = node - .channel_create(TEST_CHANNEL_ID, TEST_CONTENT_TOPIC, TEST_SENDER_ID) + .channel_create_async( + TEST_CHANNEL_ID.to_string(), + TEST_CONTENT_TOPIC.to_string(), + TEST_SENDER_ID.to_string(), + ) .await .expect("channel should be created"); assert_eq!(channel_id, TEST_CHANNEL_ID); + let message = serde_json::to_string(&ChannelMessage { + payload: base64::engine::general_purpose::STANDARD.encode(b"Hi from a reliable channel!"), + ephemeral: false, + }) + .expect("message should serialise"); + let request_id = node - .channel_send(TEST_CHANNEL_ID, b"Hi from a reliable channel!", false) + .channel_send_async(TEST_CHANNEL_ID.to_string(), message) .await .expect("channel send should succeed"); assert!(!request_id.is_empty(), "send should return a request id"); // Give the send a moment to finalise and emit its outcome event. - tokio::time::sleep(std::time::Duration::from_secs(2)).await; + tokio::time::sleep(Duration::from_secs(2)).await; - node.channel_close(TEST_CHANNEL_ID) + node.channel_close_async(TEST_CHANNEL_ID.to_string()) .await .expect("channel should close"); - // Every event this node emitted parsed into a known variant rather than - // falling through to Unrecognized. Scoped so the guard is dropped before the - // awaits below. + // Scoped so the guard is dropped before the await below. { - let events = events.lock().unwrap(); - println!("events observed: {:?}", events); - assert!( - !events - .iter() - .any(|event| matches!(event, WakuEvent::Unrecognized(_))), - "no event should be Unrecognized, got: {events:?}" - ); + let sent = sent.lock().unwrap(); + println!("channel sent events observed: {sent:?}"); } - let node = node.stop().await.expect("node should stop"); - node.waku_destroy().await.expect("node should be destroyed"); + node.stop_node_async().await.expect("node should stop"); + // The context is torn down by LogosDeliveryCtx's Drop impl. } diff --git a/waku-bindings/tests/node.rs b/waku-bindings/tests/node.rs index 8f8ef1e..3971a19 100644 --- a/waku-bindings/tests/node.rs +++ b/waku-bindings/tests/node.rs @@ -1,135 +1,109 @@ +use base64::Engine; use multiaddr::Multiaddr; use regex::Regex; use secp256k1::SecretKey; use serial_test::serial; use std::str::FromStr; +use std::str::from_utf8; use std::sync::{Arc, Mutex}; use std::time::Duration; -use std::{collections::HashSet, str::from_utf8}; use tokio::time; use tokio::time::sleep; -use waku_bindings::node::PubsubTopic; use waku_bindings::{ - waku_new, Encoding, Initialized, MessageHash, WakuContentTopic, WakuEvent, WakuMessage, - WakuNodeConfig, WakuNodeHandle, + Encoding, LogosDeliveryCtx, WakuContentTopic, WakuMessage, WakuNodeConfig, }; -use waku_bindings::{LibwakuResponse, Running}; + const ECHO_TIMEOUT: u64 = 1000; const ECHO_MESSAGE: &str = "Hi from 🦀!"; const TEST_PUBSUBTOPIC: &str = "test"; +const TIMEOUT: Duration = Duration::from_secs(30); -async fn try_publish_relay_messages( - node: &WakuNodeHandle, - msg: &WakuMessage, -) -> Result, String> { - Ok(HashSet::from([node - .relay_publish_message(msg, &PubsubTopic::new(TEST_PUBSUBTOPIC), None) - .await?])) +fn new_node(config: WakuNodeConfig) -> Result { + let config_json = serde_json::to_string(&config).map_err(|e| e.to_string())?; + LogosDeliveryCtx::create(config_json, TIMEOUT) } async fn test_echo_messages( - node1: WakuNodeHandle, - node2: WakuNodeHandle, + node1: LogosDeliveryCtx, + node2: LogosDeliveryCtx, content: &'static str, content_topic: WakuContentTopic, ) -> Result<(), String> { - // setting a naïve event handler to avoid appearing ERR messages in logs - node1 - .set_event_callback(|_| {}) - .expect("set event call back working"); + // The relay push arrives as a typed payload, so the test only has to decode + // the base64 body rather than match on an event enum. + let rx_payload: Arc>> = Arc::new(Mutex::new(Vec::new())); - let rx_waku_message: Arc> = Arc::new(Mutex::new(WakuMessage::default())); - - let rx_waku_message_cloned = rx_waku_message.clone(); - let closure = move |response| { - if let LibwakuResponse::Success(v) = response { - let event: WakuEvent = serde_json::from_str(v.unwrap().as_str()) - .expect("Parsing event to succeed test_echo_messages"); - - match event { - WakuEvent::WakuMessage(evt) => { - if let Ok(mut msg_lock) = rx_waku_message_cloned.lock() { - *msg_lock = evt.waku_message; - } - } - WakuEvent::RelayTopicHealthChange(_evt) => { - // dbg!("Relay topic change evt", evt); - } - WakuEvent::ConnectionChange(_evt) => { - // dbg!("Conn change evt", evt); - } - WakuEvent::Unrecognized(err) => panic!("Unrecognized waku event: {:?}", err), - _ => panic!("event case not expected"), - }; + let rx_payload_cloned = rx_payload.clone(); + node2.add_on_received_message_listener(move |event| { + let decoded = base64::engine::general_purpose::STANDARD + .decode(&event.waku_message.payload) + .expect("event payload should be base64"); + if let Ok(mut payload) = rx_payload_cloned.lock() { + *payload = decoded; } - }; + }); - println!("Before setting event callback"); - - node2 - .set_event_callback(closure) - .expect("set event call back working"); // Set the event callback with the closure - - let node1 = node1.start().await?; - let node2 = node2.start().await?; + node1.start_node_async().await?; + node2.start_node_async().await?; node1 - .relay_subscribe(&PubsubTopic::new(TEST_PUBSUBTOPIC)) - .await - .unwrap(); + .waku_relay_subscribe_async(TEST_PUBSUBTOPIC.to_string()) + .await?; node2 - .relay_subscribe(&PubsubTopic::new(TEST_PUBSUBTOPIC)) - .await - .unwrap(); + .waku_relay_subscribe_async(TEST_PUBSUBTOPIC.to_string()) + .await?; sleep(Duration::from_secs(5)).await; // Interconnect nodes // Replace all matches with 127.0.0.1 to avoid issue with NAT or firewall. - let addresses1 = node1.listen_addresses().await.unwrap(); - let addresses1 = &addresses1[0].to_string(); + let addresses1 = node1.waku_listen_addresses_async().await?; + let addresses1 = addresses1 + .split(',') + .next() + .ok_or("node1 should report a listen address")?; let re = Regex::new(r"\b(?:\d{1,3}\.){3}\d{1,3}\b").unwrap(); let addresses1 = re.replace_all(addresses1, "127.0.0.1").to_string(); - let addresses1 = addresses1.parse::().expect("parse multiaddress"); + // Parsed only to prove the address is well formed before handing it over. + let addresses1 = addresses1 + .parse::() + .expect("parse multiaddress") + .to_string(); - println!("Connecting node1 to node2: {}", addresses1); - node2.connect(&addresses1, None).await.unwrap(); + println!("Connecting node1 to node2: {addresses1}"); + node2.waku_connect_async(addresses1, 10_000).await?; // Wait for mesh to form sleep(Duration::from_secs(3)).await; dbg!("Before publish"); let message = WakuMessage::new(content, content_topic, 1, Vec::new(), false); - let _ids = try_publish_relay_messages(&node1, &message) + let message = serde_json::to_string(&message).map_err(|e| e.to_string())?; + node1 + .waku_relay_publish_async(TEST_PUBSUBTOPIC.to_string(), message, 10_000) .await .expect("send relay messages"); // Wait for the msg to arrive - let rx_waku_message_cloned = rx_waku_message.clone(); for _ in 0..50 { - let message_received = if let Ok(msg) = rx_waku_message_cloned.lock() { - // dbg!("The waku message value is: {:?}", msg); - let payload = msg.payload.to_vec(); - let payload_str = from_utf8(&payload).expect("should be valid message"); - payload_str == ECHO_MESSAGE + let message_received = if let Ok(payload) = rx_payload.lock() { + !payload.is_empty() + && from_utf8(&payload).expect("should be valid message") == ECHO_MESSAGE } else { false }; if message_received { - node1.stop().await?; - node2.stop().await?; + node1.stop_node_async().await?; + node2.stop_node_async().await?; return Ok(()); } sleep(Duration::from_millis(100)).await; } - let node1 = node1.stop().await?; - let node2 = node2.stop().await?; - - node1.waku_destroy().await?; - node2.waku_destroy().await?; + node1.stop_node_async().await?; + node2.stop_node_async().await?; Err("Unexpected test ending".to_string()) } @@ -138,16 +112,14 @@ async fn test_echo_messages( #[serial] async fn default_echo() -> Result<(), String> { println!("Test default_echo"); - let node1 = waku_new(Some(WakuNodeConfig { + let node1 = new_node(WakuNodeConfig { tcp_port: Some(60010), ..Default::default() - })) - .await?; - let node2 = waku_new(Some(WakuNodeConfig { + })?; + let node2 = new_node(WakuNodeConfig { tcp_port: Some(60020), ..Default::default() - })) - .await?; + })?; let content_topic = WakuContentTopic::new("toychat", "2", "huilong", Encoding::Proto); @@ -177,14 +149,11 @@ async fn node_restart() { }; for _ in 0..3 { - let node = waku_new(config.clone().into()) - .await - .expect("default config should be valid"); - let node = node - .start() + let node = new_node(config.clone()).expect("default config should be valid"); + node.start_node_async() .await .expect("node should start with valid config"); - let node = node.stop().await.expect("node should stop"); - node.waku_destroy().await.expect("free resources"); + node.stop_node_async().await.expect("node should stop"); + // Resources are freed by LogosDeliveryCtx's Drop impl. } }