surreal_sync_runtime/pipeline/
framer.rs

1//! Pluggable message framing for external transform I/O.
2
3use anyhow::Result;
4use bytes::{Bytes, BytesMut};
5
6/// Supported wire framers (v1: NDJSON only).
7///
8/// Selected per command stage via `stdio.framer` in transforms TOML and passed
9/// into child-stdio transport constructors so the config is not dead.
10#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
11pub enum FramerKind {
12    /// Newline-delimited JSON (`stdio.framer = "ndjson"`).
13    #[default]
14    Ndjson,
15}
16
17impl FramerKind {
18    /// Resolve to the concrete [`Framer`] implementation for this kind.
19    pub fn into_framer(self) -> NdjsonFramer {
20        match self {
21            FramerKind::Ndjson => NdjsonFramer,
22        }
23    }
24}
25
26/// Frames discrete payloads on a byte stream.
27///
28/// Implementations write into a reused output buffer and parse the next
29/// complete message from a [`BytesMut`] read buffer without allocating a
30/// per-line `String` (payload is returned as [`Bytes`]).
31pub trait Framer: Send + Sync {
32    /// Append one framed message for `payload` into `out` (reused across calls).
33    fn write_message(&self, payload: &[u8], out: &mut Vec<u8>);
34
35    /// Parse the next complete message from `buf`.
36    ///
37    /// Returns `Ok(None)` if the buffer does not yet contain a full message
38    /// (incomplete). On success, advances `buf` past the framed message and
39    /// returns the payload bytes (without framing).
40    fn next_message(&self, buf: &mut BytesMut) -> Result<Option<Bytes>>;
41}
42
43/// Newline-delimited framing: each message is `payload` + `\n`.
44///
45/// Payload is expected to be compact JSON when used with the external
46/// NDJSON wire protocol; this type does not parse JSON itself.
47#[derive(Debug, Default, Clone, Copy)]
48pub struct NdjsonFramer;
49
50impl Framer for NdjsonFramer {
51    fn write_message(&self, payload: &[u8], out: &mut Vec<u8>) {
52        out.extend_from_slice(payload);
53        out.push(b'\n');
54    }
55
56    fn next_message(&self, buf: &mut BytesMut) -> Result<Option<Bytes>> {
57        let Some(pos) = buf.iter().position(|&b| b == b'\n') else {
58            return Ok(None);
59        };
60        if pos > 0 && buf[pos - 1] == b'\r' {
61            // Tolerate CRLF: payload is buf[0..pos-1], consume through `\n`.
62            let mut line = buf.split_to(pos + 1);
63            let payload = line.split_to(pos - 1).freeze();
64            Ok(Some(payload))
65        } else {
66            let mut line = buf.split_to(pos + 1);
67            let payload = line.split_to(pos).freeze();
68            Ok(Some(payload))
69        }
70    }
71}
72
73#[cfg(test)]
74mod tests {
75    use super::*;
76
77    #[test]
78    fn ndjson_roundtrip_single_and_multi() {
79        let framer = NdjsonFramer;
80        let mut out = Vec::new();
81        framer.write_message(br#"{"a":1}"#, &mut out);
82        framer.write_message(br#"{"b":2}"#, &mut out);
83
84        let mut buf = BytesMut::from(out.as_slice());
85        let m1 = framer.next_message(&mut buf).unwrap().expect("msg1");
86        let m2 = framer.next_message(&mut buf).unwrap().expect("msg2");
87        assert_eq!(&m1[..], br#"{"a":1}"#);
88        assert_eq!(&m2[..], br#"{"b":2}"#);
89        assert!(framer.next_message(&mut buf).unwrap().is_none());
90        assert!(buf.is_empty());
91    }
92
93    #[test]
94    fn ndjson_incomplete_buffer() {
95        let framer = NdjsonFramer;
96        let mut buf = BytesMut::from(&b"{\"partial\":tru"[..]);
97        assert!(framer.next_message(&mut buf).unwrap().is_none());
98        assert_eq!(&buf[..], b"{\"partial\":tru");
99
100        buf.extend_from_slice(b"e}\n");
101        let m = framer.next_message(&mut buf).unwrap().expect("complete");
102        assert_eq!(&m[..], br#"{"partial":true}"#);
103        assert!(buf.is_empty());
104    }
105
106    #[test]
107    fn ndjson_empty_payload_line() {
108        let framer = NdjsonFramer;
109        let mut buf = BytesMut::from(&b"\n"[..]);
110        let m = framer.next_message(&mut buf).unwrap().expect("empty line");
111        assert!(m.is_empty());
112    }
113
114    #[test]
115    fn ndjson_crlf() {
116        let framer = NdjsonFramer;
117        let mut buf = BytesMut::from(&b"hello\r\n"[..]);
118        let m = framer.next_message(&mut buf).unwrap().expect("crlf");
119        assert_eq!(&m[..], b"hello");
120    }
121}
surreal_sync_runtime/pipeline/framer.rs

surreal_sync_runtime/pipeline/
framer.rs
