1use std::io::{self, BufRead, Write};
2
3use serde::{Deserialize, Serialize};
4use serde_json::{Map, Value};
5
6use super::util::Clock;
7
8#[derive(Debug, Clone, PartialEq)]
9pub enum Line {
10 Event(Event),
11 End(End),
12 Summary(Summary),
13}
14
15#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
16#[serde(rename_all = "lowercase")]
17pub enum State {
18 Start,
19 Progress,
20 Done,
21 Fail,
22 #[serde(other)]
23 Unknown,
24}
25
26#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
27pub struct Event {
28 #[serde(default)]
29 pub op: String,
30 #[serde(default)]
31 pub app: String,
32 #[serde(rename = "ref", default)]
33 pub reference: Option<String>,
34 pub stage: String,
35 pub state: State,
36 #[serde(default)]
37 pub fraction: Option<f64>,
38 #[serde(default)]
39 pub msg: Option<String>,
40 #[serde(default)]
41 pub at: u64,
42 #[serde(default, skip_serializing_if = "Option::is_none")]
43 pub id: Option<String>,
44}
45
46#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
47pub struct End {
48 #[serde(default)]
49 pub op: String,
50 #[serde(default)]
51 pub app: String,
52 #[serde(rename = "ref", default)]
53 pub reference: Option<String>,
54 pub ok: bool,
55 #[serde(default)]
56 pub msg: String,
57 #[serde(default)]
58 pub at: u64,
59 #[serde(default, skip_serializing_if = "Option::is_none")]
60 pub id: Option<String>,
61 #[serde(flatten)]
62 pub extra: Map<String, Value>,
63}
64
65#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
66pub struct Summary {
67 #[serde(default)]
68 pub op: String,
69 pub ok: bool,
70 #[serde(default)]
71 pub results: Vec<Outcome>,
72 #[serde(default)]
73 pub msg: Option<String>,
74 #[serde(default)]
75 pub at: u64,
76}
77
78#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
79pub struct Outcome {
80 #[serde(default)]
81 pub op: String,
82 #[serde(default)]
83 pub app: String,
84 #[serde(rename = "ref", default)]
85 pub reference: Option<String>,
86 pub ok: bool,
87 #[serde(default)]
88 pub msg: String,
89}
90
91#[derive(Serialize)]
92struct Tagged<'a, T: Serialize> {
93 v: u32,
94 #[serde(skip_serializing_if = "Option::is_none")]
95 state: Option<&'static str>,
96 #[serde(flatten)]
97 body: &'a T,
98}
99
100fn tagged<S: serde::Serializer, T: Serialize>(
101 serializer: S,
102 state: Option<&'static str>,
103 body: &T,
104) -> Result<S::Ok, S::Error> {
105 Tagged { v: 1, state, body }.serialize(serializer)
106}
107
108impl Serialize for Line {
109 fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
110 match self {
111 Line::Event(event) => tagged(serializer, None, event),
112 Line::End(end) => tagged(serializer, Some("end"), end),
113 Line::Summary(summary) => tagged(serializer, Some("summary"), summary),
114 }
115 }
116}
117
118pub(crate) fn clean(fraction: Option<f64>) -> Option<f64> {
119 fraction
120 .filter(|value| value.is_finite())
121 .map(|value| value.clamp(0.0, 1.0))
122}
123
124impl Line {
125 pub fn parse(text: &str) -> Option<Line> {
126 Line::parse_bytes(text.as_bytes())
127 }
128
129 pub fn parse_bytes(bytes: &[u8]) -> Option<Line> {
130 let Ok(Value::Object(mut map)) = serde_json::from_slice::<Value>(bytes) else {
131 return None;
132 };
133 if map.remove("v").and_then(|v| v.as_u64()) != Some(1) {
134 return None;
135 }
136 let state = map.get("state").and_then(Value::as_str)?;
137 let (end, summary) = (state == "end", state == "summary");
138 if end || summary {
139 map.remove("state");
140 }
141 let body = Value::Object(map);
142 if end {
143 serde_json::from_value(body).ok().map(Line::End)
144 } else if summary {
145 serde_json::from_value(body).ok().map(Line::Summary)
146 } else {
147 serde_json::from_value::<Event>(body).ok().map(|mut event| {
148 event.fraction = clean(event.fraction);
149 Line::Event(event)
150 })
151 }
152 }
153
154 pub fn to_json(&self) -> String {
155 serde_json::to_string(self).unwrap_or_default()
156 }
157}
158
159pub fn read(input: impl BufRead) -> impl Iterator<Item = Line> {
160 input
161 .split(b'\n')
162 .map_while(Result::ok)
163 .filter_map(|bytes| Line::parse_bytes(&bytes))
164}
165
166pub struct Writer<W: Write> {
167 out: W,
168 op: String,
169 app: String,
170 reference: Option<String>,
171 id: Option<String>,
172 ended: bool,
173 clock: Clock,
174}
175
176impl<W: Write> Writer<W> {
177 pub fn new(out: W, op: &str, app: &str) -> Writer<W> {
178 Writer::with_clock(out, op, app, Clock::system())
179 }
180
181 pub(crate) fn with_clock(out: W, op: &str, app: &str, clock: Clock) -> Writer<W> {
182 Writer {
183 out,
184 op: op.to_string(),
185 app: app.to_string(),
186 reference: None,
187 id: None,
188 ended: false,
189 clock,
190 }
191 }
192
193 pub fn reference(&mut self, reference: &str) {
194 self.reference = Some(reference.to_string());
195 }
196
197 pub fn id(&mut self, id: &str) {
198 self.id = Some(id.to_string());
199 }
200
201 pub fn event(
202 &mut self,
203 stage: &str,
204 state: State,
205 fraction: Option<f64>,
206 msg: Option<&str>,
207 ) -> io::Result<()> {
208 if self.ended {
209 return Ok(());
210 }
211 let line = Line::Event(Event {
212 op: self.op.clone(),
213 app: self.app.clone(),
214 reference: self.reference.clone(),
215 stage: stage.to_string(),
216 state,
217 fraction: clean(fraction).map(|value| (value * 1000.0).round() / 1000.0),
218 msg: msg.map(str::to_string),
219 at: self.clock.wall_ms(),
220 id: self.id.clone(),
221 });
222 self.line(&line)
223 }
224
225 pub fn end(&mut self, ok: bool, msg: &str) -> io::Result<()> {
226 if self.ended {
227 return Ok(());
228 }
229 let line = Line::End(End {
230 op: self.op.clone(),
231 app: self.app.clone(),
232 reference: self.reference.clone(),
233 ok,
234 msg: msg.to_string(),
235 at: self.clock.wall_ms(),
236 id: self.id.clone(),
237 extra: Map::new(),
238 });
239 self.line(&line)
240 }
241
242 pub fn line(&mut self, line: &Line) -> io::Result<()> {
243 if matches!(line, Line::End(_)) {
244 self.ended = true;
245 }
246 let mut bytes = line.to_json().into_bytes();
247 bytes.push(b'\n');
248 self.out.write_all(&bytes)?;
249 self.out.flush()
250 }
251}