sofie_db/contracts/
jobs.rs

1use std::any::TypeId;
2use std::collections::HashMap;
3use std::sync::Arc;
4
5use serde::Serialize;
6use serde_json::Value;
7use sofie_core::{Error, Instant, Result};
8
9use crate::worker::JobContext;
10use crate::{Call, Db, MessageKind, Outcome, Plan, Resolve, Writes};
11
12#[derive(Clone, Default)]
13pub(crate) struct JobNames(HashMap<TypeId, &'static str>);
14
15pub fn name_job<Args, J: Call<JobContext, Args>>(db: &Db, name: &'static str, _job: J) {
16    db.provided.change(|names: &mut JobNames| {
17        names.0.insert(TypeId::of::<J>(), name);
18    });
19}
20
21pub struct Jobs(pub(crate) Arc<JobNames>);
22
23impl<C: Writes> Resolve<C> for Jobs {
24    fn resolve(context: &mut C) -> impl Future<Output = Result<Self>> + Send {
25        let names = context.db().provided.get::<JobNames>().unwrap_or_default();
26        std::future::ready(Ok(Self(names)))
27    }
28}
29
30impl Jobs {
31    /// Stored when the handler's outcome commits.
32    pub fn queue<Args, J: Call<JobContext, Args>>(
33        &self,
34        job: J,
35        payload: &impl Serialize,
36    ) -> Result<Queued> {
37        Queued::new(self.name(job)?, payload, None)
38    }
39
40    /// Run no sooner than `at`; one already past runs at once.
41    pub fn queue_at<Args, J: Call<JobContext, Args>>(
42        &self,
43        job: J,
44        payload: &impl Serialize,
45        at: Instant,
46    ) -> Result<Queued> {
47        Queued::new(self.name(job)?, payload, Some(at))
48    }
49
50    /// For a package's job that queues itself again, which its own type cannot name.
51    #[doc(hidden)]
52    pub fn queue_named_at(
53        name: &'static str,
54        payload: &impl Serialize,
55        at: Instant,
56    ) -> Result<Queued> {
57        Queued::new(name, payload, Some(at))
58    }
59
60    fn name<Args, J: Call<JobContext, Args>>(&self, _job: J) -> Result<&'static str> {
61        self.0.0.get(&TypeId::of::<J>()).copied().ok_or_else(|| {
62            Error::internal_message(format!(
63                "{} is not a job; mark it `#[job]` and run `sofie gen`",
64                std::any::type_name::<J>()
65            ))
66        })
67    }
68}
69
70pub struct Queued {
71    pub(super) name: &'static str,
72    pub(super) payload: Value,
73    at: Option<Instant>,
74}
75
76impl Queued {
77    fn new(name: &'static str, payload: &impl Serialize, at: Option<Instant>) -> Result<Self> {
78        let payload = serde_json::to_value(payload)
79            .map_err(|error| Error::internal(format!("serializing job `{name}`"), error))?;
80        Ok(Self { name, payload, at })
81    }
82}
83
84impl Outcome for Queued {
85    fn plan(&self, plan: &mut Plan<'_>) -> Result<()> {
86        let message = [(self.name, self.payload.clone())];
87        match self.at {
88            Some(at) => plan.queue_at(MessageKind::Job, message, at.timestamp()),
89            None => plan.queue(MessageKind::Job, message),
90        }
91    }
92}
sofie_db/contracts/jobs.rs

sofie_db/contracts/
jobs.rs
