Update matrix-sdk-common utils that are in existing PRs already

Daniel Salinas · Daniel Salinas · commit 5a40876e57ee · 2025-05-28T13:32:14.000-04:00
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -25,6 +25,7 @@ as_variant = "1.3.0"
 assert-json-diff = "2.0.2"
 assert_matches = "1.5.0"
 assert_matches2 = "0.1.2"
+async-compat = "0.2.4"
 async-rx = "0.1.3"
 async-stream = "0.3.5"
 async-trait = "0.1.85"
diff --git a/bindings/matrix-sdk-ffi/Cargo.toml b/bindings/matrix-sdk-ffi/Cargo.toml
@@ -23,12 +23,13 @@ bundled-sqlite = ["matrix-sdk/bundled-sqlite"]
 [dependencies]
 anyhow.workspace = true
 as_variant.workspace = true
-async-compat = "0.2.4"
+async-compat.workspace = true
 extension-trait = "1.0.1"
 eyeball-im.workspace = true
 futures-util.workspace = true
 language-tags = "0.3.2"
 log-panics = { version = "2", features = ["with-backtrace"] }
+matrix-sdk-common.workspace = true
 matrix-sdk-ffi-macros.workspace = true
 matrix-sdk-ui = { workspace = true, features = ["uniffi"] }
 mime = "0.3.16"
diff --git a/bindings/matrix-sdk-ffi/src/client.rs b/bindings/matrix-sdk-ffi/src/client.rs
@@ -7,8 +7,7 @@ use std::{
 };
 
 use anyhow::{anyhow, Context as _};
