task(ui): Create TimelineSubscriber.

This patch gathers the logic of the `Timeline::subscribe` into a single
type: `TimelineSubscriber`.

The `TimelineController::subscribe_batched_and_limited` method is
renamed `subscribe` to match `Timeline::subscribe`. Things are simpler
to apprehend.

The `TimelineSubscriber` type configures the subscriber/stream in a
single place. It takes an `&ObservableItems` and a `&SkipCount`, and
configures everything. It also provides a single place to document the
behaviour of the subscriber, with the `Skip` higher-order stream.

Author: Hywan

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

@@ -38,6 +38,8 @@ pub(in crate::timeline) struct TimelineMetadata {
     /// This value is constant over the lifetime of the metadata.
     internal_id_prefix: Option<String>,
 
+    /// The `count` value for the `Skip higher-order stream used by the
+    /// `TimelineSubscriber`. See its documentation to learn more.
     pub(super) subscriber_skip_count: SkipCount,
 
     /// The hook to call whenever we run into a unable-to-decrypt event.

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

@@ -54,12 +54,10 @@ use tracing::{
     debug, error, field, field::debug, info, info_span, instrument, trace, warn, Instrument as _,
 };
 
-#[cfg(test)]
-pub(super) use self::observable_items::ObservableItems;
 pub(super) use self::{
     metadata::{RelativePosition, TimelineMetadata},
     observable_items::{
-        AllRemoteEvents, ObservableItemsEntry, ObservableItemsTransaction,
+        AllRemoteEvents, ObservableItems, ObservableItemsEntry, ObservableItemsTransaction,
         ObservableItemsTransactionEntry,
     },
     state::{FullEventMeta, PendingEdit, PendingEditKind, TimelineState},
@@ -70,6 +68,7 @@ use super::{
     event_handler::TimelineEventKind,
     event_item::{ReactionStatus, RemoteEventOrigin},
     item::TimelineUniqueId,
+    subscriber::TimelineSubscriber,
     traits::{Decryptor, RoomDataProvider},
     DateDividerMode, Error, EventSendState, EventTimelineItem, InReplyToDetails, Message,
     PaginationError, Profile, ReactionInfo, RepliedToEvent, TimelineDetails, TimelineEventItemId,
@@ -514,18 +513,14 @@ impl<P: RoomDataProvider> TimelineController<P> {
         impl Stream<Item = VectorDiff<Arc<TimelineItem>>> + SendOutsideWasm,
     ) {
         let state = self.state.read().await;
-        (state.items.clone_items(), state.items.subscribe().into_stream())
+
+        state.items.subscribe().into_values_and_stream()
     }
 
-    pub(super) async fn subscribe_batched_and_limited(
-        &self,
-    ) -> (Vector<Arc<TimelineItem>>, impl Stream<Item = Vec<VectorDiff<Arc<TimelineItem>>>>) {
+    pub(super) async fn subscribe(&self) -> (Vector<Arc<TimelineItem>>, TimelineSubscriber) {
         let state = self.state.read().await;
-        state
-            .items
-            .subscribe()
-            .into_values_and_batched_stream()
-            .dynamic_skip_with_initial_count(0, state.meta.subscriber_skip_count.subscribe())
+
+        TimelineSubscriber::new(&state.items, &state.meta.subscriber_skip_count)
     }
 
     pub(super) async fn subscribe_filter_map<U, F>(

crates/matrix-sdk-ui/src/timeline/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_and_limited().await;
+        let (items, stream) = self.controller.subscribe().await;
         let stream = WithTimelineDropHandle::new(stream, self.drop_handle.clone());
         (items, stream)
     }

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

@@ -18,10 +18,14 @@ use std::{
     task::{Context, Poll},
 };
 
+use eyeball::Subscriber;
+use eyeball_im::{VectorDiff, VectorSubscriberBatchedStream};
+use eyeball_im_util::vector::{Skip, VectorObserverExt};
 use futures_core::Stream;
+use imbl::Vector;
 use pin_project_lite::pin_project;
 
-use super::TimelineDropHandle;
+use super::{controller::ObservableItems, item::TimelineItem, TimelineDropHandle};
 
 pin_project! {
     /// A stream that wraps a [`TimelineDropHandle`] so that the `Timeline`
@@ -46,25 +50,77 @@ where
 {
     type Item = S::Item;
 
-    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        self.project().inner.poll_next(cx)
+    fn poll_next(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        self.project().inner.poll_next(context)
+    }
+}
+
+pin_project! {
+    /// A type that creates a proper `Timeline` subscriber.
+    ///
+    /// This type implements [`Stream`], so that it's entirely transparent for
+    /// all consumers expecting an `impl Stream`.
+    ///
+    /// This `Stream` pipes `VectorDiff`s from [`ObservableItems`] into a batched
+    /// stream ([`VectorSubscriberBatchedStream`]), and then apply a skip
+    /// higher-order stream ([`Skip`]).
+    ///
+    /// `Skip` works by skipping the first _n_ values, where _n_ is referred
+    /// as `count`. Here, this `count` value is defined by a `Stream<Item =
+    /// usize>` (see [`Skip::dynamic_skip_with_initial_count`]). Everytime
+    /// the `count` stream produces a value, `Skip` adjusts its output.
+    /// `count` is managed by [`SkipCount`][skip::SkipCount], and is hold in
+    /// `TimelineMetadata::subscriber_skip_count`.
+    pub(super) struct TimelineSubscriber {
+        #[pin]
+        inner: Skip<VectorSubscriberBatchedStream<Arc<TimelineItem>>, Subscriber<usize>>,
+    }
+}
+
+impl TimelineSubscriber {
+    /// Creates a [`TimelineSubscriber`], in addition to the initial values of
+    /// the subscriber.
+    pub(super) fn new(
+        observable_items: &ObservableItems,
+        observable_skip_count: &skip::SkipCount,
+    ) -> (Vector<Arc<TimelineItem>>, Self) {
+        let (initial_values, stream) = observable_items
+            .subscribe()
+            .into_values_and_batched_stream()
+            .dynamic_skip_with_initial_count(0, observable_skip_count.subscribe());
+
+        (initial_values, Self { inner: stream })
+    }
+}
+
+impl Stream for TimelineSubscriber {
+    type Item = Vec<VectorDiff<Arc<TimelineItem>>>;
+
+    fn poll_next(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        self.project().inner.poll_next(context)
     }
 }
 
 pub mod skip {
-    use eyeball::SharedObservable;
-    use futures_core::Stream;
+    use eyeball::{SharedObservable, Subscriber};
 
     use super::super::controller::TimelineFocusKind;
 
     const MAXIMUM_NUMBER_OF_INITIAL_ITEMS: usize = 20;
 
+    /// `SkipCount` helps to manage the `count` value used by the [`Skip`]
+    /// higher-order stream used by the [`TimelineSubscriber`]. See its
+    /// documentation to learn more.
+    ///
+    /// [`Skip`]: eyeball_im_util::vector::Skip
+    /// [`TimelineSubscriber`]: super::TimelineSubscriber
     #[derive(Clone, Debug)]
     pub struct SkipCount {
         count: SharedObservable<usize>,
     }
 
     impl SkipCount {
+        /// Create a [`SkipCount`] with a default `count` value set to 0.
         pub fn new() -> Self {
             Self { count: SharedObservable::new(0) }
         }
@@ -176,7 +232,7 @@ pub mod skip {
         }
 
         /// Subscribe to update of the count value.
-        pub fn subscribe(&self) -> impl Stream<Item = usize> {
+        pub fn subscribe(&self) -> Subscriber<usize> {
             self.count.subscribe()
         }
 

crates/matrix-sdk-ui/src/timeline/tests/basic.rs

@@ -461,7 +461,7 @@ async fn test_replace_with_initial_events_when_batched() {
         )
         .await;
 
-    let (items, mut stream) = timeline.controller.subscribe_batched_and_limited().await;
+    let (items, mut stream) = timeline.controller.subscribe().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");