diff --git a/crates/matrix-sdk/src/client/mod.rs b/crates/matrix-sdk/src/client/mod.rs index 7f18aacfc..87383c614 100644 --- a/crates/matrix-sdk/src/client/mod.rs +++ b/crates/matrix-sdk/src/client/mod.rs @@ -27,8 +27,10 @@ use futures_core::Stream; #[cfg(feature = "e2e-encryption")] use matrix_sdk_base::crypto::store::LockableCryptoStore; use matrix_sdk_base::{ - store::DynStateStore, sync::Notification, BaseClient, RoomInfoUpdate, RoomState, - RoomStateFilter, SendOutsideWasm, SessionMeta, SyncOutsideWasm, + store::DynStateStore, + sync::{Notification, RoomUpdates}, + BaseClient, RoomInfoUpdate, RoomState, RoomStateFilter, SendOutsideWasm, SessionMeta, + SyncOutsideWasm, }; use matrix_sdk_common::instant::Instant; #[cfg(feature = "e2e-encryption")] @@ -250,6 +252,10 @@ pub(crate) struct ClientInner { /// The sender-side of channels used to receive room updates. pub(crate) room_update_channels: StdMutex>>, + /// The sender-side of a channel used to observe all the room updates of a + /// sync response. + pub(crate) room_updates_sender: broadcast::Sender, + /// Whether the client should update its homeserver URL with the discovery /// information present in the login response. respect_login_well_known: bool, @@ -296,6 +302,9 @@ impl ClientInner { event_handlers: Default::default(), notification_handlers: Default::default(), room_update_channels: Default::default(), + // A single `RoomUpdates` is sent once per sync, so we assume that 32 is sufficient + // ballast for all observers to catch up. + room_updates_sender: broadcast::Sender::new(32), respect_login_well_known, sync_beat: event_listener::Event::new(), #[cfg(feature = "e2e-encryption")] @@ -840,6 +849,12 @@ impl Client { } } + /// Subscribe to all updates to all rooms, whenever any has been received in + /// a sync response. + pub fn subscribe_to_all_room_updates(&self) -> broadcast::Receiver { + self.inner.room_updates_sender.subscribe() + } + pub(crate) async fn notification_handlers( &self, ) -> RwLockReadGuard<'_, Vec> { diff --git a/crates/matrix-sdk/src/sync.rs b/crates/matrix-sdk/src/sync.rs index 75775e5d7..1aa3792c2 100644 --- a/crates/matrix-sdk/src/sync.rs +++ b/crates/matrix-sdk/src/sync.rs @@ -159,6 +159,9 @@ impl Client { self.handle_sync_events(HandlerKind::Presence, None, presence).await?; self.handle_sync_events(HandlerKind::ToDevice, None, to_device).await?; + // Ignore errors when there are no receivers. + let _ = self.inner.room_updates_sender.send(rooms.clone()); + for (room_id, room_info) in &rooms.join { let Some(room) = self.get_room(room_id) else { error!(?room_id, "Can't call event handler, room not found"); diff --git a/crates/matrix-sdk/tests/integration/client.rs b/crates/matrix-sdk/tests/integration/client.rs index 978f93b15..3069bb055 100644 --- a/crates/matrix-sdk/tests/integration/client.rs +++ b/crates/matrix-sdk/tests/integration/client.rs @@ -7,10 +7,14 @@ use matrix_sdk::{ media::{MediaFormat, MediaRequest, MediaThumbnailSize}, sync::RoomUpdate, }; -use matrix_sdk_base::RoomState; +use matrix_sdk_base::{sync::RoomUpdates, RoomState}; use matrix_sdk_test::{ - async_test, sync_state_event, test_json, JoinedRoomBuilder, SyncResponseBuilder, - DEFAULT_TEST_ROOM_ID, + async_test, sync_state_event, + test_json::{ + self, + sync::{MIXED_INVITED_ROOM_ID, MIXED_JOINED_ROOM_ID, MIXED_LEFT_ROOM_ID, MIXED_SYNC}, + }, + JoinedRoomBuilder, SyncResponseBuilder, DEFAULT_TEST_ROOM_ID, }; use ruma::{ api::client::{ @@ -414,7 +418,7 @@ async fn whoami() { } #[async_test] -async fn room_update_channel() { +async fn test_room_update_channel() { let (client, server) = logged_in_client().await; let mut rx = client.subscribe_to_room_updates(&DEFAULT_TEST_ROOM_ID); @@ -438,6 +442,62 @@ async fn room_update_channel() { assert_eq!(updates.unread_notifications.notification_count, 11); } +#[async_test] +async fn test_subscribe_all_room_updates() { + let (client, server) = logged_in_client().await; + + let mut rx = client.subscribe_to_all_room_updates(); + + mock_sync(&server, &*MIXED_SYNC, None).await; + let sync_settings = SyncSettings::new().timeout(Duration::from_millis(3000)); + client.sync_once(sync_settings).await.unwrap(); + + let room_updates = rx.recv().now_or_never().unwrap().unwrap(); + assert_let!(RoomUpdates { leave, join, invite } = room_updates); + + // Check the left room updates. + { + assert_eq!(leave.len(), 1); + + let (room_id, update) = leave.iter().next().unwrap(); + + assert_eq!(room_id, *MIXED_LEFT_ROOM_ID); + assert!(update.state.is_empty()); + assert_eq!(update.timeline.events.len(), 1); + assert!(update.account_data.is_empty()); + } + + // Check the joined room updates. + { + assert_eq!(join.len(), 1); + + let (room_id, update) = join.iter().next().unwrap(); + + assert_eq!(room_id, *MIXED_JOINED_ROOM_ID); + + assert_eq!(update.account_data.len(), 1); + assert_eq!(update.ephemeral.len(), 1); + assert_eq!(update.state.len(), 1); + + assert!(update.timeline.limited); + assert_eq!(update.timeline.events.len(), 1); + assert_eq!(update.timeline.prev_batch, Some("t392-516_47314_0_7_1_1_1_11444_1".to_owned())); + + assert_eq!(update.unread_notifications.highlight_count, 0); + assert_eq!(update.unread_notifications.notification_count, 11); + } + + // Check the invited room updates. + { + assert_eq!(invite.len(), 1); + + let (room_id, update) = invite.iter().next().unwrap(); + + assert_eq!(room_id, *MIXED_INVITED_ROOM_ID); + assert_eq!(update.invite_state.events.len(), 2); + } +} + // Check that the `Room::is_encrypted()` is properly deduplicated, meaning we // only make a single request to the server, and that multiple calls do return // the same result. @@ -906,7 +966,7 @@ async fn create_dm_error() { } #[async_test] -async fn ambiguity_changes() { +async fn test_ambiguity_changes() { let (client, server) = logged_in_client().await; let example_id = user_id!("@example:localhost"); diff --git a/testing/matrix-sdk-test/src/test_json/sync.rs b/testing/matrix-sdk-test/src/test_json/sync.rs index aedec342c..6a1e40bf8 100644 --- a/testing/matrix-sdk-test/src/test_json/sync.rs +++ b/testing/matrix-sdk-test/src/test_json/sync.rs @@ -1,6 +1,7 @@ //! Complete sync responses. use once_cell::sync::Lazy; +use ruma::{room_id, RoomId}; use serde_json::{json, Value as JsonValue}; use crate::DEFAULT_TEST_ROOM_ID; @@ -1230,6 +1231,181 @@ pub static LEAVE_SYNC_EVENT: Lazy = Lazy::new(|| { }) }); +/// In the [`MIXED_SYNC`], the room id of the joined room. +pub static MIXED_JOINED_ROOM_ID: Lazy<&RoomId> = + Lazy::new(|| room_id!("!SVkFJHzfwvuaIEawgC:localhost")); +/// In the [`MIXED_SYNC`], the room id of the left room. +pub static MIXED_LEFT_ROOM_ID: Lazy<&RoomId> = + Lazy::new(|| room_id!("!SVkFJHzfwvuaIEawgD:localhost")); +/// In the [`MIXED_SYNC`], the room id of the invited room. +pub static MIXED_INVITED_ROOM_ID: Lazy<&RoomId> = + Lazy::new(|| room_id!("!SVkFJHzfwvuaIEawgE:localhost")); + +/// A sync that contains updates to joined/invited/left rooms. +pub static MIXED_SYNC: Lazy = Lazy::new(|| { + json!({ + "account_data": { + "events": [] + }, + "to_device": { + "events": [] + }, + "device_lists": { + "changed": [], + "left": [] + }, + "presence": { + "events": [] + }, + "rooms": { + "join": { + *MIXED_JOINED_ROOM_ID: { + "summary": {}, + "account_data": { + "events": [ + { + "content": { + "event_id": "$someplace:example.org" + }, + "room_id": "!roomid:room.com", + "type": "m.fully_read" + } + ] + }, + "ephemeral": { + "events": [ + { + "content": { + "$151680659217152dPKjd:localhost": { + "m.read": { + "@example:localhost": { + "ts": 151680989 + } + } + } + }, + "room_id": *MIXED_JOINED_ROOM_ID, + "type": "m.receipt" + }, + ] + }, + "state": { + "events": [ + { + "content": { + "alias": "#tutorial:localhost" + }, + "event_id": "$15139375513VdeRF:localhost", + "origin_server_ts": 151393755000000_u64, + "sender": "@example:localhost", + "state_key": "", + "type": "m.room.canonical_alias", + "unsigned": { + "age": 703422 + } + }, + ] + }, + "timeline": { + "events": [ + { + "content": { + "body": "baba", + "format": "org.matrix.custom.html", + "formatted_body": "baba", + "msgtype": "m.text" + }, + "event_id": "$152037280074GZeOm:localhost", + "origin_server_ts": 152037280000000_u64, + "sender": "@example:localhost", + "type": "m.room.message", + "unsigned": { + "age": 598971425 + } + } + ], + "limited": true, + "prev_batch": "t392-516_47314_0_7_1_1_1_11444_1" + }, + "unread_notifications": { + "highlight_count": 0, + "notification_count": 11 + } + } + }, + "invite": { + *MIXED_INVITED_ROOM_ID: { + "invite_state": { + "events": [ + { + "sender": "@alice:example.com", + "type": "m.room.name", + "state_key": "", + "content": { + "name": "My Room Name" + } + }, + { + "sender": "@alice:example.com", + "type": "m.room.member", + "state_key": "@bob:example.com", + "content": { + "membership": "invite" + } + } + ] + } + } + }, + "leave": { + *MIXED_LEFT_ROOM_ID: { + "timeline": { + "events": [ + { + "content": { + "membership": "leave" + }, + "origin_server_ts": 158957809000000_u64, + "sender": "@example:localhost", + "state_key": "@example:localhost", + "type": "m.room.member", + "unsigned": { + "replaces_state": "$blahblah", + "prev_content": { + "avatar_url": null, + "displayname": "me", + "membership": "invite" + }, + "prev_sender": "@2example:localhost", + "age": 1757 + }, + "event_id": "$lQQ116Y-XqcjpSUGpuz36rNntUvOSpTjuaIvmtQ2AwA" + } + ], + "prev_batch": "toktok", + "limited": false + }, + "state": { + "events": [] + }, + "account_data": { + "events": [] + } + } + } + }, + "groups": { + "join": {}, + "invite": {}, + "leave": {} + }, + "device_one_time_keys_count": { + "signed_curve25519": 50 + }, + "next_batch": "s1380317562_757269739_1655566_503953763_334052043_1209862_55290918_65705002_101146" + }) +}); + pub static VOIP_SYNC: Lazy = Lazy::new(|| { json!({ "device_one_time_keys_count": {},