sdk: add a mechanism to get all room updates at once

(instead of having to subscribe to a single room in the event cache)

Author: bnjbvr

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<BTreeMap<OwnedRoomId, broadcast::Sender<RoomUpdate>>>,
 
+    /// The sender-side of a channel used to observe all the room updates of a
+    /// sync response.
+    pub(crate) room_updates_sender: broadcast::Sender<RoomUpdates>,
+
     /// 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<RoomUpdates> {
+        self.inner.room_updates_sender.subscribe()
+    }
+
     pub(crate) async fn notification_handlers(
         &self,
     ) -> RwLockReadGuard<'_, Vec<NotificationHandlerFn>> {

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");

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");

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<JsonValue> = 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<JsonValue> = 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": "<strong>baba</strong>",
+                                    "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<JsonValue> = Lazy::new(|| {
     json!({
         "device_one_time_keys_count": {},