use reqwest::Response; use uuid::Uuid; use crate::error::RequestError; use oneclient_events::{GroupedProgressChild, EventBus}; const PROGRESS_INTERVAL: std::time::Duration = std::time::Duration::from_millis(50); #[derive(Clone, Default)] pub struct ResponseOptions { pub notify: Option, } #[derive(Clone)] pub struct ResponseNotifyOptions { child: Option, standalone_label: Option, standalone_id: Option, done_label: Option, } impl ResponseNotifyOptions { pub fn grouped(child: GroupedProgressChild) -> Self { Self { child: Some(child), standalone_label: None, standalone_id: None, done_label: None, } } pub fn standalone(label: impl Into) -> Self { Self { child: None, standalone_label: Some(label.into()), standalone_id: None, done_label: None, } } pub fn with_id(mut self, id: Uuid) -> Self { self.standalone_id = Some(id); self } pub fn done_label(mut self, label: impl Into) -> Self { self.done_label = Some(label.into()); self } #[must_use] pub fn child(&self) -> Option<&GroupedProgressChild> { self.child.as_ref() } /// Pins the standalone notification id so a retried request keeps updating /// one notification instead of opening a fresh UI entry per attempt #[must_use] pub fn pinned(mut self) -> Self { if self.standalone_label.is_some() && self.standalone_id.is_none() { self.standalone_id = Some(Uuid::new_v4()); } self } } #[async_trait::async_trait] pub trait ResponseExt { async fn stream( self, options: ResponseOptions, events: &EventBus, ) -> Result>, RequestError>; } #[async_trait::async_trait] impl ResponseExt for Response { #[tracing::instrument(level = "debug", skip_all)] async fn stream( self, options: ResponseOptions, events: &EventBus, ) -> Result>, RequestError> { let total = self.content_length().unwrap_or(0).max(1); let mut current = 0u64; let grouped_child = options .notify .as_ref() .and_then(|n| n.child.clone()); let standalone_label = options .notify .as_ref() .and_then(|n| n.standalone_label.clone()); let done_label = options .notify .as_ref() .and_then(|n| n.done_label.clone()); let standalone_id = standalone_label .as_ref() .map(|_| options.notify.as_ref().and_then(|n| n.standalone_id).unwrap_or_else(Uuid::new_v4)); if let Some(ref child) = grouped_child { child.set_progress(0, Some(total)); } else if let (Some(id), Some(label)) = (&standalone_id, &standalone_label) { events.progress(*id, label, 0, total); } let events = events.clone(); // One event per chunk would bury the UI in tens of thousands of updates // so sample the last one always emits so the bar lands on complete let mut last_emit: Option = None; let stream = futures_lite::StreamExt::map(self.bytes_stream(), move |item| { match item { Ok(chunk) => { current += chunk.len() as u64; let now = std::time::Instant::now(); let due = current >= total || last_emit.is_none_or(|last| now.duration_since(last) >= PROGRESS_INTERVAL); if due { last_emit = Some(now); if let Some(ref child) = grouped_child { child.set_progress(current, Some(total)); } else if let (Some(id), Some(label)) = (&standalone_id, &standalone_label) { let done = current >= total; let label = if done { done_label.as_deref().unwrap_or(label) } else { label }; events.progress(*id, label, current, total); } } Ok(chunk) } Err(err) => Err(RequestError::from(err)), } }); Ok(stream) } }