| 1 | use std::future::Future; |
| 2 | |
| 3 | use futures_util::{StreamExt, stream::FuturesUnordered}; |
| 4 | use matrix_sdk::SendOutsideWasm; |
| 5 | pub(super) use matrix_sdk::executor::{AbortHandle, JoinError, JoinHandle, spawn}; |
| 6 | |
| 7 | #[cfg(not(target_family = "wasm"))] |
| 8 | pub(crate) type EngineBoxFuture<'a, T> = futures_util::future::BoxFuture<'a, T>; |
| 9 | #[cfg(target_family = "wasm")] |
| 10 | pub(crate) type EngineBoxFuture<'a, T> = futures_util::future::LocalBoxFuture<'a, T>; |
| 11 | |
| 12 | #[cfg(not(target_family = "wasm"))] |
| 13 | pub(super) fn is_panic(error: &JoinError) -> bool { |
| 14 | error.is_panic() |
| 15 | } |
| 16 | |
| 17 | /// the sdk's wasm join error has no payload, and a real panic traps before it gets here |
| 18 | #[cfg(target_family = "wasm")] |
| 19 | pub(super) fn is_panic(error: &JoinError) -> bool { |
| 20 | !error.is_cancelled() |
| 21 | } |
| 22 | |
| 23 | #[cfg(not(target_family = "wasm"))] |
| 24 | pub(super) fn resume_unwind(error: JoinError) -> ! { |
| 25 | std::panic::resume_unwind(error.into_panic()) |
| 26 | } |
| 27 | |
| 28 | #[cfg(target_family = "wasm")] |
| 29 | pub(super) fn resume_unwind(error: JoinError) -> ! { |
| 30 | panic!("{error}") |
| 31 | } |
| 32 | |
| 33 | /// a `JoinSet` that runs on wasm too, and like it aborts whatever's left when dropped |
| 34 | pub(super) struct TaskSet<T: 'static> { |
| 35 | handles: FuturesUnordered<JoinHandle<T>>, |
| 36 | } |
| 37 | |
| 38 | impl<T: 'static> TaskSet<T> { |
| 39 | pub(super) fn new() -> Self { |
| 40 | Self { |
| 41 | handles: FuturesUnordered::new(), |
| 42 | } |
| 43 | } |
| 44 | |
| 45 | pub(super) fn spawn( |
| 46 | &mut self, |
| 47 | task: impl Future<Output = T> + SendOutsideWasm + 'static, |
| 48 | ) -> AbortHandle |
| 49 | where |
| 50 | T: SendOutsideWasm, |
| 51 | { |
| 52 | let handle = spawn(task); |
| 53 | let abort = handle.abort_handle(); |
| 54 | self.handles.push(handle); |
| 55 | abort |
| 56 | } |
| 57 | |
| 58 | pub(super) fn is_empty(&self) -> bool { |
| 59 | self.handles.is_empty() |
| 60 | } |
| 61 | |
| 62 | pub(super) async fn join_next(&mut self) -> Option<Result<T, JoinError>> { |
| 63 | self.handles.next().await |
| 64 | } |
| 65 | |
| 66 | fn abort_all(&self) { |
| 67 | for handle in &self.handles { |
| 68 | handle.abort(); |
| 69 | } |
| 70 | } |
| 71 | |
| 72 | pub(super) async fn shutdown(&mut self) { |
| 73 | self.abort_all(); |
| 74 | while self.handles.next().await.is_some() {} |
| 75 | } |
| 76 | } |
| 77 | |
| 78 | impl<T: 'static> Drop for TaskSet<T> { |
| 79 | fn drop(&mut self) { |
| 80 | self.abort_all(); |
| 81 | } |
| 82 | } |
| 83 | |
| 84 | /// a store cipher's pbkdf2 holds its thread for over a second in a debug build, so natively it |
| 85 | /// runs beside the runtime and two stores can derive at once |
| 86 | #[cfg(not(target_family = "wasm"))] |
| 87 | pub(crate) async fn derive<T: Send + 'static>(work: impl FnOnce() -> T + Send + 'static) -> T { |
| 88 | tokio::task::spawn_blocking(work) |
| 89 | .await |
| 90 | .unwrap_or_else(|error| std::panic::resume_unwind(error.into_panic())) |
| 91 | } |
| 92 | |
| 93 | #[cfg(target_family = "wasm")] |
| 94 | pub(crate) async fn derive<T>(work: impl FnOnce() -> T) -> T { |
| 95 | work() |
| 96 | } |