Dima/matrix-rust-sdk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source 
(instead of having to subscribe to a single room in the event cache)
This commit is contained in:
Benjamin Bouvier
2024-02-12 18:07:45 +01:00
parent 6a81ceced0
commit a6c133369f
4 changed files with 261 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/matrix-sdk/src/client/mod.rs
+17 -2
View File
@@ -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
+3
View File
@@ -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
+65 -5
View File
@@ -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
+176
View File
@@ -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": {},