matrix-rust-sdk

public

Star0

Fork 0

Watch0

6 commits 4 branches 0 tags 0 releases16.6 MiB

Commit `cabc67b5cc05`

tarball

atagen <boss@atagen.co>2026-10-22 12:30 UTC

3 files changed+127-21

send_queue: optionally report upload progress to global listeners

Modified file: crates/matrix-sdk/src/send_queue/mod.rs+57 additions−9 deletions@@ -255,6 +255,7 @@ impl SendQueue {
             &self.client,
             owned_room_id.clone(),
             data.report_media_upload_progress.clone(),
+            data.report_global_upload_progress.clone(),
         );
 
         map.insert(owned_room_id, room_q.clone());
@@ -297,6 +298,17 @@ impl SendQueue {
         self.data().report_media_upload_progress.store(enabled, Ordering::SeqCst);
     }
 
+    /// Enable or disable sending media upload progress updates
+    /// ([`RoomSendQueueUpdate::MediaUpload`]) to the global listeners of
+    /// [`SendQueue::subscribe`], in addition to the room's listeners.
+    ///
+    /// Incremental progress is only reported if it's been enabled with
+    /// [`SendQueue::enable_upload_progress`]; otherwise, only the final
+    /// update of each upload is sent.
+    pub fn enable_global_upload_progress(&self, enabled: bool) {
+        self.data().report_global_upload_progress.store(enabled, Ordering::SeqCst);
+    }
+
     /// Subscribe to all updates for all rooms.
     ///
     /// Use [`RoomSendQueue::subscribe`] to subscribe to update for a _specific
@@ -402,6 +414,10 @@ pub(super) struct SendQueueData {
 
     /// Will media upload progress be reported via send queue updates?
     report_media_upload_progress: Arc<AtomicBool>,
+
+    /// Will media upload progress updates be sent to the global listeners
+    /// too?
+    report_global_upload_progress: Arc<AtomicBool>,
 }
 
 impl SendQueueData {
@@ -417,6 +433,7 @@ impl SendQueueData {
             error_sender,
             is_dropping: Arc::new(false.into()),
             report_media_upload_progress: Arc::new(false.into()),
+            report_global_upload_progress: Arc::new(false.into()),
         }
     }
 }
@@ -466,6 +483,7 @@ impl RoomSendQueue {
         client: &Client,
         room_id: OwnedRoomId,
         report_media_upload_progress: Arc<AtomicBool>,
+        report_global_upload_progress: Arc<AtomicBool>,
     ) -> Self {
         let (update_sender, _) = broadcast::channel(32);
 
@@ -487,6 +505,7 @@ impl RoomSendQueue {
                 global_error_sender,
                 is_dropping,
                 report_media_upload_progress,
+                report_global_upload_progress,
             ),
         );
 
