mirror of
https://github.com/logos-messaging/logos-delivery-rust-bindings.git
synced 2026-07-29 22:43:34 +00:00
feat!: expose the generated bindings, drop the hand-written FFI tier
The generated crate already provides what this tier existed to build: CBOR marshalling, the callback dance, and a typed method per proc. So macros.rs (trampoline, handle_ffi_call) and libwaku_response.rs (RET_OK/RET_ERR, LibwakuResponse) are deleted rather than ported, and node/events.rs goes with them -- its WakuEvent enum is now one generated payload struct per event, which is what the typed listeners hand back. What survives is what the generated crate does not give us: the domain layer (WakuMessage, WakuContentTopic, MessageHash, pubsubtopic) and WakuNodeConfig, which still serialises to the JSON create_node takes -- that JSON is unchanged, only its transport moved to CBOR. BREAKING CHANGE: WakuNodeHandle and its Initialized/Running typestate, waku_new, and waku_destroy are gone; callers use LogosDeliveryCtx::create and let Drop tear the node down. set_event_callback is gone too: events are per-name now, so a whole-stream callback has no equivalent -- use the typed add_on_*_listener methods, which deliver a payload struct instead of a JSON string to match on. Tests and the example move with the API. The event handling gets shorter rather than longer: no from_str, no matching five variants, no panicking on Unrecognized -- a listener only receives the event it registered for. Callers now base64-decode payloads themselves, since that was WakuMessage's serde doing it before. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Ivan FB
parent
98d005c9e1
commit
430ebc0bda
182
Cargo.lock
generated
182
Cargo.lock
generated
@ -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"
|
||||
|
||||
@ -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"
|
||||
|
||||
@ -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(())
|
||||
}
|
||||
|
||||
@ -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<String>),
|
||||
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<Self, Self::Error> {
|
||||
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<F: WakuDecode>(code: i32, result: LibwakuResponse) -> Result<F> {
|
||||
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
|
||||
),
|
||||
}
|
||||
}
|
||||
@ -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;
|
||||
|
||||
@ -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,
|
||||
};
|
||||
|
||||
@ -1,73 +0,0 @@
|
||||
use crate::general::libwaku_response::LibwakuResponse;
|
||||
|
||||
use std::{slice, str};
|
||||
use waku_sys::FFICallBack;
|
||||
|
||||
unsafe extern "C" fn trampoline<F>(
|
||||
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<F>(_closure: &F) -> FFICallBack
|
||||
where
|
||||
F: FnMut(LibwakuResponse),
|
||||
{
|
||||
Some(trampoline::<F>)
|
||||
}
|
||||
|
||||
#[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)
|
||||
}};
|
||||
}
|
||||
@ -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<String> {
|
||||
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<String> {
|
||||
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()
|
||||
)
|
||||
}
|
||||
@ -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<Mutex<Box<LibwakuResponseClosure>>>,
|
||||
}
|
||||
|
||||
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<F: FnMut(LibwakuResponse) + 'static + Sync + Send>(
|
||||
&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())
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -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<u8>,
|
||||
}
|
||||
|
||||
/// 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<S>(payload: &[u8], serializer: S) -> Result<S::Ok, S::Error>
|
||||
where
|
||||
S: Serializer,
|
||||
{
|
||||
base64::engine::general_purpose::STANDARD
|
||||
.encode(payload)
|
||||
.serialize(serializer)
|
||||
}
|
||||
|
||||
pub fn deserialize<'de, D>(deserializer: D) -> Result<Vec<u8>, 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),
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -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<WakuContentTopic>,
|
||||
) -> 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<WakuContentTopic>, // 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()
|
||||
)
|
||||
}
|
||||
@ -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<MessageHash> {
|
||||
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()
|
||||
)
|
||||
}
|
||||
@ -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<WakuNodeConfig>) -> Result<WakuNodeContext> {
|
||||
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<String> {
|
||||
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<Vec<Multiaddr>> {
|
||||
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();
|
||||
}
|
||||
}
|
||||
@ -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<State> {
|
||||
ctx: WakuNodeContext,
|
||||
_state: PhantomData<State>,
|
||||
}
|
||||
|
||||
/// 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<WakuNodeConfig>) -> Result<WakuNodeHandle<Initialized>> {
|
||||
Ok(WakuNodeHandle {
|
||||
ctx: management::waku_new(config).await?,
|
||||
_state: PhantomData,
|
||||
})
|
||||
}
|
||||
|
||||
impl<State> WakuNodeHandle<State> {
|
||||
/// Get the nwaku version
|
||||
pub async fn version(&self) -> Result<String> {
|
||||
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<Initialized> {
|
||||
/// 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<WakuNodeHandle<Running>> {
|
||||
management::waku_start(&self.ctx)
|
||||
.await
|
||||
.map(|_| WakuNodeHandle {
|
||||
ctx: self.ctx,
|
||||
_state: PhantomData,
|
||||
})
|
||||
}
|
||||
|
||||
pub fn set_event_callback<F: FnMut(LibwakuResponse) + 'static + Sync + Send>(
|
||||
&self,
|
||||
closure: F,
|
||||
) -> Result<()> {
|
||||
self.ctx.waku_set_event_callback(closure)
|
||||
}
|
||||
}
|
||||
|
||||
impl WakuNodeHandle<Running> {
|
||||
/// Stops a Waku node
|
||||
/// as per the [specification](https://rfc.vac.dev/spec/36/#extern-char-waku_stop)
|
||||
pub async fn stop(self) -> Result<WakuNodeHandle<Initialized>> {
|
||||
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<Vec<Multiaddr>> {
|
||||
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<Duration>) -> 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<String> {
|
||||
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<String> {
|
||||
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<Duration>,
|
||||
) -> Result<MessageHash> {
|
||||
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<Duration>,
|
||||
) -> Result<MessageHash> {
|
||||
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<WakuContentTopic>,
|
||||
) -> Result<()> {
|
||||
filter::waku_filter_subscribe(&self.ctx, pubsub_topic, content_topics).await
|
||||
}
|
||||
|
||||
pub async fn filter_unsubscribe(
|
||||
&self,
|
||||
pubsub_topic: &PubsubTopic,
|
||||
content_topics: Vec<WakuContentTopic>,
|
||||
) -> 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<MessageHash> {
|
||||
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<PubsubTopic>,
|
||||
content_topics: Vec<WakuContentTopic>,
|
||||
peer_addr: &str,
|
||||
include_data: bool, // is true, resp contains payload, etc. Only msg_hashes otherwise
|
||||
time_start: Option<u64>, // unix time nanoseconds
|
||||
time_end: Option<u64>, // unix time nanoseconds
|
||||
timeout_millis: Option<Duration>,
|
||||
) -> Result<Vec<StoreWakuMessageResponse>> {
|
||||
let mut cursor: Option<MessageHash> = None;
|
||||
|
||||
let mut messages: Vec<StoreWakuMessageResponse> = 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)
|
||||
}
|
||||
}
|
||||
|
||||
@ -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<Duration>,
|
||||
) -> 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)
|
||||
)
|
||||
}
|
||||
@ -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<WakuContentTopic> {
|
||||
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<Duration>,
|
||||
) -> Result<MessageHash> {
|
||||
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()
|
||||
)
|
||||
}
|
||||
@ -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<MessageHash>,
|
||||
// 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<PubsubTopic>,
|
||||
#[serde(rename = "content_topics")]
|
||||
content_topics: Vec<WakuContentTopic>,
|
||||
#[serde(rename = "time_start", skip_serializing_if = "Option::is_none")]
|
||||
time_start: Option<u64>,
|
||||
#[serde(rename = "time_end", skip_serializing_if = "Option::is_none")]
|
||||
time_end: Option<u64>,
|
||||
#[serde(rename = "message_hashes", skip_serializing_if = "Option::is_none")]
|
||||
message_hashes: Option<Vec<MessageHash>>,
|
||||
#[serde(rename = "pagination_cursor", skip_serializing_if = "Option::is_none")]
|
||||
pagination_cursor: Option<MessageHash>, // 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<u64>,
|
||||
}
|
||||
|
||||
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<PubsubTopic>) -> Self {
|
||||
self.pubsub_topic = pubsub_topic;
|
||||
self
|
||||
}
|
||||
|
||||
pub fn with_content_topics(mut self, content_topics: Vec<WakuContentTopic>) -> Self {
|
||||
self.content_topics = content_topics;
|
||||
self
|
||||
}
|
||||
|
||||
pub fn with_time_start(mut self, time_start: Option<u64>) -> Self {
|
||||
self.time_start = time_start;
|
||||
self
|
||||
}
|
||||
|
||||
pub fn with_time_end(mut self, time_end: Option<u64>) -> Self {
|
||||
self.time_end = time_end;
|
||||
self
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub fn with_message_hashes(mut self, message_hashes: Vec<MessageHash>) -> Self {
|
||||
self.message_hashes = Some(message_hashes);
|
||||
self
|
||||
}
|
||||
|
||||
pub fn with_pagination_cursor(mut self, pagination_cursor: Option<MessageHash>) -> 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<WakuStoreRespMessage>, // 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<StoreWakuMessageResponse>,
|
||||
/// Paging information in [`PagingOptions`] format from which to resume further historical queries
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub pagination_cursor: Option<MessageHash>,
|
||||
}
|
||||
|
||||
// Implement WakuDecode for Vec<Multiaddr>
|
||||
impl WakuDecode for StoreResponse {
|
||||
fn decode(input: &str) -> Result<Self> {
|
||||
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<Duration>,
|
||||
) -> Result<StoreResponse> {
|
||||
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::<Multiaddr>()
|
||||
.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
|
||||
)
|
||||
}
|
||||
@ -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<Mutex<Vec<WakuEvent>>> = 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<Mutex<Vec<ChannelMessageSentPayload>>> = 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.
|
||||
}
|
||||
|
||||
@ -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<Running>,
|
||||
msg: &WakuMessage,
|
||||
) -> Result<HashSet<MessageHash>, String> {
|
||||
Ok(HashSet::from([node
|
||||
.relay_publish_message(msg, &PubsubTopic::new(TEST_PUBSUBTOPIC), None)
|
||||
.await?]))
|
||||
fn new_node(config: WakuNodeConfig) -> Result<LogosDeliveryCtx, String> {
|
||||
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<Initialized>,
|
||||
node2: WakuNodeHandle<Initialized>,
|
||||
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<Mutex<Vec<u8>>> = Arc::new(Mutex::new(Vec::new()));
|
||||
|
||||
let rx_waku_message: Arc<Mutex<WakuMessage>> = 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::<Multiaddr>().expect("parse multiaddress");
|
||||
// Parsed only to prove the address is well formed before handing it over.
|
||||
let addresses1 = addresses1
|
||||
.parse::<Multiaddr>()
|
||||
.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.
|
||||
}
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user