downlaoder: Add global event streams

Author: cyberphantom52

Cargo.lock

@@ -29,6 +29,7 @@ thiserror = "2.0.15"
 tracing = "0.1.41"
 tonic = "0.14"
 tokio = { version = "1.47", features = ["full"] }
+tokio-stream = { version = "0.1.17", features = ["full"] }
 tokio-util = { version = "0.7", features = ["full"] }
 prost = "0.14"
 reqwest = { version = "0.12", features = ["stream"] }

Cargo.toml

@@ -9,6 +9,7 @@ path = "src/download_manager.rs"
 [dependencies]
 reqwest = { workspace = true }
 tokio = { workspace = true }
+tokio-stream = { workspace = true }
 tokio-util = { workspace = true }
 futures-core = { workspace = true }
 futures-util = { workspace = true }

crates/download-manager/Cargo.toml

@@ -1,6 +1,8 @@
 use crate::{DownloadError, DownloadEvent, DownloadID};
+use futures_core::Stream;
 use std::path::PathBuf;
 use tokio::sync::{broadcast, oneshot};
+use tokio_stream::wrappers::BroadcastStream;
 use tokio_util::sync::CancellationToken;
 
 pub struct Download {
@@ -33,6 +35,26 @@ impl Download {
     pub fn cancel(&self) {
         self.cancel_token.cancel();
     }
+
+    pub fn events(&self) -> impl Stream<Item = DownloadEvent> + 'static {
+        use tokio_stream::StreamExt as _;
+
+        let download_id = self.id;
+        BroadcastStream::new(self.events.resubscribe())
+            .filter_map(|res| res.ok())
+            .filter(move |event| {
+                let matches = match event {
+                    DownloadEvent::Queued { id, .. }
+                    | DownloadEvent::Started { id, .. }
+                    | DownloadEvent::Retrying { id, .. }
+                    | DownloadEvent::Completed { id, .. }
+                    | DownloadEvent::Failed { id, .. }
+                    | DownloadEvent::Cancelled { id, .. } => *id == download_id,
+                };
+
+                matches
+            })
+    }
 }
 
 impl std::future::Future for Download {

crates/download-manager/src/download.rs

@@ -13,12 +13,14 @@ pub use crate::{
     events::{DownloadEvent, Progress},
     request::Request,
 };
+use futures_core::Stream;
 use reqwest::Url;
 use std::{
     path::Path,
     sync::{Arc, atomic::Ordering},
 };
-use tokio::sync::mpsc;
+use tokio::sync::{broadcast, mpsc};
+use tokio_stream::wrappers::BroadcastStream;
 use tokio_util::{sync::CancellationToken, task::TaskTracker};
 
 pub struct DownloadManager {
@@ -80,6 +82,16 @@ impl DownloadManager {
     pub fn child_token(&self) -> CancellationToken {
         self.ctx.child_token()
     }
+
+    pub fn subscribe(&self) -> broadcast::Receiver<DownloadEvent> {
+        self.ctx.events.subscribe()
+    }
+
+    pub fn events(&self) -> impl Stream<Item = DownloadEvent> + 'static {
+        use tokio_stream::StreamExt as _;
+
+        BroadcastStream::new(self.subscribe()).filter_map(|res| res.ok())
+    }
 }
 
 pub struct DownloadManagerBuilder {