Fix serial IO

src/config.rs

@@ -1,6 +1,28 @@
 use clap::ValueEnum;
-use std::io::BufWriter;
-use std::sync::{Arc, Mutex as StdMutex, atomic::AtomicBool};
+use serialport::{ClearBuffer, SerialPort};
+use std::sync::{Arc, atomic::AtomicBool};
+use std::time::Duration;
+
+use crate::serial_io::WriterMsg;
+
+/// Everything needed to (re)open the serial port.
+#[derive(Clone)]
+pub struct PortSettings {
+    pub name: String,
+    pub baud: u32,
+}
+
+impl PortSettings {
+    pub fn open(&self) -> serialport::Result<Box<dyn SerialPort>> {
+        let port = serialport::new(&self.name, self.baud)
+            .timeout(Duration::from_millis(100))
+            .open()?;
+
+        // Drop any stale data buffered by the OS
+        port.clear(ClearBuffer::All)?;
+        Ok(port)
+    }
+}
 
 /// Which line ending to send when you press Enter
 #[derive(Copy, Clone, Debug, ValueEnum)]
@@ -38,6 +60,5 @@ impl LineEnding {
 pub struct UiConfig {
     pub running: Arc<AtomicBool>,
     pub line_ending: LineEnding,
-    pub tx_log: Option<Arc<StdMutex<BufWriter<std::fs::File>>>>,
-    pub log_ts: bool,
+    pub writer: std::sync::mpsc::Sender<WriterMsg>,
 }

src/logging.rs

@@ -1,32 +1,152 @@
 use anyhow::{Context, Result};
-use std::fs::OpenOptions;
-use std::io::BufWriter;
-use std::path::PathBuf;
-use std::sync::{Arc, Mutex};
-
-pub type LogWriter = Arc<Mutex<BufWriter<std::fs::File>>>;
-
-pub fn create_log_writer(path: &PathBuf, log_type: &str) -> Result<LogWriter> {
-    let file = OpenOptions::new()
-        .create(true)
-        .append(true)
-        .open(path)
-        .with_context(|| format!("Failed to open {} log file: {}", log_type, path.display()))?;
-
-    println!("Logging {} to: {}", log_type, path.display());
-    Ok(Arc::new(Mutex::new(BufWriter::new(file))))
+use chrono::Utc;
+use std::fs::{File, OpenOptions};
+use std::io::{self, BufWriter, Write};
+use std::path::Path;
+
+const HEX_BYTES_PER_LINE: usize = 16;
+
+/// A log destination that is aware of line boundaries: timestamps are written
+/// at the start of each line instead of at arbitrary read-chunk boundaries.
+pub struct LogSink<W: Write = BufWriter<File>> {
+    writer: W,
+    log_ts: bool,
+    hex: bool,
+    at_line_start: bool,
+    hex_col: usize,
 }
 
-pub fn create_rx_log_writer(path: Option<&PathBuf>) -> Result<Option<LogWriter>> {
-    match path {
-        Some(path) => Ok(Some(create_log_writer(path, "RX")?)),
-        None => Ok(None),
+impl LogSink {
+    pub fn open(path: &Path, log_type: &str, log_ts: bool, hex: bool) -> Result<Self> {
+        let file = OpenOptions::new()
+            .create(true)
+            .append(true)
+            .open(path)
+            .with_context(|| format!("Failed to open {} log file: {}", log_type, path.display()))?;
+
+        println!("Logging {} to: {}", log_type, path.display());
+        Ok(Self::new(BufWriter::new(file), log_ts, hex))
+    }
+}
+
+impl<W: Write> LogSink<W> {
+    pub fn new(writer: W, log_ts: bool, hex: bool) -> Self {
+        Self {
+            writer,
+            log_ts,
+            hex,
+            at_line_start: true,
+            hex_col: 0,
+        }
+    }
+
+    pub fn write_chunk(&mut self, bytes: &[u8]) -> io::Result<()> {
+        if self.hex {
+            self.write_hex(bytes)?;
+        } else {
+            self.write_text(bytes)?;
+        }
+        self.writer.flush()
+    }
+
+    fn write_timestamp(writer: &mut W) -> io::Result<()> {
+        write!(writer, "[{}] ", Utc::now().format("%Y-%m-%d %H:%M:%S%.3f"))
+    }
+
+    fn write_text(&mut self, bytes: &[u8]) -> io::Result<()> {
+        for &b in bytes {
+            if self.at_line_start && self.log_ts {
+                Self::write_timestamp(&mut self.writer)?;
+            }
+            self.at_line_start = b == b'\n';
+            self.writer.write_all(&[b])?;
+        }
+        Ok(())
+    }
+
+    fn write_hex(&mut self, bytes: &[u8]) -> io::Result<()> {
+        for &b in bytes {
+            if self.at_line_start {
+                if self.log_ts {
+                    Self::write_timestamp(&mut self.writer)?;
+                }
+                self.at_line_start = false;
+            } else {
+                self.writer.write_all(b" ")?;
+            }
+            write!(self.writer, "{b:02X}")?;
+            self.hex_col += 1;
+            if self.hex_col == HEX_BYTES_PER_LINE {
+                self.writer.write_all(b"\n")?;
+                self.hex_col = 0;
+                self.at_line_start = true;
+            }
+        }
+        Ok(())
     }
 }
 
-pub fn create_tx_log_writer(path: Option<&PathBuf>) -> Result<Option<LogWriter>> {
-    match path {
-        Some(path) => Ok(Some(create_log_writer(path, "TX")?)),
-        None => Ok(None),
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn output(sink: LogSink<Vec<u8>>) -> String {
+        String::from_utf8(sink.writer).expect("log output is valid UTF-8")
+    }
+
+    #[test]
+    fn text_without_timestamps_passes_bytes_through() {
+        let mut sink = LogSink::new(Vec::new(), false, false);
+        sink.write_chunk(b"hello ").expect("write");
+        sink.write_chunk(b"world\nnext").expect("write");
+        assert_eq!(output(sink), "hello world\nnext");
+    }
+
+    #[test]
+    fn text_timestamps_appear_only_at_line_starts() {
+        let mut sink = LogSink::new(Vec::new(), true, false);
+        // One line split across two chunks must produce exactly one timestamp
+        sink.write_chunk(b"first ").expect("write");
+        sink.write_chunk(b"line\nsecond line\n").expect("write");
+
+        let out = output(sink);
+        let lines: Vec<&str> = out.lines().collect();
+        assert_eq!(lines.len(), 2);
+        for line in &lines {
+            assert!(line.starts_with('['), "line missing timestamp: {line}");
+            assert_eq!(line.matches('[').count(), 1, "extra timestamp: {line}");
+        }
+        assert!(lines[0].ends_with("] first line"));
+        assert!(lines[1].ends_with("] second line"));
+    }
+
+    #[test]
+    fn hex_wraps_after_sixteen_bytes() {
+        let mut sink = LogSink::new(Vec::new(), false, true);
+        sink.write_chunk(&[0xAB; 10]).expect("write");
+        sink.write_chunk(&[0xCD; 10]).expect("write");
+
+        let out = output(sink);
+        let lines: Vec<&str> = out.lines().collect();
+        assert_eq!(
+            lines[0],
+            format!(
+                "{} {}",
+                "AB ".repeat(10).trim_end(),
+                "CD ".repeat(6).trim_end()
+            )
+        );
+        assert_eq!(lines[1], "CD CD CD CD");
+    }
+
+    #[test]
+    fn hex_timestamps_start_each_row() {
+        let mut sink = LogSink::new(Vec::new(), true, true);
+        sink.write_chunk(&[0x01; 20]).expect("write");
+
+        let out = output(sink);
+        for line in out.lines() {
+            assert!(line.starts_with('['), "row missing timestamp: {line}");
+        }
     }
 }

src/main.rs

@@ -6,21 +6,18 @@ mod ui;
 
 use anyhow::{Context, Result};
 use clap::Parser;
-use config::{LineEnding, UiConfig};
+use config::{LineEnding, PortSettings, UiConfig};
 use crossterm::terminal;
-use logging::{create_rx_log_writer, create_tx_log_writer};
+use logging::LogSink;
 use port_discovery::{choose_port_interactive, get_available_ports, print_ports};
 use ratatui::{Terminal, backend::CrosstermBackend};
-use serial_io::{SerialData, SerialReader};
-use serialport::SerialPort;
-use std::io::Read;
+use serial_io::{SerialEvent, WriterMsg, spawn_supervisor, spawn_writer};
 use std::path::PathBuf;
 use std::sync::{
     Arc, Mutex as StdMutex,
     atomic::{AtomicBool, Ordering},
 };
-use std::time::Duration;
-use tokio::sync::{Mutex, mpsc};
+use tokio::sync::mpsc;
 use ui::{UiMessage, run_ui};
 
 /// sermonizer — a tiny, friendly serial monitor
@@ -110,25 +107,27 @@ async fn main() -> Result<()> {
     }
 
     // Open port
-    let mut port = serialport::new(&port_name, baud)
-        .timeout(Duration::from_millis(100))
+    let settings = PortSettings {
+        name: port_name.clone(),
+        baud,
+    };
+    let port = settings
         .open()
         .with_context(|| format!("Failed to open serial port '{port_name}'"))?;
 
-    // Clear any stale data from the serial buffer
-    let mut discard_buf = [0u8; 1024];
-    while port.read(&mut discard_buf).is_ok() {
-        // Keep reading until timeout to flush buffer
-    }
-
     println!("Connected. Type to send; press Ctrl-C to exit.\n");
 
-    // Shared port between reader/writer
-    let port: Arc<Mutex<Box<dyn SerialPort + Send>>> = Arc::new(Mutex::new(port));
-
     // Optional log files
-    let rx_log_writer = create_rx_log_writer(args.log.as_ref())?;
-    let tx_log_writer = create_tx_log_writer(args.tx_log.as_ref())?;
+    let rx_log = args
+        .log
+        .as_deref()
+        .map(|p| LogSink::open(p, "RX", args.log_ts, args.hex))
+        .transpose()?;
+    let tx_log = args
+        .tx_log
+        .as_deref()
+        .map(|p| LogSink::open(p, "TX", args.log_ts, false))
+        .transpose()?;
 
     // Handle Ctrl-C with immediate shutdown
     let running = Arc::new(AtomicBool::new(true));
@@ -145,28 +144,30 @@ async fn main() -> Result<()> {
                 let _ = tx.send(UiMessage::Quit);
             }
         })
-        .expect("Failed to set Ctrl-C handler");
+        .context("Failed to set Ctrl-C handler")?;
     }
 
     // Communication channels for UI
     let (ui_tx, ui_rx) = mpsc::unbounded_channel::<UiMessage>();
-    let (serial_tx, serial_rx) = mpsc::unbounded_channel::<SerialData>();
+    let (event_tx, event_rx) = mpsc::unbounded_channel::<SerialEvent>();
+    let (writer_tx, writer_rx) = std::sync::mpsc::channel::<WriterMsg>();
 
     // Store UI sender for Ctrl-C handler
-    *shutdown_tx.lock().unwrap() = Some(ui_tx.clone());
+    if let Ok(mut tx_guard) = shutdown_tx.lock() {
+        *tx_guard = Some(ui_tx.clone());
+    }
 
-    // Spawn reader thread (RX) - now using the optimized SerialReader
-    let serial_reader = SerialReader::new(
-        port.clone(),
+    // Reader and writer get independent handles so writes never wait on reads;
+    // the supervisor respawns the reader after a disconnect
+    let writer_handle = spawn_writer(writer_rx, event_tx.clone(), tx_log);
+    let supervisor_handle = spawn_supervisor(
+        port,
+        settings,
         running.clone(),
-        serial_tx.clone(),
-        args.hex,
-        args.log_ts,
-        rx_log_writer.clone(),
+        event_tx.clone(),
+        writer_tx.clone(),
+        rx_log,
     );
-    let reader_handle = tokio::spawn(async move {
-        serial_reader.run().await;
-    });
 
     // Setup terminal for ratatui
     terminal::enable_raw_mode().context("Failed to enable raw mode")?;
@@ -178,20 +179,21 @@ async fn main() -> Result<()> {
     let ui_config = UiConfig {
         running: running.clone(),
         line_ending,
-        tx_log: tx_log_writer.clone(),
-        log_ts: args.log_ts,
+        writer: writer_tx.clone(),
     };
 
-    let ui_res = run_ui(&mut terminal, ui_rx, serial_rx, port.clone(), ui_config).await;
+    let ui_res = run_ui(&mut terminal, ui_rx, event_rx, ui_config).await;
 
     // Cleanup terminal
     terminal::disable_raw_mode()?;
     crossterm::execute!(terminal.backend_mut(), terminal::LeaveAlternateScreen)?;
     terminal.show_cursor()?;
 
-    // Ensure we stop and join reader
+    // Ensure we stop and join the serial threads
     running.store(false, Ordering::SeqCst);
-    let _ = reader_handle.await;
+    drop(writer_tx);
+    let _ = supervisor_handle.join();
+    let _ = writer_handle.join();
 
     if let Err(e) = ui_res {
         eprintln!("\nError: {e:?}");