sdk-ui: Add TimelineFocus::PinnedEvents

benchmarks/Cargo.toml

@@ -13,12 +13,14 @@ matrix-sdk-base = { workspace = true }
 matrix-sdk-crypto = { workspace = true }
 matrix-sdk-sqlite = { workspace = true, features = ["crypto-store"] }
 matrix-sdk-test = { workspace = true }
+matrix-sdk-ui = { workspace = true }
 matrix-sdk = { workspace = true, features = ["native-tls", "e2e-encryption", "sqlite"] }
 ruma = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 tempfile = "3.3.0"
 tokio = { workspace = true, default-features = false, features = ["rt-multi-thread"] }
+wiremock = { workspace = true }
 
 [target.'cfg(target_os = "linux")'.dependencies]
 pprof = { version = "0.13.0", features = ["flamegraph", "criterion"] }

benchmarks/benches/room_bench.rs

@@ -1,20 +1,32 @@
+use std::time::Duration;
+
 use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion, Throughput};
-use matrix_sdk::utils::IntoRawStateEventContent;
+use matrix_sdk::{
+    config::SyncSettings,
+    test_utils::{events::EventFactory, logged_in_client_with_server},
+    utils::IntoRawStateEventContent,
+};
 use matrix_sdk_base::{
     store::StoreConfig, BaseClient, RoomInfo, RoomState, SessionMeta, StateChanges, StateStore,
 };
 use matrix_sdk_sqlite::SqliteStateStore;
-use matrix_sdk_test::EventBuilder;
+use matrix_sdk_test::{EventBuilder, JoinedRoomBuilder, StateTestEvent, SyncResponseBuilder};
+use matrix_sdk_ui::{timeline::TimelineFocus, Timeline};
 use ruma::{
     api::client::membership::get_member_events,
     device_id,
     events::room::member::{RoomMemberEvent, RoomMemberEventContent},
-    owned_room_id,
+    owned_room_id, owned_user_id,
     serde::Raw,
-    user_id, OwnedUserId,
+    user_id, EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedUserId,
 };
+use serde::Serialize;
 use serde_json::json;
 use tokio::runtime::Builder;
