downloader: Add a EventBus wrapper

Author: cyberphantom52

crates/download-manager/src/context.rs

@@ -3,10 +3,10 @@ use std::sync::{
     Arc,
     atomic::{AtomicU64, AtomicUsize, Ordering},
 };
-use tokio::sync::{Semaphore, broadcast};
+use tokio::sync::Semaphore;
 use tokio_util::sync::CancellationToken;
 
-use crate::DownloadEvent;
+use crate::events::EventBus;
 
 /// Unique identifier for a download; monotonically increasing u64.
 pub type DownloadID = u64;
@@ -33,7 +33,7 @@ pub(crate) struct Context {
     pub max_concurrent: AtomicUsize,
 
     /// Global [DownloadEvent] broadcaster (buffered). Slow subscribers may miss events.
-    pub events: broadcast::Sender<DownloadEvent>,
+    pub events: EventBus,
 }
 
 impl Context {
@@ -42,15 +42,14 @@ impl Context {
     /// - Creates a root [CancellationToken] and a broadcast channel (capacity 1024).
     /// - Constructs a shared [reqwest::Client].
     pub fn new(max_concurrent: usize) -> Arc<Self> {
-        let (events, _) = broadcast::channel(1024);
         Arc::new(Self {
             semaphore: Arc::new(Semaphore::new(max_concurrent)),
             max_concurrent: AtomicUsize::new(max_concurrent),
             cancel_root: CancellationToken::new(),
             active: AtomicUsize::new(0),
             id_counter: AtomicU64::new(1),
             client: Client::new(),
-            events,
+            events: EventBus::new(),
         })
     }
 

crates/download-manager/src/download.rs

@@ -1,4 +1,4 @@
-use crate::{DownloadError, DownloadEvent, DownloadID, Progress};
+use crate::{DownloadError, DownloadID, Event, Progress};
 use futures_core::Stream;
 use std::path::PathBuf;
 use tokio::sync::{broadcast, oneshot, watch};
@@ -14,7 +14,7 @@ use tokio_util::sync::CancellationToken;
 pub struct Download {
     id: DownloadID,
     progress: watch::Receiver<Progress>,
-    events: broadcast::Receiver<DownloadEvent>,
+    events: broadcast::Receiver<Event>,
     result: oneshot::Receiver<Result<DownloadResult, DownloadError>>,
 
     cancel_token: CancellationToken,
@@ -24,7 +24,7 @@ impl Download {
     pub(crate) fn new(
         id: DownloadID,
         progress: watch::Receiver<Progress>,
-        events: broadcast::Receiver<DownloadEvent>,
+        events: broadcast::Receiver<Event>,
         result: oneshot::Receiver<Result<DownloadResult, DownloadError>>,
         cancel_token: CancellationToken,
     ) -> Self {
@@ -66,21 +66,21 @@ impl Download {
     ///
     /// Backed by a broadcast channel; lagged consumers may drop messages.
     /// This stream filters events to those whose id matches this handle.
-    pub fn events(&self) -> impl Stream<Item = DownloadEvent> + 'static {
+    pub fn events(&self) -> impl Stream<Item = Event> + '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::Probed { id, .. }
-                    | DownloadEvent::Started { id, .. }
-                    | DownloadEvent::Retrying { id, .. }
-                    | DownloadEvent::Completed { id, .. }
-                    | DownloadEvent::Failed { id, .. }
-                    | DownloadEvent::Cancelled { id, .. } => *id == download_id,
+                    Event::Queued { id, .. }
+                    | Event::Probed { id, .. }
+                    | Event::Started { id, .. }
+                    | Event::Retrying { id, .. }
+                    | Event::Completed { id, .. }
+                    | Event::Failed { id, .. }
+                    | Event::Cancelled { id, .. } => *id == download_id,
                 };
 
                 matches

crates/download-manager/src/download_manager.rs

@@ -9,7 +9,7 @@ pub mod prelude {
         context::DownloadID,
         download::{Download, DownloadResult},
         error::DownloadError,
-        events::{DownloadEvent, Progress},
+        events::{Event, Progress},
         request::Request,
     };
 }
@@ -27,7 +27,6 @@ use std::{
     sync::{Arc, atomic::Ordering},
 };
 use tokio::sync::{broadcast, mpsc};
-use tokio_stream::wrappers::BroadcastStream;
 use tokio_util::{sync::CancellationToken, task::TaskTracker};
 
 /// Entry point for scheduling, observing, and cancelling downloads.
@@ -116,17 +115,15 @@ impl DownloadManager {
     ///
     /// The underlying broadcast channel has a bounded buffer (1024). Slow consumers may lag and
     /// miss events. Consider using [DownloadManager::events()] for a stream that skips lagged messages gracefully.
-    pub fn subscribe(&self) -> broadcast::Receiver<DownloadEvent> {
+    pub fn subscribe(&self) -> broadcast::Receiver<Event> {
         self.ctx.events.subscribe()
     }
 
     /// A fallible-safe stream of global [DownloadEvent] values.
     ///
     /// Internally wraps the broadcast receiver and filters out lagged/closed errors.
-    pub fn events(&self) -> impl Stream<Item = DownloadEvent> + 'static {
-        use tokio_stream::StreamExt as _;
-
-        BroadcastStream::new(self.subscribe()).filter_map(|res| res.ok())
+    pub fn events(&self) -> impl Stream<Item = Event> + 'static {
+        self.ctx.events.events()
     }
 
     /// Gracefully stop the manager.

crates/download-manager/src/events.rs

@@ -3,9 +3,36 @@ use crate::download::RemoteInfo;
 use reqwest::Url;
 use std::path::PathBuf;
 use std::time::{Duration, Instant};
+use tokio::sync::broadcast;
 
 #[derive(Debug, Clone)]
-pub enum DownloadEvent {
+pub(crate) struct EventBus(broadcast::Sender<Event>);
+
+impl EventBus {
+    pub fn new() -> Self {
+        let (tx, _rx) = broadcast::channel(1024);
+        EventBus(tx)
+    }
+
+    pub fn subscribe(&self) -> broadcast::Receiver<Event> {
+        self.0.subscribe()
+    }
+
+    pub fn events(&self) -> impl tokio_stream::Stream<Item = Event> + 'static {
+        use tokio_stream::StreamExt as _;
+        use tokio_stream::wrappers::BroadcastStream;
+
+        BroadcastStream::new(self.subscribe()).filter_map(|res| res.ok())
+    }
+
+    pub fn send(&self, event: Event) {
+        // TODO: Log the error
+        let _ = self.0.send(event);
+    }
+}
+
+#[derive(Debug, Clone)]
+pub enum Event {
     Queued {
         id: DownloadID,
         url: Url,
@@ -40,6 +67,48 @@ pub enum DownloadEvent {
     },
 }
 
+impl std::fmt::Display for Event {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Event::Queued { id, .. } => write!(f, "[{}] Queued", id),
+            Event::Probed { id, info } => write!(f, "[{}] Probed: {:?}", id, info),
+            Event::Failed { id, error } => write!(f, "[{}] Failed: {}", id, error),
+            Event::Cancelled { id } => write!(f, "[{}] Cancelled", id),
+            Event::Started {
+                id, total_bytes, ..
+            } => {
+                if let Some(total) = total_bytes {
+                    write!(f, "[{}] Started ({} bytes)", id, total)
+                } else {
+                    write!(f, "[{}] Started", id)
+                }
+            }
+            Event::Retrying {
+                id,
+                attempt,
+                next_delay_ms,
+            } => {
+                write!(
+                    f,
+                    "[{}] Retrying: attempt {} in {} ms",
+                    id, attempt, next_delay_ms
+                )
+            }
+            Event::Completed {
+                id,
+                path,
+                bytes_downloaded,
+            } => {
+                write!(
+                    f,
+                    "[{}] Completed: {:?} ({} bytes)",
+                    id, path, bytes_downloaded
+                )
+            }
+        }
+    }
+}
+
 #[derive(Debug, Clone, Copy)]
 pub struct Progress {
     pub bytes_downloaded: u64,

crates/download-manager/src/request.rs

@@ -1,5 +1,6 @@
 use crate::{
-    Download, DownloadEvent, DownloadID, DownloadManager, Progress, scheduler::SchedulerCmd,
+    Download, DownloadID, DownloadManager, Event, Progress, events::EventBus,
+    scheduler::SchedulerCmd,
 };
 use derive_builder::Builder;
 use reqwest::{
@@ -10,7 +11,7 @@ use std::{
     path::{Path, PathBuf},
     sync::Arc,
 };
-use tokio::sync::{broadcast, oneshot, watch};
+use tokio::sync::{oneshot, watch};
 use tokio_util::sync::CancellationToken;
 
 /// Immutable description of a single download request.
@@ -26,10 +27,10 @@ pub struct Request {
     config: DownloadConfig,
 
     progress: watch::Sender<Progress>,
-    events: broadcast::Sender<DownloadEvent>,
+    events: EventBus,
 
     pub(crate) on_progress: Option<Arc<Box<dyn Fn(Progress) + Send + Sync>>>,
-    pub(crate) on_event: Option<Arc<Box<dyn Fn(DownloadEvent) + Send + Sync>>>,
+    pub(crate) on_event: Option<Arc<Box<dyn Fn(Event) + Send + Sync>>>,
 
     pub cancel_token: CancellationToken,
 }
@@ -116,9 +117,8 @@ impl Request {
         &self.config
     }
 
-    pub fn emit(&self, event: DownloadEvent) {
-        // TODO: Log the error
-        let _ = self.events.send(event.clone());
+    pub fn emit(&self, event: Event) {
+        self.events.send(event.clone());
         self.on_event.as_ref().map(|cb| cb(event));
     }
 
@@ -138,7 +138,7 @@ pub struct RequestBuilder<'a> {
     config: DownloadConfigBuilder,
 
     on_progress: Option<Arc<Box<dyn Fn(Progress) + Send + Sync>>>,
-    on_event: Option<Arc<Box<dyn Fn(DownloadEvent) + Send + Sync>>>,
+    on_event: Option<Arc<Box<dyn Fn(Event) + Send + Sync>>>,
 
     manager: &'a DownloadManager,
 }
@@ -199,7 +199,7 @@ impl RequestBuilder<'_> {
     /// Called for events emitted for this request only.
     pub fn on_event<F>(mut self, callback: F) -> Self
     where
-        F: Fn(DownloadEvent) + Send + Sync + 'static,
+        F: Fn(Event) + Send + Sync + 'static,
     {
         self.on_event = Some(Arc::new(Box::new(callback)));
         self
@@ -220,8 +220,8 @@ impl RequestBuilder<'_> {
         let (result_tx, result_rx) = oneshot::channel();
         let (progress_tx, progress_rx) = watch::channel(Progress::new(None));
         let cancel_token = self.manager.child_token();
-        let event_tx = self.manager.ctx.events.clone();
-        let event_rx = event_tx.subscribe();
+        let event_bus = self.manager.ctx.events.clone();
+        let event_rx = event_bus.subscribe();
 
         let on_progress = self.on_progress;
         let on_event = self.on_event;
@@ -235,7 +235,7 @@ impl RequestBuilder<'_> {
             on_progress,
             on_event,
 
-            events: event_tx,
+            events: event_bus,
             progress: progress_tx,
             cancel_token: cancel_token.clone(),
         };

crates/download-manager/src/scheduler.rs

@@ -14,7 +14,7 @@ use tokio::{
 use tokio_util::{task::TaskTracker, time::DelayQueue};
 
 use crate::{
-    DownloadError, DownloadEvent, DownloadID, DownloadResult, Progress, Request, context::Context,
+    DownloadError, DownloadID, DownloadResult, Event, Progress, Request, context::Context,
     download::RemoteInfo,
 };
 
@@ -88,7 +88,7 @@ impl Scheduler {
     fn schedule(&mut self, job: Job) {
         let request = &job.request;
         let id = job.id();
-        request.emit(DownloadEvent::Queued {
+        request.emit(Event::Queued {
             id,
             url: request.url().clone(),
             destination: request.destination().to_path_buf(),
@@ -233,15 +233,15 @@ impl Job {
     }
 
     fn fail(self, error: DownloadError) {
-        self.request.emit(DownloadEvent::Failed {
+        self.request.emit(Event::Failed {
             id: self.id(),
             error: error.to_string(),
         });
         self.send_result(Err(error));
     }
 
     fn finish(self, result: DownloadResult) {
-        self.request.emit(DownloadEvent::Completed {
+        self.request.emit(Event::Completed {
             id: self.id(),
             path: result.path.clone(),
             bytes_downloaded: result.bytes_downloaded,
@@ -250,7 +250,7 @@ impl Job {
     }
 
     fn retry(&self, delay: Duration) {
-        self.request.emit(DownloadEvent::Retrying {
+        self.request.emit(Event::Retrying {
             id: self.id(),
             attempt: self.attempt,
             next_delay_ms: delay.as_millis() as u64,
@@ -259,8 +259,7 @@ impl Job {
 
     fn cancel(self) {
         self.request.cancel_token.cancel();
-        self.request
-            .emit(DownloadEvent::Cancelled { id: self.id() });
+        self.request.emit(Event::Cancelled { id: self.id() });
         self.send_result(Err(DownloadError::Cancelled))
     }
 }
@@ -320,7 +319,7 @@ async fn attempt_download(
     client: Client,
 ) -> Result<DownloadResult, DownloadError> {
     if let Some(info) = probe_head(request, &client).await {
-        request.emit(DownloadEvent::Probed {
+        request.emit(Event::Probed {
             id: request.id(),
             info,
         });
@@ -344,7 +343,7 @@ async fn attempt_download(
     let total_bytes = response.content_length();
 
     let mut file = File::create(request.destination()).await?;
-    request.emit(DownloadEvent::Started {
+    request.emit(Event::Started {
         id: request.id(),
         url: request.url().clone(),
         destination: request.destination().to_path_buf(),
@@ -364,8 +363,7 @@ async fn attempt_download(
                     Ok(Some(chunk)) => {
                         file.write_all(&chunk).await?;
                         if progress.update(chunk.len() as u64) {
-                            // TODO: Log the error
-                            let _ = request.update_progress(progress);
+                            request.update_progress(progress);
                         }
                     }
                     Ok(None) => break,