lumen_server/service/
progress.rs

1use std::time::Duration;
2
3use serde::Serialize;
4
5use super::{BoxFuture, ProgressEvent, ProgressSink, ServiceError, ServiceResult};
6
7#[derive(Debug, Default)]
8pub struct NoopProgressSink;
9
10impl ProgressSink for NoopProgressSink {
11    fn publish<'a>(&'a self, _event: ProgressEvent) -> BoxFuture<'a, ServiceResult<()>> {
12        Box::pin(async { Ok(()) })
13    }
14}
15
16#[derive(Debug, Clone)]
17pub struct ProgressCallbackTarget {
18    pub url: String,
19    pub token: Option<String>,
20}
21
22#[derive(Debug, Clone)]
23pub struct CallbackProgressSink {
24    target: ProgressCallbackTarget,
25    timeout: Duration,
26}
27
28impl CallbackProgressSink {
29    pub fn new(target: ProgressCallbackTarget) -> Self {
30        Self {
31            target,
32            timeout: Duration::from_secs(5),
33        }
34    }
35
36    pub fn with_timeout(mut self, timeout: Duration) -> Self {
37        self.timeout = timeout;
38        self
39    }
40}
41
42impl ProgressSink for CallbackProgressSink {
43    fn publish<'a>(&'a self, event: ProgressEvent) -> BoxFuture<'a, ServiceResult<()>> {
44        Box::pin(async move {
45            let payload = ProgressPayload {
46                job_id: &event.job_id.0,
47                progress: event.ratio.clamp(0.0, 1.0),
48                stage: &event.stage,
49                frame: event.frame,
50                total_frames: event.total_frames,
51            };
52            let mut request = reqwest::Client::new()
53                .post(&self.target.url)
54                .json(&payload)
55                .timeout(self.timeout);
56            if let Some(token) = self.target.token.as_deref() {
57                request = request.bearer_auth(token);
58            }
59            let response = request.send().await.map_err(|err| ServiceError {
60                code: "progress_callback_failed",
61                message: err.to_string(),
62                retryable: true,
63            })?;
64            if !response.status().is_success() {
65                return Err(ServiceError {
66                    code: "progress_callback_failed",
67                    message: format!("progress callback returned status {}", response.status()),
68                    retryable: response.status().is_server_error()
69                        || response.status().as_u16() == 429,
70                });
71            }
72            Ok(())
73        })
74    }
75}
76
77#[derive(Debug, Serialize)]
78#[serde(rename_all = "camelCase")]
79struct ProgressPayload<'a> {
80    job_id: &'a str,
81    progress: f32,
82    stage: &'a str,
83    #[serde(skip_serializing_if = "Option::is_none")]
84    frame: Option<u32>,
85    #[serde(skip_serializing_if = "Option::is_none")]
86    total_frames: Option<u32>,
87}
lumen_server/service/progress.rs

lumen_server/service/
progress.rs
