matrix_sdk_ui/room_list_service/
room_list.rs

1// Copyright 2023 The Matrix.org Foundation C.I.C.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for that specific language governing permissions and
13// limitations under the License.
14
15use std::{future::ready, ops::Deref, sync::Arc};
16
17use async_cell::sync::AsyncCell;
18use async_rx::StreamExt as _;
19use async_stream::stream;
20use eyeball::{SharedObservable, Subscriber};
21use eyeball_im::{Vector, VectorDiff};
22use eyeball_im_util::vector::VectorObserverExt;
23use futures_util::{Stream, StreamExt as _, pin_mut, stream};
24use matrix_sdk::{
25    Client, Room, RoomRecencyStamp, RoomState, SlidingSync, SlidingSyncList,
26    executor::{JoinHandle, spawn},
27};
28use matrix_sdk_base::RoomInfoNotableUpdate;
29use ruma::MilliSecondsSinceUnixEpoch;
30use tokio::{
31    select,
32    sync::broadcast::{self, error::RecvError},
33};
34use tracing::{error, trace};
35
36use super::{
37    Error, State,
38    filters::BoxedFilterFn,
39    sorters::{
40        BoxedSorterFn, new_sorter_latest_event, new_sorter_lexicographic, new_sorter_name,
41        new_sorter_recency,
42    },
43};
44
45/// A `RoomList` represents a list of rooms, from a
46/// [`RoomListService`](super::RoomListService).
47#[derive(Debug)]
48pub struct RoomList {
49    client: Client,
50    sliding_sync_list: SlidingSyncList,
51    loading_state: SharedObservable<RoomListLoadingState>,
52    loading_state_task: JoinHandle<()>,
53}
54
55impl Drop for RoomList {
56    fn drop(&mut self) {
57        self.loading_state_task.abort();
58    }
59}
60
61impl RoomList {
62    pub(super) async fn new(
63        client: &Client,
64        sliding_sync: &Arc<SlidingSync>,
65        sliding_sync_list_name: &str,
66        room_list_service_state: Subscriber<State>,
67    ) -> Result<Self, Error> {
68        let sliding_sync_list = sliding_sync
69            .on_list(sliding_sync_list_name, |list| ready(list.clone()))
70            .await
71            .ok_or_else(|| Error::UnknownList(sliding_sync_list_name.to_owned()))?;
72
73        let loading_state =
74            SharedObservable::new(match sliding_sync_list.maximum_number_of_rooms() {
75                Some(maximum_number_of_rooms) => RoomListLoadingState::Loaded {
76                    maximum_number_of_rooms: Some(maximum_number_of_rooms),
77                },
78                None => RoomListLoadingState::NotLoaded,
79            });
80
81        Ok(Self {
82            client: client.clone(),
83            sliding_sync_list: sliding_sync_list.clone(),
84            loading_state: loading_state.clone(),
85            loading_state_task: spawn(async move {
86                pin_mut!(room_list_service_state);
87
88                // As soon as `RoomListService` changes its state, if it isn't
89                // `Terminated` nor `Error`, we know we have fetched something,
90                // so the room list is loaded.
91                while let Some(state) = room_list_service_state.next().await {
92                    use State::*;
93
94                    match state {
95                        Terminated { .. } | Error { .. } | Init => (),
96                        SettingUp | Recovering | Running => break,
97                    }
98                }
99
100                // Let's jump from `NotLoaded` to `Loaded`.
101                let maximum_number_of_rooms = sliding_sync_list.maximum_number_of_rooms();
102
103                loading_state.set(RoomListLoadingState::Loaded { maximum_number_of_rooms });
104
105                // Wait for updates on the maximum number of rooms to update again.
106                let mut maximum_number_of_rooms_stream =
107                    sliding_sync_list.maximum_number_of_rooms_stream();
108
109                while let Some(maximum_number_of_rooms) =
110                    maximum_number_of_rooms_stream.next().await
111                {
112                    loading_state.set(RoomListLoadingState::Loaded { maximum_number_of_rooms });
113                }
114            }),
115        })
116    }
117
118    /// Get a subscriber to the room list loading state.
119    ///
120    /// This method will send out the current loading state as the first update.
121    pub fn loading_state(&self) -> Subscriber<RoomListLoadingState> {
122        self.loading_state.subscribe_reset()
123    }
124
125    /// Get a stream of rooms.
126    fn entries(&self) -> (Vector<Room>, impl Stream<Item = Vec<VectorDiff<Room>>> + '_) {
127        self.client.rooms_stream()
128    }
129
130    /// Get a configurable stream of rooms.
131    ///
132    /// It's possible to provide a filter that will filter out room list
133    /// entries, and that it's also possible to “paginate” over the entries by
134    /// `page_size`. The rooms are also sorted.
135    ///
136    /// The returned stream will only start yielding diffs once a filter is set
137    /// through the returned [`RoomListDynamicEntriesController`]. For every
138    /// call to [`RoomListDynamicEntriesController::set_filter`], the stream
139    /// will yield a [`VectorDiff::Reset`] followed by any updates of the
140    /// room list under that filter (until the next reset).
141    pub fn entries_with_dynamic_adapters(
142        &self,
143        page_size: usize,
144    ) -> (impl Stream<Item = Vec<VectorDiff<RoomListItem>>> + '_, RoomListDynamicEntriesController)
145    {
146        self.entries_with_dynamic_adapters_impl(page_size, false)
147    }
148
149    #[doc(hidden)]
150    pub fn entries_with_dynamic_adapters_with(
151        &self,
152        page_size: usize,
153        enable_latest_event_sorter: bool,
154    ) -> (impl Stream<Item = Vec<VectorDiff<RoomListItem>>> + '_, RoomListDynamicEntriesController)
155    {
156        self.entries_with_dynamic_adapters_impl(page_size, enable_latest_event_sorter)
157    }
158
159    fn entries_with_dynamic_adapters_impl(
160        &self,
161        page_size: usize,
162        enable_latest_event_sorter: bool,
163    ) -> (impl Stream<Item = Vec<VectorDiff<RoomListItem>>> + '_, RoomListDynamicEntriesController)
164    {
165        let room_info_notable_update_receiver = self.client.room_info_notable_update_receiver();
166        let list = self.sliding_sync_list.clone();
167
168        let filter_fn_cell = AsyncCell::shared();
169
170        let limit = SharedObservable::<usize>::new(page_size);
171        let limit_stream = limit.subscribe();
172
173        let dynamic_entries_controller = RoomListDynamicEntriesController::new(
174            filter_fn_cell.clone(),
175            page_size,
176            limit,
177            list.maximum_number_of_rooms_stream(),
178        );
179
180        let stream = stream! {
181            loop {
182                let filter_fn = filter_fn_cell.take().await;
183
184                let (raw_values, raw_stream) = self.entries();
185                let values = raw_values.into_iter().map(Into::into).collect::<Vector<RoomListItem>>();
186
187                // Combine normal stream events with other updates from rooms
188                let stream = merge_stream_and_receiver(values.clone(), raw_stream, room_info_notable_update_receiver.resubscribe());
189
190                let mut sorters: Vec<BoxedSorterFn> = Vec::with_capacity(3);
191
192                if enable_latest_event_sorter {
193                    // Sort by latest event's kind, i.e. put the rooms with a
194                    // **local** latest event first.
195                    sorters.push(Box::new(new_sorter_latest_event()));
196                }
197
198                // Sort rooms by their recency (either by looking
199                // at their latest event's timestamp, or their
200                // `recency_stamp`).
201                sorters.push(Box::new(new_sorter_recency()));
202                // Finally, sort by name.
203                sorters.push(Box::new(new_sorter_name()));
204
205                let (values, stream) = (values, stream)
206                    .filter(filter_fn)
207                    .sort_by(new_sorter_lexicographic(sorters))
208                    .dynamic_head_with_initial_value(page_size, limit_stream.clone());
209
210                // Clearing the stream before chaining with the real stream.
211                yield stream::once(ready(vec![VectorDiff::Reset { values }]))
212                    .chain(stream);
213            }
214        }
215        .switch();
216
217        (stream, dynamic_entries_controller)
218    }
219}
220
221/// This function remembers the current state of the unfiltered room list, so it
222/// knows where all rooms are. When the receiver is triggered, a Set operation
223/// for the room position is inserted to the stream.
224fn merge_stream_and_receiver(
225    mut current_values: Vector<RoomListItem>,
226    raw_stream: impl Stream<Item = Vec<VectorDiff<Room>>>,
227    mut room_info_notable_update_receiver: broadcast::Receiver<RoomInfoNotableUpdate>,
228) -> impl Stream<Item = Vec<VectorDiff<RoomListItem>>> {
229    stream! {
230        pin_mut!(raw_stream);
231
232        loop {
233            select! {
234                // We want to give priority on updates from `raw_stream` as it will necessarily trigger a “refresh” of the rooms.
235                biased;
236
237                diffs = raw_stream.next() => {
238                    if let Some(diffs) = diffs {
239                        let diffs = diffs.into_iter().map(|diff| diff.map(RoomListItem::from)).collect::<Vec<_>>();
240
241                        for diff in &diffs {
242                            diff.clone().map(|room| {
243                                trace!(room = %room.room_id(), "updated in response");
244                                room
245                            }).apply(&mut current_values);
246                        }
247
248                        yield diffs;
249                    } else {
250                        // Restart immediately, don't keep on waiting for the receiver
251                        break;
252                    }
253                }
254
255                update = room_info_notable_update_receiver.recv() => {
256                    match update {
257                        Ok(update) => {
258                            // Emit a `VectorDiff::Set` for the specific rooms.
259                            if let Some(index) = current_values.iter().position(|room| room.room_id() == update.room_id) {
260                                let mut room = current_values[index].clone();
261                                room.refresh_cached_data();
262
263                                yield vec![VectorDiff::Set { index, value: room }];
264                            }
265                        }
266
267                        Err(RecvError::Closed) => {
268                            error!("Cannot receive room info notable updates because the sender has been closed");
269
270                            break;
271                        }
272
273                        Err(RecvError::Lagged(n)) => {
274                            error!(number_of_missed_updates = n, "Lag when receiving room info notable update");
275                        }
276                    }
277                }
278            }
279        }
280    }
281}
282
283/// The loading state of a [`RoomList`].
284///
285/// When a [`RoomList`] is displayed to the user, it can be in various states.
286/// This enum tries to represent those states with a correct level of
287/// abstraction.
288#[derive(Clone, Debug, PartialEq, Eq)]
289pub enum RoomListLoadingState {
290    /// The [`RoomList`] has not been loaded yet, i.e. a sync might run
291    /// or not run at all, there is nothing to show in this `RoomList` yet.
292    /// It's a good opportunity to show a placeholder to the user.
293    ///
294    /// From [`Self::NotLoaded`], it's only possible to move to
295    /// [`Self::Loaded`].
296    NotLoaded,
297
298    /// The [`RoomList`] has been loaded, i.e. a sync has been run, or more
299    /// syncs are running, there is probably something to show to the user.
300    /// Either the user has 0 room, in this case, it's a good opportunity to
301    /// show a special screen for that, or the user has multiple rooms, and it's
302    /// the classical room list.
303    ///
304    /// The number of rooms is represented by `maximum_number_of_rooms`.
305    ///
306    /// From [`Self::Loaded`], it's not possible to move back to
307    /// [`Self::NotLoaded`].
308    Loaded {
309        /// The maximum number of rooms a [`RoomList`] contains.
310        ///
311        /// It does not mean that there are exactly this many rooms to display.
312        /// Usually, the room entries are represented by [`Room`]. The room
313        /// entry might have been synced or not synced yet, but we know for sure
314        /// (from the server), that there will be this amount of rooms in the
315        /// list at the end.
316        ///
317        /// Note that it's an `Option`, because it may be possible that the
318        /// server did miss to send us this value. It's up to you, dear reader,
319        /// to know which default to adopt in case of `None`.
320        maximum_number_of_rooms: Option<u32>,
321    },
322}
323
324/// Controller for the [`RoomList`] dynamic entries.
325///
326/// To get one value of this type, use
327/// [`RoomList::entries_with_dynamic_adapters`]
328pub struct RoomListDynamicEntriesController {
329    filter: Arc<AsyncCell<BoxedFilterFn>>,
330    page_size: usize,
331    limit: SharedObservable<usize>,
332    maximum_number_of_rooms: Subscriber<Option<u32>>,
333}
334
335impl RoomListDynamicEntriesController {
336    fn new(
337        filter: Arc<AsyncCell<BoxedFilterFn>>,
338        page_size: usize,
339        limit_stream: SharedObservable<usize>,
340        maximum_number_of_rooms: Subscriber<Option<u32>>,
341    ) -> Self {
342        Self { filter, page_size, limit: limit_stream, maximum_number_of_rooms }
343    }
344
345    /// Set the filter.
346    ///
347    /// If the associated stream has been dropped, returns `false` to indicate
348    /// the operation didn't have an effect.
349    pub fn set_filter(&self, filter: BoxedFilterFn) -> bool {
350        if Arc::strong_count(&self.filter) == 1 {
351            // there is no other reference to the boxed filter fn, setting it
352            // would be pointless (no new references can be created from self,
353            // either)
354            false
355        } else {
356            self.filter.set(filter);
357            true
358        }
359    }
360
361    /// Add one page, i.e. view `page_size` more entries in the room list if
362    /// any.
363    pub fn add_one_page(&self) {
364        let Some(max) = self.maximum_number_of_rooms.get() else {
365            return;
366        };
367
368        let max: usize = max.try_into().unwrap();
369        let limit = self.limit.get();
370
371        if limit < max {
372            // With this logic, it is possible that `limit` becomes greater than `max` if
373            // `max - limit < page_size`, and that's perfectly fine. It's OK to have a
374            // `limit` greater than `max`, but it's not OK to increase the limit
375            // indefinitely.
376            self.limit.set_if_not_eq(limit + self.page_size);
377        }
378    }
379
380    /// Reset the one page, i.e. forget all pages and move back to the first
381    /// page.
382    pub fn reset_to_one_page(&self) {
383        self.limit.set_if_not_eq(self.page_size);
384    }
385}
386
387/// A facade type that derefs to [`Room`] and that caches data from
388/// [`RoomInfo`].
389///
390/// Why caching data? [`RoomInfo`] is behind a lock. Every time a filter or a
391/// sorter calls a method on [`Room`], it's likely to hit the lock in front of
392/// [`RoomInfo`]. It creates a big contention. By caching the data, it avoids
393/// hitting the lock, improving the performance greatly.
394///
395/// Data are refreshed in `merge_stream_and_receiver` (private function).
396///
397/// [`RoomInfo`]: matrix_sdk::RoomInfo
398#[derive(Clone, Debug)]
399pub struct RoomListItem {
400    /// The inner room.
401    inner: Room,
402
403    /// Cache of `Room::new_latest_event_timestamp`.
404    pub(super) cached_latest_event_timestamp: Option<MilliSecondsSinceUnixEpoch>,
405
406    /// Cache of `Room::new_latest_event_is_local`.
407    pub(super) cached_latest_event_is_local: bool,
408
409    /// Cache of `Room::recency_stamp`.
410    pub(super) cached_recency_stamp: Option<RoomRecencyStamp>,
411
412    /// Cache of `Room::cached_display_name`, already as a string.
413    pub(super) cached_display_name: Option<String>,
414
415    /// Cache of `Room::is_space`.
416    pub(super) cached_is_space: bool,
417
418    // Cache of `Room::state`.
419    pub(super) cached_state: RoomState,
420}
421
422impl RoomListItem {
423    /// Deconstruct to the inner room value.
424    pub fn into_inner(self) -> Room {
425        self.inner
426    }
427
428    /// Refresh the cached data.
429    pub(super) fn refresh_cached_data(&mut self) {
430        self.cached_latest_event_timestamp = self.inner.new_latest_event_timestamp();
431        self.cached_latest_event_is_local = self.inner.new_latest_event_is_local();
432        self.cached_recency_stamp = self.inner.recency_stamp();
433        self.cached_display_name = self.inner.cached_display_name().map(|name| name.to_string());
434        // no need to refresh `Self::is_space`.
435        self.cached_state = self.inner.state();
436    }
437}
438
439impl From<Room> for RoomListItem {
440    fn from(inner: Room) -> Self {
441        let cached_latest_event_timestamp = inner.new_latest_event_timestamp();
442        let cached_latest_event_is_local = inner.new_latest_event_is_local();
443        let cached_recency_stamp = inner.recency_stamp();
444        let cached_display_name = inner.cached_display_name().map(|name| name.to_string());
445        let cached_is_space = inner.is_space();
446        let cached_state = inner.state();
447
448        Self {
449            inner,
450            cached_latest_event_timestamp,
451            cached_latest_event_is_local,
452            cached_recency_stamp,
453            cached_display_name,
454            cached_is_space,
455            cached_state,
456        }
457    }
458}
459
460impl Deref for RoomListItem {
461    type Target = Room;
462
463    fn deref(&self) -> &Self::Target {
464        &self.inner
465    }
466}
matrix_sdk_ui/room_list_service/room_list.rs

matrix_sdk_ui/room_list_service/
room_list.rs
