diff --git a/Cargo.lock b/Cargo.lock index c775baed..f60da078 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -76,12 +76,6 @@ version = "0.2.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" -[[package]] -name = "android-tzdata" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e999941b234f3131b00bc13c22d06e8c5ff726d1b6318ac7eb276997bbb4fef0" - [[package]] name = "android_system_properties" version = "0.1.5" @@ -109,6 +103,15 @@ dependencies = [ "derive_arbitrary", ] +[[package]] +name = "archery" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eae2ed21cd55021f05707a807a5fc85695dafb98832921f6cfa06db67ca5b869" +dependencies = [ + "triomphe", +] + [[package]] name = "arrayref" version = "0.3.9" @@ -676,17 +679,16 @@ dependencies = [ [[package]] name = "chrono" -version = "0.4.41" +version = "0.4.42" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c469d952047f47f91b68d1cba3f10d63c11d73e4636f24f08daf0278abf01c4d" +checksum = "145052bdd345b87320e369255277e3fb5152762ad123a901ef5c262dd38fe8d2" dependencies = [ - "android-tzdata", "iana-time-zone", "js-sys", "num-traits", "serde", "wasm-bindgen", - "windows-link", + "windows-link 0.2.0", ] [[package]] @@ -1368,6 +1370,12 @@ version = "0.2.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d" +[[package]] +name = "fixedbitset" +version = "0.5.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99" + [[package]] name = "flate2" version = "1.1.2" @@ -1467,6 +1475,21 @@ dependencies = [ "futures-sink", ] +[[package]] +name = "futures-concurrency" +version = "7.6.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0eb68017df91f2e477ed4bea586c59eaecaa47ed885a770d0444e21e62572cd2" +dependencies = [ + "fixedbitset", + "futures-buffered", + "futures-core", + "futures-lite", + "pin-project", + "slab", + "smallvec", +] + [[package]] name = "futures-core" version = "0.3.31" @@ -2225,9 +2248,9 @@ dependencies = [ [[package]] name = "iroh" -version = "0.91.1" +version = "0.92.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c6a98c47bb5f720edeb77be502a8acd238a3c0755f0b1ad865a716224d794a59" +checksum = "135ad6b793a5851b9e5435ad36fea63df485f8fd4520a58117e7dc3326a69c15" dependencies = [ "aead", "backon", @@ -2255,7 +2278,7 @@ dependencies = [ "n0-snafu", "n0-watcher", "nested_enum_utils", - "netdev", + "netdev 0.36.0", "netwatch", "pin-project", "pkarr", @@ -2287,9 +2310,9 @@ dependencies = [ [[package]] name = "iroh-base" -version = "0.91.1" +version = "0.92.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "78bde4e612191173e8ade55e3aa719044514edfff952292ffbf581be35cbb59c" +checksum = "04ae51a14c9255a735b1db2d8cf29b875b971e96a5b23e4d0d1ee7d85bf32132" dependencies = [ "curve25519-dalek", "data-encoding", @@ -2297,12 +2320,46 @@ dependencies = [ "ed25519-dalek", "n0-snafu", "nested_enum_utils", + "postcard", "rand_core 0.6.4", "serde", "snafu", "url", ] +[[package]] +name = "iroh-gossip" +version = "0.92.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d530166ebdcfe321fb3a406dba62d777c98e1159237c7aaf0591e3758eb013c" +dependencies = [ + "blake3", + "bytes", + "data-encoding", + "derive_more 1.0.0", + "ed25519-dalek", + "futures-concurrency", + "futures-lite", + "futures-util", + "hex", + "indexmap 2.10.0", + "iroh", + "iroh-base", + "iroh-metrics", + "irpc", + "n0-future 0.1.3", + "n0-snafu", + "nested_enum_utils", + "postcard", + "rand 0.8.5", + "rand_core 0.6.4", + "serde", + "snafu", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "iroh-metrics" version = "0.35.0" @@ -2385,9 +2442,9 @@ dependencies = [ [[package]] name = "iroh-relay" -version = "0.91.1" +version = "0.92.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9f19e43de2cfc04748054e1cbf7854ae1e192ddc45a4adeae7be7076a8f1fd43" +checksum = "315cb02e660de0de339303296df9a29b27550180bb3979d0753a267649b34a7f" dependencies = [ "blake3", "bytes", @@ -2433,6 +2490,32 @@ dependencies = [ "z32", ] +[[package]] +name = "iroh-smol-kv" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2bf263d8aba7c7ec5baddcec76d9a6881b3f5470d293707b3881545f9548f31c" +dependencies = [ + "bytes", + "chrono", + "ed25519-dalek", + "hex", + "iroh", + "iroh-gossip", + "irpc", + "n0-future 0.2.0", + "peg", + "postcard", + "rand 0.8.5", + "rpds", + "serde", + "serde-big-array", + "snafu", + "sync_wrapper", + "tokio", + "tracing", +] + [[package]] name = "iroh-streamplace" version = "0.1.0" @@ -2441,14 +2524,21 @@ dependencies = [ "async-trait", "bytes", "c2pa", + "hex", "iroh", + "iroh-base", + "iroh-gossip", + "iroh-smol-kv", "irpc", "irpc-iroh", "n0-future 0.2.0", "postcard", + "rand 0.8.5", + "ref-cast", "serde", "serde_json", "snafu", + "testresult", "thiserror 2.0.16", "tokio", "tracing", @@ -2459,9 +2549,9 @@ dependencies = [ [[package]] name = "irpc" -version = "0.7.0" +version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a9f8f1d0987ea9da3d74698f921d0a817a214c83b2635a33ed4bc3efa4de1acd" +checksum = "092c0b20697bbc7de4839eebcb49be975cc09221021626d301eea55fc10bfeb7" dependencies = [ "anyhow", "futures-buffered", @@ -2482,9 +2572,9 @@ dependencies = [ [[package]] name = "irpc-derive" -version = "0.5.0" +version = "0.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3e0b26b834d401a046dd9d47bc236517c746eddbb5d25ff3e1a6075bfa4eebdb" +checksum = "209d38d83c0f7043916e90de2d3a8d01035db3a2f49ea7d5fb41b8f43e889924" dependencies = [ "proc-macro2", "quote", @@ -2493,13 +2583,14 @@ dependencies = [ [[package]] name = "irpc-iroh" -version = "0.7.0" +version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f5926531af491c6962db4d79f43ea219404cb800889922a728b1d3b92f887eda" +checksum = "dba5a31134415cdcfe65c309b4d3a9560764ca938e98e8717a25d3973f141ac1" dependencies = [ "anyhow", "getrandom 0.3.3", "iroh", + "iroh-base", "irpc", "n0-future 0.1.3", "postcard", @@ -2785,9 +2876,9 @@ dependencies = [ [[package]] name = "n0-snafu" -version = "0.2.1" +version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4fed465ff57041f29db78a9adc8864296ef93c6c16029f9e192dc303404ebd0" +checksum = "1815107e577a95bfccedb4cfabc73d709c0db6d12de3f14e0f284a8c5036dc4f" dependencies = [ "anyhow", "btparse", @@ -2836,6 +2927,23 @@ dependencies = [ "windows-sys 0.59.0", ] +[[package]] +name = "netdev" +version = "0.37.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "daa1e3eaf125c54c21e6221df12dd2a0a682784a068782dd564c836c0f281b6d" +dependencies = [ + "dlopen2", + "ipnet", + "libc", + "netlink-packet-core", + "netlink-packet-route 0.22.0", + "netlink-sys", + "once_cell", + "system-configuration", + "windows-sys 0.59.0", +] + [[package]] name = "netlink-packet-core" version = "0.7.0" @@ -2918,9 +3026,9 @@ dependencies = [ [[package]] name = "netwatch" -version = "0.8.0" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8901dbb408894af3df3fc51420ba0c6faf3a7d896077b797c39b7001e2f787bd" +checksum = "8a63d76f52f3f15ebde3ca751a2ab73a33ae156662bc04383bac8e824f84e9bb" dependencies = [ "atomic-waker", "bytes", @@ -2932,7 +3040,7 @@ dependencies = [ "n0-future 0.1.3", "n0-watcher", "nested_enum_utils", - "netdev", + "netdev 0.37.3", "netlink-packet-core", "netlink-packet-route 0.24.0", "netlink-proto", @@ -3267,6 +3375,33 @@ dependencies = [ "utf8-decode", ] +[[package]] +name = "peg" +version = "0.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9928cfca101b36ec5163e70049ee5368a8a1c3c6efc9ca9c5f9cc2f816152477" +dependencies = [ + "peg-macros", + "peg-runtime", +] + +[[package]] +name = "peg-macros" +version = "0.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6298ab04c202fa5b5d52ba03269fb7b74550b150323038878fe6c372d8280f71" +dependencies = [ + "peg-runtime", + "proc-macro2", + "quote", +] + +[[package]] +name = "peg-runtime" +version = "0.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "132dca9b868d927b35b5dd728167b2dee150eb1ad686008fc71ccb298b776fca" + [[package]] name = "pem" version = "3.0.5" @@ -3522,9 +3657,9 @@ checksum = "f84267b20a16ea918e43c6a88433c2d54fa145c92a811b5b047ccbe153674483" [[package]] name = "portmapper" -version = "0.8.0" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "62f1975debe62a70557e42b9ff9466e4890cf9d3d156d296408a711f1c5f642b" +checksum = "a9f99e8cd25cd8ee09fc7da59357fd433c0a19272956ebb4ad7443b21842988d" dependencies = [ "base64 0.22.1", "bytes", @@ -3971,18 +4106,18 @@ dependencies = [ [[package]] name = "ref-cast" -version = "1.0.24" +version = "1.0.25" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4a0ae411dbe946a674d89546582cea4ba2bb8defac896622d6496f14c23ba5cf" +checksum = "f354300ae66f76f1c85c5f84693f0ce81d747e2c3f21a45fef496d89c960bf7d" dependencies = [ "ref-cast-impl", ] [[package]] name = "ref-cast-impl" -version = "1.0.24" +version = "1.0.25" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1165225c21bff1f3bbce98f5a1f889949bc902d3575308cc7b0de30b4f6d27c7" +checksum = "b7186006dcb21920990093f30e3dea63b7d6e977bf1256be20c3563a5db070da" dependencies = [ "proc-macro2", "quote", @@ -4118,6 +4253,16 @@ dependencies = [ "serde_derive", ] +[[package]] +name = "rpds" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7f89f654d51fffdd6026289d07d1fd523244d46ae0a8bc22caa6dd7f9e8cb0b" +dependencies = [ + "archery", + "serde", +] + [[package]] name = "rsa" version = "0.9.8" @@ -4369,6 +4514,15 @@ dependencies = [ "serde_derive", ] +[[package]] +name = "serde-big-array" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11fc7cc2c76d73e0f27ee52abbd64eec84d46f370c88371120433196934e4b7f" +dependencies = [ + "serde", +] + [[package]] name = "serde-transcode" version = "1.1.1" @@ -4631,9 +4785,9 @@ checksum = "fad6c857cbab2627dcf01ec85a623ca4e7dcb5691cbaa3d7fb7653671f0d09c9" [[package]] name = "snafu" -version = "0.8.6" +version = "0.8.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "320b01e011bf8d5d7a4a4a4be966d9160968935849c83b918827f6a435e7f627" +checksum = "6e84b3f4eacbf3a1ce05eac6763b4d629d60cbc94d632e4092c54ade71f1e1a2" dependencies = [ "backtrace", "snafu-derive", @@ -4641,9 +4795,9 @@ dependencies = [ [[package]] name = "snafu-derive" -version = "0.8.6" +version = "0.8.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1961e2ef424c1424204d3a5d6975f934f56b6d50ff5732382d84ebf460e147f7" +checksum = "c1c97747dbf44bb1ca44a561ece23508e99cb592e862f22222dcf42f51d1e451" dependencies = [ "heck", "proc-macro2", @@ -4925,6 +5079,12 @@ dependencies = [ "winapi-util", ] +[[package]] +name = "testresult" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "614b328ff036a4ef882c61570f72918f7e9c5bee1da33f8e7f91e01daee7e56c" + [[package]] name = "textwrap" version = "0.16.2" @@ -5321,6 +5481,12 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a7f741b240f1a48843f9b8e0444fb55fb2a4ff67293b50a9179dfd5ea67f8d41" +[[package]] +name = "triomphe" +version = "0.1.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef8f7726da4807b58ea5c96fdc122f80702030edc33b35aff9190a51148ccc85" + [[package]] name = "try-lock" version = "0.2.5" @@ -5813,7 +5979,7 @@ dependencies = [ "windows-collections", "windows-core", "windows-future", - "windows-link", + "windows-link 0.1.3", "windows-numerics", ] @@ -5834,7 +6000,7 @@ checksum = "c0fdd3ddb90610c7638aa2b3a3ab2904fb9e5cdbecc643ddb3647212781c4ae3" dependencies = [ "windows-implement", "windows-interface", - "windows-link", + "windows-link 0.1.3", "windows-result", "windows-strings", ] @@ -5846,7 +6012,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc6a41e98427b19fe4b73c550f060b59fa592d7d686537eebf9385621bfbad8e" dependencies = [ "windows-core", - "windows-link", + "windows-link 0.1.3", "windows-threading", ] @@ -5878,6 +6044,12 @@ version = "0.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e6ad25900d524eaabdbbb96d20b4311e1e7ae1699af4fb28c17ae66c80d798a" +[[package]] +name = "windows-link" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "45e46c0661abb7180e7b9c281db115305d49ca1709ab8242adf09666d2173c65" + [[package]] name = "windows-numerics" version = "0.2.0" @@ -5885,7 +6057,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9150af68066c4c5c07ddc0ce30421554771e528bde427614c61038bc2c92c2b1" dependencies = [ "windows-core", - "windows-link", + "windows-link 0.1.3", ] [[package]] @@ -5894,7 +6066,7 @@ version = "0.3.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "56f42bd332cc6c8eac5af113fc0c1fd6a8fd2aa08a0119358686e5160d0586c6" dependencies = [ - "windows-link", + "windows-link 0.1.3", ] [[package]] @@ -5903,7 +6075,7 @@ version = "0.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "56e6c93f3a0c3b36176cb1327a4958a0353d5d166c2a35cb268ace15e91d3b57" dependencies = [ - "windows-link", + "windows-link 0.1.3", ] [[package]] @@ -5970,7 +6142,7 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b66463ad2e0ea3bbf808b7f1d371311c80e115c0b71d60efc142cafbcfb057a6" dependencies = [ - "windows-link", + "windows-link 0.1.3", ] [[package]] diff --git a/rust/export-c2pa-schema/src/main.rs b/rust/export-c2pa-schema/src/main.rs index ee1e4f32..b115a9c6 100644 --- a/rust/export-c2pa-schema/src/main.rs +++ b/rust/export-c2pa-schema/src/main.rs @@ -2,8 +2,7 @@ use std::{fs, path::Path}; use anyhow::Result; use c2pa::{Builder, ManifestDefinition, Reader}; -use schemars::JsonSchema; -use schemars::{schema::RootSchema, schema_for}; +use schemars::{schema::RootSchema, schema_for, JsonSchema}; use serde::{Deserialize, Serialize}; use serde_with::skip_serializing_none; diff --git a/rust/iroh-streamplace/Cargo.toml b/rust/iroh-streamplace/Cargo.toml index c99de17f..d67ce3ee 100644 --- a/rust/iroh-streamplace/Cargo.toml +++ b/rust/iroh-streamplace/Cargo.toml @@ -8,7 +8,7 @@ name = "iroh_streamplace" crate-type = ["staticlib", "cdylib"] [dependencies] -iroh = { version = "0.91.1", features = ["discovery-local-network"] } +iroh = { version = "0.92", features = ["discovery-local-network"] } tokio = "1.47.1" uniffi = { version = "=0.28.3", features = ["tokio"] } @@ -21,14 +21,21 @@ snafu = "0.8.6" anyhow = "1.0.99" serde = { version = "1.0.219", features = ["derive"] } postcard = { version = "1.1.3", features = ["use-std"] } -irpc = "0.7.0" -irpc-iroh = "0.7.0" +irpc = "0.8.0" +irpc-iroh = "0.8.0" c2pa = { git = "https://github.com/hyphacoop/c2pa-rs.git", rev = "1b84d40219b27340a30fc309250e774e8a7b7761", features = [ "openssl", "file_io", ] } thiserror = "2.0.16" serde_json = "1.0.145" +iroh-smol-kv = "0.2.0" +hex = "0.4.3" +iroh-gossip = "0.92.0" +ref-cast = "1.0.25" +iroh-base = "0.92.0" +rand = "0.8.5" [dev-dependencies] +testresult = "0.4.1" tracing-subscriber = { version = "0.3.19", features = ["env-filter"] } diff --git a/rust/iroh-streamplace/src/c2pa.rs b/rust/iroh-streamplace/src/c2pa.rs index 73110929..289d267b 100644 --- a/rust/iroh-streamplace/src/c2pa.rs +++ b/rust/iroh-streamplace/src/c2pa.rs @@ -1,10 +1,7 @@ -use c2pa::Builder; -use c2pa::CallbackSigner; -use c2pa::Reader; -use c2pa::settings::Settings; +use std::{io::Cursor, sync::Arc}; + +use c2pa::{Builder, CallbackSigner, Reader, settings::Settings}; use serde_json; -use std::io::Cursor; -use std::sync::Arc; #[derive(Debug, thiserror::Error, uniffi::Error)] #[uniffi(flat_error)] diff --git a/rust/iroh-streamplace/src/endpoint.rs b/rust/iroh-streamplace/src/endpoint.rs index 56e8e0e1..f81c5bf8 100644 --- a/rust/iroh-streamplace/src/endpoint.rs +++ b/rust/iroh-streamplace/src/endpoint.rs @@ -1,7 +1,6 @@ use iroh::Watcher; -use crate::error::Error; -use crate::utils::NodeAddr; +use crate::{error::Error, utils::NodeAddr}; #[derive(uniffi::Object, Debug, Clone)] pub struct Endpoint { diff --git a/rust/iroh-streamplace/src/key.rs b/rust/iroh-streamplace/src/key.rs index 37fab52f..423b53a2 100644 --- a/rust/iroh-streamplace/src/key.rs +++ b/rust/iroh-streamplace/src/key.rs @@ -8,27 +8,27 @@ use crate::error::{Error, InvalidPublicKeySnafu}; /// information that is cached for performance reasons. #[derive(Debug, Clone, Eq, uniffi::Object)] #[uniffi::export(Display)] -pub struct PublicKey { +pub struct PublicKeyOld { pub(crate) key: [u8; 32], } -impl From for PublicKey { +impl From for PublicKeyOld { fn from(key: iroh::PublicKey) -> Self { - PublicKey { + PublicKeyOld { key: *key.as_bytes(), } } } -impl From<&PublicKey> for iroh::PublicKey { - fn from(key: &PublicKey) -> Self { +impl From<&PublicKeyOld> for iroh::PublicKey { + fn from(key: &PublicKeyOld) -> Self { iroh::PublicKey::from_bytes(&key.key).unwrap() } } #[uniffi::export] -impl PublicKey { +impl PublicKeyOld { /// Returns true if the PublicKeys are equal - pub fn equal(&self, other: &PublicKey) -> bool { + pub fn equal(&self, other: &PublicKeyOld) -> bool { *self == *other } @@ -62,13 +62,13 @@ impl PublicKey { } } -impl PartialEq for PublicKey { - fn eq(&self, other: &PublicKey) -> bool { +impl PartialEq for PublicKeyOld { + fn eq(&self, other: &PublicKeyOld) -> bool { self.key == other.key } } -impl std::fmt::Display for PublicKey { +impl std::fmt::Display for PublicKeyOld { fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { iroh::PublicKey::from(self).fmt(f) } @@ -86,7 +86,7 @@ mod tests { let bytes = b"\x52\x3c\x79\x96\xba\xd7\x74\x24\xe9\x67\x86\xcf\x7a\x72\x05\x11\x53\x37\xa5\xb4\x56\x5c\xd2\x55\x06\xa0\xf2\x97\xb1\x91\xa5\xea"; // // create key from string - let key = PublicKey::from_string(key_str.clone()).unwrap(); + let key = PublicKeyOld::from_string(key_str.clone()).unwrap(); // // test methods are as expected assert_eq!(key_str, key.to_string()); @@ -94,7 +94,7 @@ mod tests { assert_eq!(fmt_str, key.fmt_short()); // // create key from bytes - let key_0 = PublicKey::from_bytes(bytes.to_vec()).unwrap(); + let key_0 = PublicKeyOld::from_bytes(bytes.to_vec()).unwrap(); // // test methods are as expected assert_eq!(key_str, key_0.to_string()); diff --git a/rust/iroh-streamplace/src/lib.rs b/rust/iroh-streamplace/src/lib.rs index 11534766..5a1dcac5 100644 --- a/rust/iroh-streamplace/src/lib.rs +++ b/rust/iroh-streamplace/src/lib.rs @@ -4,6 +4,7 @@ pub mod c2pa; pub mod endpoint; pub mod error; pub mod key; +pub mod node; pub mod receiver; pub mod sender; pub mod utils; diff --git a/rust/iroh-streamplace/src/node/db.rs b/rust/iroh-streamplace/src/node/db.rs new file mode 100644 index 00000000..e0c42bfe --- /dev/null +++ b/rust/iroh-streamplace/src/node/db.rs @@ -0,0 +1,615 @@ +use std::{ + fmt::{self, Debug}, + ops::Bound, + pin::Pin, + sync::Arc, + time::Duration, +}; + +use bytes::Bytes; +use n0_future::{Stream, StreamExt}; +use ref_cast::RefCast; +use snafu::Snafu; +use tokio::sync::Mutex; + +use super::db; + +// the files here are just copied from iroh-smol-kv-uniffi/src/code +mod kv { + mod public_key; + pub use public_key::PublicKey; + mod time_bound; + pub use time_bound::TimeBound; + mod subscribe_mode; + pub use subscribe_mode::SubscribeMode; +} +use db::util::format_bytes; +pub use kv::{PublicKey, SubscribeMode, TimeBound}; + +/// Error creating a new database node. +#[derive(Debug, Snafu, uniffi::Error)] +#[snafu(module)] +pub enum CreateError { + /// The provided private key is invalid (not 32 bytes). + PrivateKey { size: u64 }, + /// The provided gossip topic is invalid (not 32 bytes). + Topic { size: u64 }, + /// Failed to bind the iroh endpoint. + Bind { message: String }, + /// Failed to subscribe to the gossip topic. + Subscribe { message: String }, +} + +/// Error joining peers. +#[derive(Debug, Snafu, uniffi::Error)] +#[snafu(module)] +pub enum JoinPeersError { + /// Failed to parse a provided iroh node ticket. + Ticket { message: String }, + /// Error during the join peers operation. + Irpc { message: String }, +} + +/// Error putting a value into the database. +#[derive(Debug, Snafu, uniffi::Error)] +#[snafu(module)] +pub enum PutError { + /// Error during the put operation. + Irpc { message: String }, +} + +/// Configuration for an iroh-streamplace node. +#[derive(uniffi::Record, Clone)] +pub struct Config { + /// An Ed25519 secret key as a 32 byte array. + pub key: Vec, + /// The gossip topic to use. Must be 32 bytes. + /// + /// You can use e.g. a BLAKE3 hash of a topic string here. This can be used + /// as a cheap way to have a shared secret - nodes that do not know the topic + /// cannot connect to the swarm. + pub topic: Vec, + /// Maximum duration to wait for sending a stream piece to a peer. + pub max_send_duration: Duration, + /// Disable using relays, for tests. + pub disable_relay: bool, +} + +#[derive(uniffi::Enum, Debug, Clone)] +enum StreamFilter { + All, + Global, + Stream(Vec), +} + +/// A filter for subscriptions and iteration. +#[derive(uniffi::Object, Debug, Clone)] +pub struct Filter { + scope: Option>>, + stream: StreamFilter, + min_time: TimeBound, + max_time: TimeBound, +} + +#[uniffi::export] +impl Filter { + /// Creates a new filter that matches everything. + #[uniffi::constructor] + pub fn new() -> Arc { + Arc::new(Self { + scope: None, + stream: StreamFilter::All, + min_time: TimeBound::Unbounded, + max_time: TimeBound::Unbounded, + }) + } + + /// Restrict to the global namespace, no per stream data. + pub fn global(mut self: Arc) -> Arc { + let this = Arc::make_mut(&mut self); + this.stream = StreamFilter::Global; + self + } + + /// Restrict to one specific stream, no global data. + pub fn stream(mut self: Arc, stream: Vec) -> Arc { + let this = Arc::make_mut(&mut self); + this.stream = StreamFilter::Stream(stream); + self + } + + /// Restrict to a set of scopes. + pub fn scopes(mut self: Arc, scopes: Vec>) -> Arc { + let this = Arc::make_mut(&mut self); + this.scope = Some(scopes); + self + } + + /// Restrict to a single scope. + pub fn scope(self: Arc, scope: Arc) -> Arc { + self.scopes(vec![scope]) + } + + /// Restrict to a time range. + pub fn timestamps(mut self: Arc, min: TimeBound, max: TimeBound) -> Arc { + let this = Arc::make_mut(&mut self); + this.min_time = min; + this.max_time = max; + self + } + + /// Restrict to a time range given in nanoseconds since unix epoch. + pub fn time_range(self: Arc, min: u64, max: u64) -> Arc { + self.timestamps(TimeBound::Included(min), TimeBound::Excluded(max)) + } + + /// Restrict to a time range starting at min, unbounded at the top. + pub fn time_from(self: Arc, min: u64) -> Arc { + self.timestamps(TimeBound::Included(min), TimeBound::Unbounded) + } +} + +impl From for iroh_smol_kv::Filter { + fn from(value: Filter) -> Self { + let mut filter = iroh_smol_kv::Filter::ALL; + match value.stream { + // everything + StreamFilter::All => {} + // everything starting with 'g', for the global namespace + StreamFilter::Global => { + filter = filter.key_prefix(b"g".as_ref()); + } + // a specific stream, everything starting with 's' + escaped stream name + StreamFilter::Stream(t) => { + let prefix = util::encode_stream_and_key(Some(&t), &[]); + filter = filter.key_prefix(prefix); + } + }; + filter = filter.timestamps_nanos(( + Bound::::from(value.min_time), + Bound::::from(value.max_time), + )); + if let Some(scopes) = value.scope { + let keys = scopes.iter().map(|k| iroh::PublicKey::from(k.as_ref())); + filter = filter.scopes(keys); + } + filter + } +} + +/// Error getting the next item from a subscription. +#[derive(uniffi::Enum, Snafu, Debug)] +#[snafu(module)] +pub enum SubscribeNextError { + /// Error during the subscribe next operation. + Irpc { message: String }, +} + +/// Error getting the next item from a subscription. +#[derive(uniffi::Enum, Snafu, Debug)] +#[snafu(module)] +pub enum WriteError { + /// The provided private key is invalid (not 32 bytes). + PrivateKeySize { size: u64 }, +} + +/// An entry returned from the database. +#[derive(uniffi::Record, Debug, PartialEq, Eq)] +pub struct Entry { + scope: Arc, + stream: Option>, + key: Vec, + value: Vec, + timestamp: u64, +} + +/// An item returned from a subscription. +#[derive(uniffi::Enum, Debug)] +pub enum SubscribeItem { + Entry { + scope: Arc, + stream: Option>, + key: Vec, + value: Vec, + timestamp: u64, + }, + CurrentDone, + Expired { + scope: Arc, + stream: Option>, + key: Vec, + timestamp: u64, + }, + Other, +} + +fn fmt_stream(stream: &Option>) -> String { + match stream { + None => "".to_string(), + Some(s) => format_bytes(s), + } +} + +#[uniffi::export] +pub fn subscribe_item_debug(item: &SubscribeItem) -> String { + match item { + SubscribeItem::Entry { + scope, + stream, + key, + value, + timestamp, + } => format!( + "Entry {{ scope: {}, stream: {}, key: {}, value: {}, timestamp: {} }}", + scope.fmt_short(), + fmt_stream(stream), + format_bytes(key), + format_bytes(value), + timestamp + ), + SubscribeItem::CurrentDone => "CurrentDone".to_string(), + SubscribeItem::Expired { + scope, + stream, + key, + timestamp, + } => format!( + "Expired {{ scope: {}, stream: {}, key: {}, timestamp: {} }}", + scope.fmt_short(), + fmt_stream(stream), + format_bytes(key), + timestamp + ), + SubscribeItem::Other => "Other".to_string(), + } +} + +impl From for SubscribeItem { + fn from(item: iroh_smol_kv::SubscribeItem) -> Self { + match &item { + iroh_smol_kv::SubscribeItem::Entry((scope, key, value)) => { + let Some((stream, key)) = util::decode_stream_and_key(key) else { + return Self::Other; + }; + Self::Entry { + scope: Arc::new((*scope).into()), + stream, + key, + value: value.value.to_vec(), + timestamp: value.timestamp, + } + } + iroh_smol_kv::SubscribeItem::CurrentDone => Self::CurrentDone, + iroh_smol_kv::SubscribeItem::Expired((scope, topic, timestamp)) => { + let (stream, key) = util::decode_stream_and_key(topic).unwrap(); + Self::Expired { + scope: Arc::new((*scope).into()), + stream, + key, + timestamp: *timestamp, + } + } + } + } +} + +/// A response to a subscribe request. +/// +/// This can be used as a stream of [`SubscribeItem`]s. +#[derive(uniffi::Object)] +#[uniffi::export(Debug)] +#[allow(clippy::type_complexity)] +pub struct SubscribeResponse { + inner: Mutex< + Pin< + Box< + dyn Stream> + + Send + + Sync + + 'static, + >, + >, + >, +} + +impl fmt::Debug for SubscribeResponse { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("SubscribeResponse").finish_non_exhaustive() + } +} + +#[uniffi::export] +impl SubscribeResponse { + pub async fn next_raw(&self) -> Result, SubscribeNextError> { + let mut this = self.inner.lock().await; + match this.as_mut().next().await { + None => Ok(None), + Some(Ok(item)) => Ok(Some(item.into())), + Some(Err(e)) => Err(SubscribeNextError::Irpc { + message: e.to_string(), + }), + } + } +} + +/// Options for subscribing. +/// +/// `filter` specifies what to subscribe to. +/// `mode` specifies whether to get current items, new items, or both. +#[derive(uniffi::Record)] +pub struct SubscribeOpts { + pub filter: Arc, + pub mode: SubscribeMode, +} + +impl From for iroh_smol_kv::Subscribe { + fn from(opts: SubscribeOpts) -> Self { + iroh_smol_kv::Subscribe { + filter: opts.filter.as_ref().clone().into(), + mode: opts.mode.into(), + } + } +} + +/// A write scope that can be used to put values into the database. +/// +/// The default write scope is available from the [`Node::node_scope`] method. +#[derive(Clone, Debug, RefCast, uniffi::Object)] +#[repr(transparent)] +pub struct WriteScope(iroh_smol_kv::WriteScope); + +#[uniffi::export] +impl WriteScope { + pub async fn put( + &self, + stream: Option>, + key: Vec, + value: Vec, + ) -> Result<(), PutError> { + self.put_impl(stream, key, value.into()) + .await + .map_err(|e| PutError::Irpc { + message: e.to_string(), + }) + } +} + +impl WriteScope { + pub fn new(inner: iroh_smol_kv::WriteScope) -> Self { + Self(inner) + } + + /// Put a value into the database, optionally in a specific stream. + pub async fn put_impl( + &self, + stream: Option>, + key: impl AsRef<[u8]>, + value: Bytes, + ) -> Result<(), irpc::Error> { + let key = key.as_ref(); + let stream = stream.as_ref().map(|s| s.as_ref()); + let encoded = util::encode_stream_and_key(stream, key); + self.0.put(encoded, value).await?; + Ok(()) + } +} + +/// Iroh-streamplace specific metadata database. +#[derive(Debug, Clone, RefCast, uniffi::Object)] +#[repr(transparent)] +pub struct Db(iroh_smol_kv::Client); + +impl Db { + pub fn new(inner: iroh_smol_kv::Client) -> Self { + Self(inner) + } + + pub fn inner(&self) -> &iroh_smol_kv::Client { + &self.0 + } +} + +#[uniffi::export] +impl Db { + pub fn write(&self, secret: Vec) -> Result, WriteError> { + let secret = iroh::SecretKey::from_bytes(&secret.try_into().map_err(|e: Vec| { + WriteError::PrivateKeySize { + size: e.len() as u64, + } + })?); + let write = self.0.write(secret); + Ok(Arc::new(WriteScope::new(write))) + } + + pub async fn iter_with_opts( + &self, + filter: Arc, + ) -> Result, SubscribeNextError> { + let sub = self.subscribe_with_opts(SubscribeOpts { + filter, + mode: SubscribeMode::Current, + }); + let mut items = Vec::new(); + while let Some(item) = sub.next_raw().await? { + match item { + SubscribeItem::Entry { + scope, + stream, + key, + value, + timestamp, + } => { + items.push(Entry { + scope, + stream, + key, + value, + timestamp, + }); + } + _ => unreachable!("we used SubscribeMode::Current, so we should only get entries"), + } + } + Ok(items) + } + + pub fn subscribe(&self, filter: Arc) -> Arc { + self.subscribe_with_opts(SubscribeOpts { + filter, + mode: SubscribeMode::Both, + }) + } + + /// Subscribe with options. + pub fn subscribe_with_opts(&self, opts: SubscribeOpts) -> Arc { + Arc::new(SubscribeResponse { + inner: Mutex::new(Box::pin( + self.0.subscribe_with_opts(opts.into()).stream_raw(), + )), + }) + } +} + +mod util { + + pub fn encode_stream_and_key(stream: Option<&[u8]>, key: &[u8]) -> Vec { + let mut result = Vec::new(); + if let Some(s) = stream { + result.push(b's'); + escape_into([s], &mut result); + } else { + result.push(b'g'); + } + result.extend(key); + result + } + + pub fn decode_stream_and_key(encoded: &[u8]) -> Option<(Option>, Vec)> { + match encoded.split_first() { + Some((b's', mut rest)) => { + let stream = unescape_one(&mut rest)?; + Some((Some(stream), rest.to_vec())) + } + Some((b'g', rest)) => Some((None, rest.to_vec())), + _ => None, + } + } + + // these values are needed to keep the order preserved + const ESCAPE: u8 = 1; + const SEPARATOR: u8 = 0; + + /// Escape into an existing vec. + fn escape_into(components: I, result: &mut Vec) + where + I: IntoIterator, + C: AsRef<[u8]>, + { + for segment in components.into_iter() { + for &byte in segment.as_ref() { + match byte { + ESCAPE => result.extend([ESCAPE, ESCAPE]), + SEPARATOR => result.extend([ESCAPE, SEPARATOR]), + _ => result.push(byte), + } + } + result.push(SEPARATOR); + } + // you might think that the trailing separator is unnecessary, but it is needed + // to distinguish between the empty path and the path with one empty component + } + + fn unescape_one(path: &mut &[u8]) -> Option> { + let mut segment = Vec::new(); + let mut escape = false; + for (i, &byte) in path.iter().enumerate() { + if escape { + segment.push(byte); + escape = false; + } else { + match byte { + ESCAPE => escape = true, + SEPARATOR => { + *path = &path[i + 1..]; + return Some(segment); + } + _ => segment.push(byte), + } + } + } + None + } + + /// A simple version of unescape. + #[allow(dead_code)] + fn unescape(path: &[u8]) -> Vec> { + let mut components = Vec::new(); + let mut segment = Vec::new(); + let mut escape = false; + for &byte in path { + if escape { + segment.push(byte); + escape = false; + } else { + match byte { + ESCAPE => escape = true, + SEPARATOR => { + components.push(segment); + segment = Vec::new(); + } + _ => segment.push(byte), + } + } + } + components + } + + pub fn format_bytes(bytes: &[u8]) -> String { + if bytes.is_empty() { + return "\"\"".to_string(); + } + let Ok(s) = std::str::from_utf8(bytes) else { + return hex::encode(bytes); + }; + if s.chars() + .any(|c| c.is_control() && c != '\n' && c != '\t' && c != '\r') + { + return hex::encode(bytes); + } + format!("\"{}\"", escape_string(s)) + } + + pub fn escape_string(s: &str) -> String { + s.chars() + .map(|c| match c { + '"' => "\\\"".to_string(), + '\\' => "\\\\".to_string(), + '\n' => "\\n".to_string(), + '\t' => "\\t".to_string(), + '\r' => "\\r".to_string(), + c => c.to_string(), + }) + .collect() + } +} + +#[cfg(test)] +mod tests { + use super::db::util; + + #[test] + fn escape_unescape() { + let cases: Vec<(Option<&[u8]>, &[u8])> = vec![ + (None, b"key1"), + (Some(b""), b""), + (Some(b""), b"a"), + (Some(b"a"), b""), + ]; + for (stream, key) in cases { + let encoded = util::encode_stream_and_key(stream, key); + let (decoded_stream, decoded_key) = util::decode_stream_and_key(&encoded).unwrap(); + assert_eq!(decoded_stream.as_deref(), stream); + assert_eq!(decoded_key.as_slice(), key); + } + } +} diff --git a/rust/iroh-streamplace/src/node/db/kv/public_key.rs b/rust/iroh-streamplace/src/node/db/kv/public_key.rs new file mode 100644 index 00000000..5c5788d3 --- /dev/null +++ b/rust/iroh-streamplace/src/node/db/kv/public_key.rs @@ -0,0 +1,102 @@ +use std::{fmt, str::FromStr}; + +use snafu::Snafu; + +#[derive(Debug, Snafu, uniffi::Error)] +#[snafu(module)] +pub enum PublicKeyError { + Length { size: u64 }, + Invalid { message: String }, +} + +/// A public key. +/// +/// The key itself is just a 32 byte array, but a key has associated crypto +/// information that is cached for performance reasons. +#[derive(Clone, Copy, Eq, Ord, PartialOrd, uniffi::Object)] +#[uniffi::export(Display)] +pub struct PublicKey { + pub(crate) key: [u8; 32], +} + +impl fmt::Debug for PublicKey { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + iroh::PublicKey::from(self).fmt(f) + } +} + +impl From for PublicKey { + fn from(key: iroh::PublicKey) -> Self { + PublicKey { + key: *key.as_bytes(), + } + } +} + +impl From<&PublicKey> for iroh::PublicKey { + fn from(key: &PublicKey) -> Self { + iroh::PublicKey::from_bytes(&key.key).unwrap() + } +} + +#[uniffi::export] +impl PublicKey { + /// Returns true if the PublicKeys are equal + pub fn equal(&self, other: &PublicKey) -> bool { + *self == *other + } + + /// Express the PublicKey as a byte array + pub fn as_vec(&self) -> Vec { + self.key.to_vec() + } + + /// Make a PublicKey from base32 string + #[uniffi::constructor] + #[allow(clippy::result_large_err)] + pub fn from_string(s: String) -> Result { + if s.len() != 64 { + return Err(PublicKeyError::Length { + size: s.len() as u64, + }); + } + let key = iroh::PublicKey::from_str(&s).map_err(|e| PublicKeyError::Invalid { + message: e.to_string(), + })?; + Ok(key.into()) + } + + /// Make a PublicKey from byte array + #[uniffi::constructor] + #[allow(clippy::result_large_err)] + pub fn from_bytes(bytes: Vec) -> Result { + if bytes.len() != 32 { + return Err(PublicKeyError::Length { + size: bytes.len() as u64, + }); + } + let bytes: [u8; 32] = bytes.try_into().expect("checked above"); + let key = iroh::PublicKey::from_bytes(&bytes).map_err(|e| PublicKeyError::Invalid { + message: e.to_string(), + })?; + Ok(key.into()) + } + + /// Convert to a base32 string limited to the first 10 bytes for a friendly string + /// representation of the key. + pub fn fmt_short(&self) -> String { + iroh::PublicKey::from(self).fmt_short() + } +} + +impl PartialEq for PublicKey { + fn eq(&self, other: &PublicKey) -> bool { + self.key == other.key + } +} + +impl fmt::Display for PublicKey { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + iroh::PublicKey::from(self).fmt(f) + } +} diff --git a/rust/iroh-streamplace/src/node/db/kv/subscribe_mode.rs b/rust/iroh-streamplace/src/node/db/kv/subscribe_mode.rs new file mode 100644 index 00000000..7d736eae --- /dev/null +++ b/rust/iroh-streamplace/src/node/db/kv/subscribe_mode.rs @@ -0,0 +1,29 @@ +use iroh_smol_kv as w; + +/// Subscription mode for key-value subscriptions. +#[derive(uniffi::Enum, Debug, Clone, Copy)] +pub enum SubscribeMode { + Current, + Future, + Both, +} + +impl From for SubscribeMode { + fn from(m: w::SubscribeMode) -> Self { + match m { + w::SubscribeMode::Current => SubscribeMode::Current, + w::SubscribeMode::Future => SubscribeMode::Future, + w::SubscribeMode::Both => SubscribeMode::Both, + } + } +} + +impl From for w::SubscribeMode { + fn from(m: SubscribeMode) -> Self { + match m { + SubscribeMode::Current => w::SubscribeMode::Current, + SubscribeMode::Future => w::SubscribeMode::Future, + SubscribeMode::Both => w::SubscribeMode::Both, + } + } +} diff --git a/rust/iroh-streamplace/src/node/db/kv/time_bound.rs b/rust/iroh-streamplace/src/node/db/kv/time_bound.rs new file mode 100644 index 00000000..60920da5 --- /dev/null +++ b/rust/iroh-streamplace/src/node/db/kv/time_bound.rs @@ -0,0 +1,29 @@ +use std::ops::Bound; + +/// A bound on time for filtering. +#[derive(uniffi::Enum, Debug, Clone, Copy)] +pub enum TimeBound { + Unbounded, + Included(u64), + Excluded(u64), +} + +impl From> for TimeBound { + fn from(b: Bound) -> Self { + match b { + Bound::Unbounded => TimeBound::Unbounded, + Bound::Included(t) => TimeBound::Included(t), + Bound::Excluded(t) => TimeBound::Excluded(t), + } + } +} + +impl From for Bound { + fn from(b: TimeBound) -> Self { + match b { + TimeBound::Unbounded => Bound::Unbounded, + TimeBound::Included(t) => Bound::Included(t), + TimeBound::Excluded(t) => Bound::Excluded(t), + } + } +} diff --git a/rust/iroh-streamplace/src/node/mod.rs b/rust/iroh-streamplace/src/node/mod.rs new file mode 100644 index 00000000..f625025a --- /dev/null +++ b/rust/iroh-streamplace/src/node/mod.rs @@ -0,0 +1,13 @@ +use std::sync::LazyLock; + +/// We export the entire API at top level since this is what go-uniffi-bindgen will do anyway. +mod streams; +pub use streams::*; +mod db; +pub use db::*; +#[cfg(test)] +mod tests; + +/// Lazily initialized Tokio runtime for use in uniffi methods that need a runtime. +static RUNTIME: LazyLock = + LazyLock::new(|| tokio::runtime::Runtime::new().unwrap()); diff --git a/rust/iroh-streamplace/src/node/streams.rs b/rust/iroh-streamplace/src/node/streams.rs new file mode 100644 index 00000000..b5e3bff2 --- /dev/null +++ b/rust/iroh-streamplace/src/node/streams.rs @@ -0,0 +1,734 @@ +use std::{ + collections::{BTreeMap, BTreeSet, HashSet}, + str::FromStr, + sync::Arc, +}; + +use bytes::Bytes; +use iroh::{NodeId, PublicKey, RelayMode, SecretKey, Watcher}; +use iroh_base::ticket::NodeTicket; +use iroh_gossip::{net::Gossip, proto::TopicId}; +use irpc::{WithChannels, rpc::RemoteService}; +use irpc_iroh::{IrohProtocol, IrohRemoteConnection}; +use n0_future::future::Boxed; + +mod rpc { + //! Protocol API + use bytes::Bytes; + use iroh::NodeId; + use irpc::{channel::oneshot, rpc_requests}; + use serde::{Deserialize, Serialize}; + + pub const ALPN: &[u8] = b"/iroh/streamplace/1"; + + /// Subscribe to the given `key` + #[derive(Debug, Serialize, Deserialize)] + pub struct Subscribe { + pub key: String, + // TODO: verify + pub remote_id: NodeId, + } + + /// Unsubscribe from the given `key` + #[derive(Debug, Serialize, Deserialize)] + pub struct Unsubscribe { + pub key: String, + // TODO: verify + pub remote_id: NodeId, + } + + // #[derive(Debug, Serialize, Deserialize)] + // pub struct SendSegment { + // pub key: String, + // pub data: Bytes, + // } + + #[derive(Debug, Clone, Serialize, Deserialize)] + pub struct RecvSegment { + pub key: String, + pub data: Bytes, + } + + // Use the macro to generate both the Protocol and Message enums + // plus implement Channels for each type + #[rpc_requests(message = Message)] + #[derive(Serialize, Deserialize, Debug)] + pub enum Protocol { + #[rpc(tx=oneshot::Sender<()>)] + Subscribe(Subscribe), + #[rpc(tx=oneshot::Sender<()>)] + Unsubscribe(Unsubscribe), + #[rpc(tx=oneshot::Sender<()>)] + RecvSegment(RecvSegment), + } +} + +mod api { + //! Protocol API + use bytes::Bytes; + use iroh::{NodeAddr, NodeId}; + use irpc::{channel::oneshot, rpc_requests}; + use serde::{Deserialize, Serialize}; + + /// Subscribe to the given `key` + #[derive(Debug, Serialize, Deserialize)] + pub struct Subscribe { + pub key: String, + // TODO: verify + pub remote_id: NodeId, + } + + /// Unsubscribe from the given `key` + #[derive(Debug, Serialize, Deserialize)] + pub struct Unsubscribe { + pub key: String, + // TODO: verify + pub remote_id: NodeId, + } + + #[derive(Debug, Serialize, Deserialize)] + pub struct SendSegment { + pub key: String, + pub data: Bytes, + } + + #[derive(Debug, Serialize, Deserialize)] + pub struct JoinPeers { + pub peers: Vec, + } + + #[derive(Debug, Serialize, Deserialize)] + pub struct GetNodeAddr; + + // Use the macro to generate both the Protocol and Message enums + // plus implement Channels for each type + #[rpc_requests(message = Message)] + #[derive(Serialize, Deserialize, Debug)] + pub enum Protocol { + #[rpc(tx=oneshot::Sender<()>)] + Subscribe(Subscribe), + #[rpc(tx=oneshot::Sender<()>)] + Unsubscribe(Unsubscribe), + #[rpc(tx=oneshot::Sender<()>)] + SendSegment(SendSegment), + #[rpc(tx=oneshot::Sender<()>)] + JoinPeers(JoinPeers), + #[rpc(tx=oneshot::Sender)] + GetNodeAddr(GetNodeAddr), + } +} +use api::{Message as ApiMessage, Protocol as ApiProtocol}; +use n0_future::{FuturesUnordered, StreamExt}; +use rpc::{Message as RpcMessage, Protocol as RpcProtocol}; +use snafu::Snafu; +use tracing::{Instrument, debug, error, trace, trace_span, warn}; + +use super::{Config, CreateError, JoinPeersError, PutError, db, streams::rpc::RecvSegment}; + +pub(crate) enum HandlerMode { + Sender, + Forwarder, + Receiver(Box) -> Boxed<()> + Send + Sync + 'static>), +} + +impl HandlerMode { + pub fn receiver_fn(f: F) -> Self + where + F: Fn(String, Vec) -> Fut + Send + Sync + 'static, + Fut: std::future::Future + Send + 'static, + { + Self::Receiver(Box::new(move |name, data| Box::pin(f(name, data)))) + } + + pub fn receiver(handler: Arc) -> Self { + Self::receiver_fn(move |id, data| { + let handler = handler.clone(); + async move { + handler.handle_data(id, data).await; + } + }) + } + + pub fn mode_str(&self) -> &'static str { + match self { + HandlerMode::Sender => "sender", + HandlerMode::Forwarder => "forwarder", + HandlerMode::Receiver(_) => "receiver", + } + } +} + +type Tasks = FuturesUnordered)>>; + +/// Actor that contains both a kv db for metadata and a handler for the rpc protocol. +/// +/// This can be used both for sender and receiver nodes. Sender nodes will just set the +/// handler to None. +struct Actor { + /// Receiver for rpc messages from remote nodes + rpc_rx: tokio::sync::mpsc::Receiver, + /// Receiver for API messages from the user + api_rx: tokio::sync::mpsc::Receiver, + /// nodes I need to send to for each stream + subscribers: BTreeMap>, + /// nodes I am subscribed to + subscriptions: BTreeMap, + /// lightweight typed connection pool + connections: ConnectionPool, + /// How to handle incoming data + handler: HandlerMode, + /// Iroh protocol router, I need to keep it around to keep the protocol alive + router: iroh::protocol::Router, + /// Metadata db + client: db::Db, + /// Write scope for this node for the metadata db + write: db::WriteScope, + /// Ongoing tasks + tasks: Tasks, + /// Configuration, needed for timeouts etc. + config: Arc, +} + +#[derive(Debug, Clone)] +struct Connection { + id: NodeId, + rpc: irpc::Client, +} + +#[derive(Debug, Snafu)] +enum RpcTaskError { + #[snafu(transparent)] + Task { source: irpc::Error }, + #[snafu(transparent)] + Timeout { source: tokio::time::error::Elapsed }, +} + +struct ConnectionPool { + endpoint: iroh::Endpoint, + connections: BTreeMap, +} + +impl ConnectionPool { + fn new(endpoint: iroh::Endpoint) -> Self { + Self { + endpoint, + connections: BTreeMap::new(), + } + } + + /// Cheap conn pool hack + fn get(&mut self, remote: &NodeId) -> Connection { + if !self.connections.contains_key(remote) { + let conn = IrohRemoteConnection::new( + self.endpoint.clone(), + (*remote).into(), + rpc::ALPN.to_vec(), + ); + let conn = Connection { + rpc: irpc::Client::boxed(conn), + id: *remote, + }; + self.connections.insert(*remote, conn); + } + self.connections + .get_mut(remote) + .expect("just inserted") + .clone() + } + + fn remove(&mut self, remote: &NodeId) { + self.connections.remove(remote); + } +} + +impl Actor { + pub async fn spawn( + endpoint: iroh::Endpoint, + topic: iroh_gossip::proto::TopicId, + config: super::Config, + handler: HandlerMode, + ) -> Result<(Node, impl Future), iroh_gossip::api::ApiError> { + let (rpc_tx, rpc_rx) = tokio::sync::mpsc::channel::(32); + let (api_tx, api_rx) = tokio::sync::mpsc::channel::(32); + let gossip = Gossip::builder().spawn(endpoint.clone()); + let id = endpoint.node_id(); + let router = iroh::protocol::Router::builder(endpoint.clone()) + .accept(iroh_gossip::ALPN, gossip.clone()) + .accept( + rpc::ALPN, + IrohProtocol::new(rpc::Protocol::remote_handler(rpc_tx.into())), + ) + .spawn(); + let topic = gossip.subscribe(topic, vec![]).await?; + let secret = router.endpoint().secret_key().clone(); + let db_config = Default::default(); + let client = iroh_smol_kv::Client::local(topic, db_config); + let write = db::WriteScope::new(client.write(secret.clone())); + let client = db::Db::new(client); + let actor = Self { + rpc_rx, + api_rx, + subscribers: BTreeMap::new(), + subscriptions: BTreeMap::new(), + connections: ConnectionPool::new(router.endpoint().clone()), + handler, + router, + write: write.clone(), + client: client.clone(), + tasks: FuturesUnordered::new(), + config: Arc::new(config), + }; + let api = Node { + client: Arc::new(client), + write: Arc::new(write), + api: irpc::Client::local(api_tx), + }; + Ok(( + api, + actor + .run() + .instrument(trace_span!("actor", id=%id.fmt_short())), + )) + } + + async fn run(mut self) { + loop { + tokio::select! { + msg = self.rpc_rx.recv() => { + let Some(msg) = msg else { + error!("rpc channel closed"); + break; + }; + self.handle_rpc(msg).instrument(trace_span!("rpc")).await; + } + msg = self.api_rx.recv() => { + let Some(msg) = msg else { + break; + }; + self.handle_api(msg).instrument(trace_span!("api")).await; + } + res = self.tasks.next(), if !self.tasks.is_empty() => { + let Some((remote_id, res)) = res else { + error!("task finished but no result"); + break; + }; + match res { + Ok(()) => {} + Err(RpcTaskError::Timeout { source }) => { + warn!("call to {remote_id} timed out: {source}"); + } + Err(RpcTaskError::Task { source }) => { + warn!("call to {remote_id} failed: {source}"); + } + } + self.connections.remove(&remote_id); + } + } + } + } + + async fn update_subscriber_meta(&mut self, key: &str) { + let n = self + .subscribers + .get(key) + .map(|s| s.len()) + .unwrap_or_default(); + let v = n.to_string().into_bytes(); + self.write + .put_impl(Some(key.as_bytes().to_vec()), b"subscribers", v.into()) + .await + .ok(); + } + + /// Requests from remote nodes + async fn handle_rpc(&mut self, msg: RpcMessage) { + match msg { + RpcMessage::Subscribe(msg) => { + trace!("{:?}", msg.inner); + let WithChannels { + tx, + inner: rpc::Subscribe { key, remote_id }, + .. + } = msg; + self.subscribers + .entry(key.clone()) + .or_default() + .insert(remote_id); + self.update_subscriber_meta(&key).await; + tx.send(()).await.ok(); + } + RpcMessage::Unsubscribe(msg) => { + debug!("{:?}", msg.inner); + let WithChannels { + tx, + inner: rpc::Unsubscribe { key, remote_id }, + .. + } = msg; + if let Some(e) = self.subscribers.get_mut(&key) + && !e.remove(&remote_id) + { + warn!( + "unsubscribe: no subscription for {} from {}", + key, remote_id + ); + } + if let Some(subscriptions) = self.subscribers.get(&key) + && subscriptions.is_empty() + { + self.subscribers.remove(&key); + } + self.update_subscriber_meta(&key).await; + tx.send(()).await.ok(); + } + RpcMessage::RecvSegment(msg) => { + trace!("{:?}", msg.inner); + let WithChannels { + tx, + inner: rpc::RecvSegment { key, data }, + .. + } = msg; + match &self.handler { + HandlerMode::Sender => { + warn!("received segment but in sender mode"); + } + HandlerMode::Forwarder => { + if let Some(remotes) = self.subscribers.get(&key) { + Self::handle_send( + &mut self.tasks, + &mut self.connections, + &self.config, + key, + data, + remotes, + ); + } else { + trace!("no subscribers for stream {}", key); + } + } + HandlerMode::Receiver(handler) => { + if self.subscriptions.contains_key(&key) { + handler(key, data.to_vec()).await; + } else { + warn!("received segment for unsubscribed key: {}", key); + } + } + }; + tx.send(()).await.ok(); + } + } + } + + async fn handle_api(&mut self, msg: ApiMessage) { + match msg { + ApiMessage::SendSegment(msg) => { + trace!("{:?}", msg.inner); + let WithChannels { + tx, + inner: api::SendSegment { key, data }, + .. + } = msg; + if let Some(remotes) = self.subscribers.get(&key) { + Self::handle_send( + &mut self.tasks, + &mut self.connections, + &self.config, + key, + data, + remotes, + ); + } else { + trace!("no subscribers for stream {}", key); + } + tx.send(()).await.ok(); + } + ApiMessage::Subscribe(msg) => { + trace!("{:?}", msg.inner); + let WithChannels { + tx, + inner: api::Subscribe { key, remote_id }, + .. + } = msg; + let conn = self.connections.get(&remote_id); + conn.rpc + .rpc(rpc::Subscribe { + key: key.clone(), + remote_id: self.node_id(), + }) + .await + .ok(); + self.subscriptions.insert(key, remote_id); + tx.send(()).await.ok(); + } + ApiMessage::Unsubscribe(msg) => { + trace!("{:?}", msg.inner); + let WithChannels { + tx, + inner: api::Unsubscribe { key, remote_id }, + .. + } = msg; + let conn = self.connections.get(&remote_id); + conn.rpc + .rpc(rpc::Unsubscribe { + key: key.clone(), + remote_id: self.node_id(), + }) + .await + .ok(); + self.subscriptions.remove(&key); + tx.send(()).await.ok(); + } + ApiMessage::JoinPeers(msg) => { + trace!("{:?}", msg.inner); + let WithChannels { + tx, + inner: api::JoinPeers { peers }, + .. + } = msg; + let ids = peers + .iter() + .map(|a| a.node_id) + .filter(|id| *id != self.node_id()) + .collect::>(); + for addr in &peers { + self.router.endpoint().add_node_addr(addr.clone()).ok(); + } + self.client.inner().join_peers(ids).await.ok(); + tx.send(()).await.ok(); + } + ApiMessage::GetNodeAddr(msg) => { + trace!("{:?}", msg.inner); + let WithChannels { tx, .. } = msg; + if !self.config.disable_relay { + // don't await home relay if we have disabled relays, this will hang forever + self.router.endpoint().home_relay().initialized().await; + } + let addr = self.router.endpoint().node_addr().initialized().await; + tx.send(addr).await.ok(); + } + } + } + + fn handle_send( + tasks: &mut Tasks, + connections: &mut ConnectionPool, + config: &Arc, + key: String, + data: Bytes, + remotes: &BTreeSet, + ) { + let msg = rpc::RecvSegment { key, data }; + for remote in remotes { + trace!("sending to stream {}: {}", msg.key, remote); + let conn = connections.get(remote); + tasks.push(Box::pin(Self::forward_task( + config.clone(), + conn, + msg.clone(), + ))); + } + } + + async fn forward_task( + config: Arc, + conn: Connection, + msg: RecvSegment, + ) -> (NodeId, Result<(), RpcTaskError>) { + let id = conn.id; + let res = async move { + tokio::time::timeout(config.max_send_duration, conn.rpc.rpc(msg)).await??; + Ok(()) + } + .await; + (id, res) + } + + fn node_id(&self) -> PublicKey { + self.router.endpoint().node_id() + } +} + +/// Iroh-streamplace node that can send, forward or receive stream segments. +#[derive(Clone, uniffi::Object)] +pub struct Node { + client: Arc, + write: Arc, + api: irpc::Client, +} + +impl Node { + pub(crate) async fn new_in_runtime( + config: super::Config, + handler: HandlerMode, + ) -> Result, CreateError> { + let mode_str = Bytes::from(handler.mode_str()); + let secret_key = + SecretKey::from_bytes(&<[u8; 32]>::try_from(config.key.clone()).map_err(|e| { + CreateError::PrivateKey { + size: e.len() as u64, + } + })?); + let topic = + TopicId::from_bytes(<[u8; 32]>::try_from(config.topic.clone()).map_err(|e| { + CreateError::Topic { + size: e.len() as u64, + } + })?); + let relay_mode = if config.disable_relay { + RelayMode::Disabled + } else { + RelayMode::Default + }; + let endpoint = iroh::Endpoint::builder() + .secret_key(secret_key) + .relay_mode(relay_mode) + .bind() + .await + .map_err(|e| CreateError::Bind { + message: e.to_string(), + })?; + let (api, actor) = Actor::spawn(endpoint, topic, config, handler) + .await + .map_err(|e| CreateError::Subscribe { + message: e.to_string(), + })?; + api.node_scope() + .put_impl(Option::>::None, b"mode", mode_str) + .await + .ok(); + tokio::spawn(actor); + Ok(Arc::new(api)) + } +} + +/// DataHandler trait that is exported to go for receiving data callbacks. +#[uniffi::export(with_foreign)] +#[async_trait::async_trait] +pub trait DataHandler: Send + Sync { + async fn handle_data(&self, topic: String, data: Vec); +} + +#[uniffi::export] +impl Node { + /// Create a new streamplace client node. + #[uniffi::constructor] + pub async fn sender(config: super::Config) -> Result, CreateError> { + super::RUNTIME.block_on(Self::new_in_runtime(config, HandlerMode::Sender)) + } + + #[uniffi::constructor] + pub async fn forwarder(config: super::Config) -> Result, CreateError> { + super::RUNTIME.block_on(Self::new_in_runtime(config, HandlerMode::Forwarder)) + } + + #[uniffi::constructor] + pub async fn receiver( + config: super::Config, + handler: Arc, + ) -> Result, CreateError> { + super::RUNTIME.block_on(Self::new_in_runtime(config, HandlerMode::receiver(handler))) + } + + /// Get a handle to the db to watch for changes locally or globally. + pub fn db(&self) -> Arc { + self.client.clone() + } + + /// Get a handle to the write scope for this node. + /// + /// This is equivalent to calling `db.write(...)` with the secret key used to create the node. + pub fn node_scope(&self) -> Arc { + self.write.clone() + } + + /// Subscribe to updates for a given stream from a remote node. + pub async fn subscribe( + &self, + key: String, + remote_id: Arc, + ) -> Result<(), PutError> { + self.api + .rpc(api::Subscribe { + key, + remote_id: remote_id.as_ref().into(), + }) + .await + .map_err(|e| PutError::Irpc { + message: e.to_string(), + }) + } + + /// Unsubscribe from updates for a given stream from a remote node. + pub async fn unsubscribe( + &self, + key: String, + remote_id: Arc, + ) -> Result<(), PutError> { + self.api + .rpc(api::Unsubscribe { + key, + remote_id: remote_id.as_ref().into(), + }) + .await + .map_err(|e| PutError::Irpc { + message: e.to_string(), + }) + } + + /// Send a segment to all subscribers of the given stream. + pub async fn send_segment(&self, key: String, data: Vec) -> Result<(), PutError> { + self.api + .rpc(api::SendSegment { + key, + data: data.into(), + }) + .await + .map_err(|e| PutError::Irpc { + message: e.to_string(), + }) + } + + /// Join peers by their node tickets. + pub async fn join_peers(&self, peers: Vec) -> Result<(), JoinPeersError> { + let peers = peers + .iter() + .map(|p| NodeTicket::from_str(p)) + .collect::, _>>() + .map_err(|e| JoinPeersError::Ticket { + message: e.to_string(), + })?; + let addrs = peers + .iter() + .map(|t| t.node_addr().clone()) + .collect::>(); + self.api + .rpc(api::JoinPeers { peers: addrs }) + .await + .map_err(|e| JoinPeersError::Irpc { + message: e.to_string(), + }) + } + + /// Get this node's ticket. + pub async fn ticket(&self) -> Result { + let addr = self + .api + .rpc(api::GetNodeAddr) + .await + .map_err(|e| PutError::Irpc { + message: e.to_string(), + })?; + Ok(NodeTicket::from(addr).to_string()) + } + + /// Get this node's node ID. + pub async fn node_id(&self) -> Result, PutError> { + let addr = self + .api + .rpc(api::GetNodeAddr) + .await + .map_err(|e| PutError::Irpc { + message: e.to_string(), + })?; + Ok(Arc::new(addr.node_id.into())) + } +} diff --git a/rust/iroh-streamplace/src/node/tests.rs b/rust/iroh-streamplace/src/node/tests.rs new file mode 100644 index 00000000..a08e7202 --- /dev/null +++ b/rust/iroh-streamplace/src/node/tests.rs @@ -0,0 +1,299 @@ +use std::{ops::Deref, sync::Arc, time::Duration}; + +use async_trait::async_trait; +use n0_future::{BufferedStreamExt, StreamExt, stream}; +use testresult::TestResult; + +use super::{streams::HandlerMode, *}; + +struct TestNode { + node: Arc, + public: Arc, + ticket: String, + #[allow(dead_code)] + private: Vec, +} + +impl Deref for TestNode { + type Target = Node; + + fn deref(&self) -> &Self::Target { + &self.node + } +} + +impl TestNode { + /// Helper to create a test node with given config and handler mode. + async fn new(handler: HandlerMode) -> TestResult { + let config = Config { + key: vec![0_u8; 32], // will be replaced + topic: vec![0_u8; 32], // all nodes use the same topic + max_send_duration: Duration::from_secs(10), + disable_relay: false, + }; + Self::new_with_config(handler, config).await + } + + async fn new_with_config(handler: HandlerMode, mut config: Config) -> TestResult { + let key = iroh::SecretKey::generate(&mut rand::rngs::OsRng); + let key = key.to_bytes().to_vec(); + config.key = key.clone(); + let node = Node::new_in_runtime(config, handler).await?; + let public = node.node_id().await?; + let ticket = node.ticket().await?; + Ok(TestNode { + node, + private: key, + public, + ticket, + }) + } +} + +/// Helper to create multiple test nodes with given handler mode. +async fn test_nodes( + n: usize, + handler: impl Fn(usize) -> HandlerMode + Send + Sync, + disable_relay: bool, +) -> TestResult> { + const PAR: usize = 32; + let modes = (0..n).map(handler).collect::>(); + + let config = Config { + key: vec![0_u8; 32], // will be replaced + topic: vec![0_u8; 32], + max_send_duration: Duration::from_secs(10), + disable_relay, + }; + // create all nodes in parallel + let nodes = stream::iter(modes) + .map(|mode| TestNode::new_with_config(mode, config.clone())) + .buffered_unordered(PAR) + .collect::>() + .await; + let nodes = nodes.into_iter().collect::>>()?; + // join everyone to everyone + let tickets = nodes.iter().map(|n| n.ticket.clone()).collect::>(); + let res = stream::iter(&nodes) + .map(|n| n.join_peers(tickets.clone())) + .buffered_unordered(PAR) + .collect::>() + .await; + res.into_iter().collect::, _>>()?; + Ok(nodes) +} + +#[tokio::test] +async fn one_node() -> TestResult<()> { + tracing_subscriber::fmt::try_init().ok(); + let node = TestNode::new(HandlerMode::Sender).await?.node; + let write = node.node_scope(); + let db = node.db(); + println!("Ticket: {}", node.ticket().await?); + write + .put(Some(b"stream1".to_vec()), b"s".to_vec(), b"y".to_vec()) + .await?; + write + .put(Some(b"stream2".to_vec()), b"s".to_vec(), b"y".to_vec()) + .await?; + let res = db.subscribe_with_opts(SubscribeOpts { + filter: Filter::new(), + mode: SubscribeMode::Both, + }); + while let Some(item) = res.next_raw().await? { + if let SubscribeItem::CurrentDone = item { + break; + } + println!("Got item: {item:?}"); + } + let res = db + .iter_with_opts( + Filter::new() + .stream(b"stream1".to_vec()) + .scope(node.node_id().await?), + ) + .await?; + println!("Iter result: {res:?}"); + Ok(()) +} + +struct TestHandler { + info: T, + sender: tokio::sync::mpsc::Sender<(T, String, Vec)>, +} + +impl TestHandler { + fn new(info: T, sender: tokio::sync::mpsc::Sender<(T, String, Vec)>) -> Self { + Self { info, sender } + } +} + +#[async_trait] +impl DataHandler for TestHandler { + async fn handle_data(&self, topic: String, data: Vec) { + self.sender + .send((self.info.clone(), topic, data)) + .await + .ok(); + } +} + +#[tokio::test] +async fn two_nodes_send_receive() -> TestResult<()> { + tracing_subscriber::fmt::try_init().ok(); + let (tx, mut rx) = tokio::sync::mpsc::channel(32); + let handler = Arc::new(TestHandler::new((), tx)); + let sender = TestNode::new(HandlerMode::Sender).await?; + let receiver = TestNode::new(HandlerMode::receiver(handler)).await?; + // join the sender to the receiver. This will also configure the receiver endpoint to be able to dial the sender. + receiver.join_peers(vec![sender.ticket.clone()]).await?; + let stream = "teststream".to_string(); + receiver + .subscribe(stream.clone(), sender.public.clone()) + .await?; + sender.send_segment(stream, b"segment1".to_vec()).await?; + let (_, stream, data) = rx.recv().await.expect("should get data"); + assert_eq!(stream, "teststream"); + assert_eq!(data, b"segment1".to_vec()); + Ok(()) +} + +#[tokio::test] +async fn three_nodes_send_forward_receive() -> TestResult<()> { + tracing_subscriber::fmt::try_init().ok(); + let (tx, mut rx) = tokio::sync::mpsc::channel(32); + let handler = Arc::new(TestHandler::new((), tx)); + let sender = TestNode::new(HandlerMode::Sender).await?; + let forwarder = TestNode::new(HandlerMode::Forwarder).await?; + let receiver = TestNode::new(HandlerMode::receiver(handler)).await?; + // join everyone to everyone, so the receiver can reach the sender via the forwarder. + let tickets = vec![ + sender.ticket.clone(), + forwarder.ticket.clone(), + receiver.ticket.clone(), + ]; + receiver.join_peers(tickets.clone()).await?; + forwarder.join_peers(tickets.clone()).await?; + sender.join_peers(tickets).await?; + let stream = "teststream".to_string(); + receiver + .subscribe(stream.clone(), forwarder.public.clone()) + .await?; + forwarder + .subscribe(stream.clone(), sender.public.clone()) + .await?; + sender.send_segment(stream, b"segment1".to_vec()).await?; + let (_, stream, data) = rx.recv().await.expect("should get data"); + assert_eq!(stream, "teststream"); + assert_eq!(data, b"segment1".to_vec()); + Ok(()) +} + +#[tokio::test] +async fn meta_three_nodes_send_forward_receive() -> TestResult<()> { + tracing_subscriber::fmt::try_init().ok(); + let (tx, mut rx) = tokio::sync::mpsc::channel(32); + let handler = Arc::new(TestHandler::new((), tx)); + let sender = TestNode::new(HandlerMode::Sender).await?; + let forwarder = TestNode::new(HandlerMode::Forwarder).await?; + let receiver = TestNode::new(HandlerMode::receiver(handler)).await?; + // join everyone to everyone, so the receiver can reach the sender via the forwarder. + let tickets = vec![ + sender.ticket.clone(), + forwarder.ticket.clone(), + receiver.ticket.clone(), + ]; + receiver.join_peers(tickets.clone()).await?; + forwarder.join_peers(tickets.clone()).await?; + sender.join_peers(tickets).await?; + let stream = "teststream".to_string(); + receiver + .subscribe(stream.clone(), forwarder.public.clone()) + .await?; + forwarder + .subscribe(stream.clone(), sender.public.clone()) + .await?; + sender.send_segment(stream, b"segment1".to_vec()).await?; + let (_, stream, data) = rx.recv().await.expect("should get data"); + assert_eq!(stream, "teststream"); + assert_eq!(data, b"segment1".to_vec()); + let stream = receiver.db().subscribe(Filter::new()); + while let Some(item) = stream.next_raw().await? { + println!("{}", subscribe_item_debug(&item)); + } + Ok(()) +} + +async fn broadcast( + nsenders: usize, + nforwarders: usize, + nreceivers: usize, + nmsgs: usize, +) -> TestResult<()> { + let (tx, mut rx) = tokio::sync::mpsc::channel(32); + let ntotal = nsenders + nforwarders + nreceivers; + let senders = 0..nsenders; + let forwarders = nsenders..(nsenders + nforwarders); + let receivers = (nsenders + nforwarders)..ntotal; + let make_handler = |i: usize| { + if senders.contains(&i) { + HandlerMode::Sender + } else if forwarders.contains(&i) { + HandlerMode::Forwarder + } else { + HandlerMode::receiver(Arc::new(TestHandler::new(i, tx.clone()))) + } + }; + let nodes = test_nodes(ntotal, make_handler, true).await?; + let senders = &nodes[senders]; + let forwarders = &nodes[forwarders]; + let receivers = &nodes[receivers]; + let stream = "teststream".to_string(); + // subscribe all forwarders to a sender, round robin + for (i, forwarder) in forwarders.iter().enumerate() { + let sender = &senders[i % senders.len()]; + forwarder + .subscribe(stream.clone(), sender.public.clone()) + .await?; + } + // subscribe all receivers to a forwarder, round robin + for (i, receiver) in receivers.iter().enumerate() { + let forwarder = &forwarders[i % forwarders.len()]; + receiver + .subscribe(stream.clone(), forwarder.public.clone()) + .await?; + } + for _ in 0..nmsgs { + for sender in senders { + sender + .send_segment(stream.clone(), b"segment1".to_vec()) + .await?; + } + for _ in 0..receivers.len() { + let (i, stream, _) = rx.recv().await.expect("should get data"); + println!("Node {i} got data on stream {stream}"); + } + } + Ok(()) +} + +#[tokio::test] +async fn broadcast_1_2_4() -> TestResult<()> { + tracing_subscriber::fmt().try_init().ok(); + broadcast(1, 2, 4, 1).await?; + Ok(()) +} + +#[tokio::test] +async fn broadcast_1_3_9() -> TestResult<()> { + tracing_subscriber::fmt().try_init().ok(); + broadcast(1, 3, 9, 1).await?; + Ok(()) +} + +#[tokio::test] +async fn broadcast_1_4_16() -> TestResult<()> { + tracing_subscriber::fmt().try_init().ok(); + broadcast(1, 4, 16, 100).await?; + Ok(()) +} diff --git a/rust/iroh-streamplace/src/receiver.rs b/rust/iroh-streamplace/src/receiver.rs index 61743d73..9a995a57 100644 --- a/rust/iroh-streamplace/src/receiver.rs +++ b/rust/iroh-streamplace/src/receiver.rs @@ -2,11 +2,7 @@ use std::sync::Arc; use iroh::protocol::Router; -use crate::api::Api; -use crate::endpoint::Endpoint; -use crate::error::Error; -use crate::key::PublicKey; -use crate::utils::NodeAddr; +use crate::{api::Api, endpoint::Endpoint, error::Error, key::PublicKeyOld, utils::NodeAddr}; #[derive(uniffi::Object)] pub struct Receiver { @@ -21,7 +17,7 @@ impl Receiver { #[uniffi::constructor(async_runtime = "tokio")] pub async fn new( endpoint: &Endpoint, - handler: Arc, + handler: Arc, ) -> Result { let api = Api::spawn_with_handler(&endpoint.endpoint, move |id, data| { let handler = handler.clone(); @@ -42,7 +38,7 @@ impl Receiver { /// Subscribe to the given topic on the remote. #[uniffi::method(async_runtime = "tokio")] - pub async fn subscribe(&self, remote_id: Arc, topic: &str) -> Result<(), Error> { + pub async fn subscribe(&self, remote_id: Arc, topic: &str) -> Result<(), Error> { let remote_id: iroh::NodeId = remote_id.as_ref().into(); let api = Api::connect(self.endpoint.endpoint.clone(), remote_id); api.subscribe(topic.to_string(), self.endpoint.endpoint.node_id()) @@ -52,7 +48,11 @@ impl Receiver { /// Unsubscribe from this topic on the remote. #[uniffi::method(async_runtime = "tokio")] - pub async fn unsubscribe(&self, remote_id: Arc, topic: &str) -> Result<(), Error> { + pub async fn unsubscribe( + &self, + remote_id: Arc, + topic: &str, + ) -> Result<(), Error> { let remote_id: iroh::NodeId = remote_id.as_ref().into(); let api = Api::connect(self.endpoint.endpoint.clone(), remote_id); api.unsubscribe(topic.to_string(), self.endpoint.endpoint.node_id()) @@ -68,16 +68,15 @@ impl Receiver { #[uniffi::export(with_foreign)] #[async_trait::async_trait] -pub trait DataHandler: Send + Sync { +pub trait DataHandlerOld: Send + Sync { async fn handle_data(&self, topic: String, data: Vec); } #[cfg(test)] mod tests { - use crate::sender::Sender; - use super::*; + use crate::sender::Sender; #[tokio::test] async fn test_roundtrip() { @@ -96,7 +95,7 @@ mod tests { } #[async_trait::async_trait] - impl DataHandler for TestHandler { + impl DataHandlerOld for TestHandler { async fn handle_data(&self, topic: String, data: Vec) { self.messages.send((topic, data)).await.unwrap(); } @@ -109,10 +108,10 @@ mod tests { .unwrap(); let sender_addr = sender.node_addr().await; - println!("sender addr: {:?}", sender_addr); + println!("sender addr: {sender_addr:?}"); let receiver_addr = receiver.node_addr().await; - println!("recv addr: {:?}", receiver_addr); + println!("recv addr: {receiver_addr:?}"); // subscribe receiver diff --git a/rust/iroh-streamplace/src/sender.rs b/rust/iroh-streamplace/src/sender.rs index ab3b8d86..99ec5ae6 100644 --- a/rust/iroh-streamplace/src/sender.rs +++ b/rust/iroh-streamplace/src/sender.rs @@ -1,10 +1,7 @@ use bytes::Bytes; use iroh::protocol::Router; -use crate::api::Api; -use crate::endpoint::Endpoint; -use crate::error::Error; -use crate::utils::NodeAddr; +use crate::{api::Api, endpoint::Endpoint, error::Error, utils::NodeAddr}; #[derive(uniffi::Object)] pub struct Sender { diff --git a/rust/iroh-streamplace/src/utils.rs b/rust/iroh-streamplace/src/utils.rs index 2a95b731..cc02d198 100644 --- a/rust/iroh-streamplace/src/utils.rs +++ b/rust/iroh-streamplace/src/utils.rs @@ -1,13 +1,11 @@ -use std::str::FromStr; -use std::sync::Arc; +use std::{str::FromStr, sync::Arc}; -use crate::error::Error; -use crate::key::PublicKey; +use crate::{error::Error, key::PublicKeyOld}; /// A peer and it's addressing information. #[derive(Debug, Clone, PartialEq, Eq, uniffi::Object)] pub struct NodeAddr { - node_id: Arc, + node_id: Arc, relay_url: Option, addresses: Vec, } @@ -16,7 +14,7 @@ pub struct NodeAddr { impl NodeAddr { /// Create a new [`NodeAddr`] with empty [`AddrInfo`]. #[uniffi::constructor] - pub fn new(node_id: &PublicKey, derp_url: Option, addresses: Vec) -> Self { + pub fn new(node_id: &PublicKeyOld, derp_url: Option, addresses: Vec) -> Self { Self { node_id: Arc::new(node_id.clone()), relay_url: derp_url, @@ -24,7 +22,7 @@ impl NodeAddr { } } - pub fn node_id(&self) -> PublicKey { + pub fn node_id(&self) -> PublicKeyOld { self.node_id.as_ref().clone() }