task(ui): Apply the Skip higher-order stream for the Timeline.

This patch applies the `Skip` higher-order stream on the `Timeline`
subscriber. The method `TimelineController::subscribe_batched` is
renamed `subscribe_batched_and_limited` at the same time.

The `Skip` stream uses the `TimelineMetadata::subscriber_skip_count`
observable value as the `count_stream`. The initial count value is 0.

No test needs to be changed so far.

Author: Hywan

Cargo.lock

@@ -33,9 +33,9 @@ as_variant = "1.2.0"
 base64 = "0.22.1"
 byteorder = "1.5.0"
 chrono = "0.4.38"
-eyeball = { version = "0.8.8", features = ["tracing"] }
-eyeball-im = { version = "0.6.0", features = ["tracing"] }
-eyeball-im-util = "0.8.0"
+eyeball = { path = "../eyeball/eyeball/", features = ["tracing"] }
+eyeball-im = { path = "../eyeball/eyeball-im", features = ["tracing"] }
+eyeball-im-util = { path = "../eyeball/eyeball-im-util" }
 futures-core = "0.3.31"
 futures-executor = "0.3.21"
 futures-util = "0.3.31"
@@ -44,7 +44,7 @@ growable-bloom-filter = "2.1.1"
 hkdf = "0.12.4"
 hmac = "0.12.1"
 http = "1.1.0"
-imbl = "4.0.1"
+imbl = "3.0.0"
 indexmap = "2.6.0"
 insta = { version = "1.41.1", features = ["json"] }
 itertools = "0.13.0"

Cargo.toml

@@ -496,11 +496,15 @@ impl<P: RoomDataProvider> TimelineController<P> {
         (state.items.clone_items(), state.items.subscribe().into_stream())
     }
 
-    pub(super) async fn subscribe_batched(
+    pub(super) async fn subscribe_batched_and_limited(
         &self,
     ) -> (Vector<Arc<TimelineItem>>, impl Stream<Item = Vec<VectorDiff<Arc<TimelineItem>>>>) {
         let state = self.state.read().await;
-        state.items.subscribe().into_values_and_batched_stream()
+        state
+            .items
+            .subscribe()
+            .into_values_and_batched_stream()
+            .dynamic_skip_with_initial_count(0, state.meta.subscriber_skip_count.subscribe())
     }
 
     pub(super) async fn subscribe_filter_map<U, F>(

crates/matrix-sdk-ui/src/timeline/controller/mod.rs

@@ -276,7 +276,7 @@ impl Timeline {
     pub async fn subscribe(
         &self,
     ) -> (Vector<Arc<TimelineItem>>, impl Stream<Item = Vec<VectorDiff<Arc<TimelineItem>>>>) {
-        let (items, stream) = self.controller.subscribe_batched().await;
+        let (items, stream) = self.controller.subscribe_batched_and_limited().await;
         let stream = WithTimelineDropHandle::new(stream, self.drop_handle.clone());
         (items, stream)
     }

crates/matrix-sdk-ui/src/timeline/mod.rs

@@ -461,7 +461,7 @@ async fn test_replace_with_initial_events_when_batched() {
         )
         .await;
 
-    let (items, mut stream) = timeline.controller.subscribe_batched().await;
+    let (items, mut stream) = timeline.controller.subscribe_batched_and_limited().await;
     assert_eq!(items.len(), 2);
     assert!(items[0].is_date_divider());
     assert_eq!(items[1].as_event().unwrap().content().as_message().unwrap().body(), "hey");