feat(ffi): add bindings for listening to global send queue updates

Signed-off-by: Johannes Marbach <[email protected]>

Author: Johennes

bindings/matrix-sdk-ffi/CHANGELOG.md

@@ -29,6 +29,8 @@ All notable changes to this project will be documented in this file.
   ([#5786](https://github.com/matrix-org/matrix-rust-sdk/pull/5786))
 - `ComposerDraft` now includes attachments alongside the text message.
   ([#5794](https://github.com/matrix-org/matrix-rust-sdk/pull/5794))
+- Add `Client::subscribe_to_send_queue_updates` to observe global send queue updates.
+  ([#5784](https://github.com/matrix-org/matrix-rust-sdk/pull/5784))
 
 ### Features:
 

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

@@ -103,7 +103,7 @@ use crate::{
     notification::NotificationClient,
     notification_settings::NotificationSettings,
     qr_code::LoginWithQrCodeHandler,
-    room::{RoomHistoryVisibility, RoomInfoListener},
+    room::{RoomHistoryVisibility, RoomInfoListener, RoomSendQueueUpdate},
     room_directory_search::RoomDirectorySearch,
     room_preview::RoomPreview,
     ruma::{
@@ -195,6 +195,13 @@ pub trait ProgressWatcher: SyncOutsideWasm + SendOutsideWasm {
     fn transmission_progress(&self, progress: TransmissionProgress);
 }
 
+/// A listener to the global (client-wide) update reporter of the send queue.
+#[matrix_sdk_ffi_macros::export(callback_interface)]
+pub trait SendQueueRoomUpdateListener: SyncOutsideWasm + SendOutsideWasm {
+    /// Called every time the send queue emits an update for a given room.
+    fn on_update(&self, room_id: String, update: RoomSendQueueUpdate);
+}
+
 /// A listener to the global (client-wide) error reporter of the send queue.
 #[matrix_sdk_ffi_macros::export(callback_interface)]
 pub trait SendQueueRoomErrorListener: SyncOutsideWasm + SendOutsideWasm {
@@ -607,6 +614,49 @@ impl Client {
         self.inner.send_queue().enable_upload_progress(enable);
     }
 
+    /// Subscribe to the global send queue update reporter, at the
+    /// client-wide level.
+    ///
+    /// The given listener will be immediately called with
+    /// `RoomSendQueueUpdate::NewLocalEvent` for each local echo existing in
+    /// the queue.
+    pub async fn subscribe_to_send_queue_updates(
+        &self,
+        listener: Box<dyn SendQueueRoomUpdateListener>,
+    ) -> Result<Arc<TaskHandle>, ClientError> {
+        let q = self.inner.send_queue();
+        let local_echoes = q.local_echoes().await?;
+        let mut subscriber = q.subscribe();
+
+        for (room_id, local_echoes) in local_echoes {
+            for local_echo in local_echoes {
+                listener.on_update(
+                    room_id.clone().into(),
+                    RoomSendQueueUpdate::NewLocalEvent {
+                        transaction_id: local_echo.transaction_id.into(),
+                    },
+                );
+            }
+        }
+
+        Ok(Arc::new(TaskHandle::new(get_runtime_handle().spawn(async move {
+            loop {
+                match subscriber.recv().await {
+                    Ok(update) => {
+                        let room_id = update.room_id.to_string();
+                        match update.update.try_into() {
+                            Ok(update) => listener.on_update(room_id, update),
+                            Err(err) => error!("error when converting send queue update: {err}"),
+                        }
+                    }
+                    Err(err) => {
+                        error!("error when listening to the send queue update reporter: {err}");
+                    }
+                }
+            }
+        }))))
+    }
+
     /// Subscribe to the global enablement status of the send queue, at the
     /// client-wide level.
     ///

crates/matrix-sdk/src/send_queue/mod.rs

@@ -305,6 +305,31 @@ impl SendQueue {
         self.data().global_update_sender.subscribe()
     }
 
+    /// Get local echoes from all room send queues.
+    pub async fn local_echoes(
+        &self,
+    ) -> Result<BTreeMap<OwnedRoomId, Vec<LocalEcho>>, RoomSendQueueError> {
+        let room_ids =
+            self.client.state_store().load_rooms_with_unsent_requests().await.unwrap_or_else(
+                |err| {
+                    warn!("error when loading rooms with unsent requests: {err}");
+                    Vec::new()
+                },
+            );
+
+        let mut local_echoes: BTreeMap<OwnedRoomId, Vec<LocalEcho>> = BTreeMap::new();
+
+        for room_id in room_ids {
+            if let Some(room) = self.client.get_room(&room_id) {
+                let queue = self.for_room(room);
+                local_echoes
+                    .insert(room_id.to_owned(), queue.inner.queue.local_echoes(&queue).await?);
+            }
+        }
+
+        Ok(local_echoes)
+    }
+
     /// A subscriber to the enablement status (enabled or disabled) of the
     /// send queue, along with useful errors.
     pub fn subscribe_errors(&self) -> broadcast::Receiver<SendQueueRoomError> {