change(spaces): publish VectorDiffs instead of a full vectors for joined space subscriptions

Author: stefanceriu

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

@@ -14,6 +14,7 @@
 
 use std::{fmt::Debug, sync::Arc};
 
+use eyeball_im::VectorDiff;
 use futures_util::{pin_mut, StreamExt};
 use matrix_sdk_common::{SendOutsideWasm, SyncOutsideWasm};
 use matrix_sdk_ui::spaces::{
@@ -38,7 +39,7 @@ pub struct SpaceService {
 
 impl SpaceService {
     pub(crate) fn new(inner: UISpaceService) -> Self {
-        Self { inner }
+        Self { inner: inner }
     }
 }
 
@@ -57,17 +58,18 @@ impl SpaceService {
         &self,
         listener: Box<dyn SpaceServiceJoinedSpacesListener>,
     ) -> Arc<TaskHandle> {
-        let entries_stream = self.inner.subscribe_to_joined_spaces();
+        let (initial_values, mut stream) = self.inner.subscribe_to_joined_spaces();
 
-        Arc::new(TaskHandle::new(get_runtime_handle().spawn(async move {
-            pin_mut!(entries_stream);
+        listener.on_update(vec![SpaceListUpdate::Reset {
+            values: initial_values.into_iter().map(Into::into).collect(),
+        }]);
 
-            while let Some(rooms) = entries_stream.next().await {
-                listener.on_update(rooms.into_iter().map(Into::into).collect());
+        Arc::new(TaskHandle::new(get_runtime_handle().spawn(async move {
+            while let Some(diffs) = stream.next().await {
+                listener.on_update(diffs.into_iter().map(Into::into).collect());
             }
         })))
     }
-
     #[allow(clippy::unused_async)]
     // This method doesn't need to be async but if its not the FFI layer panics
     // with "there is no no reactor running, must be called from the context
@@ -150,7 +152,7 @@ pub trait SpaceRoomListEntriesListener: SendOutsideWasm + SyncOutsideWasm + Debu
 
 #[matrix_sdk_ffi_macros::export(callback_interface)]
 pub trait SpaceServiceJoinedSpacesListener: SendOutsideWasm + SyncOutsideWasm + Debug {
-    fn on_update(&self, rooms: Vec<SpaceRoom>);
+    fn on_update(&self, room_updates: Vec<SpaceListUpdate>);
 }
 
 #[derive(uniffi::Record)]
@@ -190,3 +192,44 @@ impl From<UISpaceRoom> for SpaceRoom {
         }
     }
 }
+
+#[derive(uniffi::Enum)]
+pub enum SpaceListUpdate {
+    Append { values: Vec<SpaceRoom> },
+    Clear,
+    PushFront { value: SpaceRoom },
+    PushBack { value: SpaceRoom },
+    PopFront,
+    PopBack,
+    Insert { index: u32, value: SpaceRoom },
+    Set { index: u32, value: SpaceRoom },
+    Remove { index: u32 },
+    Truncate { length: u32 },
+    Reset { values: Vec<SpaceRoom> },
+}
+
+impl From<VectorDiff<UISpaceRoom>> for SpaceListUpdate {
+    fn from(diff: VectorDiff<UISpaceRoom>) -> Self {
+        match diff {
+            VectorDiff::Append { values } => {
+                Self::Append { values: values.into_iter().map(|v| v.into()).collect() }
+            }
+            VectorDiff::Clear => Self::Clear,
+            VectorDiff::PushFront { value } => Self::PushFront { value: value.into() },
+            VectorDiff::PushBack { value } => Self::PushBack { value: value.into() },
+            VectorDiff::PopFront => Self::PopFront,
+            VectorDiff::PopBack => Self::PopBack,
+            VectorDiff::Insert { index, value } => {
+                Self::Insert { index: index as u32, value: value.into() }
+            }
+            VectorDiff::Set { index, value } => {
+                Self::Set { index: index as u32, value: value.into() }
+            }
+            VectorDiff::Remove { index } => Self::Remove { index: index as u32 },
+            VectorDiff::Truncate { length } => Self::Truncate { length: length as u32 },
+            VectorDiff::Reset { values } => {
+                Self::Reset { values: values.into_iter().map(|v| v.into()).collect() }
+            }
+        }
+    }
+}

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

@@ -16,8 +16,12 @@
 //!
 //! See [`SpaceService`] for details.
 
-use eyeball::{SharedObservable, Subscriber};
+use std::sync::Arc;
+
+use eyeball_im::{ObservableVector, VectorDiff, VectorSubscriber, VectorSubscriberBatchedStream};
+use futures_core::Stream;
 use futures_util::pin_mut;
+use imbl::Vector;
 use matrix_sdk::{Client, deserialized_responses::SyncOrStrippedState, locks::Mutex};
 use matrix_sdk_common::executor::{JoinHandle, spawn};
 use ruma::{
@@ -39,7 +43,7 @@ pub mod room_list;
 pub struct SpaceService {
     client: Client,
 
-    joined_spaces: SharedObservable<Vec<SpaceRoom>>,
+    joined_spaces: Arc<Mutex<ObservableVector<SpaceRoom>>>,
 
     room_update_handle: Mutex<Option<JoinHandle<()>>>,
 }
@@ -56,15 +60,17 @@ impl SpaceService {
     pub fn new(client: Client) -> Self {
         Self {
             client,
-            joined_spaces: SharedObservable::new(Vec::new()),
+            joined_spaces: Arc::new(Mutex::new(ObservableVector::new())),
             room_update_handle: Mutex::new(None),
         }
     }
 
-    pub fn subscribe_to_joined_spaces(&self) -> Subscriber<Vec<SpaceRoom>> {
+    pub fn subscribe_to_joined_spaces(
+        &self,
+    ) -> (Vector<SpaceRoom>, VectorSubscriberBatchedStream<SpaceRoom>) {
         if self.room_update_handle.lock().is_none() {
-            let client_clone = self.client.clone();
-            let joined_spaces_clone = self.joined_spaces.clone();
+            let client = self.client.clone();
+            let joined_spaces = Arc::clone(&self.joined_spaces);
             let all_room_updates_receiver = self.client.subscribe_to_all_room_updates();
 
             *self.room_update_handle.lock() = Some(spawn(async move {
@@ -73,10 +79,8 @@ impl SpaceService {
                 loop {
                     match all_room_updates_receiver.recv().await {
                         Ok(_) => {
-                            let new_spaces = Self::joined_spaces_for(&client_clone).await;
-                            if new_spaces != joined_spaces_clone.get() {
-                                joined_spaces_clone.set(new_spaces);
-                            }
+                            let new_spaces = Vector::from(Self::joined_spaces_for(&client).await);
+                            Self::update_joined_spaces_if_needed(new_spaces, &joined_spaces);
                         }
                         Err(err) => {
                             error!("error when listening to room updates: {err}");
@@ -86,15 +90,14 @@ impl SpaceService {
             }));
         }
 
-        self.joined_spaces.subscribe()
+        self.joined_spaces.lock().subscribe().into_values_and_batched_stream()
     }
 
     pub async fn joined_spaces(&self) -> Vec<SpaceRoom> {
         let spaces = Self::joined_spaces_for(&self.client).await;
 
-        if spaces != self.joined_spaces.get() {
-            self.joined_spaces.set(spaces.clone());
-        }
+        self.joined_spaces.lock().clear();
+        self.joined_spaces.lock().append(Vector::from(spaces.clone()));
 
         spaces
     }
@@ -103,6 +106,18 @@ impl SpaceService {
         SpaceRoomList::new(self.client.clone(), space_id)
     }
 
+    fn update_joined_spaces_if_needed(
+        new_spaces: Vector<SpaceRoom>,
+        joined_spaces: &Arc<Mutex<ObservableVector<SpaceRoom>>>,
+    ) {
+        let old_spaces = joined_spaces.lock().clone();
+
+        if new_spaces != old_spaces {
+            joined_spaces.lock().clear();
+            joined_spaces.lock().append(Vector::from(new_spaces));
+        }
+    }
+
     async fn joined_spaces_for(client: &Client) -> Vec<SpaceRoom> {
         let joined_spaces = client
             .joined_rooms()
@@ -340,7 +355,7 @@ mod tests {
 
         let space_service = SpaceService::new(client.clone());
 
-        let joined_spaces_subscriber = space_service.subscribe_to_joined_spaces();
+        let (_, joined_spaces_subscriber) = space_service.subscribe_to_joined_spaces();
         pin_mut!(joined_spaces_subscriber);
         assert_pending!(joined_spaces_subscriber);
 
@@ -349,6 +364,17 @@ mod tests {
             vec![SpaceRoom::new_from_known(client.get_room(first_space_id).unwrap(), 0)]
         );
 
+        assert_next_eq!(
+            joined_spaces_subscriber,
+            vec![VectorDiff::Append {
+                values: vec![SpaceRoom::new_from_known(
+                    client.get_room(first_space_id).unwrap(),
+                    0
+                )]
+                .into()
+            }]
+        );
+
         // Join the second space
 
         server
@@ -380,12 +406,25 @@ mod tests {
             ]
         );
 
-        // The subscriber yields new results when a space is joined
         assert_next_eq!(
             joined_spaces_subscriber,
             vec![
-                SpaceRoom::new_from_known(client.get_room(first_space_id).unwrap(), 0),
-                SpaceRoom::new_from_known(client.get_room(second_space_id).unwrap(), 1)
+                VectorDiff::Clear,
+                VectorDiff::Append {
+                    values: vec![
+                        SpaceRoom::new_from_known(client.get_room(first_space_id).unwrap(), 0),
+                        SpaceRoom::new_from_known(client.get_room(second_space_id).unwrap(), 1)
+                    ]
+                    .into()
+                },
+                VectorDiff::Clear,
+                VectorDiff::Append {
+                    values: vec![
+                        SpaceRoom::new_from_known(client.get_room(first_space_id).unwrap(), 0),
+                        SpaceRoom::new_from_known(client.get_room(second_space_id).unwrap(), 1)
+                    ]
+                    .into()
+                }
             ]
         );
 
@@ -394,7 +433,16 @@ mod tests {
         // and when one is left
         assert_next_eq!(
             joined_spaces_subscriber,
-            vec![SpaceRoom::new_from_known(client.get_room(first_space_id).unwrap(), 0)]
+            vec![
+                VectorDiff::Clear,
+                VectorDiff::Append {
+                    values: vec![SpaceRoom::new_from_known(
+                        client.get_room(first_space_id).unwrap(),
+                        0
+                    )]
+                    .into()
+                },
+            ]
         );
 
         // but it doesn't when a non-space room gets joined