From 3955fdd22a1f9caa11213a196c0c61e9ebfcaa1d Mon Sep 17 00:00:00 2001 From: Max Makarov Date: Wed, 3 Jun 2026 22:24:35 +0000 Subject: [PATCH] serial_buffer: add set_out() to retarget the downstream writer SerialBuffer owns its downstream writer privately, with no way to replace it. Buffering the Socket console requires keeping one buffer alive across client connects and disconnects and pointing it at each newly accepted client (or a discarding sink when none is connected) without dropping bytes buffered while no client was attached. Add set_out(), which swaps the writer while leaving the buffered contents intact, plus unit tests covering accumulate-while-detached, replay on connect, live pass-through, delivery of while-detached output to the next client, and that bytes already drained by one client are not resent to the next. Signed-off-by: Max Makarov Assisted-by: Claude:claude-opus-4-8 [Claude Code] --- serial_buffer/src/lib.rs | 126 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 126 insertions(+) diff --git a/serial_buffer/src/lib.rs b/serial_buffer/src/lib.rs index 471120157..b0fda8dbb 100644 --- a/serial_buffer/src/lib.rs +++ b/serial_buffer/src/lib.rs @@ -42,6 +42,11 @@ impl SerialBuffer { self.buffer.extend(buf); } + + /// Replaces the downstream writer, leaving any buffered bytes intact. + pub fn set_out(&mut self, out: Box) { + self.out = out; + } } impl Write for SerialBuffer { @@ -109,3 +114,124 @@ impl Write for SerialBuffer { self.out.flush() } } + +#[cfg(test)] +mod tests { + use std::io::Write; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::{Arc, Mutex}; + + use super::SerialBuffer; + + // A writer that appends into a shared Vec so tests can inspect what the + // buffer wrote through to its downstream sink. + #[derive(Clone)] + struct TestSink(Arc>>); + + impl TestSink { + fn new() -> Self { + TestSink(Arc::new(Mutex::new(Vec::new()))) + } + fn taken(&self) -> Vec { + self.0.lock().unwrap().clone() + } + } + + impl Write for TestSink { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(buf); + Ok(buf.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + // With no client attached (write_out == false) output is retained in the + // ring and replayed in order once a client connects (set_out + write_out). + #[test] + fn accumulates_while_detached_then_replays_on_connect() { + let write_out = Arc::new(AtomicBool::new(false)); + let mut buf = SerialBuffer::new(Box::new(std::io::sink()), write_out.clone()); + + buf.write_all(b"boot: hello\n").unwrap(); + buf.write_all(b"login: ").unwrap(); + + let sink = TestSink::new(); + buf.set_out(Box::new(sink.clone())); + write_out.store(true, Ordering::Release); + buf.flush().unwrap(); + + assert_eq!(sink.taken(), b"boot: hello\nlogin: "); + } + + // Once connected, live writes pass straight through. + #[test] + fn live_writes_pass_through_after_connect() { + let write_out = Arc::new(AtomicBool::new(false)); + let mut buf = SerialBuffer::new(Box::new(std::io::sink()), write_out.clone()); + + let sink = TestSink::new(); + buf.set_out(Box::new(sink.clone())); + write_out.store(true, Ordering::Release); + buf.write_all(b"live").unwrap(); + + assert_eq!(sink.taken(), b"live"); + } + + // Output produced while no client is attached is buffered and delivered to + // the next client to connect; bytes already drained by a previous client + // are not resent. + #[test] + fn output_while_detached_goes_to_next_client() { + let write_out = Arc::new(AtomicBool::new(false)); + let mut buf = SerialBuffer::new(Box::new(std::io::sink()), write_out.clone()); + + // First client: connects, drains "early\n", then disconnects. + let first = TestSink::new(); + buf.set_out(Box::new(first.clone())); + write_out.store(true, Ordering::Release); + buf.write_all(b"early\n").unwrap(); + assert_eq!(first.taken(), b"early\n"); + + // Disconnect: detach to a discarding sink, keep accumulating. + write_out.store(false, Ordering::Release); + buf.set_out(Box::new(std::io::sink())); + buf.write_all(b"while-away\n").unwrap(); + + // Second client: receives what was produced while no one was attached. + let second = TestSink::new(); + buf.set_out(Box::new(second.clone())); + write_out.store(true, Ordering::Release); + buf.flush().unwrap(); + assert_eq!(second.taken(), b"while-away\n"); + } + + // Bytes are delivered once: a second client connecting after the first + // already drained the backlog, with no new output in between, gets nothing. + #[test] + fn drained_bytes_are_not_resent_to_a_second_client() { + let write_out = Arc::new(AtomicBool::new(false)); + let mut buf = SerialBuffer::new(Box::new(std::io::sink()), write_out.clone()); + + buf.write_all(b"boot log\n").unwrap(); + + // First client drains the backlog. + let first = TestSink::new(); + buf.set_out(Box::new(first.clone())); + write_out.store(true, Ordering::Release); + buf.flush().unwrap(); + assert_eq!(first.taken(), b"boot log\n"); + + // First disconnects; no new output is produced while detached. + write_out.store(false, Ordering::Release); + buf.set_out(Box::new(std::io::sink())); + + // Second client connects: nothing left to deliver. + let second = TestSink::new(); + buf.set_out(Box::new(second.clone())); + write_out.store(true, Ordering::Release); + buf.flush().unwrap(); + assert!(second.taken().is_empty()); + } +}