fixup! crypto: Provide a way to subscribe to identity status changes

crates/matrix-sdk/src/room/identity_status_changes.rs

@@ -91,7 +91,10 @@ impl IdentityStatusChanges {
                 yield current_state;
             }
             while let Some(item) = unprocessed_stream.next().await {
-                yield state.room_identity_state.process_change(item).await;
+                let update = state.room_identity_state.process_change(item).await;
+                if !update.is_empty() {
+                    yield update;
+                }
             }
         }))
     }
@@ -145,9 +148,13 @@ fn wrap_room_member_events(
 
 #[cfg(test)]
 mod tests {
-    use std::pin::{pin, Pin};
+    use std::{
+        pin::{pin, Pin},
+        time::Duration,
+    };
 
     use futures_core::Stream;
+    use futures_util::FutureExt;
     use matrix_sdk_base::crypto::{IdentityState, IdentityStatusChange};
     use matrix_sdk_test::async_test;
     use test_setup::TestSetup;
@@ -233,13 +240,15 @@ mod tests {
 
         // And we are listening for identity changes
         let changes = t.subscribe_to_identity_status_changes().await;
+        let mut changes = pin!(changes);
 
         // When Bob joins the room
         t.join().await;
 
         // Then there is no notification
-        let change = next_change(&mut pin!(changes)).await;
-        assert!(change.is_empty());
+        tokio::time::sleep(Duration::from_millis(200)).await;
+        let change = changes.next().now_or_never();
+        assert!(change.is_none());
     }
 
     #[async_test]
@@ -302,34 +311,30 @@ mod tests {
 
         // ... leaves and joins again (ignored since they stay pinned) ...
         t.leave().await;
-        let change3 = next_change(&mut changes).await;
         t.join().await;
-        let change4 = next_change(&mut changes).await;
 
         // ... becomes unpinned ...
         t.unpin().await;
-        let change5 = next_change(&mut changes).await;
+        let change3 = next_change(&mut changes).await;
 
         // ... and leaves.
         t.leave().await;
-        let change6 = next_change(&mut changes).await;
+        let change4 = next_change(&mut changes).await;
 
         assert_eq!(change1[0].user_id, t.user_id());
         assert_eq!(change2[0].user_id, t.user_id());
-        assert_eq!(change5[0].user_id, t.user_id());
-        assert_eq!(change6[0].user_id, t.user_id());
+        assert_eq!(change3[0].user_id, t.user_id());
+        assert_eq!(change4[0].user_id, t.user_id());
 
         assert_eq!(change1[0].changed_to, IdentityState::PinViolation);
         assert_eq!(change2[0].changed_to, IdentityState::Pinned);
-        assert_eq!(change5[0].changed_to, IdentityState::PinViolation);
-        assert_eq!(change6[0].changed_to, IdentityState::Pinned);
+        assert_eq!(change3[0].changed_to, IdentityState::PinViolation);
+        assert_eq!(change4[0].changed_to, IdentityState::Pinned);
 
         assert_eq!(change1.len(), 1);
         assert_eq!(change2.len(), 1);
-        assert_eq!(change3.len(), 0);
-        assert_eq!(change4.len(), 0);
-        assert_eq!(change5.len(), 1);
-        assert_eq!(change6.len(), 1);
+        assert_eq!(change3.len(), 1);
+        assert_eq!(change4.len(), 1);
     }
 
     async fn next_change(