tmprl_client/ops/
schedule.rs

1//! Listing schedules.
2
3use temporalio_client::tonic::Request;
4use temporalio_common::protos::temporal::api::{
5    schedule::v1::ScheduleListEntry, workflowservice::v1::ListSchedulesRequest,
6};
7use tmprl_core::schedule::{Calendar, Interval, Range, ScheduleRow, describe_spec};
8
9use super::OpError;
10use crate::Conn;
11
12#[derive(Debug, Clone, Default)]
13pub struct SchedulePage {
14    pub rows: Vec<ScheduleRow>,
15    pub next_page_token: Vec<u8>,
16}
17
18impl SchedulePage {
19    pub fn has_more(&self) -> bool {
20        !self.next_page_token.is_empty()
21    }
22}
23
24impl Conn {
25    pub async fn list_schedules(
26        &self,
27        namespace: &str,
28        page_size: i32,
29        next_page_token: Vec<u8>,
30    ) -> Result<SchedulePage, OpError> {
31        let resp = self
32            .wf()
33            .list_schedules(Request::new(ListSchedulesRequest {
34                namespace: namespace.to_string(),
35                maximum_page_size: page_size,
36                next_page_token,
37                ..Default::default()
38            }))
39            .await
40            .map_err(|s| OpError::rpc("ListSchedules", s))?
41            .into_inner();
42
43        let mut rows: Vec<ScheduleRow> = resp
44            .schedules
45            .into_iter()
46            .map(|e| row_from(namespace, e))
47            .collect();
48        // ListSchedules gives no ordering guarantee either, and a list that reshuffles
49        // between refreshes is unusable. Schedule ids are stable, so sort on them.
50        rows.sort_by(|a, b| a.schedule_id.cmp(&b.schedule_id));
51
52        Ok(SchedulePage {
53            rows,
54            next_page_token: resp.next_page_token,
55        })
56    }
57}
58
59fn ranges(rs: &[temporalio_common::protos::temporal::api::schedule::v1::Range]) -> Vec<Range> {
60    rs.iter()
61        .map(|r| Range {
62            start: r.start,
63            end: r.end,
64            step: r.step,
65        })
66        .collect()
67}
68
69fn row_from(namespace: &str, e: ScheduleListEntry) -> ScheduleRow {
70    let info = e.info.unwrap_or_default();
71    let spec = info.spec.unwrap_or_default();
72
73    let intervals: Vec<Interval> = spec
74        .interval
75        .iter()
76        .map(|i| Interval {
77            every_secs: i.interval.as_ref().map(|d| d.seconds).unwrap_or(0),
78            offset_secs: i.phase.as_ref().map(|d| d.seconds).unwrap_or(0),
79        })
80        .collect();
81
82    let calendars: Vec<Calendar> = spec
83        .structured_calendar
84        .iter()
85        .map(|c| Calendar {
86            second: ranges(&c.second),
87            minute: ranges(&c.minute),
88            hour: ranges(&c.hour),
89            day_of_month: ranges(&c.day_of_month),
90            month: ranges(&c.month),
91            day_of_week: ranges(&c.day_of_week),
92        })
93        .collect();
94
95    ScheduleRow {
96        namespace: namespace.to_string(),
97        schedule_id: e.schedule_id,
98        workflow_type: info.workflow_type.map(|t| t.name).unwrap_or_default(),
99        paused: info.paused,
100        notes: info.notes,
101        spec: describe_spec(&spec.cron_string, &calendars, &intervals),
102        // The server returns future times ascending, but sorting costs nothing and a
103        // "next run" that is not the earliest would be wrong rather than merely odd.
104        next_run: info
105            .future_action_times
106            .iter()
107            .map(|t| t.seconds * 1000 + i64::from(t.nanos) / 1_000_000)
108            .min(),
109        recent_runs: info.recent_actions.len(),
110    }
111}
112
113#[cfg(test)]
114mod tests {
115    use super::*;
116    use temporalio_common::protos::temporal::api::{
117        common::v1::WorkflowType,
118        schedule::v1::{ScheduleListInfo, ScheduleSpec},
119    };
120
121    fn entry(info: ScheduleListInfo) -> ScheduleListEntry {
122        ScheduleListEntry {
123            schedule_id: "nightly".into(),
124            info: Some(info),
125            ..Default::default()
126        }
127    }
128
129    #[test]
130    fn a_schedule_maps_its_spec_and_state() {
131        let row = row_from(
132            "payments",
133            entry(ScheduleListInfo {
134                workflow_type: Some(WorkflowType {
135                    name: "Reconcile".into(),
136                }),
137                paused: true,
138                notes: "held during migration".into(),
139                spec: Some(ScheduleSpec {
140                    cron_string: vec!["0 2 * * *".into()],
141                    ..Default::default()
142                }),
143                ..Default::default()
144            }),
145        );
146
147        assert_eq!(row.namespace, "payments");
148        assert_eq!(row.schedule_id, "nightly");
149        assert_eq!(row.workflow_type, "Reconcile");
150        assert!(row.paused);
151        assert_eq!(row.spec, "0 2 * * *");
152        assert_eq!(row.notes, "held during migration");
153    }
154
155    #[test]
156    fn the_next_run_is_the_earliest_future_time() {
157        let row = row_from(
158            "d",
159            entry(ScheduleListInfo {
160                future_action_times: vec![
161                    prost_wkt_types::Timestamp {
162                        seconds: 300,
163                        nanos: 0,
164                    },
165                    prost_wkt_types::Timestamp {
166                        seconds: 100,
167                        nanos: 0,
168                    },
169                ],
170                ..Default::default()
171            }),
172        );
173        assert_eq!(row.next_run, Some(100_000));
174    }
175
176    #[test]
177    fn a_schedule_with_nothing_set_still_maps() {
178        // Every field of the entry is optional on the wire, and one malformed row must not
179        // take down the list.
180        let row = row_from("d", ScheduleListEntry::default());
181        assert_eq!(row.spec, "manual");
182        assert_eq!(row.next_run, None);
183        assert!(!row.paused);
184    }
185}
tmprl_client/ops/schedule.rs

tmprl_client/ops/
schedule.rs
