Add more threading utilities (extended)

crates/bitwarden-threading/Cargo.toml

@@ -9,15 +9,20 @@ repository.workspace = true
 license-file.workspace = true
 keywords.workspace = true
 
+[features]
+time = ["gloo-timers"]
+
 [dependencies]
 bitwarden-error = { workspace = true }
 log = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 thiserror = { workspace = true }
 tokio = { features = ["sync", "time", "rt"], workspace = true }
+tokio-util = { version = "0.7.15" }
 
 [target.'cfg(target_arch="wasm32")'.dependencies]
+gloo-timers = { version = "0.3.0", features = ["futures"], optional = true }
 js-sys = { workspace = true }
 tsify-next = { workspace = true }
 wasm-bindgen = { workspace = true }

crates/bitwarden-threading/src/cancellation_token.rs

@@ -0,0 +1,147 @@
+pub use tokio_util::sync::CancellationToken;
+
+#[cfg(target_arch = "wasm32")]
+pub mod wasm {
+    use tokio::select;
+    use tokio_util::sync::DropGuard;
+    use wasm_bindgen::prelude::*;
+    use wasm_bindgen_futures::spawn_local;
+
+    use super::*;
+
+    #[wasm_bindgen]
+    extern "C" {
+        #[wasm_bindgen(js_namespace = console, js_name = log)]
+        pub fn console_log(message: &str);
+
+        #[wasm_bindgen]
+        #[derive(Clone)]
+        pub type AbortController;
+
+        #[wasm_bindgen(constructor)]
+        pub fn new() -> AbortController;
+
+        #[wasm_bindgen(method, getter)]
+        pub fn signal(this: &AbortController) -> AbortSignal;
+
+        #[wasm_bindgen(method, js_name = abort)]
+        pub fn abort(this: &AbortController, reason: JsValue);
+
+        #[wasm_bindgen]
+        pub type AbortSignal;
+
+        #[wasm_bindgen(method, getter)]
+        pub fn aborted(this: &AbortSignal) -> bool;
+
+        #[wasm_bindgen(method, js_name = addEventListener)]
+        pub fn add_event_listener(
+            this: &AbortSignal,
+            event_type: &str,
+            callback: &Closure<dyn FnMut()>,
+        );
+    }
+
+    pub trait CancellationTokenExt {
+        /// Converts a `CancellationToken` to an `AbortController`.
+        /// The signal only travels in one direction: `CancellationToken` -> `AbortController`,
+        /// i.e. the `AbortController` will be aborted when the `CancellationToken` is cancelled.
+        fn to_abort_controller(self) -> AbortController;
+        fn to_bidirectional_abort_controller(self) -> (AbortController, DropGuard);
+    }
+
+    impl CancellationTokenExt for CancellationToken {
+        fn to_abort_controller(self) -> AbortController {
+            let controller = AbortController::new();
+
+            let token_clone = self.clone();
+            let controller_clone = controller.clone();
+
+            let closure_dropped_token = CancellationToken::new();
+            let drop_guard = closure_dropped_token.clone().drop_guard();
+
+            spawn_local(async move {
+                select! {
+                    _ = token_clone.cancelled() => {
+                        controller_clone.abort(JsValue::from("Rust token cancelled"));
+                    },
+                    _ = closure_dropped_token.cancelled() => {}
+                }
+            });
+
+            let closure = Closure::new({
+                let _drop_guard = drop_guard;
+                move || {
+                    // Do nothing
+                }
+            });
+            controller.signal().add_event_listener("abort", &closure);
+            closure.forget(); // Transfer ownership to the JS runtime
+
+            controller
+        }
+
+        fn to_bidirectional_abort_controller(self) -> (AbortController, DropGuard) {
+            let controller = AbortController::new();
+
+            let drop_guard = connect_token_and_controller(self.clone(), controller.clone());
+
+            (controller, drop_guard)
+        }
+    }
+
+    pub trait AbortControllerExt {
+        fn to_cancellation_token(&self) -> CancellationToken;
+        fn to_bidirectional_cancellation_token(&self) -> (CancellationToken, DropGuard);
+    }
+
+    impl AbortControllerExt for AbortController {
+        fn to_cancellation_token(&self) -> CancellationToken {
+            let token = CancellationToken::new();
+
+            let token_clone = token.clone();
+            let closure = Closure::new(move || {
+                token_clone.cancel();
+            });
+            self.signal().add_event_listener("abort", &closure);
+            closure.forget(); // Transfer ownership to the JS runtime
+
+            token
+        }
+
+        fn to_bidirectional_cancellation_token(&self) -> (CancellationToken, DropGuard) {
+            let token = CancellationToken::new();
+
+            let drop_guard = connect_token_and_controller(token.clone(), self.clone());
+
+            (token, drop_guard)
+        }
+    }
+
+    fn connect_token_and_controller(
+        token: CancellationToken,
+        controller: AbortController,
+    ) -> DropGuard {
+        let token_clone = token.clone();
+        let controller_clone = controller.clone();
+
+        let guarded_token = CancellationToken::new();
+        let drop_guard = guarded_token.clone().drop_guard();
+
+        spawn_local(async move {
+            select! {
+                _ = token_clone.cancelled() => {
+                    controller_clone.abort(JsValue::from("Rust token cancelled"));
+                },
+                _ = guarded_token.cancelled() => {}
+            }
+        });
+
+        let closure = Closure::new(move || {
+            token.cancel();
+        });
+        controller.signal().add_event_listener("abort", &closure);
+        closure.forget(); // Transfer ownership to the JS runtime
+
+        drop_guard
+    }
+}

crates/bitwarden-threading/src/lib.rs

@@ -1,3 +1,5 @@
+pub mod cancellation_token;
 mod thread_bound_runner;
+pub mod time;
 
 pub use thread_bound_runner::ThreadBoundRunner;

crates/bitwarden-threading/src/time.rs

@@ -0,0 +1,56 @@
+use std::time::Duration;
+
+#[cfg(not(target_arch = "wasm32"))]
+pub async fn sleep(duration: Duration) {
+    tokio::time::sleep(duration).await;
+}
+
+#[cfg(target_arch = "wasm32")]
+pub async fn sleep(duration: Duration) {
+    use gloo_timers::future::sleep;
+
+    sleep(duration).await;
+}
+
+#[cfg(test)]
+mod test {
+    use wasm_bindgen_test::wasm_bindgen_test;
+
+    #[wasm_bindgen_test]
+    #[allow(dead_code)] // Not actually dead, but rust-analyzer doesn't understand `wasm_bindgen_test`
+    async fn should_sleep_wasm() {
+        use js_sys::Date;
+
+        use super::*;
+
+        console_error_panic_hook::set_once();
+        let start = Date::now();
+
+        sleep(Duration::from_millis(100)).await;
+
+        let end = Date::now();
+        let elapsed = end - start;
+
+        assert!(elapsed >= 90.0, "Elapsed time was less than expected");
+    }
+
+    // #[cfg(not(target_arch = "wasm32"))]
+    #[tokio::test]
+    async fn should_sleep_tokio() {
+        use std::time::Instant;
+
+        use super::*;
+
+        let start = Instant::now();
+
+        sleep(Duration::from_millis(100)).await;
+
+        let end = Instant::now();
+        let elapsed = end.duration_since(start);
+
+        assert!(
+            elapsed >= Duration::from_millis(90),
+            "Elapsed time was less than expected"
+        );
+    }
+}

crates/bitwarden-threading/tests/cancellation_token/mod.rs

@@ -0,0 +1 @@
+mod wasm;