feat: Add support for MSC4471 event streams

Cargo.toml

@@ -71,7 +71,7 @@ rand = { version = "0.10.1", default-features = false, features = ["std", "std_r
 regex = { version = "1.12.2", default-features = false }
 reqwest = { version = "0.13.1", default-features = false }
 rmp-serde = { version = "1.3.0", default-features = false }
-ruma = { version = "0.15.1", features = [
+ruma = { version = "0.15.1", git = "https://github.com/ruma/ruma", rev = "e54e5b8048a1ff77a07f1a7a48f631b78d0a0a16", features = [
     "client-api-c",
     "compat-unset-avatar",
     "compat-upload-signatures",
@@ -89,6 +89,7 @@ ruma = { version = "0.15.1", features = [
     "unstable-msc4140",
     "unstable-msc4143",
     "unstable-msc4171",
+    "unstable-msc4195",
     "unstable-msc4278",
     "unstable-msc4286",
     "unstable-msc4306",

bindings/matrix-sdk-ffi/src/client.rs

@@ -38,12 +38,10 @@ use matrix_sdk::{
         EventEncryptionAlgorithm, RoomId, TransactionId, UInt, UserId,
         api::client::{
             account::request_openid_token,
-            discovery::{
-                discover_homeserver::RtcFocusInfo,
-                get_authorization_server_metadata::v1::Prompt as RumaOAuthPrompt,
-            },
+            discovery::get_authorization_server_metadata::v1::Prompt as RumaOAuthPrompt,
             push::{EmailPusherData, PusherIds, PusherInit, PusherKind as RumaPusherKind},
             room::{Visibility, create_room},
+            rtc::RtcTransport,
             session::get_login_types,
             user_directory::search_users,
         },
@@ -2060,7 +2058,7 @@ impl Client {
             .rtc_foci()
             .await?
             .iter()
-            .any(|focus| matches!(focus, RtcFocusInfo::LiveKit(_))))
+            .any(|focus| matches!(focus, RtcTransport::LiveKit(_))))
     }
 
     /// Get information about the homeserver's advertised map tile server, if

crates/matrix-sdk-base/src/store/integration_tests.rs

@@ -14,7 +14,7 @@ use ruma::{
     EventId, MilliSecondsSinceUnixEpoch, OwnedUserId, RoomId, TransactionId, UserId,
     api::{
         FeatureFlag, MatrixVersion,
-        client::discovery::discover_homeserver::{HomeserverInfo, RtcFocusInfo},
+        client::{discovery::discover_homeserver::HomeserverInfo, rtc::RtcTransport},
     },
     event_id,
     events::{
@@ -559,7 +559,7 @@ impl StateStoreIntegrationTests for DynStateStore {
             homeserver: HomeserverInfo::new("matrix.example.com".to_owned()),
             identity_server: None,
             tile_server: None,
-            rtc_foci: vec![RtcFocusInfo::livekit("livekit.example.com".to_owned())],
+            rtc_foci: vec![RtcTransport::livekit("livekit.example.com".to_owned())],
         };
 
         self.set_kv_data(

crates/matrix-sdk-base/src/store/mod.rs

@@ -147,7 +147,7 @@ pub enum StoreError {
     ///
     /// This should never happen.
     #[error("Redaction failed: {0}")]
-    Redaction(#[source] ruma::canonical_json::RedactionError),
+    Redaction(#[source] ruma::canonical_json::CanonicalJsonFieldError),
 
     /// The store contains invalid data.
     #[error("The store contains invalid data: {details}")]

crates/matrix-sdk-base/src/store/traits.rs

@@ -29,11 +29,12 @@ use ruma::{
     OwnedTransactionId, OwnedUserId, RoomId, TransactionId, UserId,
     api::{
         MatrixVersion, SupportedVersions,
-        client::discovery::{
-            discover_homeserver::{
-                self, HomeserverInfo, IdentityServerInfo, RtcFocusInfo, TileServerInfo,
+        client::{
+            discovery::{
+                discover_homeserver::{self, HomeserverInfo, IdentityServerInfo, TileServerInfo},
+                get_capabilities::v3::Capabilities,
             },
-            get_capabilities::v3::Capabilities,
+            rtc::RtcTransport,
         },
     },
     events::{
@@ -2070,7 +2071,7 @@ pub struct WellKnownResponse {
     pub tile_server: Option<TileServerInfo>,
 
     /// A list of the available MatrixRTC foci, ordered by priority.
-    pub rtc_foci: Vec<RtcFocusInfo>,
+    pub rtc_foci: Vec<RtcTransport>,
 }
 
 impl From<discover_homeserver::Response> for WellKnownResponse {

crates/matrix-sdk-sqlite/src/error.rs

@@ -106,7 +106,7 @@ pub enum Error {
     Unpickle,
 
     #[error("Redaction failed: {0}")]
-    Redaction(#[source] ruma::canonical_json::RedactionError),
+    Redaction(#[source] ruma::canonical_json::CanonicalJsonFieldError),
 
     #[error("An update keyed by unique ID touched more than one entry")]
     InconsistentUpdate,

crates/matrix-sdk/Cargo.toml

@@ -16,7 +16,7 @@ features = ["docsrs"]
 rustdoc-args = ["--cfg", "docsrs", "--generate-link-to-definition"]
 
 [features]
-default = ["e2e-encryption", "automatic-room-key-forwarding", "sqlite"]
+default = ["e2e-encryption", "automatic-room-key-forwarding", "sqlite", "experimental-event-streams"]
 testing = [
     "matrix-sdk-sqlite?/testing",
     "matrix-sdk-indexeddb?/testing",
@@ -81,6 +81,7 @@ federation-api = ["ruma/federation-api-c"]
 uniffi = ["dep:uniffi", "matrix-sdk-base/uniffi", "dep:matrix-sdk-ffi-macros"]
 
 experimental-widgets = ["dep:uuid", "experimental-send-custom-to-device"]
+experimental-event-streams = ["e2e-encryption", "ruma/unstable-msc4471"]
 
 docsrs = ["e2e-encryption", "sqlite", "indexeddb", "sso-login", "qrcode", "federation-api"]
 

crates/matrix-sdk/changelog.d/6607.added.md

@@ -0,0 +1,2 @@
+Add experimental support for MSC4471 event streams, allowing clients to send
+transient updates to room messages and subscribe to updates from other devices.

crates/matrix-sdk/src/client/mod.rs

@@ -14,6 +14,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#[cfg(feature = "experimental-event-streams")]
+use std::sync::OnceLock;
 use std::{
     collections::{BTreeMap, BTreeSet, btree_map},
     fmt::{self, Debug},
@@ -55,15 +57,13 @@ use ruma::{
             authenticated_media,
             device::{self, delete_devices, get_devices, update_device},
             directory::{get_public_rooms, get_public_rooms_filtered},
-            discovery::{
-                discover_homeserver::{self, RtcFocusInfo},
-                get_supported_versions,
-            },
+            discovery::{discover_homeserver, get_supported_versions},
             filter::{FilterDefinition, create_filter::v3::Request as FilterUploadRequest},
             knock::knock_room,
             media,
             membership::{join_room_by_id, join_room_by_id_or_alias},
             room::create_room,
+            rtc::RtcTransport,
             session::login::v3::DiscoveryInfo,
             sync::sync_events,
             threads::get_thread_subscriptions_changes,
@@ -87,6 +87,8 @@ use self::{
     caches::{Cache, CachedValue, ClientCaches},
     futures::SendRequest,
 };
+#[cfg(feature = "experimental-event-streams")]
+use crate::event_streams::EventStreams;
 use crate::{
     Account, AuthApi, AuthSession, Error, HttpError, Media, Pusher, RefreshTokenError, Result,
     Room, SessionTokens, TransmissionProgress,
@@ -411,6 +413,9 @@ pub(crate) struct ClientInner {
     #[cfg(feature = "e2e-encryption")]
     pub(crate) duplicate_key_upload_error_sender:
         broadcast::Sender<Option<DuplicateOneTimeKeyErrorMessage>>,
+
+    #[cfg(feature = "experimental-event-streams")]
+    event_streams: OnceLock<EventStreams>,
 }
 
 impl ClientInner {
@@ -481,6 +486,8 @@ impl ClientInner {
             task_monitor: TaskMonitor::new(),
             #[cfg(feature = "e2e-encryption")]
             duplicate_key_upload_error_sender: broadcast::channel(1).0,
+            #[cfg(feature = "experimental-event-streams")]
+            event_streams: OnceLock::new(),
         };
 
         #[allow(clippy::let_and_return)]
@@ -531,6 +538,12 @@ impl Client {
         ClientBuilder::new()
     }
 
+    /// Get the client-owned event stream manager.
+    #[cfg(feature = "experimental-event-streams")]
+    pub fn event_streams(&self) -> EventStreams {
+        self.inner.event_streams.get_or_init(|| EventStreams::new(self.clone())).clone()
+    }
+
     pub(crate) fn base_client(&self) -> &BaseClient {
         &self.inner.base_client
     }
@@ -2558,22 +2571,22 @@ impl Client {
     ///
     /// # Examples
     /// ```no_run
-    /// # use matrix_sdk::{Client, config::SyncSettings, ruma::api::client::discovery::discover_homeserver::RtcFocusInfo};
+    /// # use matrix_sdk::{Client, config::SyncSettings, ruma::api::client::rtc::RtcTransport};
     /// # use url::Url;
     /// # async {
     /// # let homeserver = Url::parse("http://localhost:8080")?;
     /// # let mut client = Client::new(homeserver).await?;
     /// let rtc_foci = client.rtc_foci().await?;
     /// let default_livekit_focus_info = rtc_foci.iter().find_map(|focus| match focus {
-    ///     RtcFocusInfo::LiveKit(info) => Some(info),
+    ///     RtcTransport::LiveKit(info) => Some(info),
     ///     _ => None,
     /// });
     /// if let Some(info) = default_livekit_focus_info {
     ///     println!("Default LiveKit service URL: {}", info.service_url);
     /// }
     /// # anyhow::Ok(()) };
     /// ```
-    pub async fn rtc_foci(&self) -> HttpResult<Vec<RtcFocusInfo>> {
+    pub async fn rtc_foci(&self) -> HttpResult<Vec<RtcTransport>> {
         let well_known = self.well_known().await;
 
         Ok(well_known.map(|well_known| well_known.rtc_foci).unwrap_or_default())
@@ -3660,10 +3673,7 @@ pub(crate) mod tests {
         RoomId, ServerName, UserId,
         api::{
             FeatureFlag, MatrixVersion,
-            client::{
-                discovery::discover_homeserver::RtcFocusInfo,
-                room::create_room::v3::Request as CreateRoomRequest,
-            },
+            client::{room::create_room::v3::Request as CreateRoomRequest, rtc::RtcTransport},
         },
         assign,
         events::{
@@ -4114,7 +4124,7 @@ pub(crate) mod tests {
         let server_url = server.uri();
         let domain = server_url.strip_prefix("http://").unwrap();
         let server_name = <&ServerName>::try_from(domain).unwrap();
-        let rtc_foci = vec![RtcFocusInfo::livekit("https://livekit.example.com".to_owned())];
+        let rtc_foci = vec![RtcTransport::livekit("https://livekit.example.com".to_owned())];
 
         let well_known_mock = server
             .mock_well_known()
@@ -4191,7 +4201,7 @@ pub(crate) mod tests {
     #[async_test]
     async fn test_missing_well_known_caching() {
         let server = MatrixMockServer::new().await;
-        let rtc_foci: Vec<RtcFocusInfo> = vec![];
+        let rtc_foci: Vec<RtcTransport> = vec![];
 
         let well_known_mock = server
             .mock_well_known()