From 05ad1c95af678576fb853d3782605b2392687c22 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Monnom?= Date: Thu, 25 May 2023 20:13:34 +0200 Subject: [PATCH] cleanup: room & dependencies (#76) --- Cargo.lock | 229 +++---- examples/Cargo.lock | 279 ++++----- examples/basic_room/src/main.rs | 3 +- examples/save_to_disk/src/main.rs | 3 +- examples/wgpu_room/Cargo.toml | 4 +- examples/wgpu_room/src/app.rs | 30 +- examples/wgpu_room/src/logo_track.rs | 10 +- examples/wgpu_room/src/sine_track.rs | 43 +- examples/wgpu_room/src/video_grid.rs | 1 + examples/wgpu_room/src/video_renderer.rs | 13 +- livekit-ffi/Cargo.toml | 3 - livekit-ffi/protocol/audio_frame.proto | 20 +- livekit-ffi/protocol/ffi.proto | 14 +- livekit-ffi/protocol/handle.proto | 4 +- livekit-ffi/protocol/room.proto | 26 +- livekit-ffi/protocol/track.proto | 6 +- livekit-ffi/protocol/video_frame.proto | 22 +- livekit-ffi/src/conversion/room.rs | 2 +- livekit-ffi/src/conversion/track.rs | 1 + livekit-ffi/src/server/audio_frame.rs | 3 +- livekit-ffi/src/server/mod.rs | 8 +- livekit-ffi/src/server/room.rs | 35 +- livekit-ffi/src/server/tests.rs | 11 +- livekit-ffi/src/server/utils.rs | 8 +- livekit-ffi/src/server/video_frame.rs | 5 +- livekit-webrtc/src/native/audio_source.rs | 5 +- livekit-webrtc/src/native/data_channel.rs | 15 +- livekit-webrtc/src/native/peer_connection.rs | 39 +- livekit/src/prelude.rs | 2 +- livekit/src/room/mod.rs | 577 +++++++++++++++++- livekit/src/room/observer.rs | 1 - .../src/room/participant/local_participant.rs | 17 +- livekit/src/room/participant/mod.rs | 1 + .../room/participant/remote_participant.rs | 14 +- livekit/src/room/publication/local.rs | 3 + livekit/src/room/room_session.rs | 563 ----------------- livekit/src/room/track/local_audio_track.rs | 13 +- livekit/src/room/track/local_video_track.rs | 13 +- livekit/src/room/track/mod.rs | 8 +- livekit/src/room/track/remote_audio_track.rs | 15 +- livekit/src/room/track/remote_video_track.rs | 15 +- livekit/src/rtc_engine/mod.rs | 13 +- livekit/src/rtc_engine/peer_transport.rs | 2 +- livekit/src/rtc_engine/rtc_events.rs | 2 +- livekit/src/rtc_engine/rtc_session.rs | 19 +- livekit/src/signal_client/mod.rs | 5 +- 46 files changed, 1094 insertions(+), 1031 deletions(-) delete mode 100644 livekit/src/room/observer.rs delete mode 100644 livekit/src/room/room_session.rs diff --git a/Cargo.lock b/Cargo.lock index cf9b6e6..bfe62ee 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10,30 +10,29 @@ checksum = "f26201604c87b1e01bd3d98f8d5d9a8fcbb815e8cedb41ffccbeb4bf593a35fe" [[package]] name = "aes" -version = "0.7.5" +version = "0.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9e8b47f52ea9bae42228d07ec09eb676433d7c4ed1ebdf0f1d1c29ed446f1ab8" +checksum = "433cfd6710c9986c576a25ca913c39d66a6474107b406f34f91d4a8923395241" dependencies = [ "cfg-if", "cipher", "cpufeatures", - "opaque-debug", ] [[package]] name = "aho-corasick" -version = "0.7.20" +version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cc936419f96fa211c1b9166887b38e5e40b19958e5b895be7c1f93adec7071ac" +checksum = "67fc08ce920c31afb70f013dcce1bfc3a3195de6a228474e45e1f145b36f8d04" dependencies = [ "memchr", ] [[package]] name = "anyhow" -version = "1.0.70" +version = "1.0.71" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7de8ce5e0f9f8d88245311066a578d72b7af3e7088f32783804676302df237e4" +checksum = "9c7d0618f0e0b7e8ff11427422b64564d5fb0be1940354bfe2e0529b18a9d9b8" [[package]] name = "autocfg" @@ -43,9 +42,9 @@ checksum = "d468802bab17cbc0cc575e9b053f41e72aa36bfa6b7f55e3529ffa43161b97fa" [[package]] name = "base64" -version = "0.21.0" +version = "0.21.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a4a4ddaa51a5bc52a6948f74c06d20aaaddb71924eab79b8c97a8c556e942d6a" +checksum = "3f1e31e207a6b8fb791a38ea3105e6cb541f55e4d029902d3039a4ad07cc4105" [[package]] name = "base64ct" @@ -70,9 +69,9 @@ dependencies = [ [[package]] name = "bumpalo" -version = "3.12.0" +version = "3.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0d261e256854913907f67ed06efbc3338dfe6179796deefc1ff763fc1aee5535" +checksum = "a3e2c3daef883ecc1b5d58c15adae93470a91d425f3532ba1695849656af3fc1" [[package]] name = "byteorder" @@ -124,11 +123,12 @@ checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd" [[package]] name = "cipher" -version = "0.3.0" +version = "0.4.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7ee52072ec15386f770805afd189a01c8841be8696bed250fa2f13c4c0d6dfb7" +checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad" dependencies = [ - "generic-array", + "crypto-common", + "inout", ] [[package]] @@ -165,9 +165,9 @@ checksum = "e496a50fda8aacccc86d7529e2c1e0892dbd0f898a6b5645b5561b89c3210efa" [[package]] name = "cpufeatures" -version = "0.2.6" +version = "0.2.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "280a9f2d8b3a38871a3c8a46fb80db65e5e5ed97da80c4d08bf27fb63e35e181" +checksum = "3e4c1eaa2012c47becbbad2ab175484c2a84d1185b566fb2cc5b8707343dfe58" dependencies = [ "libc", ] @@ -224,7 +224,7 @@ dependencies = [ "proc-macro2", "quote", "scratch", - "syn 2.0.15", + "syn 2.0.16", ] [[package]] @@ -241,7 +241,7 @@ checksum = "2345488264226bf682893e25de0769f3360aac9957980ec49361b083ddaa5bc5" dependencies = [ "proc-macro2", "quote", - "syn 2.0.15", + "syn 2.0.16", ] [[package]] @@ -265,9 +265,9 @@ checksum = "c2e66c9d817f1720209181c316d28635c050fa304f9c79e47a520882661b7308" [[package]] name = "digest" -version = "0.10.6" +version = "0.10.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8168378f4e5023e7218c89c891c0fd8ecdb5e5e4f18cb78f38cf245dd021e76f" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" dependencies = [ "block-buffer", "crypto-common", @@ -352,9 +352,9 @@ checksum = "0ce7134b9999ecaf8bcd65542e436736ef32ddca1b3e06094cb6ec5755203b80" [[package]] name = "flate2" -version = "1.0.25" +version = "1.0.26" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a8a2db397cb1c8772f31494cb8917e48cd1e64f0fa7efac59fbd741a0a8ce841" +checksum = "3b9429470923de8e8cbd4d2dc513535400b4b3fef0319fb5c4e1f520a7bef743" dependencies = [ "crc32fast", "miniz_oxide", @@ -446,7 +446,7 @@ checksum = "89ca545a94061b6365f2c7355b4b32bd20df3ff95f02da9329b34ccc3bd6ee72" dependencies = [ "proc-macro2", "quote", - "syn 2.0.15", + "syn 2.0.16", ] [[package]] @@ -508,9 +508,9 @@ checksum = "d2fabcfbdc87f4758337ca535fb41a6d701b65693ce38287d856d1674551ec9b" [[package]] name = "h2" -version = "0.3.18" +version = "0.3.19" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "17f8a914c2987b688368b5138aa05321db91f4090cf26118185672ad588bce21" +checksum = "d357c7ae988e7d2182f7d7871d0b963962420b0678b0997ce7de72001aeab782" dependencies = [ "bytes", "fnv", @@ -658,6 +658,15 @@ dependencies = [ "hashbrown", ] +[[package]] +name = "inout" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a0c10553d664a4d0bcff9f4215d0aac67a639cc68ef660840afe309b807bc9f5" +dependencies = [ + "generic-array", +] + [[package]] name = "instant" version = "0.1.12" @@ -669,9 +678,9 @@ dependencies = [ [[package]] name = "io-lifetimes" -version = "1.0.10" +version = "1.0.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9c66c74d2ae7e79a5a8f7ac924adbe38ee42a859c6539ad869eb51f0b52dc220" +checksum = "eae7b9aee968036d54dce06cebaefd919e4472e753296daccd6d344e3e2df0c2" dependencies = [ "hermit-abi 0.3.1", "libc", @@ -722,9 +731,9 @@ dependencies = [ [[package]] name = "js-sys" -version = "0.3.61" +version = "0.3.63" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "445dde2150c55e483f3d8416706b97ec8e8237c307e5b7b4b8dd15e6af2a0730" +checksum = "2f37a4a5928311ac501dee68b3c7613a1037d0edb30c8e5427bd832d55d1b790" dependencies = [ "wasm-bindgen", ] @@ -749,9 +758,9 @@ checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646" [[package]] name = "libc" -version = "0.2.141" +version = "0.2.144" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3304a64d199bb964be99741b7a14d26972741915b3649639149b2479bb46f4b5" +checksum = "2b00cc1c228a6782d0f076e7b232802e0c5689d41bb5df366f2a6b6621cfdfe1" [[package]] name = "link-cplusplus" @@ -764,9 +773,9 @@ dependencies = [ [[package]] name = "linux-raw-sys" -version = "0.3.2" +version = "0.3.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3f508063cc7bb32987c71511216bd5a32be15bccb6a80b52df8b9d7f01fc3aa2" +checksum = "ef53942eb7bf7ff43a617b3e2c1c4a5ecf5944a7c1bc12d7ee39bbb15e5c1519" [[package]] name = "livekit" @@ -888,9 +897,9 @@ checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" [[package]] name = "miniz_oxide" -version = "0.6.2" +version = "0.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b275950c28b37e794e8c55d88aeb5e139d0ce23fdbbeda68f8d7174abdf9e8fa" +checksum = "e7810e0be55b428ada41041c41f32c9f1a42817901b4ccf45fa3d4b6561e74c7" dependencies = [ "adler", ] @@ -947,17 +956,11 @@ version = "1.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b7e5500299e16ebb147ae15a00a942af264cf3688f47923b8fc2cd5858f23ad3" -[[package]] -name = "opaque-debug" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "624a8340c38c1b80fd549087862da4ba43e08858af025b236e509b6649fc13d5" - [[package]] name = "openssl" -version = "0.10.50" +version = "0.10.52" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7e30d8bc91859781f0a943411186324d580f2bbeb71b452fe91ae344806af3f1" +checksum = "01b8574602df80f7b85fdfc5392fa884a4e3b3f4f35402c070ab34c3d3f78d56" dependencies = [ "bitflags", "cfg-if", @@ -976,7 +979,7 @@ checksum = "a948666b637a0f465e8564c73e89d4dde00d72d4d473cc972f390fc3dcee7d9c" dependencies = [ "proc-macro2", "quote", - "syn 2.0.15", + "syn 2.0.16", ] [[package]] @@ -987,9 +990,9 @@ checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf" [[package]] name = "openssl-sys" -version = "0.9.85" +version = "0.9.87" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0d3d193fb1488ad46ffe3aaabc912cc931d02ee8518fe2959aea8ef52718b0c0" +checksum = "8e17f59264b2809d77ae94f0e1ebabc434773f370d6ca667bd223ea10e06cc7e" dependencies = [ "cc", "libc", @@ -1073,9 +1076,9 @@ checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" [[package]] name = "pkg-config" -version = "0.3.26" +version = "0.3.27" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6ac9a59f73473f1b8d852421e59e64809f025994837ef743615c6d0c5b305160" +checksum = "26072860ba924cbfa98ea39c8c19b4dd6a4a25423dbdf219c1eca91aa0cf6964" [[package]] name = "ppv-lite86" @@ -1095,9 +1098,9 @@ dependencies = [ [[package]] name = "proc-macro2" -version = "1.0.56" +version = "1.0.58" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2b63bdb0cd06f1f4dedf69b254734f9b45af66e4a031e42a7480257d9898b435" +checksum = "fa1fb82fc0c281dd9671101b66b771ebbe1eaf967b96ac8740dcba4b70005ca8" dependencies = [ "unicode-ident", ] @@ -1158,9 +1161,9 @@ dependencies = [ [[package]] name = "quote" -version = "1.0.26" +version = "1.0.27" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4424af4bf778aae2051a77b60283332f386554255d722233d09fbfc7e30da2fc" +checksum = "8f4f29d145265ec1c483c7c654450edde0bfe043d3938d6972630663356d9500" dependencies = [ "proc-macro2", ] @@ -1215,9 +1218,9 @@ dependencies = [ [[package]] name = "regex" -version = "1.7.3" +version = "1.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b1f693b24f6ac912f4893ef08244d70b6067480d2f1a46e950c9691e6749d1d" +checksum = "d1a59b5d8e97dee33696bf13c5ba8ab85341c002922fba050069326b9c498974" dependencies = [ "aho-corasick", "memchr", @@ -1226,15 +1229,15 @@ dependencies = [ [[package]] name = "regex-syntax" -version = "0.6.29" +version = "0.7.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f162c6dd7b008981e4d40210aca20b4bd0f9b60ca9271061b07f78537722f2e1" +checksum = "436b050e76ed2903236f032a59761c1eb99e1b0aead2c257922771dab1fc8c78" [[package]] name = "reqwest" -version = "0.11.16" +version = "0.11.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "27b71749df584b7f4cac2c426c127a7c785a5106cc98f7a8feb044115f0fa254" +checksum = "cde824a14b7c14f85caff81225f411faacc04a2013f41670f41443742b1c1c55" dependencies = [ "base64", "bytes", @@ -1284,9 +1287,9 @@ dependencies = [ [[package]] name = "rustix" -version = "0.37.12" +version = "0.37.19" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "722529a737f5a942fdbac3a46cee213053196737c5eaa3386d52e85b786f2659" +checksum = "acf8729d8542766f1b2cf77eb034d52f40d375bb8b615d0b147089946e16613d" dependencies = [ "bitflags", "errno", @@ -1378,9 +1381,9 @@ dependencies = [ [[package]] name = "security-framework" -version = "2.8.2" +version = "2.9.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a332be01508d814fed64bf28f798a146d73792121129962fdf335bb3c49a4254" +checksum = "1fc758eb7bffce5b308734e9b0c1468893cae9ff70ebf13e7090be8dcbcc83a8" dependencies = [ "bitflags", "core-foundation", @@ -1391,9 +1394,9 @@ dependencies = [ [[package]] name = "security-framework-sys" -version = "2.8.0" +version = "2.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "31c9bb296072e961fcbd8853511dd39c2d8be2deb1e17c6860b1d30732b323b4" +checksum = "f51d0c0d83bec45f16480d0ce0058397a69e48fcdc52d1dc8855fb68acbd31a7" dependencies = [ "core-foundation-sys", "libc", @@ -1401,22 +1404,22 @@ dependencies = [ [[package]] name = "serde" -version = "1.0.160" +version = "1.0.163" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bb2f3770c8bce3bcda7e149193a069a0f4365bda1fa5cd88e03bca26afc1216c" +checksum = "2113ab51b87a539ae008b5c6c02dc020ffa39afd2d83cffcb3f4eb2722cebec2" dependencies = [ "serde_derive", ] [[package]] name = "serde_derive" -version = "1.0.160" +version = "1.0.163" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "291a097c63d8497e00160b166a967a4a79c64f3facdd01cbd7502231688d77df" +checksum = "8c805777e3930c8883389c602315a24224bcc738b63905ef87cd1420353ea93e" dependencies = [ "proc-macro2", "quote", - "syn 2.0.15", + "syn 2.0.16", ] [[package]] @@ -1506,9 +1509,9 @@ checksum = "6e63cff320ae2c57904679ba7cb63280a3dc4613885beafb148ee7bf9aa9042d" [[package]] name = "subtle" -version = "2.4.1" +version = "2.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6bdef32e8150c2a081110b42772ffe7d7c9032b606bc226c8260fd97e0976601" +checksum = "81cdd64d312baedb58e21336b31bc043b77e01cc99033ce76ef539f78e965ebc" [[package]] name = "syn" @@ -1523,9 +1526,9 @@ dependencies = [ [[package]] name = "syn" -version = "2.0.15" +version = "2.0.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a34fcf3e8b60f57e6a14301a2e916d323af98b0ea63c599441eec8558660c822" +checksum = "a6f671d4b5ffdb8eadec19c0ae67fe2639df8684bd7bc4b83d986b8db549cf01" dependencies = [ "proc-macro2", "quote", @@ -1582,14 +1585,14 @@ checksum = "f9456a42c5b0d803c8cd86e73dd7cc9edd429499f37a3550d286d5e86720569f" dependencies = [ "proc-macro2", "quote", - "syn 2.0.15", + "syn 2.0.16", ] [[package]] name = "time" -version = "0.3.20" +version = "0.3.21" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cd0cbfecb4d19b5ea75bb31ad904eb5b9fa13f21079c3b92017ebdf4999a5890" +checksum = "8f3403384eaacbca9923fa06940178ac13e4edb725486d70e8e15881d0c836cc" dependencies = [ "serde", "time-core", @@ -1597,9 +1600,9 @@ dependencies = [ [[package]] name = "time-core" -version = "0.1.0" +version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2e153e1f1acaef8acc537e68b44906d2db6436e2b35ac2c6b42640fff91f00fd" +checksum = "7300fbefb4dadc1af235a9cef3737cea692a9d97e1b9cbcd4ebdae6f8868e6fb" [[package]] name = "tinyvec" @@ -1618,9 +1621,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" [[package]] name = "tokio" -version = "1.27.0" +version = "1.28.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d0de47a4eecbe11f498978a9b29d792f0d2692d1dd003650c24c76510e3bc001" +checksum = "0aa32867d44e6f2ce3385e89dceb990188b8bb0fb25b0cf576647a6f98ac5105" dependencies = [ "autocfg", "bytes", @@ -1632,18 +1635,18 @@ dependencies = [ "signal-hook-registry", "socket2", "tokio-macros", - "windows-sys 0.45.0", + "windows-sys 0.48.0", ] [[package]] name = "tokio-macros" -version = "2.0.0" +version = "2.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "61a573bdc87985e9d6ddeed1b3d864e8a302c847e40d647746df2f1de209d1ce" +checksum = "630bdcf245f78637c13ec01ffae6187cca34625e8c63150d424b59e55af2675e" dependencies = [ "proc-macro2", "quote", - "syn 2.0.15", + "syn 2.0.16", ] [[package]] @@ -1668,9 +1671,9 @@ dependencies = [ [[package]] name = "tokio-stream" -version = "0.1.12" +version = "0.1.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8fb52b74f05dbf495a8fba459fdc331812b96aa086d9eb78101fa0d4569c3313" +checksum = "397c988d37662c7dda6d2208364a706264bf3d6138b11d436cbac0ad38832842" dependencies = [ "futures-core", "pin-project-lite", @@ -1694,9 +1697,9 @@ dependencies = [ [[package]] name = "tokio-util" -version = "0.7.7" +version = "0.7.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5427d89453009325de0d8f342c9490009f76e999cb7672d77e46267448f7e6b2" +checksum = "806fe8c2c87eccc8b3267cbae29ed3ab2d0bd37fca70ab622e46aaa9375ddb7d" dependencies = [ "bytes", "futures-core", @@ -1726,20 +1729,20 @@ dependencies = [ [[package]] name = "tracing-attributes" -version = "0.1.23" +version = "0.1.24" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4017f8f45139870ca7e672686113917c71c7a6e02d4924eda67186083c03081a" +checksum = "0f57e3ca2a01450b1a921183a9c9cbfda207fd822cef4ccb00a65402cbba7a74" dependencies = [ "proc-macro2", "quote", - "syn 1.0.109", + "syn 2.0.16", ] [[package]] name = "tracing-core" -version = "0.1.30" +version = "0.1.31" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "24eb03ba0eab1fd845050058ce5e616558e8f8d8fca633e6b163fe25c797213a" +checksum = "0955b8137a1df6f1a2e9a37d8a6656291ff0297c1a97c24e0d8425fe2312f79a" dependencies = [ "once_cell", ] @@ -1785,9 +1788,9 @@ checksum = "92888ba5573ff080736b3648696b70cafad7d250551175acbaa4e0385b3e1460" [[package]] name = "unicode-ident" -version = "1.0.8" +version = "1.0.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e5464a87b239f13a63a501f2701565754bae92d243d4bb7eb12f6d57d2269bf4" +checksum = "b15811caf2415fb889178633e7724bad2509101cde276048e013b9def5e51fa0" [[package]] name = "unicode-normalization" @@ -1857,9 +1860,9 @@ checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423" [[package]] name = "wasm-bindgen" -version = "0.2.84" +version = "0.2.86" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "31f8dcbc21f30d9b8f2ea926ecb58f6b91192c17e9d33594b3df58b2007ca53b" +checksum = "5bba0e8cb82ba49ff4e229459ff22a191bbe9a1cb3a341610c9c33efc27ddf73" dependencies = [ "cfg-if", "wasm-bindgen-macro", @@ -1867,24 +1870,24 @@ dependencies = [ [[package]] name = "wasm-bindgen-backend" -version = "0.2.84" +version = "0.2.86" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "95ce90fd5bcc06af55a641a86428ee4229e44e07033963a2290a8e241607ccb9" +checksum = "19b04bc93f9d6bdee709f6bd2118f57dd6679cf1176a1af464fca3ab0d66d8fb" dependencies = [ "bumpalo", "log", "once_cell", "proc-macro2", "quote", - "syn 1.0.109", + "syn 2.0.16", "wasm-bindgen-shared", ] [[package]] name = "wasm-bindgen-futures" -version = "0.4.34" +version = "0.4.36" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f219e0d211ba40266969f6dbdd90636da12f75bee4fc9d6c23d1260dadb51454" +checksum = "2d1985d03709c53167ce907ff394f5316aa22cb4e12761295c5dc57dacb6297e" dependencies = [ "cfg-if", "js-sys", @@ -1894,9 +1897,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro" -version = "0.2.84" +version = "0.2.86" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4c21f77c0bedc37fd5dc21f897894a5ca01e7bb159884559461862ae90c0b4c5" +checksum = "14d6b024f1a526bb0234f52840389927257beb670610081360e5a03c5df9c258" dependencies = [ "quote", "wasm-bindgen-macro-support", @@ -1904,28 +1907,28 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro-support" -version = "0.2.84" +version = "0.2.86" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2aff81306fcac3c7515ad4e177f521b5c9a15f2b08f4e32d823066102f35a5f6" +checksum = "e128beba882dd1eb6200e1dc92ae6c5dbaa4311aa7bb211ca035779e5efc39f8" dependencies = [ "proc-macro2", "quote", - "syn 1.0.109", + "syn 2.0.16", "wasm-bindgen-backend", "wasm-bindgen-shared", ] [[package]] name = "wasm-bindgen-shared" -version = "0.2.84" +version = "0.2.86" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0046fef7e28c3804e5e38bfa31ea2a0f73905319b677e57ebe37e49358989b5d" +checksum = "ed9d5b4305409d1fc9482fee2d7f9bcbf24b3972bf59817ef757e23982242a93" [[package]] name = "web-sys" -version = "0.3.61" +version = "0.3.63" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e33b99f4b23ba3eec1a53ac264e35a755f00e966e0065077d6027c0f575b0b97" +checksum = "3bdd9ef4e984da1187bf8110c5cf5b845fbc87a23602cdf912386a76fcd3a7c2" dependencies = [ "js-sys", "wasm-bindgen", @@ -2166,9 +2169,9 @@ dependencies = [ [[package]] name = "zip" -version = "0.6.4" +version = "0.6.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0445d0fbc924bb93539b4316c11afb121ea39296f99a3c4c9edad09e3658cdef" +checksum = "760394e246e4c28189f19d488c058bf16f564016aefac5d32bb1f3b51d5e9261" dependencies = [ "aes", "byteorder", diff --git a/examples/Cargo.lock b/examples/Cargo.lock index ca0b9ab..7c60bb5 100644 --- a/examples/Cargo.lock +++ b/examples/Cargo.lock @@ -35,14 +35,13 @@ checksum = "f26201604c87b1e01bd3d98f8d5d9a8fcbb815e8cedb41ffccbeb4bf593a35fe" [[package]] name = "aes" -version = "0.7.5" +version = "0.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9e8b47f52ea9bae42228d07ec09eb676433d7c4ed1ebdf0f1d1c29ed446f1ab8" +checksum = "433cfd6710c9986c576a25ca913c39d66a6474107b406f34f91d4a8923395241" dependencies = [ "cfg-if", "cipher", "cpufeatures", - "opaque-debug", ] [[package]] @@ -183,9 +182,9 @@ dependencies = [ [[package]] name = "base64" -version = "0.21.0" +version = "0.21.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a4a4ddaa51a5bc52a6948f74c06d20aaaddb71924eab79b8c97a8c556e942d6a" +checksum = "3f1e31e207a6b8fb791a38ea3105e6cb541f55e4d029902d3039a4ad07cc4105" [[package]] name = "base64ct" @@ -232,9 +231,9 @@ checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a" [[package]] name = "bitflags" -version = "2.2.1" +version = "2.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "24a6904aef64d73cf10ab17ebace7befb918b82164785cb89907993be7f83813" +checksum = "6776fc96284a0bb647b615056fc496d1fe1644a7ab01829818a6d91cae888b84" [[package]] name = "block" @@ -272,9 +271,9 @@ dependencies = [ [[package]] name = "bumpalo" -version = "3.12.1" +version = "3.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9b1ce199063694f33ffb7dd4e0ee620741495c32833cde5aa08f02a0bf96f0c8" +checksum = "a3e2c3daef883ecc1b5d58c15adae93470a91d425f3532ba1695849656af3fc1" [[package]] name = "bytemuck" @@ -293,7 +292,7 @@ checksum = "fdde5c9cd29ebd706ce1b35600920a33550e402fc998a2e53ad3b42c3c47a192" dependencies = [ "proc-macro2", "quote", - "syn 2.0.15", + "syn 2.0.16", ] [[package]] @@ -371,11 +370,12 @@ checksum = "fd16c4719339c4530435d38e511904438d07cce7950afa3718a84ac36c10e89e" [[package]] name = "cipher" -version = "0.3.0" +version = "0.4.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7ee52072ec15386f770805afd189a01c8841be8696bed250fa2f13c4c0d6dfb7" +checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad" dependencies = [ - "generic-array", + "crypto-common", + "inout", ] [[package]] @@ -569,7 +569,7 @@ dependencies = [ "proc-macro2", "quote", "scratch", - "syn 2.0.15", + "syn 2.0.16", ] [[package]] @@ -586,7 +586,7 @@ checksum = "2345488264226bf682893e25de0769f3360aac9957980ec49361b083ddaa5bc5" dependencies = [ "proc-macro2", "quote", - "syn 2.0.15", + "syn 2.0.16", ] [[package]] @@ -608,35 +608,15 @@ checksum = "c2e66c9d817f1720209181c316d28635c050fa304f9c79e47a520882661b7308" [[package]] name = "digest" -version = "0.10.6" +version = "0.10.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8168378f4e5023e7218c89c891c0fd8ecdb5e5e4f18cb78f38cf245dd021e76f" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" dependencies = [ "block-buffer", "crypto-common", "subtle", ] -[[package]] -name = "dirs" -version = "4.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ca3aa72a6f96ea37bbc5aa912f6788242832f75369bdfdadcb0e38423f100059" -dependencies = [ - "dirs-sys", -] - -[[package]] -name = "dirs-sys" -version = "0.3.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1b1d1d91c932ef41c0f2663aa8b0ca0342d444d842c06914aa0a7e352d0bada6" -dependencies = [ - "libc", - "redox_users", - "winapi", -] - [[package]] name = "dispatch" version = "0.2.0" @@ -660,16 +640,16 @@ checksum = "9ea835d29036a4087793836fa931b08837ad5e957da9e23886b29586fb9b6650" [[package]] name = "ecolor" -version = "0.21.0" -source = "git+https://github.com/emilk/egui#e9fa6c8ff68a0257ce0f2801446ba701917b7ae9" +version = "0.22.0" +source = "git+https://github.com/emilk/egui#307565efa55158cfa6b82d2e8fdc4c4914b954ed" dependencies = [ "bytemuck", ] [[package]] name = "egui" -version = "0.21.0" -source = "git+https://github.com/emilk/egui#e9fa6c8ff68a0257ce0f2801446ba701917b7ae9" +version = "0.22.0" +source = "git+https://github.com/emilk/egui#307565efa55158cfa6b82d2e8fdc4c4914b954ed" dependencies = [ "ahash 0.8.3", "epaint", @@ -679,8 +659,8 @@ dependencies = [ [[package]] name = "egui-wgpu" -version = "0.21.0" -source = "git+https://github.com/emilk/egui#e9fa6c8ff68a0257ce0f2801446ba701917b7ae9" +version = "0.22.0" +source = "git+https://github.com/emilk/egui#307565efa55158cfa6b82d2e8fdc4c4914b954ed" dependencies = [ "bytemuck", "epaint", @@ -693,8 +673,8 @@ dependencies = [ [[package]] name = "egui-winit" -version = "0.21.1" -source = "git+https://github.com/emilk/egui#e9fa6c8ff68a0257ce0f2801446ba701917b7ae9" +version = "0.22.0" +source = "git+https://github.com/emilk/egui#307565efa55158cfa6b82d2e8fdc4c4914b954ed" dependencies = [ "arboard", "egui", @@ -714,8 +694,8 @@ checksum = "7fcaabb2fef8c910e7f4c7ce9f67a1283a1715879a7c230ca9d6d1ae31f16d91" [[package]] name = "emath" -version = "0.21.0" -source = "git+https://github.com/emilk/egui#e9fa6c8ff68a0257ce0f2801446ba701917b7ae9" +version = "0.22.0" +source = "git+https://github.com/emilk/egui#307565efa55158cfa6b82d2e8fdc4c4914b954ed" dependencies = [ "bytemuck", ] @@ -744,8 +724,8 @@ dependencies = [ [[package]] name = "epaint" -version = "0.21.0" -source = "git+https://github.com/emilk/egui#e9fa6c8ff68a0257ce0f2801446ba701917b7ae9" +version = "0.22.0" +source = "git+https://github.com/emilk/egui#307565efa55158cfa6b82d2e8fdc4c4914b954ed" dependencies = [ "ab_glyph", "ahash 0.8.3", @@ -950,7 +930,7 @@ checksum = "89ca545a94061b6365f2c7355b4b32bd20df3ff95f02da9329b34ccc3bd6ee72" dependencies = [ "proc-macro2", "quote", - "syn 2.0.15", + "syn 2.0.16", ] [[package]] @@ -1104,9 +1084,9 @@ dependencies = [ [[package]] name = "h2" -version = "0.3.18" +version = "0.3.19" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "17f8a914c2987b688368b5138aa05321db91f4090cf26118185672ad588bce21" +checksum = "d357c7ae988e7d2182f7d7871d0b963962420b0678b0997ce7de72001aeab782" dependencies = [ "bytes", "fnv", @@ -1190,6 +1170,15 @@ dependencies = [ "digest", ] +[[package]] +name = "home" +version = "0.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5444c27eef6923071f7ebcc33e3444508466a76f7a2b93da00ed6e19f30c1ddb" +dependencies = [ + "windows-sys 0.48.0", +] + [[package]] name = "http" version = "0.2.9" @@ -1306,6 +1295,15 @@ dependencies = [ "hashbrown", ] +[[package]] +name = "inout" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a0c10553d664a4d0bcff9f4215d0aac67a639cc68ef660840afe309b807bc9f5" +dependencies = [ + "generic-array", +] + [[package]] name = "instant" version = "0.1.12" @@ -1320,9 +1318,9 @@ dependencies = [ [[package]] name = "io-lifetimes" -version = "1.0.10" +version = "1.0.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9c66c74d2ae7e79a5a8f7ac924adbe38ee42a859c6539ad869eb51f0b52dc220" +checksum = "eae7b9aee968036d54dce06cebaefd919e4472e753296daccd6d344e3e2df0c2" dependencies = [ "hermit-abi 0.3.1", "libc", @@ -1404,9 +1402,9 @@ dependencies = [ [[package]] name = "js-sys" -version = "0.3.61" +version = "0.3.63" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "445dde2150c55e483f3d8416706b97ec8e8237c307e5b7b4b8dd15e6af2a0730" +checksum = "2f37a4a5928311ac501dee68b3c7613a1037d0edb30c8e5427bd832d55d1b790" dependencies = [ "wasm-bindgen", ] @@ -1471,9 +1469,9 @@ dependencies = [ [[package]] name = "linux-raw-sys" -version = "0.3.7" +version = "0.3.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ece97ea872ece730aed82664c424eb4c8291e1ff2480247ccf7409044bc6479f" +checksum = "ef53942eb7bf7ff43a617b3e2c1c4a5ecf5944a7c1bc12d7ee39bbb15e5c1519" [[package]] name = "livekit" @@ -1652,9 +1650,9 @@ checksum = "e5ce46fe64a9d73be07dcbe690a38ce1b293be448fd8ce1e6c1b8062c9f72c6a" [[package]] name = "naga" -version = "0.12.0" +version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f00ce114f2867153c079d4489629dbd27aa4b5387a8ba5341bd3f6dfe870688f" +checksum = "94d3edd593521f4a1dfd9b25193ed0224764572905f013d30ca5fbb85e010876" dependencies = [ "bit-set", "bitflags 1.3.2", @@ -1918,12 +1916,6 @@ version = "1.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b7e5500299e16ebb147ae15a00a942af264cf3688f47923b8fc2cd5858f23ad3" -[[package]] -name = "opaque-debug" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "624a8340c38c1b80fd549087862da4ba43e08858af025b236e509b6649fc13d5" - [[package]] name = "openssl" version = "0.10.52" @@ -1947,7 +1939,7 @@ checksum = "a948666b637a0f465e8564c73e89d4dde00d72d4d473cc972f390fc3dcee7d9c" dependencies = [ "proc-macro2", "quote", - "syn 2.0.15", + "syn 2.0.16", ] [[package]] @@ -2056,22 +2048,22 @@ dependencies = [ [[package]] name = "pin-project" -version = "1.0.12" +version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ad29a609b6bcd67fee905812e544992d216af9d755757c05ed2d0e15a74c6ecc" +checksum = "c95a7476719eab1e366eaf73d0260af3021184f18177925b07f54b30089ceead" dependencies = [ "pin-project-internal", ] [[package]] name = "pin-project-internal" -version = "1.0.12" +version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "069bdb1e05adc7a8990dce9cc75370895fbe4e3d58b9b73bf1aee56359344a55" +checksum = "39407670928234ebc5e6e580247dd567ad73a3578460c5990f9503df207e8f07" dependencies = [ "proc-macro2", "quote", - "syn 1.0.109", + "syn 2.0.16", ] [[package]] @@ -2133,9 +2125,9 @@ dependencies = [ [[package]] name = "proc-macro2" -version = "1.0.56" +version = "1.0.58" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2b63bdb0cd06f1f4dedf69b254734f9b45af66e4a031e42a7480257d9898b435" +checksum = "fa1fb82fc0c281dd9671101b66b771ebbe1eaf967b96ac8740dcba4b70005ca8" dependencies = [ "unicode-ident", ] @@ -2300,22 +2292,11 @@ dependencies = [ "bitflags 1.3.2", ] -[[package]] -name = "redox_users" -version = "0.4.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b033d837a7cf162d7993aded9304e30a83213c648b6e389db233191f891e5c2b" -dependencies = [ - "getrandom", - "redox_syscall 0.2.16", - "thiserror", -] - [[package]] name = "regex" -version = "1.8.1" +version = "1.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "af83e617f331cc6ae2da5443c602dfa5af81e517212d9d611a5b3ba1777b5370" +checksum = "d1a59b5d8e97dee33696bf13c5ba8ab85341c002922fba050069326b9c498974" dependencies = [ "aho-corasick", "memchr", @@ -2324,9 +2305,9 @@ dependencies = [ [[package]] name = "regex-syntax" -version = "0.7.1" +version = "0.7.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a5996294f19bd3aae0453a862ad728f60e6600695733dd5df01da90c54363a3c" +checksum = "436b050e76ed2903236f032a59761c1eb99e1b0aead2c257922771dab1fc8c78" [[package]] name = "renderdoc-sys" @@ -2336,9 +2317,9 @@ checksum = "216080ab382b992234dda86873c18d4c48358f5cfcb70fd693d7f6f2131b628b" [[package]] name = "reqwest" -version = "0.11.17" +version = "0.11.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "13293b639a097af28fc8a90f22add145a9c954e49d77da06263d58cf44d5fb91" +checksum = "cde824a14b7c14f85caff81225f411faacc04a2013f41670f41443742b1c1c55" dependencies = [ "base64", "bytes", @@ -2465,9 +2446,9 @@ dependencies = [ [[package]] name = "security-framework" -version = "2.8.2" +version = "2.9.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a332be01508d814fed64bf28f798a146d73792121129962fdf335bb3c49a4254" +checksum = "1fc758eb7bffce5b308734e9b0c1468893cae9ff70ebf13e7090be8dcbcc83a8" dependencies = [ "bitflags 1.3.2", "core-foundation", @@ -2478,9 +2459,9 @@ dependencies = [ [[package]] name = "security-framework-sys" -version = "2.8.0" +version = "2.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "31c9bb296072e961fcbd8853511dd39c2d8be2deb1e17c6860b1d30732b323b4" +checksum = "f51d0c0d83bec45f16480d0ce0058397a69e48fcdc52d1dc8855fb68acbd31a7" dependencies = [ "core-foundation-sys", "libc", @@ -2488,22 +2469,22 @@ dependencies = [ [[package]] name = "serde" -version = "1.0.162" +version = "1.0.163" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "71b2f6e1ab5c2b98c05f0f35b236b22e8df7ead6ffbf51d7808da7f8817e7ab6" +checksum = "2113ab51b87a539ae008b5c6c02dc020ffa39afd2d83cffcb3f4eb2722cebec2" dependencies = [ "serde_derive", ] [[package]] name = "serde_derive" -version = "1.0.162" +version = "1.0.163" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a2a0814352fd64b58489904a44ea8d90cb1a91dcb6b4f5ebabc32c8318e93cb6" +checksum = "8c805777e3930c8883389c602315a24224bcc738b63905ef87cd1420353ea93e" dependencies = [ "proc-macro2", "quote", - "syn 2.0.15", + "syn 2.0.16", ] [[package]] @@ -2677,9 +2658,9 @@ checksum = "9df65f20698aeed245efdde3628a6b559ea1239bbb871af1b6e3b58c413b2bd1" [[package]] name = "subtle" -version = "2.4.1" +version = "2.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6bdef32e8150c2a081110b42772ffe7d7c9032b606bc226c8260fd97e0976601" +checksum = "81cdd64d312baedb58e21336b31bc043b77e01cc99033ce76ef539f78e965ebc" [[package]] name = "syn" @@ -2694,9 +2675,9 @@ dependencies = [ [[package]] name = "syn" -version = "2.0.15" +version = "2.0.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a34fcf3e8b60f57e6a14301a2e916d323af98b0ea63c599441eec8558660c822" +checksum = "a6f671d4b5ffdb8eadec19c0ae67fe2639df8684bd7bc4b83d986b8db549cf01" dependencies = [ "proc-macro2", "quote", @@ -2753,7 +2734,7 @@ checksum = "f9456a42c5b0d803c8cd86e73dd7cc9edd429499f37a3550d286d5e86720569f" dependencies = [ "proc-macro2", "quote", - "syn 2.0.15", + "syn 2.0.16", ] [[package]] @@ -2835,9 +2816,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" [[package]] name = "tokio" -version = "1.28.0" +version = "1.28.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c3c786bf8134e5a3a166db9b29ab8f48134739014a3eca7bc6bfa95d673b136f" +checksum = "0aa32867d44e6f2ce3385e89dceb990188b8bb0fb25b0cf576647a6f98ac5105" dependencies = [ "autocfg", "bytes", @@ -2860,7 +2841,7 @@ checksum = "630bdcf245f78637c13ec01ffae6187cca34625e8c63150d424b59e55af2675e" dependencies = [ "proc-macro2", "quote", - "syn 2.0.15", + "syn 2.0.16", ] [[package]] @@ -2914,15 +2895,15 @@ dependencies = [ [[package]] name = "toml_datetime" -version = "0.6.1" +version = "0.6.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3ab8ed2edee10b50132aed5f331333428b011c99402b5a534154ed15746f9622" +checksum = "5a76a9312f5ba4c2dec6b9161fdf25d87ad8a09256ccea5a556fef03c706a10f" [[package]] name = "toml_edit" -version = "0.19.8" +version = "0.19.10" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "239410c8609e8125456927e6707163a3b1fdb40561e4b803bc041f466ccfdc13" +checksum = "2380d56e8670370eee6566b0bfd4265f65b3f432e8c6d85623f728d4fa31f739" dependencies = [ "indexmap", "toml_datetime", @@ -2955,14 +2936,14 @@ checksum = "0f57e3ca2a01450b1a921183a9c9cbfda207fd822cef4ccb00a65402cbba7a74" dependencies = [ "proc-macro2", "quote", - "syn 2.0.15", + "syn 2.0.16", ] [[package]] name = "tracing-core" -version = "0.1.30" +version = "0.1.31" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "24eb03ba0eab1fd845050058ce5e616558e8f8d8fca633e6b163fe25c797213a" +checksum = "0955b8137a1df6f1a2e9a37d8a6656291ff0297c1a97c24e0d8425fe2312f79a" dependencies = [ "once_cell", "valuable", @@ -3048,9 +3029,9 @@ checksum = "92888ba5573ff080736b3648696b70cafad7d250551175acbaa4e0385b3e1460" [[package]] name = "unicode-ident" -version = "1.0.8" +version = "1.0.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e5464a87b239f13a63a501f2701565754bae92d243d4bb7eb12f6d57d2269bf4" +checksum = "b15811caf2415fb889178633e7724bad2509101cde276048e013b9def5e51fa0" [[package]] name = "unicode-normalization" @@ -3142,9 +3123,9 @@ checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423" [[package]] name = "wasm-bindgen" -version = "0.2.84" +version = "0.2.86" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "31f8dcbc21f30d9b8f2ea926ecb58f6b91192c17e9d33594b3df58b2007ca53b" +checksum = "5bba0e8cb82ba49ff4e229459ff22a191bbe9a1cb3a341610c9c33efc27ddf73" dependencies = [ "cfg-if", "wasm-bindgen-macro", @@ -3152,24 +3133,24 @@ dependencies = [ [[package]] name = "wasm-bindgen-backend" -version = "0.2.84" +version = "0.2.86" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "95ce90fd5bcc06af55a641a86428ee4229e44e07033963a2290a8e241607ccb9" +checksum = "19b04bc93f9d6bdee709f6bd2118f57dd6679cf1176a1af464fca3ab0d66d8fb" dependencies = [ "bumpalo", "log", "once_cell", "proc-macro2", "quote", - "syn 1.0.109", + "syn 2.0.16", "wasm-bindgen-shared", ] [[package]] name = "wasm-bindgen-futures" -version = "0.4.34" +version = "0.4.36" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f219e0d211ba40266969f6dbdd90636da12f75bee4fc9d6c23d1260dadb51454" +checksum = "2d1985d03709c53167ce907ff394f5316aa22cb4e12761295c5dc57dacb6297e" dependencies = [ "cfg-if", "js-sys", @@ -3179,9 +3160,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro" -version = "0.2.84" +version = "0.2.86" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4c21f77c0bedc37fd5dc21f897894a5ca01e7bb159884559461862ae90c0b4c5" +checksum = "14d6b024f1a526bb0234f52840389927257beb670610081360e5a03c5df9c258" dependencies = [ "quote", "wasm-bindgen-macro-support", @@ -3189,22 +3170,22 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro-support" -version = "0.2.84" +version = "0.2.86" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2aff81306fcac3c7515ad4e177f521b5c9a15f2b08f4e32d823066102f35a5f6" +checksum = "e128beba882dd1eb6200e1dc92ae6c5dbaa4311aa7bb211ca035779e5efc39f8" dependencies = [ "proc-macro2", "quote", - "syn 1.0.109", + "syn 2.0.16", "wasm-bindgen-backend", "wasm-bindgen-shared", ] [[package]] name = "wasm-bindgen-shared" -version = "0.2.84" +version = "0.2.86" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0046fef7e28c3804e5e38bfa31ea2a0f73905319b677e57ebe37e49358989b5d" +checksum = "ed9d5b4305409d1fc9482fee2d7f9bcbf24b3972bf59817ef757e23982242a93" [[package]] name = "wayland-client" @@ -3281,9 +3262,9 @@ dependencies = [ [[package]] name = "web-sys" -version = "0.3.61" +version = "0.3.63" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e33b99f4b23ba3eec1a53ac264e35a755f00e966e0065077d6027c0f575b0b97" +checksum = "3bdd9ef4e984da1187bf8110c5cf5b845fbc87a23602cdf912386a76fcd3a7c2" dependencies = [ "js-sys", "wasm-bindgen", @@ -3291,12 +3272,12 @@ dependencies = [ [[package]] name = "webbrowser" -version = "0.8.9" +version = "0.8.10" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b692165700260bbd40fbc5ff23766c03e339fbaca907aeea5cb77bf0a553ca83" +checksum = "fd222aa310eb7532e3fd427a5d7db7e44bc0b0cf1c1e21139c345325511a85b6" dependencies = [ "core-foundation", - "dirs", + "home", "jni", "log", "ndk-context", @@ -3329,9 +3310,9 @@ checksum = "9193164d4de03a926d909d3bc7c30543cecb35400c02114792c2cae20d5e2dbb" [[package]] name = "wgpu" -version = "0.16.0" +version = "0.16.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "13edd72c7b08615b7179dd7e778ee3f0bdc870ef2de9019844ff2cceeee80b11" +checksum = "3059ea4ddec41ca14f356833e2af65e7e38c0a8f91273867ed526fb9bafcca95" dependencies = [ "arrayvec", "cfg-if", @@ -3353,13 +3334,13 @@ dependencies = [ [[package]] name = "wgpu-core" -version = "0.16.0" +version = "0.16.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "625bea30a0ba50d88025f95c80211d1a85c86901423647fb74f397f614abbd9a" +checksum = "8f478237b4bf0d5b70a39898a66fa67ca3a007d79f2520485b8b0c3dfc46f8c2" dependencies = [ "arrayvec", "bit-vec", - "bitflags 2.2.1", + "bitflags 2.3.1", "codespan-reporting", "log", "naga", @@ -3384,7 +3365,7 @@ dependencies = [ "arrayvec", "ash", "bit-set", - "bitflags 2.2.1", + "bitflags 2.3.1", "block", "core-graphics-types", "d3d12", @@ -3422,7 +3403,7 @@ version = "0.16.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5bd33a976130f03dcdcd39b3810c0c3fc05daf86f0aaf867db14bfb7c4a9a32b" dependencies = [ - "bitflags 2.2.1", + "bitflags 2.3.1", "js-sys", "web-sys", ] @@ -3660,9 +3641,9 @@ checksum = "1a515f5799fe4961cb532f983ce2b23082366b898e52ffbce459c86f67c8378a" [[package]] name = "winit" -version = "0.28.5" +version = "0.28.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "94c9651471cd576737671fbf7081edfea43de3e06846dd9bd4e49ea803c9f55f" +checksum = "866db3f712fffba75d31bf0cdecf357c8aeafd158c5b7ab51dba2a2b2d47f196" dependencies = [ "android-activity", "bitflags 1.3.2", @@ -3764,15 +3745,15 @@ dependencies = [ [[package]] name = "xml-rs" -version = "0.8.8" +version = "0.8.13" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4f20f14e2bd1fef6ec891d50dbb37c7290a0568a6bbd9020bf62010d02b6f80e" +checksum = "2d8f380ae16a37b30e6a2cf67040608071384b1450c189e61bea3ff57cde922d" [[package]] name = "zip" -version = "0.6.5" +version = "0.6.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7e92305c174683d78035cbf1b70e18db6329cc0f1b9cae0a52ca90bf5bfe7125" +checksum = "760394e246e4c28189f19d488c058bf16f564016aefac5d32bb1f3b51d5e9261" dependencies = [ "aes", "byteorder", diff --git a/examples/basic_room/src/main.rs b/examples/basic_room/src/main.rs index 0c27c0a..32f2f40 100644 --- a/examples/basic_room/src/main.rs +++ b/examples/basic_room/src/main.rs @@ -12,8 +12,7 @@ async fn main() { let token = env::var("LIVEKIT_TOKEN").expect("LIVEKIT_TOKEN is not set"); let (room, mut rx) = Room::connect(&url, &token).await.unwrap(); - let session = room.session(); - log::info!("Connected to room: {} - {}", session.name(), session.sid()); + log::info!("Connected to room: {} - {}", room.name(), room.sid()); while let Some(msg) = rx.recv().await { log::info!("Event: {:?}", msg); diff --git a/examples/save_to_disk/src/main.rs b/examples/save_to_disk/src/main.rs index a9499f6..c7b6e30 100644 --- a/examples/save_to_disk/src/main.rs +++ b/examples/save_to_disk/src/main.rs @@ -90,8 +90,7 @@ async fn main() { let token = env::var("LIVEKIT_TOKEN").expect("LIVEKIT_TOKEN is not set"); let (room, mut rx) = Room::connect(&url, &token).await.unwrap(); - let session = room.session(); - println!("Connected to room: {} - {}", session.name(), session.sid()); + println!("Connected to room: {} - {}", room.name(), room.sid()); while let Some(msg) = rx.recv().await { match msg { diff --git a/examples/wgpu_room/Cargo.toml b/examples/wgpu_room/Cargo.toml index 1753798..ea9cea9 100644 --- a/examples/wgpu_room/Cargo.toml +++ b/examples/wgpu_room/Cargo.toml @@ -12,9 +12,7 @@ futures = "0.3" wgpu = "0.16" winit = "0.28" parking_lot = "0.12.1" - +image = "0.24" egui = { git = "https://github.com/emilk/egui" } egui-wgpu = { git = "https://github.com/emilk/egui", features = ["winit"] } egui-winit = { git = "https://github.com/emilk/egui" } - -image = "0.24" diff --git a/examples/wgpu_room/src/app.rs b/examples/wgpu_room/src/app.rs index 544913b..026cf37 100644 --- a/examples/wgpu_room/src/app.rs +++ b/examples/wgpu_room/src/app.rs @@ -6,14 +6,8 @@ use crate::{events::AsyncCmd, video_grid::VideoGrid}; use egui::{Rounding, Stroke}; use egui_wgpu::WgpuConfiguration; use futures::StreamExt; - - use livekit::prelude::*; use livekit::webrtc::audio_stream::native::NativeAudioStream; - -use livekit::webrtc::video_frame::native::I420BufferExt; - - use livekit::SimulateScenario; use parking_lot::Mutex; use std::collections::HashMap; @@ -37,7 +31,7 @@ use winit::{ }; struct Session { - room: Room, + room: Arc, logo_track: LogoTrack, sine_track: SineTrack, close_tx: oneshot::Sender<()>, @@ -114,9 +108,10 @@ pub fn run(rt: tokio::runtime::Runtime) { let res = Room::connect(&url, &token).await; if let Ok((room, room_events)) = res { + let room = Arc::new(room); let (close_tx, close_rx) = oneshot::channel(); - let logo_track = LogoTrack::new(room.session()); - let sine_track = SineTrack::new(room.session()); + let logo_track = LogoTrack::new(room.clone()); + let sine_track = SineTrack::new(room.clone()); let handle = tokio::spawn(room_task( state.clone(), room_events, @@ -148,7 +143,7 @@ pub fn run(rt: tokio::runtime::Runtime) { } AsyncCmd::SimulateScenario { scenario } => { if let Some(session) = state.session.lock().as_ref() { - let _ = session.room.session().simulate_scenario(scenario).await; + let _ = session.room.simulate_scenario(scenario).await; } } AsyncCmd::ToggleLogo => { @@ -164,7 +159,11 @@ pub fn run(rt: tokio::runtime::Runtime) { AsyncCmd::ToggleSine => { if let Some(session) = state.session.lock().as_mut() { let sine_track = &mut session.sine_track; - sine_track.publish().await.unwrap(); + if !sine_track.is_published() { + sine_track.publish().await.unwrap(); + } else { + sine_track.unpublish().await.unwrap(); + } } } } @@ -401,15 +400,15 @@ impl App { { // Room Info if let Some(session) = self.state.session.lock().as_ref() { - ui.label(format!("Name: {}", session.room.session().name())); - ui.label(format!("SID: {}", session.room.session().sid())); + ui.label(format!("Name: {}", session.room.name())); + ui.label(format!("SID: {}", session.room.sid())); ui.label(format!( "ConnectionState: {:?}", - session.room.session().connection_state() + session.room.connection_state() )); ui.label(format!( "ParticipantCount: {:?}", - session.room.session().participants().len() + 1 + session.room.participants().len() + 1 )); } } @@ -459,7 +458,6 @@ impl App { self.state.session.lock().as_ref().and_then(|session| { session .room - .session() .participants() .get(participant_sid) .map(|p| p.name()) diff --git a/examples/wgpu_room/src/logo_track.rs b/examples/wgpu_room/src/logo_track.rs index 94c1861..9be3d59 100644 --- a/examples/wgpu_room/src/logo_track.rs +++ b/examples/wgpu_room/src/logo_track.rs @@ -38,15 +38,15 @@ struct TrackHandle { pub struct LogoTrack { rtc_source: NativeVideoSource, - session: RoomSession, + room: Arc, handle: Option, } impl LogoTrack { - pub fn new(session: RoomSession) -> Self { + pub fn new(room: Arc) -> Self { Self { rtc_source: NativeVideoSource::default(), - session, + room, handle: None, } } @@ -67,7 +67,7 @@ impl LogoTrack { let task = tokio::spawn(Self::track_task(close_rx, self.rtc_source.clone())); - self.session + self.room .local_participant() .publish_track( LocalTrack::Video(track.clone()), @@ -93,7 +93,7 @@ impl LogoTrack { let _ = handle.close_tx.send(()); let _ = handle.task.await; - self.session + self.room .local_participant() .unpublish_track(handle.track.sid(), true) .await?; diff --git a/examples/wgpu_room/src/sine_track.rs b/examples/wgpu_room/src/sine_track.rs index ba5e606..4148790 100644 --- a/examples/wgpu_room/src/sine_track.rs +++ b/examples/wgpu_room/src/sine_track.rs @@ -1,7 +1,6 @@ use livekit::options::{AudioCaptureOptions, TrackPublishOptions}; use livekit::webrtc::audio_frame::AudioFrame; use livekit::{prelude::*, webrtc::audio_source::native::NativeAudioSource}; -use parking_lot::Mutex; use std::sync::Arc; use std::time::Duration; use tokio::sync::oneshot; @@ -27,7 +26,6 @@ impl Default for FrameData { } struct TrackHandle { - frame_data: Arc>, close_tx: oneshot::Sender<()>, track: LocalAudioTrack, task: JoinHandle<()>, @@ -35,19 +33,23 @@ struct TrackHandle { pub struct SineTrack { rtc_source: NativeAudioSource, - session: RoomSession, + room: Arc, handle: Option, } impl SineTrack { - pub fn new(session: RoomSession) -> Self { + pub fn new(room: Arc) -> Self { Self { rtc_source: NativeAudioSource::default(), - session, + room, handle: None, } } + pub fn is_published(&self) -> bool { + self.handle.is_some() + } + pub async fn publish(&mut self) -> Result<(), RoomError> { let (close_tx, close_rx) = oneshot::channel(); let track = LocalAudioTrack::create_audio_track( @@ -60,14 +62,9 @@ impl SineTrack { self.rtc_source.clone(), ); - let data = Arc::new(Mutex::new(FrameData::default())); - let task = tokio::spawn(Self::track_task( - close_rx, - self.rtc_source.clone(), - data.clone(), - )); + let task = tokio::spawn(Self::track_task(close_rx, self.rtc_source.clone())); - self.session + self.room .local_participant() .publish_track( LocalTrack::Audio(track.clone()), @@ -79,7 +76,6 @@ impl SineTrack { .await?; let handle = TrackHandle { - frame_data: data, close_tx, track, task, @@ -89,11 +85,21 @@ impl SineTrack { Ok(()) } - async fn track_task( - _close_rx: oneshot::Receiver<()>, - rtc_source: NativeAudioSource, - frame_options: Arc>, - ) { + pub async fn unpublish(&mut self) -> Result<(), RoomError> { + if let Some(handle) = self.handle.take() { + handle.close_tx.send(()).ok(); + handle.task.await.ok(); + self.room + .local_participant() + .unpublish_track(handle.track.sid(), true) + .await?; + } + + Ok(()) + } + + async fn track_task(_close_rx: oneshot::Receiver<()>, rtc_source: NativeAudioSource) { + let mut data = FrameData::default(); let mut interval = tokio::time::interval(Duration::from_millis(10)); let mut samples_10ms = Vec::::new(); @@ -102,7 +108,6 @@ impl SineTrack { interval.tick().await; - let mut data = frame_options.lock(); let samples_count_10ms = (data.sample_rate / 100) as usize * NUM_CHANNELS; if samples_10ms.capacity() != samples_count_10ms { diff --git a/examples/wgpu_room/src/video_grid.rs b/examples/wgpu_room/src/video_grid.rs index 30f509c..4219191 100644 --- a/examples/wgpu_room/src/video_grid.rs +++ b/examples/wgpu_room/src/video_grid.rs @@ -141,6 +141,7 @@ impl VideoGrid { } } +#[allow(dead_code)] impl VideoGrid { pub fn min_video_size(mut self, min_video_size: egui::Vec2) -> Self { self.min_video_size = min_video_size; diff --git a/examples/wgpu_room/src/video_renderer.rs b/examples/wgpu_room/src/video_renderer.rs index 5226783..00e741e 100644 --- a/examples/wgpu_room/src/video_renderer.rs +++ b/examples/wgpu_room/src/video_renderer.rs @@ -2,15 +2,14 @@ use futures::StreamExt; use livekit::webrtc::native::yuv_helper; use livekit::webrtc::prelude::*; use livekit::webrtc::video_stream::native::NativeVideoStream; - -use std::{ - ops::DerefMut, - sync::{Arc, Mutex}, -}; +use parking_lot::Mutex; +use std::{ops::DerefMut, sync::Arc}; use tracing::debug_span; pub struct VideoRenderer { internal: Arc>, + + #[allow(dead_code)] rtc_track: RtcVideoTrack, } @@ -109,7 +108,7 @@ impl VideoRenderer { let span = debug_span!("texture_upload"); let _enter = span.enter(); - let mut internal = internal.lock().unwrap(); + let mut internal = internal.lock(); let buffer = frame.buffer.to_i420(); let width: u32 = buffer.width().try_into().unwrap(); @@ -174,6 +173,6 @@ impl VideoRenderer { } pub fn texture_id(&self) -> Option { - self.internal.lock().unwrap().egui_texture.clone() + self.internal.lock().egui_texture.clone() } } diff --git a/livekit-ffi/Cargo.toml b/livekit-ffi/Cargo.toml index a13141a..8500ebb 100644 --- a/livekit-ffi/Cargo.toml +++ b/livekit-ffi/Cargo.toml @@ -28,6 +28,3 @@ livekit-api = { path = "../livekit-api", version = "0.1.0" } [lib] crate-type = ["cdylib", "staticlib"] - -[profile.release] -opt-level = "z" diff --git a/livekit-ffi/protocol/audio_frame.proto b/livekit-ffi/protocol/audio_frame.proto index ab1f581..f84b44f 100644 --- a/livekit-ffi/protocol/audio_frame.proto +++ b/livekit-ffi/protocol/audio_frame.proto @@ -18,7 +18,7 @@ message AllocAudioBufferResponse { AudioFrameBufferInfo buffer = 1; } // Create a new AudioStream // AudioStream is used to receive audio frames from a track message NewAudioStreamRequest { - FFIHandleId room_handle = 1; + FfiHandleId room_handle = 1; string participant_sid = 2; string track_sid = 3; AudioStreamType type = 4; @@ -31,21 +31,21 @@ message NewAudioSourceResponse { AudioSourceInfo source = 1; } // Push a frame to an AudioSource message CaptureAudioFrameRequest { - FFIHandleId source_handle = 1; - FFIHandleId buffer_handle = 2; + FfiHandleId source_handle = 1; + FfiHandleId buffer_handle = 2; } message CaptureAudioFrameResponse {} // Create a new AudioResampler message NewAudioResamplerRequest {} message NewAudioResamplerResponse { - FFIHandleId handle = 1; + FfiHandleId handle = 1; } // Remix and resample an audio frame message RemixAndResampleRequest { - FFIHandleId resampler_handle = 1; - FFIHandleId buffer_handle = 2; + FfiHandleId resampler_handle = 1; + FfiHandleId buffer_handle = 2; uint32 num_channels = 3; uint32 sample_rate = 4; } @@ -59,7 +59,7 @@ message RemixAndResampleResponse { /// message AudioFrameBufferInfo { - FFIHandleId handle = 1; + FfiHandleId handle = 1; uint64 data_ptr = 2; // *const i16 uint32 num_channels = 3; uint32 sample_rate = 4; @@ -76,13 +76,13 @@ enum AudioStreamType { } message AudioStreamInfo { - FFIHandleId handle = 1; + FfiHandleId handle = 1; AudioStreamType type = 2; string track_sid = 3; } message AudioStreamEvent { - FFIHandleId handle = 1; + FfiHandleId handle = 1; oneof message { AudioFrameReceived frame_received = 2; } } @@ -99,6 +99,6 @@ enum AudioSourceType { } message AudioSourceInfo { - FFIHandleId handle = 1; + FfiHandleId handle = 1; AudioSourceType type = 2; } diff --git a/livekit-ffi/protocol/ffi.proto b/livekit-ffi/protocol/ffi.proto index df21e9f..9779c94 100644 --- a/livekit-ffi/protocol/ffi.proto +++ b/livekit-ffi/protocol/ffi.proto @@ -12,7 +12,7 @@ import "audio_frame.proto"; /// This is the input of livekit_ffi_request function /// We always expect a response (FFIResponse) -message FFIRequest { +message FfiRequest { oneof message { InitializeRequest initialize = 1; DisposeRequest dispose = 2; @@ -33,7 +33,7 @@ message FFIRequest { NewVideoSourceRequest new_video_source = 11; CaptureVideoFrameRequest capture_video_frame = 12; ToI420Request to_i420 = 13; - ToARGBRequest to_argb = 14; + ToArgbRequest to_argb = 14; // Audio AllocAudioBufferRequest alloc_audio_buffer = 15; @@ -46,7 +46,7 @@ message FFIRequest { } /// This is the output of livekit_ffi_request function. -message FFIResponse { +message FfiResponse { oneof message { InitializeResponse initialize = 1; DisposeResponse dispose = 2; @@ -67,7 +67,7 @@ message FFIResponse { NewVideoSourceResponse new_video_source = 11; CaptureVideoFrameResponse capture_video_frame = 12; ToI420Response to_i420 = 13; - ToARGBResponse to_argb = 14; + ToArgbResponse to_argb = 14; // Audio AllocAudioBufferResponse alloc_audio_buffer = 15; @@ -79,7 +79,7 @@ message FFIResponse { } } -message FFIEvent { +message FfiEvent { oneof message { RoomEvent room_event = 1; TrackEvent track_event = 2; @@ -104,11 +104,11 @@ message DisposeRequest { } message DisposeResponse { - optional FFIAsyncId async_id = 1; // None if sync + optional FfiAsyncId async_id = 1; // None if sync } message DisposeCallback { - FFIAsyncId async_id = 1; + FfiAsyncId async_id = 1; } // TODO(theomonnom): Debug messages (Print handles, forward logs). diff --git a/livekit-ffi/protocol/handle.proto b/livekit-ffi/protocol/handle.proto index 63fdb21..e0bb30a 100644 --- a/livekit-ffi/protocol/handle.proto +++ b/livekit-ffi/protocol/handle.proto @@ -7,11 +7,11 @@ option csharp_namespace = "LiveKit.Proto"; /// The foreign language is responsable for disposing handles /// Forgetting to dispose the handle may lead to memory leaks /// Messages in this file can contain an FFIHandle -message FFIHandleId { +message FfiHandleId { uint64 id = 1; } /// Link the request/response of an asynchronous call -message FFIAsyncId { +message FfiAsyncId { uint64 id = 1; } diff --git a/livekit-ffi/protocol/room.proto b/livekit-ffi/protocol/room.proto index e8154c5..00722a3 100644 --- a/livekit-ffi/protocol/room.proto +++ b/livekit-ffi/protocol/room.proto @@ -15,42 +15,42 @@ message ConnectRequest { RoomOptions options = 3; } message ConnectResponse { - FFIAsyncId async_id = 1; + FfiAsyncId async_id = 1; } message ConnectCallback { - FFIAsyncId async_id = 1; + FfiAsyncId async_id = 1; optional string error = 2; RoomInfo room = 3; } // Disconnect from the a room -message DisconnectRequest { FFIHandleId room_handle = 1; } -message DisconnectResponse { FFIAsyncId async_id = 1; } +message DisconnectRequest { FfiHandleId room_handle = 1; } +message DisconnectResponse { FfiAsyncId async_id = 1; } message DisconnectCallback { } // Publish a track to the room message PublishTrackRequest { - FFIHandleId room_handle = 1; - FFIHandleId track_handle = 2; + FfiHandleId room_handle = 1; + FfiHandleId track_handle = 2; TrackPublishOptions options = 3; } message PublishTrackResponse { - FFIAsyncId async_id = 1; + FfiAsyncId async_id = 1; } message PublishTrackCallback { - FFIAsyncId async_id = 1; + FfiAsyncId async_id = 1; optional string error = 2; TrackPublicationInfo publication = 3; } // Unpublish a track from the room message UnpublishTrackRequest { - FFIHandleId room_handle = 1; + FfiHandleId room_handle = 1; string track_sid = 2; bool stop_on_unpublish = 3; } message UnpublishTrackResponse { - FFIAsyncId async_id = 1; + FfiAsyncId async_id = 1; } message UnpublishTrackCallback { optional string error = 1; @@ -109,7 +109,7 @@ enum DataPacketKind { } message RoomEvent { - FFIHandleId room_handle = 1; + FfiHandleId room_handle = 1; oneof message { ParticipantConnected participant_connected = 2; ParticipantDisconnected participant_disconnected = 3; @@ -131,7 +131,7 @@ message RoomEvent { } message RoomInfo { - FFIHandleId handle = 1; + FfiHandleId handle = 1; string sid = 2; string name = 3; string metadata = 4; @@ -140,7 +140,7 @@ message RoomInfo { } message DataReceived { - FFIHandleId handle = 1; + FfiHandleId handle = 1; optional string participant_sid = 2; uint64 data_ptr = 3; uint64 data_size = 4; diff --git a/livekit-ffi/protocol/track.proto b/livekit-ffi/protocol/track.proto index 1a34327..3f150e2 100644 --- a/livekit-ffi/protocol/track.proto +++ b/livekit-ffi/protocol/track.proto @@ -21,7 +21,7 @@ message AudioCaptureOptions { message CreateVideoTrackRequest { string name = 1; VideoCaptureOptions options = 2; - FFIHandleId source_handle = 3; + FfiHandleId source_handle = 3; } message CreateVideoTrackResponse { TrackInfo track = 1; @@ -31,7 +31,7 @@ message CreateVideoTrackResponse { message CreateAudioTrackRequest { string name = 1; AudioCaptureOptions options = 2; - FFIHandleId source_handle = 3; + FfiHandleId source_handle = 3; } message CreateAudioTrackResponse { TrackInfo track = 1; @@ -80,7 +80,7 @@ message TrackPublicationInfo { message TrackInfo { // Tracks created/owned by the client will have a handle - FFIHandleId opt_handle = 1; + FfiHandleId opt_handle = 1; string sid = 2; string name = 3; TrackKind kind = 4; diff --git a/livekit-ffi/protocol/video_frame.proto b/livekit-ffi/protocol/video_frame.proto index 797e8df..8af8bb2 100644 --- a/livekit-ffi/protocol/video_frame.proto +++ b/livekit-ffi/protocol/video_frame.proto @@ -16,7 +16,7 @@ message AllocVideoBufferResponse { VideoFrameBufferInfo buffer = 1; } // Create a new VideoStream // VideoStream is used to receive video frames from a track message NewVideoStreamRequest { - FFIHandleId room_handle = 1; + FfiHandleId room_handle = 1; string participant_sid = 2; string track_sid = 3; VideoStreamType type = 4; @@ -30,9 +30,9 @@ message NewVideoSourceResponse { VideoSourceInfo source = 1; } // Push a frame to a VideoSource message CaptureVideoFrameRequest { - FFIHandleId source_handle = 1; + FfiHandleId source_handle = 1; VideoFrameInfo frame = 2; - FFIHandleId buffer_handle = 3; + FfiHandleId buffer_handle = 3; } message CaptureVideoFrameResponse {} @@ -42,15 +42,15 @@ message ToI420Request { bool flip_y = 1; oneof from { ARGBBufferInfo argb = 2; - FFIHandleId buffer = 3; + FfiHandleId buffer = 3; } } message ToI420Response { VideoFrameBufferInfo buffer = 1; } // Convert a YUV frame to a RGBA frame // Only I420 is supported atm -message ToARGBRequest { - FFIHandleId buffer = 1; +message ToArgbRequest { + FfiHandleId buffer = 1; uint64 dst_ptr = 2; VideoFormatType dst_format = 3; uint32 dst_stride = 4; @@ -58,7 +58,7 @@ message ToARGBRequest { uint32 dst_height = 6; bool flip_y = 7; } -message ToARGBResponse {} +message ToArgbResponse {} /// /// VideoFrame buffers /// @@ -115,7 +115,7 @@ message VideoFrameInfo { } message VideoFrameBufferInfo { - FFIHandleId handle = 1; + FfiHandleId handle = 1; VideoFrameBufferType buffer_type = 2; uint32 width = 3; uint32 height = 4; @@ -166,13 +166,13 @@ enum VideoStreamType { } message VideoStreamInfo { - FFIHandleId handle = 1; + FfiHandleId handle = 1; VideoStreamType type = 2; string track_sid = 3; } message VideoStreamEvent { - FFIHandleId handle = 1; + FfiHandleId handle = 1; oneof message { VideoFrameReceived frame_received = 2; } } @@ -192,7 +192,7 @@ enum VideoSourceType { message VideoSourceInfo { // # SAFETY // This handle must not be dropped if a track is currently using it - FFIHandleId handle = 1; + FfiHandleId handle = 1; VideoSourceType type = 2; } diff --git a/livekit-ffi/src/conversion/room.rs b/livekit-ffi/src/conversion/room.rs index e4f4be5..17975c0 100644 --- a/livekit-ffi/src/conversion/room.rs +++ b/livekit-ffi/src/conversion/room.rs @@ -66,7 +66,7 @@ impl proto::RoomEvent { } impl proto::RoomInfo { - pub fn from_session(handle_id: FfiHandleId, session: &RoomSession) -> Self { + pub fn from_room(handle_id: FfiHandleId, session: &Room) -> Self { Self { handle: Some(handle_id.into()), sid: session.sid().into(), diff --git a/livekit-ffi/src/conversion/track.rs b/livekit-ffi/src/conversion/track.rs index e314318..1c81af3 100644 --- a/livekit-ffi/src/conversion/track.rs +++ b/livekit-ffi/src/conversion/track.rs @@ -60,6 +60,7 @@ impl_publication_into!(&TrackPublication); macro_rules! impl_track_into { ($fnc:ident, $t:ty) => { impl proto::TrackInfo { + #[allow(dead_code)] pub fn $fnc(handle_id: FfiHandleId, track: $t) -> Self { Self { opt_handle: Some(handle_id.into()), diff --git a/livekit-ffi/src/server/audio_frame.rs b/livekit-ffi/src/server/audio_frame.rs index eb61cb4..bac35de 100644 --- a/livekit-ffi/src/server/audio_frame.rs +++ b/livekit-ffi/src/server/audio_frame.rs @@ -157,8 +157,7 @@ impl FfiAudioSource { proto::AudioSourceType::AudioSourceNative => { let audio_source = NativeAudioSource::default(); Ok::(AudioSource::Native(audio_source)) - } - _ => return Err(FfiError::InvalidRequest("unsupported audio source type")), + } //_ => return Err(FfiError::InvalidRequest("unsupported audio source type")), }?; let audio_source = Self { diff --git a/livekit-ffi/src/server/mod.rs b/livekit-ffi/src/server/mod.rs index 10ec5c5..c932741 100644 --- a/livekit-ffi/src/server/mod.rs +++ b/livekit-ffi/src/server/mod.rs @@ -209,12 +209,12 @@ impl FfiServer { .ok_or(FfiError::InvalidRequest("room_handle is empty"))? .id as FfiHandleId; - let room = self + let ffi_room = self .ffi_handles .get(&room_handle) .ok_or(FfiError::InvalidRequest("room not found"))?; - let room = room + let ffi_room = ffi_room .downcast_ref::() .ok_or(FfiError::InvalidRequest("room is not a FfiRoom"))?; @@ -233,8 +233,8 @@ impl FfiServer { .downcast_ref::() .ok_or(FfiError::InvalidRequest("track is not a LocalTrack"))?; - let publication = room - .session() + let publication = ffi_room + .room() .local_participant() .publish_track( track.clone(), diff --git a/livekit-ffi/src/server/room.rs b/livekit-ffi/src/server/room.rs index a80b085..f4970f6 100644 --- a/livekit-ffi/src/server/room.rs +++ b/livekit-ffi/src/server/room.rs @@ -1,12 +1,12 @@ use crate::server::FfiServer; use crate::{proto, FfiHandleId, FfiResult}; use livekit::prelude::*; +use std::sync::Arc; use tokio::sync::{mpsc, oneshot}; use tokio::task::JoinHandle; pub struct FfiRoom { - room: Room, - handle_id: FfiHandleId, + room: Arc, handle: JoinHandle<()>, close_tx: oneshot::Sender<()>, } @@ -17,45 +17,42 @@ impl FfiRoom { connect: proto::ConnectRequest, ) -> FfiResult { let (room, events) = Room::connect(&connect.url, &connect.token).await?; + let room = Arc::new(room); let (close_tx, close_rx) = oneshot::channel(); - let session = room.session(); let next_id = server.next_id() as FfiHandleId; - let handle = server.async_runtime.spawn(room_task( - server, - session.clone(), - next_id, - events, - close_rx, - )); - let room_info = proto::RoomInfo::from_session(next_id, &session); + let handle = + server + .async_runtime + .spawn(room_task(server, room.clone(), next_id, events, close_rx)); + let room_info = proto::RoomInfo::from_room(next_id, &room); let ffi_room = Self { - handle_id: next_id, - room, + room: room.clone(), handle, close_tx, }; + server.ffi_handles().insert(next_id, Box::new(ffi_room)); - server.rooms().lock().insert(session.sid(), next_id); + server.rooms().lock().insert(room.sid(), next_id); Ok(room_info) } pub async fn close(self) { - self.room.close().await; + let _ = self.room.close().await; let _ = self.close_tx.send(()); let _ = self.handle.await; } - pub fn session(&self) -> RoomSession { - self.room.session() + pub fn room(&self) -> &Arc { + &self.room } } async fn room_task( server: &'static FfiServer, - session: RoomSession, + room: Arc, room_handle: FfiHandleId, mut events: mpsc::UnboundedReceiver, mut close_rx: oneshot::Receiver<()>, @@ -63,7 +60,7 @@ async fn room_task( server .async_runtime .spawn(participant_task(Participant::Local( - session.local_participant(), + room.local_participant(), ))); loop { diff --git a/livekit-ffi/src/server/tests.rs b/livekit-ffi/src/server/tests.rs index 5e32ee6..9994743 100644 --- a/livekit-ffi/src/server/tests.rs +++ b/livekit-ffi/src/server/tests.rs @@ -12,8 +12,8 @@ mod client { INVALID_HANDLE, }; use lazy_static::lazy_static; + use parking_lot::Mutex; use prost::Message; - use std::sync::Mutex; use tokio::sync::mpsc; lazy_static! { @@ -31,7 +31,7 @@ mod client { impl Default for FfiClient { fn default() -> Self { let (event_tx, event_rx) = mpsc::unbounded_channel(); - *EVENT_TX.lock().unwrap() = Some(event_tx); + *EVENT_TX.lock() = Some(event_tx); Self { event_rx } } } @@ -88,7 +88,6 @@ mod client { let event = proto::FfiEvent::decode(data).unwrap(); EVENT_TX .lock() - .unwrap() .as_ref() .unwrap() .send(event.message.unwrap()) @@ -99,9 +98,9 @@ mod client { struct TestScope {} impl TestScope { - fn new() -> (Self, std::sync::MutexGuard<'static, client::FfiClient>) { + fn new() -> (Self, parking_lot::MutexGuard<'static, client::FfiClient>) { // Run one test at a time - let client = client::FFI_CLIENT.lock().unwrap(); + let client = client::FFI_CLIENT.lock(); (TestScope {}, client) } @@ -176,7 +175,7 @@ fn create_i420_buffer() { #[test] #[ignore] // Ignore for now ( need to setup GHA ) fn publish_video_track() { - let (test, mut client) = TestScope::new(); + let (_test, mut client) = TestScope::new(); let (lk_url, lk_api_key, lk_api_secret) = test_env(); tokio::runtime::Builder::new_multi_thread() diff --git a/livekit-ffi/src/server/utils.rs b/livekit-ffi/src/server/utils.rs index cb1a484..12de89b 100644 --- a/livekit-ffi/src/server/utils.rs +++ b/livekit-ffi/src/server/utils.rs @@ -7,17 +7,17 @@ pub fn find_remote_track( participant_sid: &ParticipantSid, room_handle: FfiHandleId, ) -> FfiResult { - let room = server + let ffi_room = server .ffi_handles() .get(&room_handle) .ok_or(FfiError::InvalidRequest("room not found"))?; - let room = room + let ffi_room = ffi_room .downcast_ref::() .ok_or(FfiError::InvalidRequest("room is not ffi room"))?; - let session = room.session(); - let participants = session.participants(); + let room = ffi_room.room(); + let participants = room.participants(); let participant = participants .get(participant_sid) .ok_or(FfiError::InvalidRequest("participant not found"))?; diff --git a/livekit-ffi/src/server/video_frame.rs b/livekit-ffi/src/server/video_frame.rs index 157c047..c772dc6 100644 --- a/livekit-ffi/src/server/video_frame.rs +++ b/livekit-ffi/src/server/video_frame.rs @@ -160,10 +160,9 @@ impl FfiVideoSource { let source_inner = match source_type { proto::VideoSourceType::VideoSourceNative => { let video_source = NativeVideoSource::default(); - Ok(VideoSource::Native(video_source)) + VideoSource::Native(video_source) } - _ => Err(FfiError::InvalidRequest("unsupported video source type")), - }?; + }; let video_source = Self { handle_id: server.next_id(), diff --git a/livekit-webrtc/src/native/audio_source.rs b/livekit-webrtc/src/native/audio_source.rs index 617ce24..e25fa3e 100644 --- a/livekit-webrtc/src/native/audio_source.rs +++ b/livekit-webrtc/src/native/audio_source.rs @@ -1,6 +1,7 @@ use crate::audio_frame::AudioFrame; use cxx::SharedPtr; -use std::sync::{Arc, Mutex}; +use parking_lot::Mutex; +use std::sync::Arc; use webrtc_sys::media_stream as sys_ms; #[derive(Clone)] @@ -32,7 +33,7 @@ impl NativeAudioSource { } pub fn capture_frame(&self, frame: &AudioFrame) { - let mut inner = self.inner.lock().unwrap(); + let mut inner = self.inner.lock(); let samples_10ms = (frame.sample_rate / 100 * frame.num_channels) as usize; if inner.sample_rate != frame.sample_rate || inner.num_channels != frame.num_channels { inner.buf.resize(samples_10ms as usize, 0); diff --git a/livekit-webrtc/src/native/data_channel.rs b/livekit-webrtc/src/native/data_channel.rs index cbe2289..a0314e7 100644 --- a/livekit-webrtc/src/native/data_channel.rs +++ b/livekit-webrtc/src/native/data_channel.rs @@ -3,8 +3,9 @@ use crate::data_channel::{ OnStateChange, }; use cxx::SharedPtr; +use parking_lot::Mutex; use std::str; -use std::sync::{Arc, Mutex}; +use std::sync::Arc; use webrtc_sys::data_channel as sys_dc; impl From for DataState { @@ -94,15 +95,15 @@ impl DataChannel { } pub fn on_state_change(&self, handler: Option) { - *self.observer.state_change_handler.lock().unwrap() = handler; + *self.observer.state_change_handler.lock() = handler; } pub fn on_message(&self, handler: Option) { - *self.observer.message_handler.lock().unwrap() = handler; + *self.observer.message_handler.lock() = handler; } pub fn on_buffered_amount_change(&self, handler: Option) { - *self.observer.buffered_amount_change_handler.lock().unwrap() = handler; + *self.observer.buffered_amount_change_handler.lock() = handler; } } @@ -115,21 +116,21 @@ struct DataChannelObserver { impl sys_dc::DataChannelObserver for DataChannelObserver { fn on_state_change(&self, state: sys_dc::ffi::DataState) { - let mut handler = self.state_change_handler.lock().unwrap(); + let mut handler = self.state_change_handler.lock(); if let Some(f) = handler.as_mut() { f(state.into()); } } fn on_message(&self, data: &[u8], binary: bool) { - let mut handler = self.message_handler.lock().unwrap(); + let mut handler = self.message_handler.lock(); if let Some(f) = handler.as_mut() { f(DataBuffer { data, binary }); } } fn on_buffered_amount_change(&self, sent_data_size: u64) { - let mut handler = self.buffered_amount_change_handler.lock().unwrap(); + let mut handler = self.buffered_amount_change_handler.lock(); if let Some(f) = handler.as_mut() { f(sent_data_size); } diff --git a/livekit-webrtc/src/native/peer_connection.rs b/livekit-webrtc/src/native/peer_connection.rs index dfd1a9c..67b1f0c 100644 --- a/livekit-webrtc/src/native/peer_connection.rs +++ b/livekit-webrtc/src/native/peer_connection.rs @@ -23,8 +23,9 @@ use crate::MediaType; use crate::{session_description::SessionDescription, RtcError}; use cxx::{SharedPtr, UniquePtr}; use futures::channel::oneshot; +use parking_lot::Mutex; use std::mem::ManuallyDrop; -use std::sync::{Arc, Mutex}; +use std::sync::Arc; use webrtc_sys::data_channel as sys_dc; use webrtc_sys::jsep as sys_jsep; use webrtc_sys::peer_connection as sys_pc; @@ -372,39 +373,39 @@ impl PeerConnection { } pub fn on_connection_state_change(&self, f: Option) { - *self.observer.connection_change_handler.lock().unwrap() = f; + *self.observer.connection_change_handler.lock() = f; } pub fn on_data_channel(&self, f: Option) { - *self.observer.data_channel_handler.lock().unwrap() = f; + *self.observer.data_channel_handler.lock() = f; } pub fn on_ice_candidate(&self, f: Option) { - *self.observer.ice_candidate_handler.lock().unwrap() = f; + *self.observer.ice_candidate_handler.lock() = f; } pub fn on_ice_candidate_error(&self, f: Option) { - *self.observer.ice_candidate_error_handler.lock().unwrap() = f; + *self.observer.ice_candidate_error_handler.lock() = f; } pub fn on_ice_connection_state_change(&self, f: Option) { - *self.observer.ice_connection_change_handler.lock().unwrap() = f; + *self.observer.ice_connection_change_handler.lock() = f; } pub fn on_ice_gathering_state_change(&self, f: Option) { - *self.observer.ice_gathering_change_handler.lock().unwrap() = f; + *self.observer.ice_gathering_change_handler.lock() = f; } pub fn on_negotiation_needed(&self, f: Option) { - *self.observer.negotiation_needed_handler.lock().unwrap() = f; + *self.observer.negotiation_needed_handler.lock() = f; } pub fn on_signaling_state_change(&self, f: Option) { - *self.observer.signaling_change_handler.lock().unwrap() = f; + *self.observer.signaling_change_handler.lock() = f; } pub fn on_track(&self, f: Option) { - *self.observer.track_handler.lock().unwrap() = f; + *self.observer.track_handler.lock() = f; } } @@ -451,7 +452,7 @@ pub struct PeerObserver { impl sys_pc::PeerConnectionObserver for PeerObserver { fn on_signaling_change(&self, new_state: sys_pc::ffi::SignalingState) { - if let Some(f) = self.signaling_change_handler.lock().unwrap().as_mut() { + if let Some(f) = self.signaling_change_handler.lock().as_mut() { f(new_state.into()); } } @@ -461,7 +462,7 @@ impl sys_pc::PeerConnectionObserver for PeerObserver { fn on_remove_stream(&self, _stream: SharedPtr) {} fn on_data_channel(&self, data_channel: SharedPtr) { - if let Some(f) = self.data_channel_handler.lock().unwrap().as_mut() { + if let Some(f) = self.data_channel_handler.lock().as_mut() { f(DataChannel { handle: imp_dc::DataChannel::configure(data_channel), }); @@ -471,7 +472,7 @@ impl sys_pc::PeerConnectionObserver for PeerObserver { fn on_renegotiation_needed(&self) {} fn on_negotiation_needed_event(&self, event: u32) { - if let Some(f) = self.negotiation_needed_handler.lock().unwrap().as_mut() { + if let Some(f) = self.negotiation_needed_handler.lock().as_mut() { f(event); } } @@ -479,25 +480,25 @@ impl sys_pc::PeerConnectionObserver for PeerObserver { fn on_ice_connection_change(&self, _new_state: sys_pc::ffi::IceConnectionState) {} fn on_standardized_ice_connection_change(&self, new_state: sys_pc::ffi::IceConnectionState) { - if let Some(f) = self.ice_connection_change_handler.lock().unwrap().as_mut() { + if let Some(f) = self.ice_connection_change_handler.lock().as_mut() { f(new_state.into()); } } fn on_connection_change(&self, new_state: sys_pc::ffi::PeerConnectionState) { - if let Some(f) = self.connection_change_handler.lock().unwrap().as_mut() { + if let Some(f) = self.connection_change_handler.lock().as_mut() { f(new_state.into()); } } fn on_ice_gathering_change(&self, new_state: sys_pc::ffi::IceGatheringState) { - if let Some(f) = self.ice_gathering_change_handler.lock().unwrap().as_mut() { + if let Some(f) = self.ice_gathering_change_handler.lock().as_mut() { f(new_state.into()); } } fn on_ice_candidate(&self, candidate: SharedPtr) { - if let Some(f) = self.ice_candidate_handler.lock().unwrap().as_mut() { + if let Some(f) = self.ice_candidate_handler.lock().as_mut() { f(IceCandidate { handle: imp_ic::IceCandidate { sys_handle: candidate, @@ -514,7 +515,7 @@ impl sys_pc::PeerConnectionObserver for PeerObserver { error_code: i32, error_text: String, ) { - if let Some(f) = self.ice_candidate_error_handler.lock().unwrap().as_mut() { + if let Some(f) = self.ice_candidate_error_handler.lock().as_mut() { f(IceCandidateError { address, port, @@ -547,7 +548,7 @@ impl sys_pc::PeerConnectionObserver for PeerObserver { } fn on_track(&self, transceiver: SharedPtr) { - if let Some(f) = self.track_handler.lock().unwrap().as_mut() { + if let Some(f) = self.track_handler.lock().as_mut() { let receiver = transceiver.receiver(); let streams = receiver.streams(); let track = receiver.track(); diff --git a/livekit/src/prelude.rs b/livekit/src/prelude.rs index f8c204b..5c9a311 100644 --- a/livekit/src/prelude.rs +++ b/livekit/src/prelude.rs @@ -1,6 +1,6 @@ pub use crate::participant::{LocalParticipant, Participant, ParticipantEvent, RemoteParticipant}; -pub use crate::{ConnectionState, Room, RoomError, RoomEvent, RoomResult, RoomSession}; +pub use crate::{ConnectionState, Room, RoomError, RoomEvent, RoomResult}; pub use crate::publication::{LocalTrackPublication, RemoteTrackPublication, TrackPublication}; diff --git a/livekit/src/room/mod.rs b/livekit/src/room/mod.rs index b7b3fc9..144d0d9 100644 --- a/livekit/src/room/mod.rs +++ b/livekit/src/room/mod.rs @@ -2,11 +2,19 @@ use self::track::RemoteTrack; use crate::participant::ConnectionQuality; use crate::prelude::*; use crate::rtc_engine::EngineError; +use crate::rtc_engine::{EngineEvent, EngineEvents, EngineResult, RtcEngine}; +use crate::signal_client::SignalOptions; use livekit_protocol as proto; +use livekit_protocol::observer::Dispatcher; +use parking_lot::{Mutex, RwLock, RwLockReadGuard}; +use std::collections::HashMap; use std::fmt::Debug; +use std::sync::atomic::{AtomicU8, Ordering}; use std::sync::Arc; use thiserror::Error; -use tokio::sync::mpsc; +use tokio::sync::{mpsc, oneshot}; +use tokio::task::JoinHandle; +use tracing::{error, info, instrument, trace, Level}; pub use crate::rtc_engine::SimulateScenario; @@ -14,11 +22,8 @@ pub mod id; pub mod options; pub mod participant; pub mod publication; -pub mod room_session; pub mod track; -pub use room_session::*; - pub type RoomResult = Result; #[derive(Error, Debug)] @@ -29,6 +34,8 @@ pub enum RoomError { Internal(String), #[error("this track or a track of the same source is already published")] TrackAlreadyPublished, + #[error("already closed")] + AlreadyClosed, } #[derive(Clone, Debug)] @@ -85,9 +92,32 @@ pub enum RoomEvent { Reconnected, } -#[derive(Debug)] +#[derive(Debug, Clone, Copy, Eq, PartialEq)] +pub enum ConnectionState { + Disconnected, + Connected, + Reconnecting, + Unknown, +} + +struct RoomHandle { + session_task: JoinHandle<()>, + close_emitter: oneshot::Sender<()>, +} + pub struct Room { - handle: SessionHandle, + inner: Arc, + handle: Mutex>, +} + +impl Debug for Room { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Room") + .field("sid", &self.sid()) + .field("name", &self.name()) + .field("connection_state", &self.connection_state()) + .finish() + } } impl Room { @@ -95,21 +125,538 @@ impl Room { url: &str, token: &str, ) -> RoomResult<(Self, mpsc::UnboundedReceiver)> { - let handle = SessionHandle::connect(url, token).await?; - let events = handle.subscribe(); - Ok((Self { handle }, events)) + let (rtc_engine, engine_events) = RtcEngine::new(); + let rtc_engine = Arc::new(rtc_engine); + rtc_engine + .connect(url, token, SignalOptions::default()) + .await?; + + let join_response = rtc_engine.join_response().unwrap(); + let pi = join_response.participant.unwrap().clone(); + let local_participant = LocalParticipant::new( + rtc_engine.clone(), + pi.sid.into(), + pi.identity.into(), + pi.name, + pi.metadata, + ); + + let room_info = join_response.room.unwrap(); + let inner = Arc::new(SessionInner { + state: AtomicU8::new(ConnectionState::Disconnected as u8), + sid: Mutex::new(room_info.sid.into()), + name: Mutex::new(room_info.name), + metadata: Mutex::new(room_info.metadata), + participants: Default::default(), + participants_tasks: Default::default(), + active_speakers: Default::default(), + rtc_engine, + local_participant, + dispatcher: Default::default(), + }); + + for pi in join_response.other_participants { + let participant = { + let pi = pi.clone(); + inner.create_participant(pi.sid.into(), pi.identity.into(), pi.name, pi.metadata) + }; + participant.update_info(pi.clone()); + } + + let (close_emitter, close_receiver) = oneshot::channel(); + let session_task = tokio::spawn(inner.clone().room_task(engine_events, close_receiver)); + + inner.update_connection_state(ConnectionState::Connected); + + let session = Self { + inner, + handle: Mutex::new(Some(RoomHandle { + session_task, + close_emitter, + })), + }; + + let events = session.subscribe(); + Ok((session, events)) } - pub async fn close(self) { - self.handle.close().await; + pub async fn close(&self) -> RoomResult<()> { + if let Some(handle) = self.handle.lock().take() { + self.inner.close().await; + handle.close_emitter.send(()).ok(); + handle.session_task.await.ok(); + Ok(()) + } else { + Err(RoomError::AlreadyClosed) + } } - /// Allow multiple subscribers/observers to receive events pub fn subscribe(&self) -> mpsc::UnboundedReceiver { - self.handle.subscribe() + self.inner.dispatcher.register() } - pub fn session(&self) -> RoomSession { - self.handle.session() + pub fn sid(&self) -> RoomSid { + self.inner.sid.lock().clone() + } + + pub fn name(&self) -> String { + self.inner.name.lock().clone() + } + + pub fn metadata(&self) -> String { + self.inner.metadata.lock().clone() + } + + pub fn local_participant(&self) -> LocalParticipant { + self.inner.local_participant.clone() + } + + pub fn connection_state(&self) -> ConnectionState { + self.inner.state.load(Ordering::Acquire).try_into().unwrap() + } + + pub fn participants(&self) -> RwLockReadGuard> { + self.inner.participants.read() + } + + pub async fn simulate_scenario(&self, scenario: SimulateScenario) -> EngineResult<()> { + self.inner.rtc_engine.simulate_scenario(scenario).await + } +} + +struct SessionInner { + state: AtomicU8, // ConnectionState + sid: Mutex, + name: Mutex, + metadata: Mutex, + participants: RwLock>, + participants_tasks: RwLock, oneshot::Sender<()>)>>, + active_speakers: RwLock>, + rtc_engine: Arc, + local_participant: LocalParticipant, + dispatcher: Dispatcher, +} + +impl Debug for SessionInner { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("SessionInner") + .field("sid", &self.sid) + .field("name", &self.name) + .field("rtc_engine", &self.rtc_engine) + .finish() + } +} + +impl SessionInner { + #[instrument(level = Level::DEBUG)] + async fn room_task( + self: Arc, + mut engine_events: EngineEvents, + mut close_receiver: oneshot::Receiver<()>, + ) { + loop { + tokio::select! { + res = engine_events.recv() => { + if let Some(event) = res { + if let Err(err) = self.on_engine_event(event).await { + error!("failed to handle engine event: {:?}", err); + } + } + }, + _ = &mut close_receiver => { + trace!("closing room_task"); + break; + } + } + } + } + + /// Listen to the Participant events and forward them to the Room Dispatcher + #[instrument(level = Level::DEBUG)] + async fn participant_task( + self: Arc, + participant: Participant, + mut participant_events: mpsc::UnboundedReceiver, + mut close_rx: oneshot::Receiver<()>, + ) { + loop { + tokio::select! { + res = participant_events.recv() => { + if let Some(event) = res { + if let Err(err) = self.on_participant_event(&participant, event).await { + error!("failed to handle participant event for {:?}: {:?}", participant.sid(), err); + } + } + }, + _ = &mut close_rx => { + trace!("closing participant_task for {:?}", participant.sid()); + break; + }, + } + } + } + + #[instrument(level = Level::DEBUG)] + async fn on_participant_event( + self: &Arc, + participant: &Participant, + event: ParticipantEvent, + ) -> RoomResult<()> { + if let Participant::Remote(remote_participant) = participant { + match event { + ParticipantEvent::TrackPublished { publication } => { + self.dispatcher.dispatch(&RoomEvent::TrackPublished { + participant: remote_participant.clone(), + publication, + }); + } + ParticipantEvent::TrackUnpublished { publication } => { + self.dispatcher.dispatch(&RoomEvent::TrackUnpublished { + participant: remote_participant.clone(), + publication, + }); + } + ParticipantEvent::TrackSubscribed { track, publication } => { + self.dispatcher.dispatch(&RoomEvent::TrackSubscribed { + participant: remote_participant.clone(), + track, + publication, + }); + } + ParticipantEvent::TrackUnsubscribed { track, publication } => { + self.dispatcher.dispatch(&RoomEvent::TrackUnsubscribed { + participant: remote_participant.clone(), + track, + publication, + }); + } + _ => {} + }; + } + + Ok(()) + } + + #[instrument(level = Level::DEBUG)] + async fn on_engine_event(self: &Arc, event: EngineEvent) -> RoomResult<()> { + match event { + EngineEvent::ParticipantUpdate { updates } => self.handle_participant_update(updates), + EngineEvent::MediaTrack { + track, + stream, + receiver: _, + } => { + let stream_id = stream.id(); + let lk_stream_id = unpack_stream_id(&stream_id); + if lk_stream_id.is_none() { + Err(RoomError::Internal(format!( + "MediaTrack event with invalid track_id: {:?}", + &stream_id + )))?; + } + + let (participant_sid, track_sid) = lk_stream_id.unwrap(); + let track_sid = track_sid.to_owned().into(); + let remote_participant = self.get_participant(&participant_sid.to_string().into()); + + if let Some(remote_participant) = remote_participant { + tokio::spawn(async move { + remote_participant + .add_subscribed_media_track(track_sid, track) + .await; + }); + } else { + // The server should send participant updates before sending a new offer + // So this should never happen. + Err(RoomError::Internal(format!( + "AddTrack event with invalid participant_sid: {:?}", + participant_sid + )))?; + } + } + EngineEvent::Resuming => { + if self.update_connection_state(ConnectionState::Reconnecting) { + self.dispatcher.dispatch(&RoomEvent::Reconnecting); + } + } + EngineEvent::Resumed => { + self.update_connection_state(ConnectionState::Connected); + self.dispatcher.dispatch(&RoomEvent::Reconnected); + + // TODO(theomonnom): Update subscriptions settings + // TODO(theomonnom): Send sync state + } + EngineEvent::Restarting => self.handle_restarting(), + EngineEvent::Restarted => self.handle_restarted(), + EngineEvent::Disconnected => self.handle_disconnected(), + EngineEvent::Data { + payload, + kind, + participant_sid, + } => { + let payload = Arc::new(payload); + if let Some(participant) = self.get_participant(&participant_sid.into()) { + self.dispatcher.dispatch(&RoomEvent::DataReceived { + payload: payload.clone(), + kind, + participant: participant.clone(), + }); + + participant.on_data_received(payload, kind); + } + } + EngineEvent::SpeakersChanged { speakers } => self.handle_speakers_changed(speakers), + EngineEvent::ConnectionQuality { updates } => { + self.handle_connection_quality_update(updates) + } + } + + Ok(()) + } + + #[instrument(level = Level::DEBUG)] + async fn close(&self) { + self.rtc_engine.close().await; + } + + /// Change the connection state and emit an event + /// Does nothing if the state is already the same + #[instrument(level = Level::DEBUG)] + fn update_connection_state(&self, state: ConnectionState) -> bool { + let old_state = self.state.load(Ordering::Acquire); + if old_state == state as u8 { + return false; + } + + self.state.store(state as u8, Ordering::Release); + self.dispatcher + .dispatch(&RoomEvent::ConnectionStateChanged(state)); + return true; + } + + /// Update the participants inside a Room. + /// It'll create, update or remove a participant + /// It also update the participant tracks. + #[instrument(level = Level::DEBUG)] + fn handle_participant_update(self: &Arc, updates: Vec) { + for pi in updates { + if pi.sid == self.local_participant.sid() + || pi.identity == self.local_participant.identity() + { + self.local_participant.clone().update_info(pi); + continue; + } + + let remote_participant = self.get_participant(&pi.sid.clone().into()); + + if let Some(remote_participant) = remote_participant { + if pi.state == proto::participant_info::State::Disconnected as i32 { + // Participant disconnected + info!("Participant disconnected: {}", pi.sid); + self.clone() + .handle_participant_disconnect(remote_participant) + } else { + // Participant is already connected, update the it + remote_participant.update_info(pi.clone()); + } + } else { + // Create a new participant + info!("Participant connected: {}", pi.sid); + let remote_participant = { + let pi = pi.clone(); + self.create_participant(pi.sid.into(), pi.identity.into(), pi.name, pi.metadata) + }; + + let _ = self + .dispatcher + .dispatch(&RoomEvent::ParticipantConnected(remote_participant.clone())); + + remote_participant.update_info(pi.clone()); + } + } + } + + /// Active speakers changed + /// Update the participants & sort the active_speakers by audio_level + #[instrument(level = Level::DEBUG)] + fn handle_speakers_changed(&self, speakers_info: Vec) { + let mut speakers = Vec::new(); + + for speaker in speakers_info { + let participant = { + if speaker.sid == self.local_participant.sid() { + Participant::Local(self.local_participant.clone()) + } else { + if let Some(participant) = self.get_participant(&speaker.sid.into()) { + Participant::Remote(participant) + } else { + continue; + } + } + }; + + participant.set_speaking(speaker.active); + participant.set_audio_level(speaker.level); + + if speaker.active { + speakers.push(participant); + } + } + + speakers.sort_by(|a, b| a.audio_level().partial_cmp(&b.audio_level()).unwrap()); + *self.active_speakers.write() = speakers.clone(); + + let _ = self + .dispatcher + .dispatch(&RoomEvent::ActiveSpeakersChanged { speakers }); + } + + /// Handle a connection quality update + /// Emit ConnectionQualityChanged event for the concerned participants + #[instrument(level = Level::DEBUG)] + fn handle_connection_quality_update(&self, updates: Vec) { + for update in updates { + let participant = { + if update.participant_sid == self.local_participant.sid() { + Participant::Local(self.local_participant.clone()) + } else { + if let Some(participant) = self.get_participant(&update.participant_sid.into()) + { + Participant::Remote(participant) + } else { + continue; + } + } + }; + + let quality: ConnectionQuality = proto::ConnectionQuality::from_i32(update.quality) + .unwrap() + .into(); + + participant.set_connection_quality(quality); + self.dispatcher + .dispatch(&RoomEvent::ConnectionQualityChanged { + participant, + quality, + }); + } + } + + #[instrument(level = Level::DEBUG)] + fn handle_restarting(self: &Arc) { + // Remove existing participants/subscriptions on full reconnect + for (_, participant) in self.participants.read().iter() { + self.clone() + .handle_participant_disconnect(participant.clone()); + } + + if self.update_connection_state(ConnectionState::Reconnecting) { + self.dispatcher.dispatch(&RoomEvent::Reconnecting); + } + } + + #[instrument(level = Level::DEBUG)] + fn handle_restarted(self: &Arc) { + // Full reconnect succeeded! + let join_response = self.rtc_engine.join_response().unwrap(); + + self.update_connection_state(ConnectionState::Connected); + self.dispatcher.dispatch(&RoomEvent::Reconnected); + + if let Some(pi) = join_response.participant { + self.local_participant.update_info(pi); // The sid may have changed + } + + self.handle_participant_update(join_response.other_participants); + + // TODO(theomonnom): unpublish & republish tracks + } + + #[instrument(level = Level::DEBUG)] + fn handle_disconnected(&self) { + if self.state.load(Ordering::Acquire) == ConnectionState::Disconnected as u8 { + return; + } + + self.update_connection_state(ConnectionState::Disconnected); + self.dispatcher.dispatch(&RoomEvent::Disconnected); + } + + /// Create a new participant + /// Also add it to the participants list + #[instrument(level = Level::DEBUG)] + fn create_participant( + self: &Arc, + sid: ParticipantSid, + identity: ParticipantIdentity, + name: String, + metadata: String, + ) -> RemoteParticipant { + let participant = RemoteParticipant::new(sid.clone(), identity, name, metadata); + + // Create the participant task + let (close_tx, close_rx) = oneshot::channel(); + let participant_task = tokio::spawn(self.clone().participant_task( + Participant::Remote(participant.clone()), + participant.register_observer(), + close_rx, + )); + self.participants_tasks + .write() + .insert(sid.clone(), (participant_task, close_tx)); + + self.participants.write().insert(sid, participant.clone()); + participant + } + + /// A participant has disconnected + /// Cleanup the participant and emit an event + #[instrument(level = Level::DEBUG)] + fn handle_participant_disconnect(self: Arc, remote_participant: RemoteParticipant) { + tokio::spawn(async move { + for (sid, _) in &*remote_participant.tracks() { + remote_participant.unpublish_track(&sid); + } + + // Close the participant task + if let Some((task, close_tx)) = self + .participants_tasks + .write() + .remove(&remote_participant.sid()) + { + let _ = close_tx.send(()); + let _ = task.await; + } + + self.participants.write().remove(&remote_participant.sid()); + self.dispatcher + .dispatch(&RoomEvent::ParticipantDisconnected(remote_participant)); + }); + } + + fn get_participant(&self, sid: &ParticipantSid) -> Option { + self.participants.read().get(sid).cloned() + } +} + +fn unpack_stream_id(stream_id: &str) -> Option<(&str, &str)> { + let split: Vec<&str> = stream_id.split('|').collect(); + if split.len() == 2 { + let participant_sid = split.get(0).unwrap(); + let track_sid = split.get(1).unwrap(); + Some((participant_sid, track_sid)) + } else { + None + } +} + +impl From for ConnectionState { + fn from(value: u8) -> Self { + match value { + 0 => ConnectionState::Disconnected, + 1 => ConnectionState::Connected, + 2 => ConnectionState::Reconnecting, + _ => ConnectionState::Unknown, + } } } diff --git a/livekit/src/room/observer.rs b/livekit/src/room/observer.rs deleted file mode 100644 index 8b13789..0000000 --- a/livekit/src/room/observer.rs +++ /dev/null @@ -1 +0,0 @@ - diff --git a/livekit/src/room/participant/local_participant.rs b/livekit/src/room/participant/local_participant.rs index bd3085b..70f5fdd 100644 --- a/livekit/src/room/participant/local_participant.rs +++ b/livekit/src/room/participant/local_participant.rs @@ -9,16 +9,27 @@ use livekit_protocol as proto; use livekit_webrtc::rtp_parameters::RtpEncodingParameters; use parking_lot::RwLockReadGuard; use std::collections::HashMap; +use std::fmt::Debug; use std::sync::Arc; use tokio::sync::mpsc; -use tracing::{debug, trace}; +use tracing::debug; -#[derive(Debug, Clone)] +#[derive(Clone)] pub struct LocalParticipant { inner: Arc, rtc_engine: Arc, } +impl Debug for LocalParticipant { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("LocalParticipant") + .field("sid", &self.sid()) + .field("identity", &self.identity()) + .field("name", &self.name()) + .finish() + } +} + impl LocalParticipant { pub(crate) fn new( rtc_engine: Arc, @@ -110,7 +121,7 @@ impl LocalParticipant { pub async fn unpublish_track( &self, track: TrackSid, - stop_on_unpublish: bool, + _stop_on_unpublish: bool, ) -> RoomResult { let mut tracks = self.inner.tracks.write(); if let Some(TrackPublication::Local(publication)) = tracks.remove(&track) { diff --git a/livekit/src/room/participant/mod.rs b/livekit/src/room/participant/mod.rs index c799855..be6f71a 100644 --- a/livekit/src/room/participant/mod.rs +++ b/livekit/src/room/participant/mod.rs @@ -5,6 +5,7 @@ use livekit_protocol::enum_dispatch; use livekit_protocol::observer::Dispatcher; use parking_lot::{Mutex, RwLock, RwLockReadGuard}; use std::collections::HashMap; +use std::fmt::Debug; use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU8, Ordering}; use std::sync::Arc; use tokio::sync::mpsc; diff --git a/livekit/src/room/participant/remote_participant.rs b/livekit/src/room/participant/remote_participant.rs index 9d4c820..a52babe 100644 --- a/livekit/src/room/participant/remote_participant.rs +++ b/livekit/src/room/participant/remote_participant.rs @@ -6,6 +6,7 @@ use livekit_webrtc as rtc; use parking_lot::RwLockReadGuard; use rtc::prelude::MediaStreamTrack; use std::collections::{HashMap, HashSet}; +use std::fmt::Debug; use std::sync::Arc; use std::time::Duration; use tokio::sync::mpsc; @@ -14,11 +15,21 @@ use tracing::{debug, error, instrument, Level}; const ADD_TRACK_TIMEOUT: Duration = Duration::from_secs(5); -#[derive(Debug, Clone)] +#[derive(Clone)] pub struct RemoteParticipant { inner: Arc, } +impl Debug for RemoteParticipant { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("RemoteParticipant") + .field("sid", &self.sid()) + .field("identity", &self.identity()) + .field("name", &self.name()) + .finish() + } +} + impl RemoteParticipant { pub(crate) fn new( sid: ParticipantSid, @@ -99,7 +110,6 @@ impl RemoteParticipant { unreachable!() } } - _ => unreachable!(), }; debug!("starting track: {:?}", sid); diff --git a/livekit/src/room/publication/local.rs b/livekit/src/room/publication/local.rs index 07986a3..1ad8f56 100644 --- a/livekit/src/room/publication/local.rs +++ b/livekit/src/room/publication/local.rs @@ -9,6 +9,8 @@ use std::sync::Arc; #[derive(Debug)] struct LocalTrackPublicationInner { publication_inner: TrackPublicationInner, + + #[allow(unused)] // TODO(theomonnom) options: Mutex, } @@ -89,6 +91,7 @@ impl LocalTrackPublication { self.inner.publication_inner.update_track(track); } + #[allow(dead_code)] #[inline] pub(crate) fn update_info(&self, info: proto::TrackInfo) { self.inner.publication_inner.update_info(info); diff --git a/livekit/src/room/room_session.rs b/livekit/src/room/room_session.rs deleted file mode 100644 index c9689da..0000000 --- a/livekit/src/room/room_session.rs +++ /dev/null @@ -1,563 +0,0 @@ -use crate::participant::ConnectionQuality; -use crate::prelude::*; -use crate::rtc_engine::{EngineEvent, EngineEvents, EngineResult, RtcEngine}; -use crate::signal_client::SignalOptions; -use crate::{RoomError, RoomEvent, RoomResult, SimulateScenario}; -use livekit_protocol as proto; -use livekit_protocol::observer::Dispatcher; -use parking_lot::{Mutex, RwLock, RwLockReadGuard}; -use std::collections::HashMap; -use std::sync::atomic::{AtomicU8, Ordering}; -use std::sync::Arc; -use tokio::sync::{mpsc, oneshot}; -use tokio::task::JoinHandle; -use tracing::{error, info, instrument, trace, Level}; - -#[derive(Debug, Clone, Copy, Eq, PartialEq)] -pub enum ConnectionState { - Disconnected, - Connected, - Reconnecting, - Unknown, -} - -/// Internal representation of a RoomSession -#[derive(Debug)] -struct SessionInner { - state: AtomicU8, // ConnectionState - sid: Mutex, - name: Mutex, - metadata: Mutex, - participants: RwLock>, - participants_tasks: RwLock, oneshot::Sender<()>)>>, - active_speakers: RwLock>, - rtc_engine: Arc, - local_participant: LocalParticipant, - dispatcher: Dispatcher, -} - -#[derive(Debug)] -pub(crate) struct SessionHandle { - session: RoomSession, - session_task: JoinHandle<()>, - close_emitter: oneshot::Sender<()>, -} - -/// RoomSession represents a connection to a room. -/// It can be cloned and shared across threads. -#[derive(Debug, Clone)] -pub struct RoomSession { - inner: Arc, -} - -impl SessionHandle { - pub async fn connect(url: &str, token: &str) -> RoomResult { - let (rtc_engine, engine_events) = RtcEngine::new(); - let rtc_engine = Arc::new(rtc_engine); - rtc_engine - .connect(url, token, SignalOptions::default()) - .await?; - - let join_response = rtc_engine.join_response().unwrap(); - let pi = join_response.participant.unwrap().clone(); - let local_participant = LocalParticipant::new( - rtc_engine.clone(), - pi.sid.into(), - pi.identity.into(), - pi.name, - pi.metadata, - ); - - let room_info = join_response.room.unwrap(); - let inner = Arc::new(SessionInner { - state: AtomicU8::new(ConnectionState::Disconnected as u8), - sid: Mutex::new(room_info.sid.into()), - name: Mutex::new(room_info.name), - metadata: Mutex::new(room_info.metadata), - participants: Default::default(), - participants_tasks: Default::default(), - active_speakers: Default::default(), - rtc_engine, - local_participant, - dispatcher: Default::default(), - }); - - for pi in join_response.other_participants { - let participant = { - let pi = pi.clone(); - inner.create_participant(pi.sid.into(), pi.identity.into(), pi.name, pi.metadata) - }; - participant.update_info(pi.clone()); - } - - let (close_emitter, close_receiver) = oneshot::channel(); - let session_task = tokio::spawn(inner.clone().room_task(engine_events, close_receiver)); - - inner.update_connection_state(ConnectionState::Connected); - - let session = Self { - session: RoomSession { inner }, - session_task, - close_emitter, - }; - Ok(session) - } - - pub async fn close(self) { - self.session.inner.close().await; - let _ = self.close_emitter.send(()); - let _ = self.session_task.await; - } - - pub fn subscribe(&self) -> mpsc::UnboundedReceiver { - self.session.inner.dispatcher.register() - } - - pub fn session(&self) -> RoomSession { - self.session.clone() - } -} - -impl RoomSession { - pub fn sid(&self) -> RoomSid { - self.inner.sid.lock().clone() - } - - pub fn name(&self) -> String { - self.inner.name.lock().clone() - } - - pub fn metadata(&self) -> String { - self.inner.metadata.lock().clone() - } - - pub fn local_participant(&self) -> LocalParticipant { - self.inner.local_participant.clone() - } - - pub fn connection_state(&self) -> ConnectionState { - self.inner.state.load(Ordering::Acquire).try_into().unwrap() - } - - pub fn participants(&self) -> RwLockReadGuard> { - self.inner.participants.read() - } - - pub async fn simulate_scenario(&self, scenario: SimulateScenario) -> EngineResult<()> { - self.inner.rtc_engine.simulate_scenario(scenario).await - } -} - -impl SessionInner { - #[instrument(level = Level::DEBUG)] - async fn room_task( - self: Arc, - mut engine_events: EngineEvents, - mut close_receiver: oneshot::Receiver<()>, - ) { - loop { - tokio::select! { - res = engine_events.recv() => { - if let Some(event) = res { - if let Err(err) = self.on_engine_event(event).await { - error!("failed to handle engine event: {:?}", err); - } - } - }, - _ = &mut close_receiver => { - trace!("closing room_task"); - break; - } - } - } - } - - /// Listen to the Participant events and forward them to the Room Dispatcher - #[instrument(level = Level::DEBUG)] - async fn participant_task( - self: Arc, - participant: Participant, - mut participant_events: mpsc::UnboundedReceiver, - mut close_rx: oneshot::Receiver<()>, - ) { - loop { - tokio::select! { - res = participant_events.recv() => { - if let Some(event) = res { - if let Err(err) = self.on_participant_event(&participant, event).await { - error!("failed to handle participant event for {:?}: {:?}", participant.sid(), err); - } - } - }, - _ = &mut close_rx => { - trace!("closing participant_task for {:?}", participant.sid()); - break; - }, - } - } - } - - #[instrument(level = Level::DEBUG)] - async fn on_participant_event( - self: &Arc, - participant: &Participant, - event: ParticipantEvent, - ) -> RoomResult<()> { - if let Participant::Remote(remote_participant) = participant { - match event { - ParticipantEvent::TrackPublished { publication } => { - self.dispatcher.dispatch(&RoomEvent::TrackPublished { - participant: remote_participant.clone(), - publication, - }); - } - ParticipantEvent::TrackUnpublished { publication } => { - self.dispatcher.dispatch(&RoomEvent::TrackUnpublished { - participant: remote_participant.clone(), - publication, - }); - } - ParticipantEvent::TrackSubscribed { track, publication } => { - self.dispatcher.dispatch(&RoomEvent::TrackSubscribed { - participant: remote_participant.clone(), - track, - publication, - }); - } - ParticipantEvent::TrackUnsubscribed { track, publication } => { - self.dispatcher.dispatch(&RoomEvent::TrackUnsubscribed { - participant: remote_participant.clone(), - track, - publication, - }); - } - _ => {} - }; - } - - Ok(()) - } - - #[instrument(level = Level::DEBUG)] - async fn on_engine_event(self: &Arc, event: EngineEvent) -> RoomResult<()> { - match event { - EngineEvent::ParticipantUpdate { updates } => self.handle_participant_update(updates), - EngineEvent::MediaTrack { - track, - stream, - receiver: _, - } => { - let stream_id = stream.id(); - let lk_stream_id = unpack_stream_id(&stream_id); - if lk_stream_id.is_none() { - Err(RoomError::Internal(format!( - "MediaTrack event with invalid track_id: {:?}", - &stream_id - )))?; - } - - let (participant_sid, track_sid) = lk_stream_id.unwrap(); - let track_sid = track_sid.to_owned().into(); - let remote_participant = self.get_participant(&participant_sid.to_string().into()); - - if let Some(remote_participant) = remote_participant { - tokio::spawn(async move { - remote_participant - .add_subscribed_media_track(track_sid, track) - .await; - }); - } else { - // The server should send participant updates before sending a new offer - // So this should never happen. - Err(RoomError::Internal(format!( - "AddTrack event with invalid participant_sid: {:?}", - participant_sid - )))?; - } - } - EngineEvent::Resuming => { - if self.update_connection_state(ConnectionState::Reconnecting) { - self.dispatcher.dispatch(&RoomEvent::Reconnecting); - } - } - EngineEvent::Resumed => { - self.update_connection_state(ConnectionState::Connected); - self.dispatcher.dispatch(&RoomEvent::Reconnected); - - // TODO(theomonnom): Update subscriptions settings - // TODO(theomonnom): Send sync state - } - EngineEvent::Restarting => self.handle_restarting(), - EngineEvent::Restarted => self.handle_restarted(), - EngineEvent::Disconnected => self.handle_disconnected(), - EngineEvent::Data { - payload, - kind, - participant_sid, - } => { - let payload = Arc::new(payload); - if let Some(participant) = self.get_participant(&participant_sid.into()) { - self.dispatcher.dispatch(&RoomEvent::DataReceived { - payload: payload.clone(), - kind, - participant: participant.clone(), - }); - - participant.on_data_received(payload, kind); - } - } - EngineEvent::SpeakersChanged { speakers } => self.handle_speakers_changed(speakers), - EngineEvent::ConnectionQuality { updates } => { - self.handle_connection_quality_update(updates) - } - } - - Ok(()) - } - - #[instrument(level = Level::DEBUG)] - async fn close(&self) { - self.rtc_engine.close().await; - } - - /// Change the connection state and emit an event - /// Does nothing if the state is already the same - #[instrument(level = Level::DEBUG)] - fn update_connection_state(&self, state: ConnectionState) -> bool { - let old_state = self.state.load(Ordering::Acquire); - if old_state == state as u8 { - return false; - } - - self.state.store(state as u8, Ordering::Release); - self.dispatcher - .dispatch(&RoomEvent::ConnectionStateChanged(state)); - return true; - } - - /// Update the participants inside a Room. - /// It'll create, update or remove a participant - /// It also update the participant tracks. - #[instrument(level = Level::DEBUG)] - fn handle_participant_update(self: &Arc, updates: Vec) { - for pi in updates { - if pi.sid == self.local_participant.sid() - || pi.identity == self.local_participant.identity() - { - self.local_participant.clone().update_info(pi); - continue; - } - - let remote_participant = self.get_participant(&pi.sid.clone().into()); - - if let Some(remote_participant) = remote_participant { - if pi.state == proto::participant_info::State::Disconnected as i32 { - // Participant disconnected - info!("Participant disconnected: {}", pi.sid); - self.clone() - .handle_participant_disconnect(remote_participant) - } else { - // Participant is already connected, update the it - remote_participant.update_info(pi.clone()); - } - } else { - // Create a new participant - info!("Participant connected: {}", pi.sid); - let remote_participant = { - let pi = pi.clone(); - self.create_participant(pi.sid.into(), pi.identity.into(), pi.name, pi.metadata) - }; - - let _ = self - .dispatcher - .dispatch(&RoomEvent::ParticipantConnected(remote_participant.clone())); - - remote_participant.update_info(pi.clone()); - } - } - } - - /// Active speakers changed - /// Update the participants & sort the active_speakers by audio_level - #[instrument(level = Level::DEBUG)] - fn handle_speakers_changed(&self, speakers_info: Vec) { - let mut speakers = Vec::new(); - - for speaker in speakers_info { - let participant = { - if speaker.sid == self.local_participant.sid() { - Participant::Local(self.local_participant.clone()) - } else { - if let Some(participant) = self.get_participant(&speaker.sid.into()) { - Participant::Remote(participant) - } else { - continue; - } - } - }; - - participant.set_speaking(speaker.active); - participant.set_audio_level(speaker.level); - - if speaker.active { - speakers.push(participant); - } - } - - speakers.sort_by(|a, b| a.audio_level().partial_cmp(&b.audio_level()).unwrap()); - *self.active_speakers.write() = speakers.clone(); - - let _ = self - .dispatcher - .dispatch(&RoomEvent::ActiveSpeakersChanged { speakers }); - } - - /// Handle a connection quality update - /// Emit ConnectionQualityChanged event for the concerned participants - #[instrument(level = Level::DEBUG)] - fn handle_connection_quality_update(&self, updates: Vec) { - for update in updates { - let participant = { - if update.participant_sid == self.local_participant.sid() { - Participant::Local(self.local_participant.clone()) - } else { - if let Some(participant) = self.get_participant(&update.participant_sid.into()) - { - Participant::Remote(participant) - } else { - continue; - } - } - }; - - let quality: ConnectionQuality = proto::ConnectionQuality::from_i32(update.quality) - .unwrap() - .into(); - - participant.set_connection_quality(quality); - self.dispatcher - .dispatch(&RoomEvent::ConnectionQualityChanged { - participant, - quality, - }); - } - } - - #[instrument(level = Level::DEBUG)] - fn handle_restarting(self: &Arc) { - // Remove existing participants/subscriptions on full reconnect - for (_, participant) in self.participants.read().iter() { - self.clone() - .handle_participant_disconnect(participant.clone()); - } - - if self.update_connection_state(ConnectionState::Reconnecting) { - self.dispatcher.dispatch(&RoomEvent::Reconnecting); - } - } - - #[instrument(level = Level::DEBUG)] - fn handle_restarted(self: &Arc) { - // Full reconnect succeeded! - let join_response = self.rtc_engine.join_response().unwrap(); - - self.update_connection_state(ConnectionState::Connected); - self.dispatcher.dispatch(&RoomEvent::Reconnected); - - if let Some(pi) = join_response.participant { - self.local_participant.update_info(pi); // The sid may have changed - } - - self.handle_participant_update(join_response.other_participants); - - // TODO(theomonnom): unpublish & republish tracks - } - - #[instrument(level = Level::DEBUG)] - fn handle_disconnected(&self) { - if self.state.load(Ordering::Acquire) == ConnectionState::Disconnected as u8 { - return; - } - - self.update_connection_state(ConnectionState::Disconnected); - self.dispatcher.dispatch(&RoomEvent::Disconnected); - } - - /// Create a new participant - /// Also add it to the participants list - #[instrument(level = Level::DEBUG)] - fn create_participant( - self: &Arc, - sid: ParticipantSid, - identity: ParticipantIdentity, - name: String, - metadata: String, - ) -> RemoteParticipant { - let participant = RemoteParticipant::new(sid.clone(), identity, name, metadata); - - // Create the participant task - let (close_tx, close_rx) = oneshot::channel(); - let participant_task = tokio::spawn(self.clone().participant_task( - Participant::Remote(participant.clone()), - participant.register_observer(), - close_rx, - )); - self.participants_tasks - .write() - .insert(sid.clone(), (participant_task, close_tx)); - - self.participants.write().insert(sid, participant.clone()); - participant - } - - /// A participant has disconnected - /// Cleanup the participant and emit an event - #[instrument(level = Level::DEBUG)] - fn handle_participant_disconnect(self: Arc, remote_participant: RemoteParticipant) { - tokio::spawn(async move { - for (sid, _) in &*remote_participant.tracks() { - remote_participant.unpublish_track(&sid); - } - - // Close the participant task - if let Some((task, close_tx)) = self - .participants_tasks - .write() - .remove(&remote_participant.sid()) - { - let _ = close_tx.send(()); - let _ = task.await; - } - - self.participants.write().remove(&remote_participant.sid()); - self.dispatcher - .dispatch(&RoomEvent::ParticipantDisconnected(remote_participant)); - }); - } - - fn get_participant(&self, sid: &ParticipantSid) -> Option { - self.participants.read().get(sid).cloned() - } -} - -fn unpack_stream_id(stream_id: &str) -> Option<(&str, &str)> { - let split: Vec<&str> = stream_id.split('|').collect(); - if split.len() == 2 { - let participant_sid = split.get(0).unwrap(); - let track_sid = split.get(1).unwrap(); - Some((participant_sid, track_sid)) - } else { - None - } -} - -impl From for ConnectionState { - fn from(value: u8) -> Self { - match value { - 0 => ConnectionState::Disconnected, - 1 => ConnectionState::Connected, - 2 => ConnectionState::Reconnecting, - _ => ConnectionState::Unknown, - } - } -} diff --git a/livekit/src/room/track/local_audio_track.rs b/livekit/src/room/track/local_audio_track.rs index 161f08a..2158d1d 100644 --- a/livekit/src/room/track/local_audio_track.rs +++ b/livekit/src/room/track/local_audio_track.rs @@ -7,6 +7,7 @@ use livekit_protocol as proto; use livekit_webrtc as rtc; use parking_lot::Mutex; use rtc::audio_source::native::NativeAudioSource; +use std::fmt::Debug; use std::sync::Arc; use tokio::sync::mpsc; @@ -16,11 +17,21 @@ pub struct LocalAudioTrackInner { capture_options: Mutex, } -#[derive(Clone, Debug)] +#[derive(Clone)] pub struct LocalAudioTrack { inner: Arc, } +impl Debug for LocalAudioTrack { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("LocalAudioTrack") + .field("sid", &self.sid()) + .field("name", &self.name()) + .field("source", &self.source()) + .finish() + } +} + impl LocalAudioTrack { pub(crate) fn new( name: String, diff --git a/livekit/src/room/track/local_video_track.rs b/livekit/src/room/track/local_video_track.rs index 53a7e2b..05b6ceb 100644 --- a/livekit/src/room/track/local_video_track.rs +++ b/livekit/src/room/track/local_video_track.rs @@ -6,6 +6,7 @@ use livekit_webrtc as rtc; use livekit_webrtc::peer_connection_factory::native::PeerConnectionFactoryExt; use parking_lot::Mutex; use rtc::video_source::native::NativeVideoSource; +use std::fmt::Debug; use std::sync::Arc; use tokio::sync::mpsc; @@ -15,11 +16,21 @@ struct LocalVideoTrackInner { capture_options: Mutex, } -#[derive(Clone, Debug)] +#[derive(Clone)] pub struct LocalVideoTrack { inner: Arc, } +impl Debug for LocalVideoTrack { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("LocalVideoTrack") + .field("sid", &self.sid()) + .field("name", &self.name()) + .field("source", &self.source()) + .finish() + } +} + impl LocalVideoTrack { pub fn new( name: String, diff --git a/livekit/src/room/track/mod.rs b/livekit/src/room/track/mod.rs index 70098dc..46504d2 100644 --- a/livekit/src/room/track/mod.rs +++ b/livekit/src/room/track/mod.rs @@ -9,10 +9,10 @@ use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use thiserror::Error; use tokio::sync::mpsc; -pub mod local_audio_track; -pub mod local_video_track; -pub mod remote_audio_track; -pub mod remote_video_track; +mod local_audio_track; +mod local_video_track; +mod remote_audio_track; +mod remote_video_track; pub use local_audio_track::*; pub use local_video_track::*; diff --git a/livekit/src/room/track/remote_audio_track.rs b/livekit/src/room/track/remote_audio_track.rs index 8fd5042..eb88acd 100644 --- a/livekit/src/room/track/remote_audio_track.rs +++ b/livekit/src/room/track/remote_audio_track.rs @@ -2,14 +2,25 @@ use super::TrackInner; use crate::prelude::*; use livekit_protocol as proto; use livekit_webrtc as rtc; +use std::fmt::Debug; use std::sync::Arc; use tokio::sync::mpsc; -#[derive(Clone, Debug)] +#[derive(Clone)] pub struct RemoteAudioTrack { pub(crate) inner: Arc, } +impl Debug for RemoteAudioTrack { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("RemoteAudioTrack") + .field("sid", &self.sid()) + .field("name", &self.name()) + .field("source", &self.source()) + .finish() + } +} + impl RemoteAudioTrack { pub(crate) fn new( sid: TrackSid, @@ -90,12 +101,14 @@ impl RemoteAudioTrack { true } + #[allow(dead_code)] #[inline] pub(crate) fn transceiver(&self) -> Option { self.inner.transceiver() } #[inline] + #[allow(dead_code)] pub(crate) fn update_transceiver( &self, transceiver: Option, diff --git a/livekit/src/room/track/remote_video_track.rs b/livekit/src/room/track/remote_video_track.rs index f7ed25b..c6d9e89 100644 --- a/livekit/src/room/track/remote_video_track.rs +++ b/livekit/src/room/track/remote_video_track.rs @@ -2,14 +2,25 @@ use super::TrackInner; use crate::prelude::*; use livekit_protocol as proto; use livekit_webrtc as rtc; +use std::fmt::Debug; use std::sync::Arc; use tokio::sync::mpsc; -#[derive(Clone, Debug)] +#[derive(Clone)] pub struct RemoteVideoTrack { pub(crate) inner: Arc, } +impl Debug for RemoteVideoTrack { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("RemoteVideoTrack") + .field("sid", &self.sid()) + .field("name", &self.name()) + .field("source", &self.source()) + .finish() + } +} + impl RemoteVideoTrack { pub(crate) fn new( sid: TrackSid, @@ -90,11 +101,13 @@ impl RemoteVideoTrack { true } + #[allow(dead_code)] #[inline] pub(crate) fn transceiver(&self) -> Option { self.inner.transceiver() } + #[allow(dead_code)] #[inline] pub(crate) fn update_transceiver( &self, diff --git a/livekit/src/rtc_engine/mod.rs b/livekit/src/rtc_engine/mod.rs index e022135..7b3dbf6 100644 --- a/livekit/src/rtc_engine/mod.rs +++ b/livekit/src/rtc_engine/mod.rs @@ -7,6 +7,7 @@ use livekit_protocol as proto; use livekit_webrtc::prelude::*; use livekit_webrtc::session_description::SdpParseError; use parking_lot::Mutex; +use std::fmt::Debug; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::time::Duration; @@ -99,7 +100,6 @@ struct EngineHandle { close_sender: oneshot::Sender<()>, } -#[derive(Debug)] struct EngineInner { lk_runtime: Arc, session_info: Mutex>, // Last/Current Sessioninfo @@ -113,6 +113,17 @@ struct EngineInner { reconnect_interval: Mutex, } +impl Debug for EngineInner { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("EngineInner") + .field("session_info", &self.session_info) + .field("opened", &self.opened) + .field("reconnecting", &self.reconnecting) + .field("full_reconnect", &self.full_reconnect) + .finish() + } +} + #[derive(Debug)] pub struct RtcEngine { inner: Arc, diff --git a/livekit/src/rtc_engine/peer_transport.rs b/livekit/src/rtc_engine/peer_transport.rs index 176a9eb..9ad8ba4 100644 --- a/livekit/src/rtc_engine/peer_transport.rs +++ b/livekit/src/rtc_engine/peer_transport.rs @@ -4,7 +4,7 @@ use std::fmt::{Debug, Formatter}; use std::time::Duration; use tracing::{event, Level}; -const NEGOTIATION_FREQUENCY: Duration = Duration::from_millis(150); +const _NEGOTIATION_FREQUENCY: Duration = Duration::from_millis(150); pub type OnOfferCreated = Box; diff --git a/livekit/src/rtc_engine/rtc_events.rs b/livekit/src/rtc_engine/rtc_events.rs index 0a4d3d7..97b811c 100644 --- a/livekit/src/rtc_engine/rtc_events.rs +++ b/livekit/src/rtc_engine/rtc_events.rs @@ -3,7 +3,7 @@ use crate::rtc_engine::peer_transport::OnOfferCreated; use livekit_protocol as proto; use livekit_webrtc::{self as rtc, prelude::*}; use tokio::sync::mpsc; -use tracing::{debug, error}; +use tracing::{error}; pub type RtcEmitter = mpsc::UnboundedSender; pub type RtcEvents = mpsc::UnboundedReceiver; diff --git a/livekit/src/rtc_engine/rtc_session.rs b/livekit/src/rtc_engine/rtc_session.rs index c9de2bc..599684c 100644 --- a/livekit/src/rtc_engine/rtc_session.rs +++ b/livekit/src/rtc_engine/rtc_session.rs @@ -14,6 +14,7 @@ use prost::Message; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::convert::TryInto; +use std::fmt::Debug; use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use std::sync::Arc; use std::time::Duration; @@ -104,7 +105,6 @@ pub struct SessionInfo { } /// Fields shared with rtc_task and signal_task -#[derive(Debug)] struct SessionInner { info: SessionInfo, signal_client: Arc, @@ -128,6 +128,18 @@ struct SessionInner { closed: AtomicBool, emitter: SessionEmitter, } + +impl Debug for SessionInner { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("SessionInner") + .field("info", &self.info) + .field("pc_state", &self.pc_state) + .field("has_published", &self.has_published) + .field("closed", &self.closed) + .finish() + } +} + /// This struct holds a WebRTC session /// The session changes at every reconnection /// @@ -317,6 +329,7 @@ impl RtcSession { &self.inner.info } + #[allow(dead_code)] #[inline] pub fn state(&self) -> PeerState { self.inner @@ -326,21 +339,25 @@ impl RtcSession { .unwrap() } + #[allow(dead_code)] #[inline] pub fn publisher(&self) -> &AsyncMutex { &self.inner.publisher_pc } + #[allow(dead_code)] #[inline] pub fn subscriber(&self) -> &AsyncMutex { &self.inner.subscriber_pc } + #[allow(dead_code)] #[inline] pub fn signal_client(&self) -> &Arc { &self.inner.signal_client } + #[allow(dead_code)] #[inline] pub fn data_channel(&self, kind: proto::data_packet::Kind) -> &DataChannel { &self.inner.data_channel(kind) diff --git a/livekit/src/signal_client/mod.rs b/livekit/src/signal_client/mod.rs index 5ac3804..b429435 100644 --- a/livekit/src/signal_client/mod.rs +++ b/livekit/src/signal_client/mod.rs @@ -1,6 +1,6 @@ use crate::signal_client::signal_stream::SignalStream; use livekit_protocol as proto; -use livekit_webrtc::prelude::*; + use parking_lot::RwLock; use std::fmt::Debug; use std::time::Duration; @@ -104,9 +104,10 @@ impl SignalClient { // TODO(theomonnom): enqueue message } + /*#[allow(dead_code)] pub async fn clear_queue(&self) { // TODO(theomonnom): impl - } + }*/ #[instrument(level = Level::DEBUG)] pub async fn flush_queue(&self) {