omgbase_sync/
pipe.rs

1//! An in-memory pipe: a `Write` half that hands byte chunks to a blocking
2//! `Read` half over a channel. Lets a test or a conformance runner connect an
3//! [`crate::ExternalSource`] to a scripted adapter running on a thread
4//! without spawning a process. Dropping the writer is EOF for the reader.
5
6use std::io::{Read, Write};
7use std::sync::mpsc::{Receiver, Sender, channel};
8
9/// The writing end.
10#[derive(Debug)]
11pub struct PipeWriter {
12    tx: Sender<Vec<u8>>,
13}
14
15/// The reading end (blocks until bytes arrive or the writer is dropped).
16#[derive(Debug)]
17pub struct PipeReader {
18    rx: Receiver<Vec<u8>>,
19    pending: Vec<u8>,
20    at: usize,
21}
22
23/// A connected `(writer, reader)` pair.
24#[must_use]
25pub fn pipe() -> (PipeWriter, PipeReader) {
26    let (tx, rx) = channel();
27    (
28        PipeWriter { tx },
29        PipeReader {
30            rx,
31            pending: Vec::new(),
32            at: 0,
33        },
34    )
35}
36
37impl Write for PipeWriter {
38    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
39        if buf.is_empty() {
40            return Ok(0);
41        }
42        self.tx
43            .send(buf.to_vec())
44            .map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "reader dropped"))?;
45        Ok(buf.len())
46    }
47
48    fn flush(&mut self) -> std::io::Result<()> {
49        Ok(())
50    }
51}
52
53impl Read for PipeReader {
54    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
55        if self.at >= self.pending.len() {
56            match self.rx.recv() {
57                Ok(chunk) => {
58                    self.pending = chunk;
59                    self.at = 0;
60                }
61                Err(_) => return Ok(0),
62            }
63        }
64        let n = buf.len().min(self.pending.len() - self.at);
65        buf[..n].copy_from_slice(&self.pending[self.at..self.at + n]);
66        self.at += n;
67        Ok(n)
68    }
69}
70
71/// Who speaks a transcript line: the adapter (`In`, written to the engine)
72/// or the engine (`Out`, a request the adapter waits for).
73#[derive(Clone, Copy, Debug, PartialEq, Eq)]
74pub enum Dir {
75    In,
76    Out,
77}
78
79/// A scripted adapter on a thread (the reference's `fake-adapter.mjs`): it
80/// writes every leading `In` line, then for each `Out` entry waits for one
81/// request line from the engine (recording it) before playing the `In` lines
82/// that follow. It stays alive until the engine closes its end, then returns
83/// the request lines it received.
84#[derive(Debug)]
85pub struct ScriptedAdapter {
86    handle: std::thread::JoinHandle<Vec<String>>,
87}
88
89impl ScriptedAdapter {
90    /// Spawn over a fresh pipe pair; returns the adapter and the engine's
91    /// ends (`from_adapter`, `to_adapter`) for [`crate::ExternalSource::connect`].
92    #[must_use]
93    pub fn spawn(transcript: Vec<(Dir, String)>) -> (Self, PipeReader, PipeWriter) {
94        let (to_engine, from_adapter) = pipe();
95        let (to_adapter, from_engine) = pipe();
96        let handle = std::thread::spawn(move || {
97            let mut out = to_engine;
98            let mut i = 0;
99            let mut received = Vec::new();
100            let play = |i: &mut usize, out: &mut PipeWriter| {
101                while *i < transcript.len() && transcript[*i].0 == Dir::In {
102                    if writeln!(out, "{}", transcript[*i].1).is_err() {
103                        return;
104                    }
105                    *i += 1;
106                }
107            };
108            play(&mut i, &mut out);
109            use std::io::BufRead;
110            for line in std::io::BufReader::new(from_engine).lines() {
111                let Ok(line) = line else { break };
112                if line.trim().is_empty() {
113                    continue;
114                }
115                received.push(line);
116                if i < transcript.len() && transcript[i].0 == Dir::Out {
117                    i += 1;
118                }
119                play(&mut i, &mut out);
120            }
121            received
122        });
123        (Self { handle }, from_adapter, to_adapter)
124    }
125
126    /// The request lines the adapter received, once the engine has closed.
127    #[must_use]
128    pub fn received(self) -> Vec<String> {
129        self.handle.join().unwrap_or_default()
130    }
131}
132
133#[cfg(test)]
134mod tests {
135    use super::*;
136    use std::io::BufRead;
137
138    #[test]
139    fn lines_cross_and_drop_is_eof() {
140        let (mut w, r) = pipe();
141        let reader = std::thread::spawn(move || {
142            let mut lines = Vec::new();
143            for line in std::io::BufReader::new(r).lines() {
144                lines.push(line.unwrap());
145            }
146            lines
147        });
148        w.write_all(b"one\ntw").unwrap();
149        w.write_all(b"o\n").unwrap();
150        w.flush().unwrap();
151        w.write_all(b"three").unwrap();
152        drop(w);
153        assert_eq!(reader.join().unwrap(), ["one", "two", "three"]);
154        let (mut w, r) = pipe();
155        drop(r);
156        assert!(w.write_all(b"x").is_err());
157    }
158}
omgbase_sync/pipe.rs

omgbase_sync/
pipe.rs
