bash_interop/rig/wire/
message.rs

1//! What travels: a shell's account of itself, the messages it sends, and the
2//! answers it is sent.
3//!
4//! Each is a bash array literal. The protocol puts its own words in front — a
5//! verb where there is one, then the sending shell's clock as `at=` — and the
6//! reader shifts exactly those back off, leaving what the shell wrote.
7
8use std::fmt;
9use std::time::{SystemTime, UNIX_EPOCH};
10use std::vec;
11
12use serde::{Deserialize, Serialize};
13
14use crate::failure::{Doing, Failure};
15use bash_strings::{emit_array, parse_array};
16
17#[derive(Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Debug, Serialize, Deserialize)]
18pub struct Micros(pub u64);
19
20impl Micros {
21    /// The run's own clock, to sit beside the sending shell's `$EPOCHREALTIME`.
22    pub(crate) fn now() -> Result<Self, Failure> {
23        SystemTime::now()
24            .duration_since(UNIX_EPOCH)
25            .map(|since| Self(since.as_micros() as u64))
26            .doing(|| "reading the run's clock".into())
27    }
28
29    /// `$EPOCHREALTIME`: seconds, the locale's decimal separator, six digits.
30    fn parse_epoch(text: &str) -> Option<Self> {
31        let (seconds, micros) = text.split_once(['.', ','])?;
32        if micros.len() != 6 {
33            return None;
34        }
35
36        Some(Self(
37            seconds.parse::<u64>().ok()? * 1_000_000 + micros.parse::<u64>().ok()?,
38        ))
39    }
40}
41
42#[derive(Copy, Clone, PartialEq, Eq, Hash, Debug, Serialize, Deserialize)]
43pub struct Pid(pub u32);
44
45impl fmt::Display for Pid {
46    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
47        write!(f, "{}", self.0)
48    }
49}
50
51/// Whether the shell is waiting for something back. A word outside this set is
52/// a defect in the bash, never a client's choice: a client's own tag is a
53/// payload word, and the protocol never reads one.
54#[derive(Copy, Clone, PartialEq, Eq, Debug, Serialize)]
55#[serde(rename_all = "UPPERCASE")]
56pub enum Verb {
57    Say,
58    Ask,
59}
60
61/// When one line was written and when it was read.
62#[derive(Copy, Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
63pub struct Stamp {
64    /// The sending shell's `$EPOCHREALTIME`.
65    pub sent_at: Micros,
66
67    /// The run's clock at the read that completed the line.
68    pub heard_at: Micros,
69}
70
71/// What one shell's client said, once.
72#[derive(Debug, Clone, Serialize)]
73pub struct Message {
74    pub verb: Verb,
75    pub stamp: Stamp,
76
77    /// The client's arglist, and nothing of the protocol's.
78    pub words: Vec<String>,
79}
80
81impl Message {
82    /// The words after `lead`, if this message begins with it — how a decoder
83    /// claims one family of messages and declines the rest.
84    pub fn behind(&self, lead: &str) -> Option<&[String]> {
85        match self.words.split_first() {
86            Some((first, rest)) if first == lead => Some(rest),
87            _ => None,
88        }
89    }
90
91    /// One line off a shell's pipe: the verb, the clock, the words.
92    pub(crate) fn read(line: Line) -> Result<Self, Failure> {
93        let refused = |why: &str| {
94            Failure::new(
95                format!("reading the line {:?}", line.text),
96                why,
97            )
98        };
99        let mut ahead = Ahead::over(&line.text).map_err(|why| refused(&why))?;
100
101        let verb = match ahead.word().map_err(refused)?.as_str() {
102            "SAY" => Verb::Say,
103            "ASK" => Verb::Ask,
104            other => {
105                return Err(refused(&format!(
106                    "{other} is not a verb"
107                )));
108            }
109        };
110        let sent_at = ahead.clock().map_err(refused)?;
111
112        Ok(Self {
113            verb,
114            stamp: Stamp {
115                sent_at,
116                heard_at: line.heard_at,
117            },
118            words: ahead.rest(),
119        })
120    }
121}
122
123/// A shell's account of itself, as it announces itself: the clock, then the
124/// pairs [`Shell::of`](crate::shell::Shell::of) reads.
125#[derive(Debug)]
126pub(crate) struct Account {
127    pub stamp: Stamp,
128    pub words: Vec<String>,
129}
130
131impl Account {
132    pub(crate) fn read(text: &str, heard_at: Micros) -> Result<Self, Failure> {
133        let refused = |why: &str| {
134            Failure::new(
135                format!("reading the account {text:?}"),
136                why,
137            )
138        };
139        let mut ahead = Ahead::over(text).map_err(|why| refused(&why))?;
140        let sent_at = ahead.clock().map_err(refused)?;
141
142        Ok(Self {
143            stamp: Stamp { sent_at, heard_at },
144            words: ahead.rest(),
145        })
146    }
147}
148
149/// One line as read off a shell's pipe, with the run's clock at the read.
150#[derive(Debug)]
151pub(crate) struct Line {
152    pub text: String,
153    pub heard_at: Micros,
154}
155
156/// The protocol's own words, taken off the front one at a time.
157struct Ahead(vec::IntoIter<String>);
158
159impl Ahead {
160    fn over(text: &str) -> Result<Self, String> {
161        parse_array(text)
162            .map(|words| Self(words.into_iter()))
163            .map_err(|why| why.to_string())
164    }
165
166    fn word(&mut self) -> Result<String, &'static str> {
167        self.0.next().ok_or("the line ended early")
168    }
169
170    /// The `at=` header: the sender's `$EPOCHREALTIME`.
171    fn clock(&mut self) -> Result<Micros, &'static str> {
172        match self.word()?.split_once('=') {
173            Some(("at", value)) => Micros::parse_epoch(value).ok_or("bad at"),
174            _ => Err("no at= header"),
175        }
176    }
177
178    fn rest(self) -> Vec<String> {
179        self.0.collect()
180    }
181}
182
183/// Value of the first `key value` pair with this key — a convention clients may
184/// write their payload in, unrelated to the `key=value` headers the protocol
185/// puts in front of one.
186pub fn field<'a>(words: &'a [String], key: &str) -> Option<&'a str> {
187    words
188        .chunks_exact(2)
189        .find(|pair| pair[0] == key)
190        .map(|pair| pair[1].as_str())
191}
192
193/// What a blocked shell is told to run next: one command, as an arglist — the
194/// same shape a message has, encoded the same way.
195#[derive(Debug)]
196pub struct Answer(Vec<String>);
197
198impl Answer {
199    /// A command and the arguments it is given. The command word stands apart
200    /// because a command of no words is not one.
201    pub fn of(command: impl Into<String>, args: impl IntoIterator<Item = impl Into<String>>) -> Self {
202        let mut words = vec![command.into()];
203        words.extend(args.into_iter().map(Into::into));
204
205        Self(words)
206    }
207
208    /// The command `return code`.
209    pub fn status(code: u8) -> Self {
210        Self::of("return", [code.to_string()])
211    }
212
213    /// A word this rig has no answer for: `return 127`, bash's own "command
214    /// not found".
215    pub fn unknown() -> Self {
216        Self::status(127)
217    }
218}
219
220/// One line, whatever it carries: the bash array literal a shell reads back
221/// with `declare -a` — how an answer travels the reply pipe.
222impl fmt::Display for Answer {
223    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
224        write!(f, "{}", emit_array(&self.0))
225    }
226}
227
228#[cfg(test)]
229mod tests {
230    use super::*;
231
232    fn words(items: &[&str]) -> Vec<String> {
233        items.iter().map(|item| item.to_string()).collect()
234    }
235
236    fn line(kind: &str, payload: &[&str]) -> Line {
237        let mut all = words(&[kind, "at=1.000002"]);
238        all.extend(words(payload));
239
240        Line {
241            text: emit_array(&all),
242            heard_at: Micros(50),
243        }
244    }
245
246    fn spoke(payload: &[&str]) -> Message {
247        Message::read(line("SAY", payload)).expect("a message")
248    }
249
250    #[test]
251    fn the_protocols_words_come_off_and_the_clients_remain() {
252        let message = spoke(&["REC", "a space", ""]);
253
254        assert_eq!(message.verb, Verb::Say);
255        assert_eq!(message.stamp.sent_at, Micros(1_000_002));
256        assert_eq!(
257            message.stamp.heard_at,
258            Micros(50),
259            "the run's clock, from the read"
260        );
261        assert_eq!(
262            message.words,
263            words(&["REC", "a space", ""]),
264            "the payload alone"
265        );
266        assert_eq!(
267            message.behind("REC"),
268            Some(words(&["a space", ""]).as_slice())
269        );
270    }
271
272    #[test]
273    fn a_message_may_carry_nothing_of_its_own() {
274        let message = spoke(&[]);
275
276        assert!(message.words.is_empty());
277        assert_eq!(message.behind("REC"), None);
278    }
279
280    /// An account has no verb: the clock comes first.
281    #[test]
282    fn an_account_is_the_clock_and_the_pairs() {
283        let text = emit_array(&words(&[
284            "at=1.000002",
285            "zero",
286            "x.bash",
287            "flags",
288            "hB",
289        ]));
290        let account = Account::read(&text, Micros(50)).unwrap();
291
292        assert_eq!(
293            account.stamp,
294            Stamp {
295                sent_at: Micros(1_000_002),
296                heard_at: Micros(50)
297            }
298        );
299        assert_eq!(
300            field(&account.words, "zero"),
301            Some("x.bash")
302        );
303
304        let verbed = emit_array(&words(&["JOIN", "at=1.000002"]));
305        assert!(
306            Account::read(&verbed, Micros(0)).is_err(),
307            "a verb where the clock goes"
308        );
309    }
310
311    #[test]
312    fn a_header_the_protocol_did_not_write_is_an_error() {
313        let bad = [
314            emit_array(&words(&["MUMBLE", "at=1.000002"])),
315            emit_array(&words(&["JOIN", "at=1.000002"])),
316            emit_array(&words(&["SAY", "when=1.000002"])),
317            emit_array(&words(&["SAY", "at=1.0"])),
318            emit_array(&words(&["SAY"])),
319            "(unquoted".to_string(),
320        ];
321        for text in bad {
322            let refused = Message::read(Line {
323                text: text.clone(),
324                heard_at: Micros(0),
325            });
326            assert!(
327                refused.is_err(),
328                "{text} should not read"
329            );
330        }
331    }
332
333    /// The shell reads an answer with `read -r`, which stops at a newline, so a
334    /// word carrying one arrives escaped and the delimiter is the only newline
335    /// on that pipe.
336    #[test]
337    fn an_answer_is_one_line_whatever_it_carries() {
338        let carried = ["%s", "two\nlines", "a\ttab", "\u{ff}", "it's", ""];
339        let message = Answer::of("printf", carried).to_string();
340
341        assert!(
342            !message.contains('\n'),
343            "a raw newline would truncate the read: {message}"
344        );
345
346        let mut expected = vec!["printf".to_string()];
347        expected.extend(carried.iter().map(|word| word.to_string()));
348        assert_eq!(
349            parse_array(&message).unwrap(),
350            expected,
351            "and it still reads back"
352        );
353    }
354
355    /// A word may itself be a message, decoded one level at a time.
356    #[test]
357    fn messages_round_trip() {
358        let nested = emit_array(&words(&["INNER", "x y"]));
359        let message = spoke(&["TAG", "quote'inside", "two\nlines", &nested]);
360
361        assert_eq!(
362            message.words,
363            words(&["TAG", "quote'inside", "two\nlines", &nested])
364        );
365        assert_eq!(
366            parse_array(&nested).unwrap(),
367            words(&["INNER", "x y"])
368        );
369    }
370}
bash_interop/rig/wire/message.rs

bash_interop/rig/wire/
message.rs
