fixup! feat(room): create a cleanup task in `Room::subscribe_to_knock…

…_requests`

crates/matrix-sdk-base/src/rooms/normal.rs

@@ -1222,8 +1222,6 @@ impl Room {
         }
 
         let current_seen_events_guard = self.get_write_guarded_current_knock_request_ids().await?;
-        // We're not calling `get_seen_join_request_ids` here because we need to keep
-        // the Mutex's guard until we've updated the data
         let mut current_seen_events = current_seen_events_guard.clone().unwrap_or_default();
 
         current_seen_events.extend(event_to_user_ids);
@@ -1237,8 +1235,6 @@ impl Room {
     /// current room members.
     pub async fn remove_outdated_seen_knock_requests_ids(&self) -> StoreResult<()> {
         let current_seen_events_guard = self.get_write_guarded_current_knock_request_ids().await?;
-        // We're not calling `get_seen_join_request_ids` here because we need to keep
-        // the Mutex's guard until we've updated the data
         let mut current_seen_events = current_seen_events_guard.clone().unwrap_or_default();
 
         // Get and deserialize the member events for the seen knock requests
@@ -2145,7 +2141,6 @@ mod tests {
     use std::{
         collections::BTreeSet,
         ops::{Not, Sub},
-        pin::pin,
         str::FromStr,
         sync::Arc,
         time::Duration,
@@ -2196,7 +2191,7 @@ mod tests {
         store::{IntoStateStore, MemoryStore, StateChanges, StateStore, StoreConfig},
         test_utils::logged_in_base_client,
         BaseClient, MinimalStateEvent, OriginalMinimalStateEvent, RoomDisplayName,
-        RoomInfoNotableUpdateReasons, RoomMembersUpdate, RoomStateFilter, SessionMeta,
+        RoomInfoNotableUpdateReasons, RoomStateFilter, SessionMeta,
     };
 
     #[test]
@@ -3789,27 +3784,4 @@ mod tests {
             ]
         );
     }
-
-    #[async_test]
-    async fn test_room_member_updates_sender_and_receiver() {
-        use assert_matches::assert_matches;
-
-        let client = logged_in_base_client(None).await;
-        let room = client.get_or_create_room(room_id!("!a:b.c"), RoomState::Joined);
-
-        let mut receiver = room.room_member_updates_sender.subscribe();
-
-        assert!(receiver.is_empty());
-
-        room.room_member_updates_sender
-            .send(RoomMembersUpdate::FullReload)
-            .expect("broadcasting a room members update failed");
-
-        let recv = pin!(receiver.recv());
-        let next = matrix_sdk_common::timeout::timeout(recv, Duration::from_secs(1))
-            .await
-            .expect("receiving a room members update timed out")
-            .expect("failed receiving a room members update");
-        assert_matches!(next, RoomMembersUpdate::FullReload);
-    }
 }

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

@@ -117,6 +117,18 @@ macro_rules! assert_next_with_timeout {
     }};
 }
 
+/// Asserts the next item in a `Receiver` can be loaded in the given timeout in
+/// milliseconds.
+#[macro_export]
+macro_rules! assert_recv_with_timeout {
+    ($receiver:expr, $timeout_ms:expr) => {{
+        tokio::time::timeout(std::time::Duration::from_millis($timeout_ms), $receiver.recv())
+            .await
+            .expect("Next event timed out")
+            .expect("No next event received")
+    }};
+}
+
 /// Assert the next item in a `Stream` or `Subscriber` matches the provided
 /// pattern in the given timeout in milliseconds.
 ///

crates/matrix-sdk/tests/integration/room/joined.rs

@@ -1,16 +1,18 @@
 use std::{
+    collections::BTreeSet,
     sync::{Arc, Mutex},
     time::Duration,
 };
 
+use assert_matches2::assert_let;
 use futures_util::{future::join_all, pin_mut};
 use matrix_sdk::{
-    assert_next_with_timeout,
+    assert_next_with_timeout, assert_recv_with_timeout,
     config::SyncSettings,
     room::{edit::EditedContent, Receipts, ReportedContentScore, RoomMemberRole},
     test_utils::mocks::MatrixMockServer,
 };
-use matrix_sdk_base::RoomState;
+use matrix_sdk_base::{RoomMembersUpdate, RoomState};
 use matrix_sdk_test::{
     async_test,
     event_factory::EventFactory,
@@ -1156,3 +1158,63 @@ async fn test_subscribe_to_knock_requests_clears_seen_ids_on_member_reload() {
 
     handle.abort();
 }
+
+#[async_test]
+async fn test_room_member_updates_sender_on_full_member_reload() {
+    use assert_matches::assert_matches;
+    let server = MatrixMockServer::new().await;
+    let client = server.client_builder().build().await;
+
+    let room_id = room_id!("!a:b.c");
+    let room = server.sync_joined_room(&client, room_id).await;
+
+    let mut receiver = room.room_member_updates_sender.subscribe();
+    assert!(receiver.is_empty());
+
+    // When loading the full room member list
+    let user_id = user_id!("@alice:b.c");
+    let joined_event = EventFactory::new()
+        .room(room_id)
+        .event(RoomMemberEventContent::new(MembershipState::Join))
+        .sender(user_id)
+        .state_key(user_id)
+        .into_raw_timeline()
+        .cast();
+    server.mock_get_members().ok(vec![joined_event]).mock_once().mount().await;
+    room.sync_members().await.expect("could not reload room members");
+
+    // The member updates sender emits a full reload
+    let next = assert_recv_with_timeout!(receiver, 100);
+    assert_matches!(next, RoomMembersUpdate::FullReload);
+}
+
+#[async_test]
+async fn test_room_member_updates_sender_on_partial_members_update() {
+    let server = MatrixMockServer::new().await;
+    let client = server.client_builder().build().await;
+
+    let room_id = room_id!("!a:b.c");
+    let room = server.sync_joined_room(&client, room_id).await;
+
+    let mut receiver = room.room_member_updates_sender.subscribe();
+    assert!(receiver.is_empty());
+
+    // When loading a few room member updates
+    let user_id = user_id!("@alice:b.c");
+    let joined_event = EventFactory::new()
+        .room(room_id)
+        .event(RoomMemberEventContent::new(MembershipState::Join))
+        .sender(user_id)
+        .state_key(user_id)
+        .into_raw_sync()
+        .cast();
+    server
+        .sync_room(&client, JoinedRoomBuilder::new(room_id).add_state_bulk(vec![joined_event]))
+        .await;
+
+    // The member updates sender emits a partial update with the user ids of the
+    // members
+    let next = assert_recv_with_timeout!(receiver, 100);
+    assert_let!(RoomMembersUpdate::Partial(user_ids) = next);
+    assert_eq!(user_ids, BTreeSet::from_iter(vec![user_id!("@alice:b.c").to_owned()]));
+}