alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::io;
2
3use gix_packetline::{Channel, blocking_io::encode};
4
5pub const MAX_BAND: usize = 65515;
6
7pub enum Frame<'a> {
8 Data(&'a [u8]),
9 Flush,
10 Delim,
11 ResponseEnd,
12}
13
14fn hex4(prefix: &[u8]) -> io::Result<u16> {
15 std::str::from_utf8(prefix)
16 .ok()
17 .and_then(|text| u16::from_str_radix(text, 16).ok())
18 .ok_or_else(|| io::Error::other("invalid pkt-line length prefix"))
19}
20
21pub fn frames(
22 input: &[u8],
23 stop_after_flushes: Option<usize>,
24) -> impl Iterator<Item = io::Result<(Frame<'_>, usize)>> + '_ {
25 let mut pos = 0usize;
26 let mut flushes = 0usize;
27 let mut stopped = false;
28 std::iter::from_fn(move || {
29 (!stopped && pos + 4 <= input.len()).then(|| {
30 let frame = hex4(&input[pos..pos + 4]).and_then(|len| {
31 pos += 4;
32 match len {
33 0 => {
34 flushes += 1;
35 stopped = stop_after_flushes == Some(flushes);
36 Ok((Frame::Flush, pos))
37 }
38 1 => Ok((Frame::Delim, pos)),
39 2 => Ok((Frame::ResponseEnd, pos)),
40 3 => Err(io::Error::other("invalid pkt-line length 3")),
41 n => {
42 let end = pos - 4 + usize::from(n);
43 (end <= input.len())
44 .then(|| {
45 let payload = &input[pos..end];
46 pos = end;
47 (Frame::Data(payload), end)
48 })
49 .ok_or_else(|| io::Error::other("truncated pkt-line"))
50 }
51 }
52 });
53 stopped |= frame.is_err();
54 frame
55 })
56 })
57}
58
59pub fn data_payloads(input: &[u8]) -> io::Result<Vec<&[u8]>> {
60 collect_data(input, Some(1))
61}
62
63pub fn data_payloads_all(input: &[u8]) -> io::Result<Vec<&[u8]>> {
64 collect_data(input, None)
65}
66
67fn collect_data(input: &[u8], stop_after_flushes: Option<usize>) -> io::Result<Vec<&[u8]>> {
68 frames(input, stop_after_flushes)
69 .filter_map(|item| match item {
70 Ok((Frame::Data(payload), _)) => Some(Ok(payload)),
71 Ok(_) => None,
72 Err(err) => Some(Err(err)),
73 })
74 .collect()
75}
76
77#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
78pub struct Caps {
79 pub atomic: bool,
80 pub side_band_64k: bool,
81 pub push_options: bool,
82}
83
84pub(crate) fn first_command(input: &[u8]) -> Option<&[u8]> {
85 frames(input, Some(1)).find_map(|item| match item {
86 Ok((Frame::Data(payload), _)) if !is_preamble(payload) => Some(payload),
87 _ => None,
88 })
89}
90
91fn is_preamble(line: &[u8]) -> bool {
92 line.starts_with(b"shallow ")
93}
94
95pub(crate) fn parse_caps(first_command: &[u8]) -> Caps {
96 let caps = first_command
97 .split(|byte| *byte == 0)
98 .nth(1)
99 .and_then(|caps| std::str::from_utf8(caps).ok())
100 .unwrap_or_default();
101 let has = |needle: &str| caps.split_whitespace().any(|cap| cap == needle);
102 Caps {
103 atomic: has("atomic"),
104 side_band_64k: has("side-band-64k"),
105 push_options: has("push-options"),
106 }
107}
108
109pub struct Receive<'a> {
110 pub commands: Vec<&'a [u8]>,
111 pub options: Vec<&'a [u8]>,
112 pub pack: &'a [u8],
113 pub caps: Caps,
114}
115
116pub fn split_receive(input: &[u8]) -> io::Result<Receive<'_>> {
117 let caps = first_command(input).map(parse_caps).unwrap_or_default();
118 let boundary = if caps.push_options { 2 } else { 1 };
119 frames(input, Some(boundary))
120 .try_fold(
121 (Vec::new(), Vec::new(), 0usize, None),
122 |(mut commands, mut options, flushes, end), item| {
123 item.map(|(frame, at)| match frame {
124 Frame::Data(payload) if flushes == 0 && !is_preamble(payload) => {
125 commands.push(payload);
126 (commands, options, flushes, end)
127 }
128 Frame::Data(payload) if flushes == 1 && caps.push_options => {
129 options.push(payload);
130 (commands, options, flushes, end)
131 }
132 Frame::Data(_) => (commands, options, flushes, end),
133 Frame::Flush => (commands, options, flushes + 1, Some(at)),
134 _ => (commands, options, flushes, end),
135 })
136 },
137 )
138 .map(|(commands, options, _flushes, end)| Receive {
139 commands,
140 options,
141 caps,
142 pack: &input[end.unwrap_or(input.len())..],
143 })
144}
145
146pub fn write_data(buf: &mut Vec<u8>, payload: &[u8]) -> io::Result<()> {
147 encode::data_to_write(payload, buf).map(|_| ())
148}
149
150pub fn write_flush(buf: &mut Vec<u8>) -> io::Result<()> {
151 encode::flush_to_write(buf).map(|_| ())
152}
153
154pub fn write_delim(buf: &mut Vec<u8>) -> io::Result<()> {
155 encode::delim_to_write(buf).map(|_| ())
156}
157
158pub fn write_band(buf: &mut Vec<u8>, chunk: &[u8]) -> io::Result<()> {
159 encode::band_to_write(Channel::Data, chunk, buf).map(|_| ())
160}
161
162pub fn write_band_progress(buf: &mut Vec<u8>, message: &[u8]) -> io::Result<()> {
163 encode::band_to_write(Channel::Progress, message, buf).map(|_| ())
164}
165
166pub fn write_band_error(buf: &mut Vec<u8>, message: &[u8]) -> io::Result<()> {
167 encode::band_to_write(Channel::Error, message, buf).map(|_| ())
168}
169
170pub fn frame_report(report: &[u8], messages: &[String], side_band: bool) -> Vec<u8> {
171 if !side_band {
172 return report.to_vec();
173 }
174 let mut buf = Vec::new();
175 report
176 .chunks(MAX_BAND)
177 .for_each(|chunk| write_band(&mut buf, chunk).expect("band write to vec never fails"));
178 messages.iter().for_each(|message| {
179 format!("{message}\n")
180 .into_bytes()
181 .chunks(MAX_BAND)
182 .for_each(|chunk| {
183 write_band_progress(&mut buf, chunk).expect("band write to vec never fails")
184 });
185 });
186 write_flush(&mut buf).expect("flush write to vec never fails");
187 buf
188}
189
190#[cfg(test)]
191mod tests {
192 use super::*;
193
194 fn command_line(caps: &str) -> Vec<u8> {
195 let mut line = b"\
196 0000000000000000000000000000000000000000 \
197 1111111111111111111111111111111111111111 refs/heads/main"
198 .to_vec();
199 line.push(0);
200 line.extend_from_slice(caps.as_bytes());
201 line.push(b'\n');
202 line
203 }
204
205 #[test]
206 fn a_malformed_length_prefix_terminates_instead_of_spinning() {
207 let garbage = b"zzzz this is not a pkt-line stream at all";
208 assert_eq!(
209 frames(garbage, None).count(),
210 1,
211 "bad length prefix yields one error frame then the stream ends"
212 );
213 assert!(
214 first_command(garbage).is_none(),
215 "no command is parsed out of garbage, and scan does not loop"
216 );
217 assert!(
218 split_receive(garbage).is_err(),
219 "malformed prefix is a parse error, never an infinite loop"
220 );
221 }
222
223 #[test]
224 fn split_receive_skips_the_push_options_section_before_the_pack() {
225 let mut body = Vec::new();
226 write_data(
227 &mut body,
228 &command_line("report-status side-band-64k push-options"),
229 )
230 .unwrap();
231 write_flush(&mut body).unwrap();
232 write_data(&mut body, b"ci-skip").unwrap();
233 write_data(&mut body, b"verbose-ci").unwrap();
234 write_flush(&mut body).unwrap();
235 body.extend_from_slice(b"PACKreal-pack-bytes");
236
237 let parsed = split_receive(&body).unwrap();
238 assert!(parsed.caps.push_options);
239 assert!(parsed.caps.side_band_64k);
240 assert_eq!(parsed.commands.len(), 1);
241 assert_eq!(parsed.options, vec![&b"ci-skip"[..], &b"verbose-ci"[..]]);
242 assert_eq!(parsed.pack, b"PACKreal-pack-bytes");
243 }
244
245 #[test]
246 fn split_receive_reads_caps_past_a_shallow_preamble_line() {
247 let mut body = Vec::new();
248 write_data(
249 &mut body,
250 b"shallow 1111111111111111111111111111111111111111\n",
251 )
252 .unwrap();
253 write_data(&mut body, &command_line("report-status side-band-64k")).unwrap();
254 write_flush(&mut body).unwrap();
255 body.extend_from_slice(b"PACKbytes");
256
257 let parsed = split_receive(&body).unwrap();
258 assert!(
259 parsed.caps.side_band_64k,
260 "capabilities come from the command line, not the shallow preamble"
261 );
262 assert_eq!(
263 parsed.commands.len(),
264 1,
265 "the shallow line is not a command"
266 );
267 assert_eq!(parsed.pack, b"PACKbytes");
268 }
269
270 #[test]
271 fn split_receive_without_push_options_starts_the_pack_after_the_command_flush() {
272 let mut body = Vec::new();
273 write_data(&mut body, &command_line("report-status side-band-64k")).unwrap();
274 write_flush(&mut body).unwrap();
275 body.extend_from_slice(b"PACKbytes");
276
277 let parsed = split_receive(&body).unwrap();
278 assert!(!parsed.caps.push_options);
279 assert!(parsed.options.is_empty());
280 assert_eq!(parsed.pack, b"PACKbytes");
281 }
282
283 #[test]
284 fn frame_report_muxes_the_report_on_band_one_and_messages_on_band_two() {
285 let report = b"unpack ok\n";
286 let messages = vec!["hello there".to_string()];
287 let framed = frame_report(report, &messages, true);
288
289 let bands: Vec<(u8, Vec<u8>)> = frames(&framed, None)
290 .filter_map(|item| match item {
291 Ok((Frame::Data(payload), _)) => Some((payload[0], payload[1..].to_vec())),
292 _ => None,
293 })
294 .collect();
295 assert_eq!(bands[0].0, 1, "report rides band 1");
296 assert_eq!(bands[0].1, report);
297 assert_eq!(bands[1].0, 2, "message rides band 2");
298 assert_eq!(bands[1].1, b"hello there\n");
299 assert!(framed.ends_with(b"0000"), "outer flush closes the stream");
300 }
301
302 #[test]
303 fn frame_report_passes_through_raw_without_side_band() {
304 let report = b"unpack ok\n0000";
305 assert_eq!(
306 frame_report(report, &["dropped".to_string()], false),
307 report
308 );
309 }
310}