-use async_compat::get_runtime_handle;
-use futures_util::{pin_mut, StreamExt};
+use futures_util::pin_mut;
 use matrix_sdk::{
     authentication::oauth::{
         AccountManagementActionFull, ClientId, OAuthAuthorizationData, OAuthSession,
@@ -41,6 +40,8 @@ use matrix_sdk::{
     AuthApi, AuthSession, Client as MatrixClient, SessionChange, SessionTokens,
     STATE_STORE_DATABASE_NAME,
 };
+use matrix_sdk_common::runtime::get_runtime_handle;
+use matrix_sdk_common::stream::StreamExt;
 use matrix_sdk_common::{AsyncTraitDeps, SendOutsideWasm, SyncOutsideWasm};
 use matrix_sdk_ui::{
     notification_client::{
diff --git a/bindings/matrix-sdk-ffi/src/client_builder.rs b/bindings/matrix-sdk-ffi/src/client_builder.rs
@@ -2,7 +2,6 @@ use std::{fs, num::NonZeroUsize, path::Path, sync::Arc, time::Duration};
 
 use matrix_sdk_common::{SendOutsideWasm, SyncOutsideWasm};
 
-use async_compat::get_runtime_handle;
 use futures_util::StreamExt;
 use matrix_sdk::{
     authentication::oauth::qrcode::{self, DeviceCodeErrorResponseType, LoginFailureReason},
@@ -21,6 +20,7 @@ use matrix_sdk::{
     Client as MatrixClient, ClientBuildError as MatrixClientBuildError, HttpError, IdParseError,
     RumaApiError, SqliteStoreConfig,
 };
+use matrix_sdk_common::runtime::get_runtime_handle;
 use ruma::api::error::{DeserializationError, FromHttpResponseError};
 use tracing::{debug, error};
 use zeroize::Zeroizing;
diff --git a/bindings/matrix-sdk-ffi/src/encryption.rs b/bindings/matrix-sdk-ffi/src/encryption.rs
@@ -1,11 +1,11 @@
 use std::sync::Arc;
 
-use async_compat::get_runtime_handle;
 use futures_util::StreamExt;
 use matrix_sdk::{
     encryption,
     encryption::{backups, recovery},
 };
+use matrix_sdk_common::runtime::get_runtime_handle;
 use matrix_sdk_common::{SendOutsideWasm, SyncOutsideWasm};
 use thiserror::Error;
 use tracing::{error, info};
diff --git a/bindings/matrix-sdk-ffi/src/room.rs b/bindings/matrix-sdk-ffi/src/room.rs
@@ -1,7 +1,6 @@
 use std::{collections::HashMap, pin::pin, sync::Arc};
 
 use anyhow::{Context, Result};
-use async_compat::get_runtime_handle;
 use futures_util::{pin_mut, StreamExt};
 use matrix_sdk::{
     crypto::LocalTrust,
@@ -13,6 +12,7 @@ use matrix_sdk::{
     PredecessorRoom as SdkPredecessorRoom, RoomHero as SdkRoomHero, RoomMemberships, RoomState,
     SuccessorRoom as SdkSuccessorRoom,
 };
+use matrix_sdk_common::runtime::get_runtime_handle;
 use matrix_sdk_ui::{
     timeline::{default_event_filter, RoomExt, TimelineBuilder},
     unable_to_decrypt_hook::UtdHookManager,
diff --git a/bindings/matrix-sdk-ffi/src/room_directory_search.rs b/bindings/matrix-sdk-ffi/src/room_directory_search.rs
@@ -15,10 +15,10 @@
 
 use std::{fmt::Debug, sync::Arc};
 
-use async_compat::get_runtime_handle;
 use eyeball_im::VectorDiff;
 use futures_util::StreamExt;
 use matrix_sdk::room_directory_search::RoomDirectorySearch as SdkRoomDirectorySearch;
+use matrix_sdk_common::runtime::get_runtime_handle;
 use matrix_sdk_common::{SendOutsideWasm, SyncOutsideWasm};
 use ruma::ServerName;
 use tokio::sync::RwLock;
diff --git a/bindings/matrix-sdk-ffi/src/room_list.rs b/bindings/matrix-sdk-ffi/src/room_list.rs
@@ -2,7 +2,6 @@
 
 use std::{fmt::Debug, mem::MaybeUninit, ptr::addr_of_mut, sync::Arc, time::Duration};
 
-use async_compat::get_runtime_handle;
 use eyeball_im::VectorDiff;
 use futures_util::{pin_mut, StreamExt};
 use matrix_sdk::{
@@ -12,6 +11,7 @@ use matrix_sdk::{
     },
     Room as SdkRoom,
 };
+use matrix_sdk_common::runtime::get_runtime_handle;
 use matrix_sdk_common::{SendOutsideWasm, SyncOutsideWasm};
 use matrix_sdk_ui::{
     room_list_service::filters::{
diff --git a/bindings/matrix-sdk-ffi/src/session_verification.rs b/bindings/matrix-sdk-ffi/src/session_verification.rs
@@ -1,6 +1,5 @@
 use std::sync::{Arc, RwLock};
 
-use async_compat::get_runtime_handle;
 use futures_util::StreamExt;
 use matrix_sdk::{
     encryption::{
@@ -11,6 +10,7 @@ use matrix_sdk::{
     ruma::events::key::verification::VerificationMethod,
     Account,
 };
+use matrix_sdk_common::runtime::get_runtime_handle;
 use matrix_sdk_common::{SendOutsideWasm, SyncOutsideWasm};
 use ruma::UserId;
 use tracing::{error, warn};
diff --git a/bindings/matrix-sdk-ffi/src/sync_service.rs b/bindings/matrix-sdk-ffi/src/sync_service.rs
@@ -14,9 +14,9 @@
 
 use std::{fmt::Debug, sync::Arc};
 
-use async_compat::get_runtime_handle;
 use futures_util::pin_mut;
 use matrix_sdk::Client;
+use matrix_sdk_common::runtime::get_runtime_handle;
 use matrix_sdk_common::{SendOutsideWasm, SyncOutsideWasm};
 use matrix_sdk_ui::{
     sync_service::{
diff --git a/bindings/matrix-sdk-ffi/src/timeline/mod.rs b/bindings/matrix-sdk-ffi/src/timeline/mod.rs
@@ -16,7 +16,6 @@ use std::{collections::HashMap, fmt::Write as _, fs, panic, sync::Arc};
 
 use anyhow::{Context, Result};
 use as_variant::as_variant;
-use async_compat::get_runtime_handle;
 use eyeball_im::VectorDiff;
 use futures_util::{pin_mut, StreamExt as _};
 use matrix_sdk::{
@@ -31,6 +30,7 @@ use matrix_sdk::{
         reply::{EnforceThread, Reply},
     },
 };
+use matrix_sdk_common::runtime::get_runtime_handle;
 use matrix_sdk_ui::timeline::{
     self, AttachmentSource, EventItemOrigin, Profile, TimelineDetails,
     TimelineUniqueId as SdkTimelineUniqueId,
diff --git a/bindings/matrix-sdk-ffi/src/utils.rs b/bindings/matrix-sdk-ffi/src/utils.rs
@@ -14,7 +14,7 @@
 
 use std::{mem::ManuallyDrop, ops::Deref};
 
-use async_compat::get_runtime_handle;
+use matrix_sdk_common::runtime::get_runtime_handle;
 use ruma::{MilliSecondsSinceUnixEpoch, UInt};
 use tracing::warn;
 
diff --git a/bindings/matrix-sdk-ffi/src/widget.rs b/bindings/matrix-sdk-ffi/src/widget.rs
@@ -1,11 +1,11 @@
 use std::sync::{Arc, Mutex};
 
-use async_compat::get_runtime_handle;
 use language_tags::LanguageTag;
 use matrix_sdk::{
     async_trait,
     widget::{MessageLikeEventFilter, StateEventFilter, ToDeviceEventFilter},
 };
+use matrix_sdk_common::runtime::get_runtime_handle;
 use matrix_sdk_common::{SendOutsideWasm, SyncOutsideWasm};
 use ruma::events::MessageLikeEventType;
 use tracing::error;
diff --git a/crates/matrix-sdk-common/Cargo.toml b/crates/matrix-sdk-common/Cargo.toml
@@ -40,6 +40,7 @@ uniffi = { workspace = true, optional = true }
 
 [target.'cfg(not(target_family = "wasm"))'.dependencies]
 # Enable the test macro.
+async-compat.workspace = true
 tokio = { workspace = true, features = ["rt", "macros"] }
 
 [target.'cfg(target_family = "wasm")'.dependencies]
diff --git a/crates/matrix-sdk-common/src/executor.rs b/crates/matrix-sdk-common/src/executor.rs
@@ -12,88 +12,126 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! Abstraction over an executor so we can spawn tasks under WASM the same way
+//! Abstraction over an executor so we can spawn tasks under Wasm the same way
 //! we do usually.
 
-#[cfg(target_family = "wasm")]
-use std::{
-    future::Future,
-    pin::Pin,
-    task::{Context, Poll},
-};
+//! On non Wasm platforms, this re-exports parts of tokio directly.  For Wasm,
+//! we wrap the n0-future package's AbortHandle + spawn method in order to
+//! provide an interface that is similar to tokio's.
 
-#[cfg(target_family = "wasm")]
-pub use futures_util::future::Aborted as JoinError;
-#[cfg(target_family = "wasm")]
-use futures_util::{
-    future::{AbortHandle, Abortable, RemoteHandle},
-    FutureExt,
-};
 #[cfg(not(target_family = "wasm"))]
-pub use tokio::task::{spawn, JoinError, JoinHandle};
+mod sys {
+    pub use tokio::task::{spawn, AbortHandle, JoinError, JoinHandle};
 
-#[cfg(target_family = "wasm")]
-pub fn spawn<F, T>(future: F) -> JoinHandle<T>
-where
-    F: Future<Output = T> + 'static,
-{
-    let (future, remote_handle) = future.remote_handle();
-    let (abort_handle, abort_registration) = AbortHandle::new_pair();
-    let future = Abortable::new(future, abort_registration);
-
-    wasm_bindgen_futures::spawn_local(async {
-        // Poll the future, and ignore the result (either it's `Ok(())`, or it's
-        // `Err(Aborted)`).
-        let _ = future.await;
-    });
-
-    JoinHandle { remote_handle: Some(remote_handle), abort_handle }
+    /// A handle to a runtime for executing async tasks and futures.
+    pub type Handle = tokio::runtime::Handle;
+    pub type Runtime = tokio::runtime::Runtime;
 }
 
 #[cfg(target_family = "wasm")]
-#[derive(Debug)]
-pub struct JoinHandle<T> {
-    remote_handle: Option<RemoteHandle<T>>,
-    abort_handle: AbortHandle,
-}
+mod sys {
+    use std::{
+        future::Future,
+        pin::Pin,
+        task::{Context, Poll},
+    };
+
+    pub use futures_util::future::AbortHandle;
+    use futures_util::{
+        future::{Abortable, RemoteHandle},
+        FutureExt,
+    };
+
+    #[derive(Debug)]
+    pub enum JoinError {
+        Cancelled,
+        Panic,
+    }
 
-#[cfg(target_family = "wasm")]
-impl<T> JoinHandle<T> {
-    pub fn abort(&self) {
-        self.abort_handle.abort();
+    impl JoinError {
+        /// Returns true if the error was caused by the task being cancelled.
+        ///
+        /// See [the module level docs] for more information on cancellation.
+        ///
+        /// [the module level docs]: crate::task#cancellation
+        pub fn is_cancelled(&self) -> bool {
+            matches!(self, JoinError::Cancelled)
+        }
     }
 
-    pub fn is_finished(&self) -> bool {
-        self.abort_handle.is_aborted()
+    impl std::fmt::Display for JoinError {
+        fn fmt(&self, fmt: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+            match &self {
+                JoinError::Cancelled => write!(fmt, "task was cancelled"),
+                JoinError::Panic => write!(fmt, "task panicked"),
+            }
+        }
     }
-}
 
-#[cfg(target_family = "wasm")]
-impl<T> Drop for JoinHandle<T> {
-    fn drop(&mut self) {
-        // don't abort the spawned future
-        if let Some(h) = self.remote_handle.take() {
-            h.forget();
+    #[derive(Debug)]
+    pub struct JoinHandle<T> {
+        remote_handle: Option<RemoteHandle<T>>,
+        abort_handle: AbortHandle,
+    }
+
+    impl<T> JoinHandle<T> {
+        pub fn abort(&self) {
+            self.abort_handle.abort();
+        }
+
+        pub fn abort_handle(&self) -> AbortHandle {
+            self.abort_handle.clone()
+        }
+
+        pub fn is_finished(&self) -> bool {
+            self.abort_handle.is_aborted()
         }
     }
-}
 
-#[cfg(target_family = "wasm")]
-impl<T: 'static> Future for JoinHandle<T> {
-    type Output = Result<T, JoinError>;
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        if self.abort_handle.is_aborted() {
-            // The future has been aborted. It is not possible to poll it again.
-            Poll::Ready(Err(JoinError))
-        } else if let Some(handle) = self.remote_handle.as_mut() {
-            Pin::new(handle).poll(cx).map(Ok)
-        } else {
-            Poll::Ready(Err(JoinError))
+    impl<T> Drop for JoinHandle<T> {
+        fn drop(&mut self) {
+            // don't abort the spawned future
+            if let Some(h) = self.remote_handle.take() {
+                h.forget();
+            }
         }
     }
+
+    impl<T: 'static> Future for JoinHandle<T> {
+        type Output = Result<T, JoinError>;
+
+        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+            if self.abort_handle.is_aborted() {
+                // The future has been aborted. It is not possible to poll it again.
+                Poll::Ready(Err(JoinError::Cancelled))
+            } else if let Some(handle) = self.remote_handle.as_mut() {
+                Pin::new(handle).poll(cx).map(Ok)
+            } else {
+                Poll::Ready(Err(JoinError::Panic))
+            }
+        }
+    }
+
+    pub fn spawn<F, T>(future: F) -> JoinHandle<T>
+    where
+        F: Future<Output = T> + 'static,
+    {
+        let (future, remote_handle) = future.remote_handle();
+        let (abort_handle, abort_registration) = AbortHandle::new_pair();
+        let future = Abortable::new(future, abort_registration);
+
+        wasm_bindgen_futures::spawn_local(async {
+            // Poll the future, and ignore the result (either it's `Ok(())`, or it's
+            // `Err(Aborted)`).
+            let _ = future.await;
+        });
+
+        JoinHandle { remote_handle: Some(remote_handle), abort_handle }
+    }
 }
 
+pub use sys::*;
+
 #[cfg(test)]
 mod tests {
     use assert_matches::assert_matches;
diff --git a/crates/matrix-sdk-common/src/lib.rs b/crates/matrix-sdk-common/src/lib.rs
diff --git a/crates/matrix-sdk-common/src/runtime.rs b/crates/matrix-sdk-common/src/runtime.rs
diff --git a/crates/matrix-sdk-common/src/stream.rs b/crates/matrix-sdk-common/src/stream.rs
