lumen-server 0.2.2

Embeddable render service primitives and local HTTP server for Lumen.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

use std::{sync::Arc, time::Duration};

use super::{
    ArtifactStore, ArtifactWrite, ProgressSink, RenderExecutor, RenderQueue, ServiceError,
    ServiceResult, WorkerId,
};

pub struct RenderService<Q, E, A, P> {
    pub queue: Arc<Q>,
    pub executor: Arc<E>,
    pub artifacts: Arc<A>,
    pub progress: Arc<P>,
}

impl<Q, E, A, P> RenderService<Q, E, A, P>
where
    Q: RenderQueue,
    E: RenderExecutor,
    A: ArtifactStore,
    P: ProgressSink,
{
    pub fn new(queue: Q, executor: E, artifacts: A, progress: P) -> Self {
        Self {
            queue: Arc::new(queue),
            executor: Arc::new(executor),
            artifacts: Arc::new(artifacts),
            progress: Arc::new(progress),
        }
    }

    pub async fn process_next(
        &self,
        worker_id: &WorkerId,
        lease_ttl: Duration,
    ) -> ServiceResult<ProcessNextOutcome> {
        let Some(lease) = self.queue.lease(worker_id, lease_ttl).await? else {
            return Ok(ProcessNextOutcome::NoJob);
        };
        let lease_id = lease.id.clone();
        let job_id = lease.job.id.clone();

        match self
            .executor
            .execute(lease.job, self.progress.as_ref())
            .await
        {
            Ok(output) => {
                let artifact = match self
                    .artifacts
                    .put(ArtifactWrite {
                        job_id,
                        bytes: output.bytes,
                        content_type: output.content_type,
                    })
                    .await
                {
                    Ok(artifact) => artifact,
                    Err(error) => {
                        self.queue.nack(lease_id, error.clone()).await?;
                        return Err(error);
                    }
                };
                self.queue.ack(lease_id).await?;
                Ok(ProcessNextOutcome::Completed {
                    artifact_uri: artifact.uri,
                })
            }
            Err(error) => {
                self.queue.nack(lease_id, error.clone()).await?;
                Err(error)
            }
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ProcessNextOutcome {
    NoJob,
    Completed { artifact_uri: String },
}

impl From<anyhow::Error> for ServiceError {
    fn from(error: anyhow::Error) -> Self {
        Self {
            code: "service_error",
            message: error.to_string(),
            retryable: true,
        }
    }
}