chore: Replace Timeline::subscribe by Timeline::subscribe_batched.

This patch changes all calls to `Timeline::subscribe` to replace them by
`Timeline::subscribe_batched`. Most of them are in tests. It's the first
step of a plan to remove `Timeline::subscribe`.

The rest of the patch updates all the tests to use
`Timeline::subscribe_batched`.

Author: Hywan

benchmarks/benches/room_bench.rs

@@ -198,7 +198,7 @@ pub fn load_pinned_events_benchmark(c: &mut Criterion) {
                 .await
                 .expect("Could not create timeline");
 
-            let (items, _) = timeline.subscribe().await;
+            let (items, _) = timeline.subscribe_batched().await;
             assert_eq!(items.len(), PINNED_EVENTS_COUNT + 1);
             timeline.clear().await;
         });

crates/matrix-sdk-ui/tests/integration/room_list_service.rs

@@ -2430,7 +2430,7 @@ async fn test_room_timeline() -> Result<(), Error> {
     room.init_timeline_with_builder(room.default_room_timeline_builder().await.unwrap()).await?;
     let timeline = room.timeline().unwrap();
 
-    let (previous_timeline_items, mut timeline_items_stream) = timeline.subscribe().await;
+    let (previous_timeline_items, mut timeline_items_stream) = timeline.subscribe_batched().await;
 
     sync_then_assert_request_and_fake_response! {
         [server, room_list, sync]
@@ -2493,7 +2493,7 @@ async fn test_room_empty_timeline() {
     // The room wasn't synced, but it will be available
     let room = room_list.room(&room_id).unwrap();
     let timeline = room.default_room_timeline_builder().await.unwrap().build().await.unwrap();
-    let (prev_items, _) = timeline.subscribe().await;
+    let (prev_items, _) = timeline.subscribe_batched().await;
 
     // However, since the room wasn't synced its timeline won't have any initial
     // items

crates/matrix-sdk-ui/tests/integration/timeline/echo.rs

@@ -19,8 +19,8 @@ use assert_matches2::assert_let;
 use eyeball_im::VectorDiff;
 use futures_util::StreamExt;
 use matrix_sdk::{
-    assert_next_matches_with_timeout, config::SyncSettings, executor::spawn,
-    ruma::MilliSecondsSinceUnixEpoch, test_utils::logged_in_client_with_server,
+    config::SyncSettings, executor::spawn, ruma::MilliSecondsSinceUnixEpoch,
+    test_utils::logged_in_client_with_server,
 };
 use matrix_sdk_test::{
     async_test, event_factory::EventFactory, mocks::mock_encryption_state, JoinedRoomBuilder,
@@ -33,7 +33,7 @@ use ruma::{
     room_id, uint, user_id,
 };
 use serde_json::json;
-use stream_assert::assert_next_matches;
+use stream_assert::{assert_next_matches, assert_pending};
 use tokio::task::yield_now;
 use wiremock::{
     matchers::{header, method, path_regex},
@@ -65,7 +65,7 @@ async fn test_echo() {
             .await
             .unwrap(),
     );
-    let (_, mut timeline_stream) = timeline.subscribe().await;
+    let (_, mut timeline_stream) = timeline.subscribe_batched().await;
 
     let event_id = event_id!("$ev");
 
@@ -83,7 +83,10 @@ async fn test_echo() {
         timeline.send(RoomMessageEventContent::text_plain("Hello, World!").into()).await
     });
 
-    assert_let!(Some(VectorDiff::PushBack { value: local_echo }) = timeline_stream.next().await);
+    assert_let!(Some(timeline_updates) = timeline_stream.next().await);
+    assert_eq!(timeline_updates.len(), 2);
+
+    assert_let!(VectorDiff::PushBack { value: local_echo } = &timeline_updates[0]);
     let item = local_echo.as_event().unwrap();
     assert_matches!(item.send_state(), Some(EventSendState::NotSentYet));
     assert_let!(TimelineItemContent::Message(msg) = item.content());
@@ -92,15 +95,16 @@ async fn test_echo() {
     assert!(item.event_id().is_none());
     let txn_id = item.transaction_id().unwrap();
 
-    assert_let!(Some(VectorDiff::PushFront { value: date_divider }) = timeline_stream.next().await);
+    assert_let!(VectorDiff::PushFront { value: date_divider } = &timeline_updates[1]);
     assert!(date_divider.is_date_divider());
 
     // Wait for the sending to finish and assert everything was successful
     send_hdl.await.unwrap().unwrap();
 
-    assert_let!(
-        Some(VectorDiff::Set { index: 1, value: sent_confirmation }) = timeline_stream.next().await
-    );
+    assert_let!(Some(timeline_updates) = timeline_stream.next().await);
+    assert_eq!(timeline_updates.len(), 1);
+
+    assert_let!(VectorDiff::Set { index: 1, value: sent_confirmation } = &timeline_updates[0]);
     let item = sent_confirmation.as_event().unwrap();
     assert_matches!(item.send_state(), Some(EventSendState::Sent { .. }));
     assert_eq!(item.event_id(), Some(event_id));
@@ -120,19 +124,24 @@ async fn test_echo() {
     let _response = client.sync_once(sync_settings.clone()).await.unwrap();
     server.reset().await;
 
+    assert_let!(Some(timeline_updates) = timeline_stream.next().await);
+    assert_eq!(timeline_updates.len(), 4);
+
     // Local echo is replaced with the remote echo.
-    assert_next_matches!(timeline_stream, VectorDiff::Remove { index: 1 });
-    let remote_echo =
-        assert_next_matches!(timeline_stream, VectorDiff::PushFront { value } => value);
+    assert_let!(VectorDiff::Remove { index: 1 } = &timeline_updates[0]);
+
+    assert_let!(VectorDiff::PushFront { value: remote_echo } = &timeline_updates[1]);
     let item = remote_echo.as_event().unwrap();
     assert!(item.is_own());
     assert_eq!(item.timestamp(), MilliSecondsSinceUnixEpoch(uint!(152038280)));
 
     // The date divider is also replaced.
-    let date_divider =
-        assert_next_matches!(timeline_stream, VectorDiff::PushFront { value } => value);
+    assert_let!(VectorDiff::PushFront { value: date_divider } = &timeline_updates[2]);
     assert!(date_divider.is_date_divider());
-    assert_next_matches!(timeline_stream, VectorDiff::Remove { index: 2 });
+
+    assert_let!(VectorDiff::Remove { index: 2 } = &timeline_updates[3]);
+
+    assert_pending!(timeline_stream);
 }
 
 #[async_test]
@@ -229,7 +238,7 @@ async fn test_dedup_by_event_id_late() {
 
     let room = client.get_room(room_id).unwrap();
     let timeline = Arc::new(room.timeline().await.unwrap());
-    let (_, mut timeline_stream) = timeline.subscribe().await;
+    let (_, mut timeline_stream) = timeline.subscribe_batched().await;
 
     let event_id = event_id!("$wWgymRfo7ri1uQx0NXO40vLJ");
 
@@ -251,14 +260,16 @@ async fn test_dedup_by_event_id_late() {
 
     timeline.send(RoomMessageEventContent::text_plain("Hello, World!").into()).await.unwrap();
 
+    assert_let!(Some(timeline_updates) = timeline_stream.next().await);
+    assert_eq!(timeline_updates.len(), 2);
+
     // Timeline: [local echo]
-    let local_echo =
-        assert_next_matches_with_timeout!(timeline_stream, VectorDiff::PushBack { value } => value);
+    assert_let!(VectorDiff::PushBack { value: local_echo } = &timeline_updates[0]);
     let item = local_echo.as_event().unwrap();
     assert_matches!(item.send_state(), Some(EventSendState::NotSentYet));
 
     // Timeline: [date-divider, local echo]
-    let date_divider = assert_next_matches_with_timeout!( timeline_stream, VectorDiff::PushFront { value } => value);
+    assert_let!(VectorDiff::PushFront { value: date_divider } = &timeline_updates[1]);
     assert!(date_divider.is_date_divider());
 
     let f = EventFactory::new();
@@ -275,21 +286,29 @@ async fn test_dedup_by_event_id_late() {
     mock_sync(&server, sync_builder.build_json_sync_response(), None).await;
     let _response = client.sync_once(sync_settings.clone()).await.unwrap();
 
+    assert_let!(Some(timeline_updates) = timeline_stream.next().await);
+    assert_eq!(timeline_updates.len(), 2);
+
     // Timeline: [remote-echo, date-divider, local echo]
-    let remote_echo =
-        assert_next_matches!(timeline_stream, VectorDiff::PushFront { value } => value);
+    assert_let!(VectorDiff::PushFront { value: remote_echo } = &timeline_updates[0]);
     let item = remote_echo.as_event().unwrap();
     assert_eq!(item.event_id(), Some(event_id));
 
     // Timeline: [date-divider, remote-echo, date-divider, local echo]
-    let date_divider = assert_next_matches_with_timeout!(timeline_stream, VectorDiff::PushFront { value } => value);
+    assert_let!(VectorDiff::PushFront { value: date_divider } = &timeline_updates[1]);
     assert!(date_divider.is_date_divider());
 
+    assert_let!(Some(timeline_updates) = timeline_stream.next().await);
+    assert_eq!(timeline_updates.len(), 2);
+
     // Local echo and its date divider are removed.
     // Timeline: [date-divider, remote-echo, date-divider]
-    assert_matches!(timeline_stream.next().await, Some(VectorDiff::Remove { index: 3 }));
+    assert_let!(VectorDiff::Remove { index: 3 } = &timeline_updates[0]);
+
     // Timeline: [date-divider, remote-echo]
-    assert_matches!(timeline_stream.next().await, Some(VectorDiff::Remove { index: 2 }));
+    assert_let!(VectorDiff::Remove { index: 2 } = &timeline_updates[1]);
+
+    assert_pending!(timeline_stream);
 }
 
 #[async_test]