From 3ebd70009d7acb302a49fcd5e9537a1a3cea2f7f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=ABl=20Schulz-Andres?= Date: Sun, 5 Jul 2026 17:44:29 +0200 Subject: [PATCH 1/5] fix: clear stale serial input via ClearBuffer instead of unbounded read loop --- src/main.rs | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/src/main.rs b/src/main.rs index cad5d26..35f3f3e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -13,7 +13,6 @@ 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 std::path::PathBuf; use std::sync::{ Arc, Mutex as StdMutex, @@ -110,16 +109,14 @@ async fn main() -> Result<()> { } // Open port - let mut port = serialport::new(&port_name, baud) + let port = serialport::new(&port_name, baud) .timeout(Duration::from_millis(100)) .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 - } + // Clear any stale data from the serial buffers + port.clear(serialport::ClearBuffer::All) + .context("Failed to clear serial buffers")?; println!("Connected. Type to send; press Ctrl-C to exit.\n"); From be301621cbb82dda0187113c5e90706ca0f3ed07 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=ABl=20Schulz-Andres?= Date: Sun, 5 Jul 2026 20:28:15 +0200 Subject: [PATCH 2/5] refactor: propagate Ctrl-C handler setup errors instead of panicking --- src/main.rs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/main.rs b/src/main.rs index 35f3f3e..9364f0d 100644 --- a/src/main.rs +++ b/src/main.rs @@ -142,7 +142,7 @@ 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 @@ -150,7 +150,9 @@ async fn main() -> Result<()> { let (serial_tx, serial_rx) = mpsc::unbounded_channel::(); // 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( From fce347ddd41545c0ffef54ae232f03a12b3ede61 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=ABl=20Schulz-Andres?= Date: Sun, 5 Jul 2026 20:37:39 +0200 Subject: [PATCH 3/5] feat: add line-aware LogSink with timestamps at line starts --- src/logging.rs | 154 ++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 151 insertions(+), 3 deletions(-) diff --git a/src/logging.rs b/src/logging.rs index 6fe5987..5589d47 100644 --- a/src/logging.rs +++ b/src/logging.rs @@ -1,9 +1,12 @@ use anyhow::{Context, Result}; -use std::fs::OpenOptions; -use std::io::BufWriter; -use std::path::PathBuf; +use chrono::Utc; +use std::fs::{File, OpenOptions}; +use std::io::{self, BufWriter, Write}; +use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; +const HEX_BYTES_PER_LINE: usize = 16; + pub type LogWriter = Arc>>; pub fn create_log_writer(path: &PathBuf, log_type: &str) -> Result { @@ -30,3 +33,148 @@ pub fn create_tx_log_writer(path: Option<&PathBuf>) -> Result> None => Ok(None), } } + +/// 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> { + writer: W, + log_ts: bool, + hex: bool, + at_line_start: bool, + hex_col: usize, +} + +impl LogSink { + pub fn open(path: &Path, log_type: &str, log_ts: bool, hex: bool) -> Result { + 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 LogSink { + 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(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn output(sink: LogSink>) -> 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}"); + } + } +} From d0ab00b97c2ce0d6254546924c785d55257b6d76 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=ABl=20Schulz-Andres?= Date: Sun, 5 Jul 2026 20:40:20 +0200 Subject: [PATCH 4/5] refactor: replace shared port mutex with dedicated reader/writer threads --- src/config.rs | 8 +- src/logging.rs | 30 +----- src/main.rs | 59 ++++++------ src/serial_io.rs | 216 +++++++++++++++++--------------------------- src/ui/app_state.rs | 11 +++ src/ui/mod.rs | 86 +++++++----------- 6 files changed, 161 insertions(+), 249 deletions(-) diff --git a/src/config.rs b/src/config.rs index 1d24402..34efdc4 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,6 +1,7 @@ use clap::ValueEnum; -use std::io::BufWriter; -use std::sync::{Arc, Mutex as StdMutex, atomic::AtomicBool}; +use std::sync::{Arc, atomic::AtomicBool}; + +use crate::serial_io::WriterMsg; /// Which line ending to send when you press Enter #[derive(Copy, Clone, Debug, ValueEnum)] @@ -38,6 +39,5 @@ impl LineEnding { pub struct UiConfig { pub running: Arc, pub line_ending: LineEnding, - pub tx_log: Option>>>, - pub log_ts: bool, + pub writer: std::sync::mpsc::Sender, } diff --git a/src/logging.rs b/src/logging.rs index 5589d47..916bb36 100644 --- a/src/logging.rs +++ b/src/logging.rs @@ -2,38 +2,10 @@ use anyhow::{Context, Result}; use chrono::Utc; use std::fs::{File, OpenOptions}; use std::io::{self, BufWriter, Write}; -use std::path::{Path, PathBuf}; -use std::sync::{Arc, Mutex}; +use std::path::Path; const HEX_BYTES_PER_LINE: usize = 16; -pub type LogWriter = Arc>>; - -pub fn create_log_writer(path: &PathBuf, log_type: &str) -> Result { - 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)))) -} - -pub fn create_rx_log_writer(path: Option<&PathBuf>) -> Result> { - match path { - Some(path) => Ok(Some(create_log_writer(path, "RX")?)), - None => Ok(None), - } -} - -pub fn create_tx_log_writer(path: Option<&PathBuf>) -> Result> { - match path { - Some(path) => Ok(Some(create_log_writer(path, "TX")?)), - None => Ok(None), - } -} - /// 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> { diff --git a/src/main.rs b/src/main.rs index 9364f0d..91351dd 100644 --- a/src/main.rs +++ b/src/main.rs @@ -4,22 +4,21 @@ mod port_discovery; mod serial_io; mod ui; -use anyhow::{Context, Result}; +use anyhow::{Context, Result, bail}; use clap::Parser; use config::{LineEnding, 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 serial_io::{SerialEvent, WriterMsg, spawn_reader, 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 @@ -120,12 +119,17 @@ async fn main() -> Result<()> { println!("Connected. Type to send; press Ctrl-C to exit.\n"); - // Shared port between reader/writer - let port: Arc>> = 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)); @@ -147,25 +151,23 @@ async fn main() -> Result<()> { // Communication channels for UI let (ui_tx, ui_rx) = mpsc::unbounded_channel::(); - let (serial_tx, serial_rx) = mpsc::unbounded_channel::(); + let (event_tx, event_rx) = mpsc::unbounded_channel::(); + let (writer_tx, writer_rx) = std::sync::mpsc::channel::(); // Store UI sender for Ctrl-C handler 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(), - running.clone(), - serial_tx.clone(), - args.hex, - args.log_ts, - rx_log_writer.clone(), - ); - let reader_handle = tokio::spawn(async move { - serial_reader.run().await; - }); + // Reader and writer get independent handles so writes never wait on reads + let writer_port = port + .try_clone() + .context("Failed to clone serial port handle")?; + let writer_handle = spawn_writer(writer_rx, event_tx.clone(), tx_log); + if writer_tx.send(WriterMsg::NewPort(writer_port)).is_err() { + bail!("Writer thread terminated unexpectedly"); + } + let reader_handle = spawn_reader(port, running.clone(), event_tx.clone(), rx_log); // Setup terminal for ratatui terminal::enable_raw_mode().context("Failed to enable raw mode")?; @@ -177,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 _ = reader_handle.join(); + let _ = writer_handle.join(); if let Err(e) = ui_res { eprintln!("\nError: {e:?}"); diff --git a/src/serial_io.rs b/src/serial_io.rs index 5ba3e9a..d630a65 100644 --- a/src/serial_io.rs +++ b/src/serial_io.rs @@ -1,151 +1,99 @@ -use anyhow::Result; -use chrono::Utc; use serialport::SerialPort; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; -use tokio::sync::{Mutex, mpsc}; +use std::thread::JoinHandle; +use tokio::sync::mpsc; +use crate::logging::LogSink; + +const READ_BUF_SIZE: usize = 4096; + +/// Events sent from the serial threads to the UI. #[derive(Debug, Clone)] -pub enum SerialData { - Received(String), +pub enum SerialEvent { + Data(Vec), + Error(String), + Disconnected(String), } -pub struct SerialReader { - port: Arc>>, - running: Arc, - sender: mpsc::UnboundedSender, - hex_mode: bool, - log_ts: bool, - rx_log_writer: Option>>>, - // No cached timestamp needed with chrono - buffer: Vec, // Pre-allocated buffer +/// Messages consumed by the writer thread. +pub enum WriterMsg { + Data(Vec), + NewPort(Box), } -impl SerialReader { - pub fn new( - port: Arc>>, - running: Arc, - sender: mpsc::UnboundedSender, - hex_mode: bool, - log_ts: bool, - rx_log_writer: Option>>>, - ) -> Self { - Self { - port, - running, - sender, - hex_mode, - log_ts, - rx_log_writer, - // No cached timestamp initialization needed - buffer: vec![0u8; 4096], // Pre-allocate buffer to avoid allocations - } - } - - pub async fn run(mut self) { - while self.running.load(Ordering::SeqCst) { - let n = { - let mut guard = self.port.lock().await; - match guard.read(&mut self.buffer) { - Ok(n) => n, - Err(ref e) if e.kind() == std::io::ErrorKind::TimedOut => 0, - Err(_) => break, +/// Reads from the port until shutdown or a fatal error. Returns the RX log +/// sink so a future connection can keep appending to it. +pub fn spawn_reader( + mut port: Box, + running: Arc, + events: mpsc::UnboundedSender, + mut rx_log: Option, +) -> JoinHandle> { + std::thread::spawn(move || { + let mut buf = [0u8; READ_BUF_SIZE]; + while running.load(Ordering::SeqCst) { + match port.read(&mut buf) { + Ok(0) => { + let _ = events.send(SerialEvent::Disconnected("port returned EOF".into())); + break; + } + Ok(n) => { + if let Some(log) = rx_log.as_mut() + && let Err(e) = log.write_chunk(&buf[..n]) + { + let _ = events.send(SerialEvent::Error(format!( + "RX log write failed, logging disabled: {e}" + ))); + rx_log = None; + } + if events.send(SerialEvent::Data(buf[..n].to_vec())).is_err() { + break; + } + } + Err(ref e) if e.kind() == std::io::ErrorKind::TimedOut => {} + Err(e) => { + let _ = events.send(SerialEvent::Disconnected(e.to_string())); + break; } - }; - - if n > 0 { - // Make a copy to avoid borrow checker issues - let bytes = self.buffer[..n].to_vec(); - self.process_received_data(&bytes).await; - } else { - // Small async yield to prevent busy waiting - tokio::task::yield_now().await; - } - } - } - - async fn process_received_data(&mut self, bytes: &[u8]) { - // Format the data - optimized to avoid multiple allocations - let display_text = if self.hex_mode { - self.format_hex_data(bytes) - } else { - self.format_text_data(bytes) - }; - - // Send to UI - let _ = self.sender.send(SerialData::Received(display_text)); - - // Write to RX log file if configured - self.write_to_log(bytes).await; - } - - fn format_hex_data(&mut self, bytes: &[u8]) -> String { - let capacity = if self.log_ts { 32 } else { 0 } + bytes.len() * 3; // Estimate capacity - let mut hex_str = String::with_capacity(capacity); - - if self.log_ts { - hex_str.push('['); - hex_str.push_str(&Utc::now().format("%Y-%m-%d %H:%M:%S%.3f").to_string()); - hex_str.push_str("] "); - } - - // Optimize hex formatting with pre-allocated string - for (i, b) in bytes.iter().enumerate() { - if i > 0 { - hex_str.push(' '); } - hex_str.push_str(&format!("{:02X}", b)); - } - - hex_str - } - - fn format_text_data(&mut self, bytes: &[u8]) -> String { - let capacity = if self.log_ts { 32 } else { 0 } + bytes.len(); - let mut text = String::with_capacity(capacity); - - if self.log_ts { - text.push('['); - text.push_str(&Utc::now().format("%Y-%m-%d %H:%M:%S%.3f").to_string()); - text.push_str("] "); } + rx_log + }) +} - // Use from_utf8_lossy but avoid extra allocations where possible - text.push_str(&String::from_utf8_lossy(bytes)); - text - } - - async fn write_to_log(&mut self, bytes: &[u8]) { - if let Some(w) = &self.rx_log_writer - && let Ok(mut lw) = w.lock() - { - use std::io::Write; - - if self.log_ts { - let _ = write!(lw, "[{}] ", Utc::now().format("%Y-%m-%d %H:%M:%S%.3f")); - } - - if self.hex_mode { - for (i, b) in bytes.iter().enumerate() { - let separator = if i + 1 == bytes.len() { "" } else { " " }; - let _ = write!(lw, "{:02X}{}", b, separator); +/// Owns the write half of the port. Write errors are reported to the UI +/// instead of terminating the application. +pub fn spawn_writer( + messages: std::sync::mpsc::Receiver, + events: mpsc::UnboundedSender, + mut tx_log: Option, +) -> JoinHandle<()> { + std::thread::spawn(move || { + let mut port: Option> = None; + while let Ok(msg) = messages.recv() { + match msg { + WriterMsg::NewPort(p) => port = Some(p), + WriterMsg::Data(bytes) => { + let Some(p) = port.as_mut() else { + let _ = + events.send(SerialEvent::Error("not connected, input dropped".into())); + continue; + }; + if let Err(e) = p.write_all(&bytes).and_then(|_| p.flush()) { + let _ = events.send(SerialEvent::Error(format!("write failed: {e}"))); + continue; + } + if let Some(log) = tx_log.as_mut() + && let Err(e) = log.write_chunk(&bytes) + { + let _ = events.send(SerialEvent::Error(format!( + "TX log write failed, logging disabled: {e}" + ))); + tx_log = None; + } } - let _ = writeln!(lw); - } else { - let _ = lw.write_all(bytes); } - - let _ = lw.flush(); } - } -} - -pub async fn write_bytes_async( - port: &Arc>>, - bytes: &[u8], -) -> Result<()> { - let mut guard = port.lock().await; - guard.write_all(bytes)?; - guard.flush()?; - Ok(()) + }) } diff --git a/src/ui/app_state.rs b/src/ui/app_state.rs index eac7325..cef7da6 100644 --- a/src/ui/app_state.rs +++ b/src/ui/app_state.rs @@ -60,6 +60,17 @@ impl AppState { } } + /// Push a complete status line (bypasses partial-line assembly). + pub fn add_notice(&mut self, message: String) { + self.output_lines.push(message); + if self.output_lines.len() > 1000 { + self.output_lines.drain(..self.output_lines.len() - 1000); + } + self.auto_scroll_state + .select(Some(self.output_lines.len() - 1)); + self.needs_render = true; + } + pub fn scroll_up(&mut self) { if self.output_lines.is_empty() { return; diff --git a/src/ui/mod.rs b/src/ui/mod.rs index cd81a82..1be2e72 100644 --- a/src/ui/mod.rs +++ b/src/ui/mod.rs @@ -7,14 +7,12 @@ pub use rendering::draw_ui; use anyhow::Result; use crossterm::event::{self, Event, KeyCode, KeyEventKind, KeyModifiers}; use ratatui::{Terminal, backend::Backend}; -use std::sync::Arc; use std::sync::atomic::Ordering; use std::time::Duration; use tokio::sync::mpsc; use crate::config::UiConfig; -use crate::serial_io::{SerialData, write_bytes_async}; -use chrono::Utc; +use crate::serial_io::{SerialEvent, WriterMsg}; #[derive(Debug)] pub enum UiMessage { @@ -24,12 +22,10 @@ pub enum UiMessage { pub async fn run_ui( terminal: &mut Terminal, mut ui_rx: mpsc::UnboundedReceiver, - mut serial_rx: mpsc::UnboundedReceiver, - port: Arc>>, + mut serial_rx: mpsc::UnboundedReceiver, ui_config: UiConfig, ) -> Result<()> { let mut app_state = AppState::new(); - // No cached timestamp needed with chrono while ui_config.running.load(Ordering::SeqCst) && !app_state.should_quit { tokio::select! { @@ -45,14 +41,10 @@ pub async fn run_ui( } } - // Serial data - data = serial_rx.recv() => { - if let Some(data) = data { - match data { - SerialData::Received(line) => { - app_state.add_output(line); - } - } + // Serial events + event = serial_rx.recv() => { + if let Some(event) = event { + handle_serial_event(event, &mut app_state); } } @@ -67,7 +59,7 @@ pub async fn run_ui( } => { if let Ok(Event::Key(k)) = key_result && k.kind == KeyEventKind::Press { - handle_key_event(k, &mut app_state, &port, &ui_config).await?; + handle_key_event(k, &mut app_state, &ui_config); } } } @@ -83,12 +75,25 @@ pub async fn run_ui( Ok(()) } -async fn handle_key_event( +fn handle_serial_event(event: SerialEvent, app_state: &mut AppState) { + match event { + SerialEvent::Data(bytes) => { + app_state.add_output(String::from_utf8_lossy(&bytes).into_owned()); + } + SerialEvent::Error(message) => { + app_state.add_notice(format!("[sermonizer] {message}")); + } + SerialEvent::Disconnected(reason) => { + app_state.add_notice(format!("[sermonizer] device disconnected: {reason}")); + } + } +} + +fn handle_key_event( key: crossterm::event::KeyEvent, app_state: &mut AppState, - port: &Arc>>, ui_config: &UiConfig, -) -> Result<()> { +) { match key.code { KeyCode::Char(c) if key.modifiers.contains(KeyModifiers::CONTROL) && (c == 'c' || c == 'd') => @@ -106,7 +111,7 @@ async fn handle_key_event( app_state.update_input(c); } KeyCode::Enter => { - handle_enter_key(app_state, port, ui_config).await?; + handle_enter_key(app_state, ui_config); } KeyCode::Backspace => { app_state.backspace_input(); @@ -131,46 +136,19 @@ async fn handle_key_event( } _ => {} } - Ok(()) } -async fn handle_enter_key( - app_state: &mut AppState, - port: &Arc>>, - ui_config: &UiConfig, -) -> Result<()> { +fn handle_enter_key(app_state: &mut AppState, ui_config: &UiConfig) { let input = app_state.clear_input(); - // Send the complete line to serial port - if !input.is_empty() { - write_bytes_async(port, input.as_bytes()).await?; - if let Some(w) = &ui_config.tx_log - && let Ok(mut lw) = w.lock() - { - use std::io::Write; - if ui_config.log_ts { - let _ = write!(lw, "[{}] ", Utc::now().format("%Y-%m-%d %H:%M:%S%.3f")); - } - let _ = lw.write_all(input.as_bytes()); - let _ = lw.flush(); - } + // Send input and line ending as a single write + let mut bytes = input.into_bytes(); + bytes.extend_from_slice(ui_config.line_ending.bytes()); + if bytes.is_empty() { + return; } - // Send line ending - let end = ui_config.line_ending.bytes(); - if !end.is_empty() { - write_bytes_async(port, end).await?; - if let Some(w) = &ui_config.tx_log - && let Ok(mut lw) = w.lock() - { - use std::io::Write; - if ui_config.log_ts && input.is_empty() { - let _ = write!(lw, "[{}] ", Utc::now().format("%Y-%m-%d %H:%M:%S%.3f")); - } - let _ = lw.write_all(end); - let _ = lw.flush(); - } + if ui_config.writer.send(WriterMsg::Data(bytes)).is_err() { + app_state.add_notice("[sermonizer] writer stopped, input dropped".to_string()); } - - Ok(()) } From 148d825586f661d5ba528afc31fae7403a1b3e0a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=ABl=20Schulz-Andres?= Date: Sun, 5 Jul 2026 20:41:36 +0200 Subject: [PATCH 5/5] feat: reconnect automatically when device disappears --- src/config.rs | 21 +++++++++++++++++ src/main.rs | 38 +++++++++++++++--------------- src/serial_io.rs | 60 +++++++++++++++++++++++++++++++++++++++++++++++- src/ui/mod.rs | 7 +++++- 4 files changed, 105 insertions(+), 21 deletions(-) diff --git a/src/config.rs b/src/config.rs index 34efdc4..101e12c 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,8 +1,29 @@ use clap::ValueEnum; +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> { + 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)] pub enum LineEnding { diff --git a/src/main.rs b/src/main.rs index 91351dd..3e0cb1f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -4,20 +4,19 @@ mod port_discovery; mod serial_io; mod ui; -use anyhow::{Context, Result, bail}; +use anyhow::{Context, Result}; use clap::Parser; -use config::{LineEnding, UiConfig}; +use config::{LineEnding, PortSettings, UiConfig}; use crossterm::terminal; use logging::LogSink; use port_discovery::{choose_port_interactive, get_available_ports, print_ports}; use ratatui::{Terminal, backend::CrosstermBackend}; -use serial_io::{SerialEvent, WriterMsg, spawn_reader, spawn_writer}; +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::mpsc; use ui::{UiMessage, run_ui}; @@ -108,15 +107,14 @@ async fn main() -> Result<()> { } // Open port - let 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 buffers - port.clear(serialport::ClearBuffer::All) - .context("Failed to clear serial buffers")?; - println!("Connected. Type to send; press Ctrl-C to exit.\n"); // Optional log files @@ -159,15 +157,17 @@ async fn main() -> Result<()> { *tx_guard = Some(ui_tx.clone()); } - // Reader and writer get independent handles so writes never wait on reads - let writer_port = port - .try_clone() - .context("Failed to clone serial port handle")?; + // 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); - if writer_tx.send(WriterMsg::NewPort(writer_port)).is_err() { - bail!("Writer thread terminated unexpectedly"); - } - let reader_handle = spawn_reader(port, running.clone(), event_tx.clone(), rx_log); + let supervisor_handle = spawn_supervisor( + port, + settings, + running.clone(), + event_tx.clone(), + writer_tx.clone(), + rx_log, + ); // Setup terminal for ratatui terminal::enable_raw_mode().context("Failed to enable raw mode")?; @@ -192,7 +192,7 @@ async fn main() -> Result<()> { // Ensure we stop and join the serial threads running.store(false, Ordering::SeqCst); drop(writer_tx); - let _ = reader_handle.join(); + let _ = supervisor_handle.join(); let _ = writer_handle.join(); if let Err(e) = ui_res { diff --git a/src/serial_io.rs b/src/serial_io.rs index d630a65..0e2dff0 100644 --- a/src/serial_io.rs +++ b/src/serial_io.rs @@ -2,11 +2,15 @@ use serialport::SerialPort; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::thread::JoinHandle; +use std::time::Duration; use tokio::sync::mpsc; +use crate::config::PortSettings; use crate::logging::LogSink; const READ_BUF_SIZE: usize = 4096; +const RECONNECT_POLL: Duration = Duration::from_millis(100); +const RECONNECT_RETRY_TICKS: u32 = 5; /// Events sent from the serial threads to the UI. #[derive(Debug, Clone)] @@ -14,6 +18,7 @@ pub enum SerialEvent { Data(Vec), Error(String), Disconnected(String), + Reconnected, } /// Messages consumed by the writer thread. @@ -22,9 +27,62 @@ pub enum WriterMsg { NewPort(Box), } +/// Supervises the reader thread: when the device disappears it keeps trying +/// to reopen the port and resumes reading once it comes back. +pub fn spawn_supervisor( + first_port: Box, + settings: PortSettings, + running: Arc, + events: mpsc::UnboundedSender, + writer: std::sync::mpsc::Sender, + mut rx_log: Option, +) -> JoinHandle<()> { + std::thread::spawn(move || { + let mut port = Some(first_port); + while running.load(Ordering::SeqCst) { + let Some(p) = port.take() else { break }; + + // Hand the writer its own handle to the fresh connection + match p.try_clone() { + Ok(w) => { + if writer.send(WriterMsg::NewPort(w)).is_err() { + break; + } + } + Err(e) => { + let _ = events.send(SerialEvent::Error(format!( + "cannot clone port handle, sending disabled: {e}" + ))); + } + } + + let reader = spawn_reader(p, running.clone(), events.clone(), rx_log.take()); + rx_log = reader.join().unwrap_or(None); + if !running.load(Ordering::SeqCst) { + break; + } + + // Reader exited while we are still running: the device is gone. + // Poll until the port can be reopened. + let mut ticks = 0u32; + while running.load(Ordering::SeqCst) && port.is_none() { + std::thread::sleep(RECONNECT_POLL); + ticks += 1; + if !ticks.is_multiple_of(RECONNECT_RETRY_TICKS) { + continue; + } + if let Ok(p) = settings.open() { + let _ = events.send(SerialEvent::Reconnected); + port = Some(p); + } + } + } + }) +} + /// Reads from the port until shutdown or a fatal error. Returns the RX log /// sink so a future connection can keep appending to it. -pub fn spawn_reader( +fn spawn_reader( mut port: Box, running: Arc, events: mpsc::UnboundedSender, diff --git a/src/ui/mod.rs b/src/ui/mod.rs index 1be2e72..ae2f5a0 100644 --- a/src/ui/mod.rs +++ b/src/ui/mod.rs @@ -84,7 +84,12 @@ fn handle_serial_event(event: SerialEvent, app_state: &mut AppState) { app_state.add_notice(format!("[sermonizer] {message}")); } SerialEvent::Disconnected(reason) => { - app_state.add_notice(format!("[sermonizer] device disconnected: {reason}")); + app_state.add_notice(format!( + "[sermonizer] device disconnected: {reason} — reconnecting (Ctrl+C to quit)" + )); + } + SerialEvent::Reconnected => { + app_state.add_notice("[sermonizer] device reconnected".to_string()); } } }