+use wiremock::{
+    matchers::{header, method, path, path_regex, query_param, query_param_is_missing},
+    Mock, MockServer, Request, ResponseTemplate,
+};
 
 pub fn receive_all_members_benchmark(c: &mut Criterion) {
     const MEMBERS_IN_ROOM: usize = 100000;
@@ -99,6 +111,120 @@ pub fn receive_all_members_benchmark(c: &mut Criterion) {
     group.finish();
 }
 
+pub fn load_pinned_events_benchmark(c: &mut Criterion) {
+    const PINNED_EVENTS_COUNT: usize = 100;
+
+    let runtime = Builder::new_multi_thread().enable_all().build().expect("Can't create runtime");
+    let room_id = owned_room_id!("!room:example.com");
+    let sender_id = owned_user_id!("@sender:example.com");
+
+    let f = EventFactory::new().room(&room_id).sender(&sender_id);
+    let (client, server) = runtime.block_on(logged_in_client_with_server());
+
+    let mut sync_response_builder = SyncResponseBuilder::new();
+    let mut joined_room_builder = JoinedRoomBuilder::new(&room_id);
+
+    let pinned_event_ids: Vec<OwnedEventId> = (0..PINNED_EVENTS_COUNT)
+        .map(|i| EventId::parse(format!("${i}")).expect("Invalid event id"))
+        .collect();
+    joined_room_builder = joined_room_builder.add_state_event(StateTestEvent::Custom(json!(
+        {
+            "content": {
+                "pinned": pinned_event_ids
+            },
+            "event_id": "$15139375513VdeRF:localhost",
+            "origin_server_ts": 151393755,
+            "sender": "@example:localhost",
+            "state_key": "",
+            "type": "m.room.pinned_events",
+            "unsigned": {
+                "age": 703422
+            }
+        }
+    )));
+    let response_json =
+        sync_response_builder.add_joined_room(joined_room_builder).build_json_sync_response();
+    runtime.block_on(mock_sync(&server, response_json, None));
+
+    let sync_settings = SyncSettings::default();
+    runtime.block_on(client.sync_once(sync_settings)).expect("Could not sync");
+    runtime.block_on(server.reset());
+
+    runtime.block_on(
+        Mock::given(method("GET"))
+            .and(path_regex(r"/_matrix/client/r0/rooms/.*/event/.*"))
+            .respond_with(move |r: &Request| {
+                let segments: Vec<&str> = r.url.path_segments().expect("Invalid path").collect();
+                let event_id_str = segments[6];
+                // let f = EventFactory::new().room(&room_id)
+                let event_id = EventId::parse(event_id_str).expect("Invalid event id in response");
+                let event = f
+                    .text_msg(format!("Message {event_id_str}"))
+                    .event_id(&event_id)
+                    .server_ts(MilliSecondsSinceUnixEpoch::now())
+                    .into_raw_sync();
+                ResponseTemplate::new(200)
+                    .set_delay(Duration::from_millis(50))
+                    .set_body_json(event.json())
+            })
+            .mount(&server),
+    );
+    // runtime.block_on(server.reset());
+
+    let room = client.get_room(&room_id).expect("Room not found");
+    assert!(!room.pinned_event_ids().is_empty());
+    assert_eq!(room.pinned_event_ids().len(), PINNED_EVENTS_COUNT);
+
+    let count = PINNED_EVENTS_COUNT;
+    let name = format!("{count} pinned events");
+    let mut group = c.benchmark_group("Test");
+    group.throughput(Throughput::Elements(count as u64));
+    group.sample_size(10);
+
+    group.bench_function(BenchmarkId::new("load_pinned_events", name), |b| {
+        b.to_async(&runtime).iter(|| async {
+            assert!(!room.pinned_event_ids().is_empty());
+            assert_eq!(room.pinned_event_ids().len(), PINNED_EVENTS_COUNT);
+
+            let timeline = Timeline::builder(&room)
+                .with_focus(TimelineFocus::PinnedEvents { max_events_to_load: 100 })
+                .build()
+                .await
+                .expect("Could not create timeline");
+
+            let (items, _) = timeline.subscribe().await;
+            assert_eq!(items.len(), PINNED_EVENTS_COUNT + 1);
+            timeline.clear().await;
+        });
+    });
+
+    {
+        let _guard = runtime.enter();
+        runtime.block_on(server.reset());
+        drop(client);
+        drop(server);
+    }
+
+    group.finish();
+}
+
+async fn mock_sync(server: &MockServer, response_body: impl Serialize, since: Option<String>) {
+    let mut mock_builder = Mock::given(method("GET"))
+        .and(path("/_matrix/client/r0/sync"))
+        .and(header("authorization", "Bearer 1234"));
+
+    if let Some(since) = since {
+        mock_builder = mock_builder.and(query_param("since", since));
+    } else {
+        mock_builder = mock_builder.and(query_param_is_missing("since"));
+    }
+
+    mock_builder
+        .respond_with(ResponseTemplate::new(200).set_body_json(response_body))
+        .mount(server)
+        .await;
+}
+
 fn criterion() -> Criterion {
     #[cfg(target_os = "linux")]
     let criterion = Criterion::default().with_profiler(pprof::criterion::PProfProfiler::new(
@@ -114,6 +240,6 @@ fn criterion() -> Criterion {
 criterion_group! {
     name = room;
     config = criterion();
-    targets = receive_all_members_benchmark,
+    targets = receive_all_members_benchmark, load_pinned_events_benchmark,
 }
 criterion_main!(room);

bindings/matrix-sdk-ffi/src/room.rs

@@ -228,6 +228,25 @@ impl Room {
         Ok(Timeline::new(timeline))
     }
 
+    pub async fn pinned_events_timeline(
+        &self,
+        internal_id_prefix: Option<String>,
+        max_events_to_load: u16,
+    ) -> Result<Arc<Timeline>, ClientError> {
+        let room = &self.inner;
+
+        let mut builder = matrix_sdk_ui::timeline::Timeline::builder(room);
+
+        if let Some(internal_id_prefix) = internal_id_prefix {
+            builder = builder.with_internal_id_prefix(internal_id_prefix);
+        }
+
+        let timeline =
+            builder.with_focus(TimelineFocus::PinnedEvents { max_events_to_load }).build().await?;
+
+        Ok(Timeline::new(timeline))
+    }
+
     pub fn is_encrypted(&self) -> Result<bool, ClientError> {
         Ok(RUNTIME.block_on(self.inner.is_encrypted())?)
     }

bindings/matrix-sdk-ffi/src/room_info.rs

@@ -66,7 +66,7 @@ impl RoomInfo {
         for (id, level) in power_levels_map.iter() {
             user_power_levels.insert(id.to_string(), *level);
         }
-        let pinned_event_ids = room.pinned_events().iter().map(|id| id.to_string()).collect();
+        let pinned_event_ids = room.pinned_event_ids().iter().map(|id| id.to_string()).collect();
 
         Ok(Self {
             id: room.room_id().to_string(),

crates/matrix-sdk-base/src/rooms/normal.rs

@@ -22,6 +22,7 @@ use std::{
 
 use bitflags::bitflags;
 use eyeball::{SharedObservable, Subscriber};
+use futures_util::{Stream, StreamExt};
 #[cfg(all(feature = "e2e-encryption", feature = "experimental-sliding-sync"))]
 use matrix_sdk_common::ring_buffer::RingBuffer;
 #[cfg(feature = "experimental-sliding-sync")]
@@ -948,9 +949,17 @@ impl Room {
         self.inner.read().recency_stamp
     }
 
-    /// Get the list of event ids for pinned events in this room.
-    pub fn pinned_events(&self) -> Vec<OwnedEventId> {
-        self.inner.get().base_info.pinned_events.map(|content| content.pinned).unwrap_or_default()
+    /// Get a `Stream` of loaded pinned events for this room.
+    /// If no pinned events are found a single empty `Vec` will be returned.
+    pub fn pinned_event_ids_stream(&self) -> impl Stream<Item = Vec<OwnedEventId>> {
+        self.inner
+            .subscribe()
+            .map(|i| i.base_info.pinned_events.map(|c| c.pinned).unwrap_or_default())
+    }
+
+    /// Returns the current pinned event ids for this room.
+    pub fn pinned_event_ids(&self) -> Vec<OwnedEventId> {
+        self.inner.read().pinned_event_ids()
     }
 }
 
@@ -1471,6 +1480,24 @@ impl RoomInfo {
     pub(crate) fn update_recency_stamp(&mut self, stamp: u64) {
         self.recency_stamp = Some(stamp);
     }
+
+    /// Returns the current pinned event ids for this room.
+    pub fn pinned_event_ids(&self) -> Vec<OwnedEventId> {
+        self.base_info.pinned_events.clone().map(|c| c.pinned).unwrap_or_default()
+    }
+
+    /// Checks if an `EventId` is currently pinned.
+    /// It avoids having to clone the whole list of event ids to check a single
+    /// value.
+    ///
+    /// Returns `true` if the provided `event_id` is pinned, `false` otherwise.
+    pub fn is_pinned_event(&self, event_id: &EventId) -> bool {
+        self.base_info
+            .pinned_events
+            .as_ref()
+            .map(|p| p.pinned.contains(&event_id.to_owned()))
+            .unwrap_or_default()
+    }
 }
 
 #[cfg(feature = "experimental-sliding-sync")]

crates/matrix-sdk-base/src/sliding_sync/mod.rs

@@ -2194,7 +2194,7 @@ mod tests {
 
         // The newly created room has no pinned event ids
         let room = client.get_room(room_id).unwrap();
-        let pinned_event_ids = room.pinned_events();
+        let pinned_event_ids = room.pinned_event_ids();
         assert!(pinned_event_ids.is_empty());
 
         // Load new pinned event id
@@ -2208,7 +2208,7 @@ mod tests {
         let response = response_with_room(room_id, room_response);
         client.process_sliding_sync(&response, &(), true).await.expect("Failed to process sync");
 
-        let pinned_event_ids = room.pinned_events();
+        let pinned_event_ids = room.pinned_event_ids();
         assert_eq!(pinned_event_ids.len(), 1);
         assert_eq!(pinned_event_ids[0], pinned_event_id);
 
@@ -2222,7 +2222,7 @@ mod tests {
         ));
         let response = response_with_room(room_id, room_response);
         client.process_sliding_sync(&response, &(), true).await.expect("Failed to process sync");
-        let pinned_event_ids = room.pinned_events();
+        let pinned_event_ids = room.pinned_event_ids();
         assert!(pinned_event_ids.is_empty());
     }