dataflow_rs/def/
ser.rs

1//! `Serialize` for the definition types: the v4 spelling, which
2//! [`parse`](super::parse) reads back with the compat reader off.
3//!
4//! Only v4 keys are written, in the known-key table's order. A key holding
5//! its default is left out (`priority` 0, `channel` `"default"`, `status`
6//! active, `version` 1, the default policies, `terminal: false`, `init` 0,
7//! `increment` 1, `max_concurrency` 1), and so is an empty one (`None`, an
8//! empty `tags`, an empty `ext`). `ext` is written as `"ext"`, a function's
9//! `input` as it was, a condition or `over` as its raw JSON, and a timestamp
10//! in RFC 3339.
11
12use super::Rollout;
13use super::types::{
14    FailurePolicy, ForEachDef, FunctionRef, GroupDef, Logic, LoopDef, NodeHeader, RejectionPolicy,
15    Status, StepDef, TaskDef, WorkflowDef, WorkflowFailurePolicy, defaults,
16};
17use chrono::{DateTime, Utc};
18use serde::ser::{Serialize, SerializeMap, Serializer};
19use std::convert::Infallible;
20
21/// Formats a timestamp as RFC 3339 only in the write pass, not in [`Count`].
22struct Rfc3339<'a>(&'a DateTime<Utc>);
23
24impl Serialize for Rfc3339<'_> {
25    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
26        serializer.serialize_str(&self.0.to_rfc3339())
27    }
28}
29
30/// Where a node's present fields go. Each node lists them once, and that list
31/// is run twice: counted, then written, so the map's length is exact for a
32/// format that needs it up front.
33trait Fields {
34    type Error;
35    fn field<T: Serialize + ?Sized>(
36        &mut self,
37        key: &'static str,
38        value: &T,
39    ) -> Result<(), Self::Error>;
40
41    /// The field, when `value` is set.
42    fn opt<T: Serialize>(
43        &mut self,
44        key: &'static str,
45        value: Option<&T>,
46    ) -> Result<(), Self::Error> {
47        value.map_or(Ok(()), |value| self.field(key, value))
48    }
49
50    /// The field, when `present`: a key holding its default is left out.
51    fn field_if<T: Serialize + ?Sized>(
52        &mut self,
53        present: bool,
54        key: &'static str,
55        value: &T,
56    ) -> Result<(), Self::Error> {
57        if present {
58            self.field(key, value)
59        } else {
60            Ok(())
61        }
62    }
63}
64
65struct Count(usize);
66
67impl Fields for Count {
68    type Error = Infallible;
69    fn field<T: Serialize + ?Sized>(&mut self, _: &'static str, _: &T) -> Result<(), Infallible> {
70        self.0 += 1;
71        Ok(())
72    }
73}
74
75struct Write<M>(M);
76
77impl<M: SerializeMap> Fields for Write<M> {
78    type Error = M::Error;
79    fn field<T: Serialize + ?Sized>(
80        &mut self,
81        key: &'static str,
82        value: &T,
83    ) -> Result<(), M::Error> {
84        self.0.serialize_entry(key, value)
85    }
86}
87
88/// A node written as a map of its present fields.
89trait Node {
90    fn fields<F: Fields>(&self, out: &mut F) -> Result<(), F::Error>;
91}
92
93fn node<N: Node, S: Serializer>(node: &N, serializer: S) -> Result<S::Ok, S::Error> {
94    let mut count = Count(0);
95    let Ok(()) = node.fields(&mut count);
96    let mut out = Write(serializer.serialize_map(Some(count.0))?);
97    node.fields(&mut out)?;
98    out.0.end()
99}
100
101/// `name` and `description`, when set.
102fn names<F: Fields>(header: &NodeHeader, out: &mut F) -> Result<(), F::Error> {
103    out.opt("name", header.name.as_ref())?;
104    out.opt("description", header.description.as_ref())
105}
106
107/// `ext`, when it holds anything: the last key a node writes.
108fn ext<F: Fields>(header: &NodeHeader, out: &mut F) -> Result<(), F::Error> {
109    out.field_if(!header.ext.is_empty(), "ext", &header.ext)
110}
111
112impl Node for WorkflowDef {
113    fn fields<F: Fields>(&self, out: &mut F) -> Result<(), F::Error> {
114        out.field("id", &self.id)?;
115        names(&self.header, out)?;
116        out.field_if(
117            self.priority != defaults::PRIORITY,
118            "priority",
119            &self.priority,
120        )?;
121        out.opt("condition", self.condition.as_ref())?;
122        out.field("tasks", &self.tasks)?;
123        out.field_if(
124            self.on_failure != WorkflowFailurePolicy::default(),
125            "on_failure",
126            &self.on_failure,
127        )?;
128        out.field_if(self.channel != defaults::CHANNEL, "channel", &self.channel)?;
129        out.field_if(self.status != Status::default(), "status", &self.status)?;
130        out.field_if(self.version != defaults::VERSION, "version", &self.version)?;
131        out.opt("rollout", self.rollout.as_ref())?;
132        out.opt("loop", self.r#loop.as_ref())?;
133        out.field_if(!self.tags.is_empty(), "tags", &self.tags)?;
134        if let Some(at) = &self.created_at {
135            out.field("created_at", &Rfc3339(at))?;
136        }
137        if let Some(at) = &self.updated_at {
138            out.field("updated_at", &Rfc3339(at))?;
139        }
140        ext(&self.header, out)
141    }
142}
143
144impl Node for TaskDef {
145    fn fields<F: Fields>(&self, out: &mut F) -> Result<(), F::Error> {
146        out.field("id", &self.id)?;
147        names(&self.header, out)?;
148        out.opt("condition", self.condition.as_ref())?;
149        out.field("function", &self.function)?;
150        out.field_if(
151            self.on_failure != FailurePolicy::default(),
152            "on_failure",
153            &self.on_failure,
154        )?;
155        out.field_if(
156            self.on_rejection != RejectionPolicy::default(),
157            "on_rejection",
158            &self.on_rejection,
159        )?;
160        out.field_if(self.terminal, "terminal", &true)?;
161        out.opt("for_each", self.for_each.as_ref())?;
162        ext(&self.header, out)
163    }
164}
165
166impl Node for GroupDef {
167    fn fields<F: Fields>(&self, out: &mut F) -> Result<(), F::Error> {
168        out.field("id", &self.id)?;
169        names(&self.header, out)?;
170        out.opt("condition", self.condition.as_ref())?;
171        out.field_if(self.terminal, "terminal", &true)?;
172        out.field("tasks", &self.tasks)?;
173        ext(&self.header, out)
174    }
175}
176
177impl Node for LoopDef {
178    fn fields<F: Fields>(&self, out: &mut F) -> Result<(), F::Error> {
179        out.opt("counter", self.counter.as_ref())?;
180        out.field_if(self.init != defaults::INIT, "init", &self.init)?;
181        out.field_if(
182            self.increment != defaults::INCREMENT,
183            "increment",
184            &self.increment,
185        )?;
186        out.field("max", &self.max)?;
187        out.field_if(!self.setup.is_empty(), "setup", &self.setup)?;
188        out.opt("over", self.over.as_ref())?;
189        out.opt("as", self.r#as.as_ref())?;
190        out.opt("scratch", self.scratch.as_ref())
191    }
192}
193
194impl Node for ForEachDef {
195    fn fields<F: Fields>(&self, out: &mut F) -> Result<(), F::Error> {
196        out.field("over", &self.over)?;
197        out.field("as", &self.r#as)?;
198        out.field_if(
199            self.max_concurrency != defaults::MAX_CONCURRENCY,
200            "max_concurrency",
201            &self.max_concurrency,
202        )?;
203        out.opt("collect", self.collect.as_ref())?;
204        out.opt("into", self.into.as_ref())
205    }
206}
207
208impl Node for FunctionRef {
209    fn fields<F: Fields>(&self, out: &mut F) -> Result<(), F::Error> {
210        out.field("name", &self.name)?;
211        out.field("input", &self.input)
212    }
213}
214
215impl Node for Rollout {
216    fn fields<F: Fields>(&self, out: &mut F) -> Result<(), F::Error> {
217        out.field("bucket_start", &self.bucket_start)?;
218        out.field("bucket_end", &self.bucket_end)
219    }
220}
221
222impl Node for NodeHeader {
223    fn fields<F: Fields>(&self, out: &mut F) -> Result<(), F::Error> {
224        names(self, out)?;
225        ext(self, out)
226    }
227}
228
229/// `Serialize` through [`Node`], for each type that is written as a map.
230macro_rules! serialize_as_node {
231    ($($ty:ty),* $(,)?) => {$(
232        impl Serialize for $ty {
233            fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
234                node(self, serializer)
235            }
236        }
237    )*};
238}
239
240serialize_as_node!(
241    WorkflowDef,
242    TaskDef,
243    GroupDef,
244    LoopDef,
245    ForEachDef,
246    FunctionRef,
247    Rollout,
248    NodeHeader,
249);
250
251impl Serialize for StepDef {
252    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
253        match self {
254            Self::Task(task) => task.serialize(serializer),
255            Self::Group(group) => group.serialize(serializer),
256        }
257    }
258}
259
260impl Serialize for Logic {
261    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
262        self.0.serialize(serializer)
263    }
264}
265
266/// `Serialize` as the authored spelling, for each policy and status.
267macro_rules! serialize_as_spelling {
268    ($($ty:ty),* $(,)?) => {$(
269        impl Serialize for $ty {
270            fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
271                serializer.serialize_str(self.as_str())
272            }
273        }
274    )*};
275}
276
277serialize_as_spelling!(
278    Status,
279    FailurePolicy,
280    RejectionPolicy,
281    WorkflowFailurePolicy,
282);
dataflow_rs/def/ser.rs

dataflow_rs/def/
ser.rs