@@ -665,6 +684,7 @@ impl RoomSendQueue {
         global_error_sender: broadcast::Sender<SendQueueRoomError>,
         is_dropping: Arc<AtomicBool>,
         report_media_upload_progress: Arc<AtomicBool>,
+        report_global_upload_progress: Arc<AtomicBool>,
     ) {
         trace!("spawned the sending task");
 
@@ -756,7 +776,10 @@ impl RoomSendQueue {
                             let progress = RoomSendQueue::create_media_upload_progress_observable(
                                 &media_upload_progress_info,
                                 related_to,
+                                room_id,
+                                &global_update_sender,
                                 &update_sender,
+                                &report_global_upload_progress,
                             );
 
                             (Some(media_upload_progress_info), Some(progress))
@@ -883,14 +906,20 @@ impl RoomSendQueue {
                                 })
                                 .unwrap_or(AbstractProgress { current: 1, total: 1 });
 
-                            // Purposefully don't use `send_update` here, because we don't want to
-                            // notify the global listeners about an upload progress update.
-                            let _ = update_sender.send(RoomSendQueueUpdate::MediaUpload {
-                                related_to: related_txn_id.as_ref().unwrap_or(&txn_id).clone(),
-                                file: Some(sent_media_info.file),
-                                index,
-                                progress,
-                            });
+                            // Purposefully don't use `send_update` here, because the global
+                            // listeners only get upload progress updates when they asked for them.
+                            send_upload_progress(
+                                &global_update_sender,
+                                &update_sender,
+                                room_id,
+                                &report_global_upload_progress,
+                                RoomSendQueueUpdate::MediaUpload {
+                                    related_to: related_txn_id.as_ref().unwrap_or(&txn_id).clone(),
+                                    file: Some(sent_media_info.file),
+                                    index,
+                                    progress,
+                                },
+                            );
                         }
 
                         SentRequestKey::Redaction { event_id, redacts, reason } => {
@@ -1252,6 +1281,23 @@ fn send_update(
     let _ = global_update_sender.send(SendQueueUpdate { room_id: room_id.to_owned(), update });
 }
 
+/// Send a media upload progress update to the room's listeners, and to the
+/// global listeners only if [`SendQueue::enable_global_upload_progress`] was
+/// called.
+fn send_upload_progress(
+    global_update_sender: &broadcast::Sender<SendQueueUpdate>,
+    update_sender: &broadcast::Sender<RoomSendQueueUpdate>,
+    room_id: &RoomId,
+    report_global_upload_progress: &AtomicBool,
+    update: RoomSendQueueUpdate,
+) {
+    if report_global_upload_progress.load(Ordering::SeqCst) {
+        send_update(global_update_sender, update_sender, room_id, update);
+    } else {
+        let _ = update_sender.send(update);
+    }
+}
+
 impl From<&crate::Error> for QueueWedgeError {
     fn from(value: &crate::Error) -> Self {
         match value {
@@ -3388,7 +3434,9 @@ mod tests {
         let react_id = ChildTransactionId::new();
         let react = DependentQueuedRequest {
             own_transaction_id: react_id.clone(),
-            kind: DependentQueuedRequestKind::ReactEvent { key: "🧠".to_owned(), shortcode: None },
+            kind: DependentQueuedRequestKind::ReactEvent {
+                key: "🧠".to_owned(), shortcode: None
+            },
             parent_transaction_id: txn.clone(),
             parent_key: None,
             created_at: MilliSecondsSinceUnixEpoch::now(),

Modified file: crates/matrix-sdk/src/send_queue/progress.rs+30 additions−12 deletions@@ -14,20 +14,26 @@
 
 //! Progress facilities for the media upload system.
 
-use std::ops::Add;
+use std::{
+    ops::Add,
+    sync::{Arc, atomic::AtomicBool},
+};
 
 use eyeball::SharedObservable;
 #[cfg(feature = "unstable-msc4274")]
 use matrix_sdk_base::store::AccumulatedSentMediaInfo;
 use matrix_sdk_base::{media::MediaRequestParameters, store::DependentQueuedRequestKind};
 use matrix_sdk_common::executor::spawn;
-use ruma::{TransactionId, events::room::MediaSource};
+use ruma::{RoomId, TransactionId, events::room::MediaSource};
 use tokio::sync::broadcast;
 use tracing::warn;
 
 use crate::{
     Room, TransmissionProgress,
-    send_queue::{QueueStorage, RoomSendQueue, RoomSendQueueStorageError, RoomSendQueueUpdate},
+    send_queue::{
+        QueueStorage, RoomSendQueue, RoomSendQueueStorageError, RoomSendQueueUpdate,
+        SendQueueUpdate, send_upload_progress,
+    },
 };
 
 /// Progress of an operation in abstract units.
@@ -208,13 +214,19 @@ impl RoomSendQueue {
     pub(super) fn create_media_upload_progress_observable(
         media_upload_info: &MediaUploadProgressInfo,
         related_txn_id: &TransactionId,
+        room_id: &RoomId,
+        global_update_sender: &broadcast::Sender<SendQueueUpdate>,
         update_sender: &broadcast::Sender<RoomSendQueueUpdate>,
+        report_global_upload_progress: &Arc<AtomicBool>,
     ) -> SharedObservable<TransmissionProgress> {
         let progress: SharedObservable<TransmissionProgress> = Default::default();
         let mut subscriber = progress.subscribe();
 
         let related_txn_id = related_txn_id.to_owned();
+        let room_id = room_id.to_owned();
+        let global_update_sender = global_update_sender.clone();
         let update_sender = update_sender.clone();
+        let report_global_upload_progress = report_global_upload_progress.clone();
         let media_upload_info = *media_upload_info;
 
         // Watch and communicate the progress on a detached background task. Once
@@ -222,15 +234,21 @@ impl RoomSendQueue {
         // task will end.
         spawn(async move {
             while let Some(progress) = subscriber.next().await {
-                // Purposefully don't use `send_update` here, because we don't want to notify
-                // the global listeners about an upload progress update.
-                let _ = update_sender.send(RoomSendQueueUpdate::MediaUpload {
-                    related_to: related_txn_id.clone(),
-                    file: None,
-                    index: media_upload_info.index,
-                    progress: estimate_media_upload_progress(progress, media_upload_info.bytes)
-                        + media_upload_info.offsets,
-                });
+                // Purposefully don't use `send_update` here, because the global listeners
+                // only get upload progress updates when they asked for them.
+                send_upload_progress(
+                    &global_update_sender,
+                    &update_sender,
+                    &room_id,
+                    &report_global_upload_progress,
+                    RoomSendQueueUpdate::MediaUpload {
+                        related_to: related_txn_id.clone(),
+                        file: None,
+                        index: media_upload_info.index,
+                        progress: estimate_media_upload_progress(progress, media_upload_info.bytes)
+                            + media_upload_info.offsets,
+                    },
+                );
             }
         });
 

Modified file: crates/matrix-sdk/tests/integration/send_queue.rs+40 additions@@ -2991,6 +2991,46 @@ async fn abort_and_verify(
 }
 
 #[async_test]
+async fn test_global_upload_progress() {
+    let mock = MatrixMockServer::new().await;
+
+    let room_id = room_id!("!a:b.c");
+    let client = mock.client_builder().build().await;
+    let room = mock.sync_joined_room(&client, room_id).await;
+
+    // Enabling global upload progress applies to an existing room queue.
+    let q = room.send_queue();
+    client.send_queue().enable_upload_progress(true);
+    client.send_queue().enable_global_upload_progress(true);
+    let mut global_watch = client.send_queue().subscribe();
+
+    mock.mock_authenticated_media_config().ok_default().mount().await;
+    mock.mock_room_state_encryption().plain().mount().await;
+    mock.mock_upload().ok(mxc_uri!("mxc://sdk.rs/media")).mock_once().mount().await;
+    mock.mock_room_send().ok(event_id!("$media")).mock_once().mount().await;
+
+    queue_attachment_no_thumbnail(&q).await;
+
+    let mut uploaded_file = None;
+    loop {
+        let update = timeout(Duration::from_secs(5), global_watch.recv()).await.unwrap().unwrap();
+        assert_eq!(update.room_id, room_id);
+        match update.update {
+            RoomSendQueueUpdate::MediaUpload { file, .. } => {
+                if file.is_some() {
+                    uploaded_file = file;
+                }
+            }
+            RoomSendQueueUpdate::SentEvent { .. } => break,
+            _ => {}
+        }
+    }
+
+    assert_let!(Some(MediaSource::Plain(mxc)) = uploaded_file);
+    assert_eq!(mxc, mxc_uri!("mxc://sdk.rs/media"));
+}
+
+#[async_test]
 async fn test_media_event_is_sent_in_order() {
     // Test that despite happening in multiple requests, sending a media maintains
     // the ordering.

matrix-rust-sdk

Commit cabc67b5cc05

Commit `cabc67b5cc05`