From 8594c661e544d4528a6d976bb21ec2523e0ba992 Mon Sep 17 00:00:00 2001 From: gyanano <1624055384@qq.com> Date: Wed, 2 Sep 2026 05:43:41 -0700 Subject: [PATCH] feat(serial): event push replaces 100ms log polling (RFC #3 Step 4) Backend: - New bus.rs: Frame {session, seq, dir, t_mono_ns, t_wall, Arc data} with per-session seq from 1 shared by TX/RX; FrameBus with bounded per-subscriber queues (drop-oldest + dropped_before prefix-gap accounting) and Nagle batching (GUI_DEFAULT 16ms/64KiB/512 frames/4096 queue; lone frame for an idle consumer pushes immediately past a 4ms anti-storm interval). - Read loop and send_data publish frames after the buffer write (dual-write coexistence: log buffer/text+raw recording/stats unchanged, all golden tests intact). - New get_logs_snapshot command {epoch, session, entries}; clear_logs returns a bumped epoch. Bridge (main.rs): app-lifetime pump thread consumes batches with bounded blocking recv, decorates (display_text/timestamp) and emits serial://frames; the wire DTO carries no raw bytes (seq/dir/len + display text). Frontend: - useSerialLogs hook: snapshot alignment, incremental append with rAF throttling, seq dedupe, session-change resync, epoch guard so a snapshot racing a clear never resurrects entries, dropped_before placeholder rows, trim mirroring the backend 1000-entry cap; LogViewer keys rows by session-seq. - Rollback switch: localStorage serialEventPush=0 restores the legacy 100ms polling loop. Tests: 40/40 green (5 new bus unit tests incl. drop-oldest prefix-gap accounting; harness test asserting contiguous seq across RX/TX through the real reader thread). Hardware-verified on FTDI loopback: burst sending, clear-during-stream, search/autoscroll/counters, unplug. --- frontend/src/App.tsx | 24 +- frontend/src/components/LogViewer.tsx | 2 +- frontend/src/hooks/useSerialLogs.ts | 172 +++++++++++ frontend/src/i18n/translations/en.json | 3 +- frontend/src/i18n/translations/zh-CN.json | 3 +- frontend/src/types.ts | 6 + src-tauri/src/bus.rs | 345 ++++++++++++++++++++++ src-tauri/src/main.rs | 93 +++++- src-tauri/src/serial_manager.rs | 101 ++++++- src-tauri/src/types.rs | 17 ++ 10 files changed, 748 insertions(+), 18 deletions(-) create mode 100644 frontend/src/hooks/useSerialLogs.ts create mode 100644 src-tauri/src/bus.rs diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 164f5d2..4424652 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -10,6 +10,7 @@ import SettingsModal from './components/SettingsModal'; import { SerialPortInfo, SerialConfig, LogEntry, ConnectionStatus, DataFormat, ChecksumConfig, QuickCommandList, QuickCommand, LineEnding, TextEncoding, FrameSegmentationConfig } from './types'; import { useTheme } from './contexts/ThemeContext'; import { useTranslation } from './i18n'; +import { useSerialLogs } from './hooks/useSerialLogs'; import { appendChecksum } from './utils/checksum'; import { loadTimezone, formatDateForFilename, getSystemTimezoneOffset, parseUtcOffset } from './utils/timezone'; import { Toaster } from './components/ui/sonner'; @@ -135,7 +136,12 @@ function App() { bytes_received: 0, connection_time: null, }); - const [logs, setLogs] = useState([]); + const [polledLogs, setPolledLogs] = useState([]); + // RFC #3 Step 4: event push is the default log path; set + // localStorage 'serialEventPush' = '0' to roll back to 100ms polling. + const [eventPush] = useState(() => localStorage.getItem('serialEventPush') !== '0'); + const { logs: pushedLogs, clearLogs: clearPushedLogs } = useSerialLogs(eventPush); + const logs = eventPush ? pushedLogs : polledLogs; const [sendText, setSendText] = useState(''); const [sendFormat, setSendFormat] = useState('Text'); const [checksumConfig, setChecksumConfig] = useState({ @@ -417,15 +423,17 @@ function App() { // Set up intervals for updating status, logs, and ports const statusInterval = setInterval(updateStatus, 1000); - const logsInterval = setInterval(updateLogs, 100); // More frequent log updates + // Event push (RFC #3 Step 4) replaces the 100ms full-clone log polling; + // the polling interval only runs in rollback mode. + const logsInterval = eventPush ? null : setInterval(updateLogs, 100); const portsInterval = setInterval(loadPorts, 3000); // Check for new ports every 3 seconds return () => { clearInterval(statusInterval); - clearInterval(logsInterval); + if (logsInterval !== null) clearInterval(logsInterval); clearInterval(portsInterval); }; - }, [loadPorts]); // 依赖loadPorts函数 + }, [loadPorts, eventPush]); // 依赖loadPorts函数 const handlePortSelect = (port: string) => { setSelectedPort(port); @@ -451,7 +459,7 @@ function App() { const updateLogs = async () => { try { const newLogs = await invoke('get_logs'); - setLogs(newLogs); + setPolledLogs(newLogs); } catch (error) { console.error('Failed to get logs:', error); } @@ -552,9 +560,13 @@ function App() { }; const handleClearLogs = async () => { + if (eventPush) { + await clearPushedLogs(); + return; + } try { await invoke('clear_logs'); - setLogs([]); + setPolledLogs([]); } catch (error) { console.error('Failed to clear logs:', error); } diff --git a/frontend/src/components/LogViewer.tsx b/frontend/src/components/LogViewer.tsx index dd56c00..eec0626 100644 --- a/frontend/src/components/LogViewer.tsx +++ b/frontend/src/components/LogViewer.tsx @@ -664,7 +664,7 @@ const LogViewer: React.FC = ({ logs, onClear, onExport, isConnec
{logs.map((log, index) => (
{ logEntryRefs.current[index] = el; }} className="py-1 px-2 rounded-[4px] transition-colors duration-150" style={{ diff --git a/frontend/src/hooks/useSerialLogs.ts b/frontend/src/hooks/useSerialLogs.ts new file mode 100644 index 0000000..4c35f71 --- /dev/null +++ b/frontend/src/hooks/useSerialLogs.ts @@ -0,0 +1,172 @@ +import { useState, useEffect, useRef, useCallback } from 'react'; +import { invoke } from '@tauri-apps/api/core'; +import { listen } from '@tauri-apps/api/event'; +import { LogEntry } from '../types'; +import { useTranslation } from '../i18n'; + +/** + * Event-driven serial log state (RFC #3 Step 4): replaces the 100 ms + * full-clone polling of `get_logs` with a one-shot snapshot + incremental + * `serial://frames` batches. + * + * Invariants: + * - Snapshot carries (epoch, session); a snapshot that raced a `clear_logs` + * (epoch mismatch) or a newer snapshot (ticket) is discarded — cleared + * logs never resurrect. + * - Batches dedupe by seq: frames already covered by the snapshot or a + * prior batch are skipped; frames from before a clear are skipped. + * - A batch whose session differs from the current one triggers a resync + * snapshot instead of an append (seq restarts at 1 per session). + * - `dropped_before > 0` inserts a placeholder row so channel overload is + * visible instead of silent. + */ + +interface FrameDto { + session: number; + seq: number; + direction: 'Sent' | 'Received'; + len: number; + timestamp: string; + display_text: string; + timestamp_formatted: string | null; +} + +interface FrameBatchDto { + session: number; + first_seq: number; + dropped_before: number; + frames: FrameDto[]; +} + +interface LogsSnapshot { + epoch: number; + session: number; + entries: LogEntry[]; +} + +/** Mirrors the backend default (`SerialManager::new`, clamped 100..10000). */ +const MAX_ENTRIES = 1000; + +export function useSerialLogs(enabled: boolean) { + const { t } = useTranslation(); + const [logs, setLogs] = useState([]); + + const epochRef = useRef(0); + const sessionRef = useRef(0); + const lastSeqRef = useRef(0); + const clearedSeqRef = useRef(0); + const pendingRef = useRef([]); + const rafRef = useRef(null); + const snapTicketRef = useRef(0); + + const flushPending = useCallback(() => { + rafRef.current = null; + if (pendingRef.current.length === 0) return; + const batch = pendingRef.current; + pendingRef.current = []; + setLogs((prev) => { + const next = [...prev, ...batch]; + return next.length > MAX_ENTRIES ? next.slice(next.length - MAX_ENTRIES) : next; + }); + }, []); + + const scheduleFlush = useCallback(() => { + if (rafRef.current === null) { + rafRef.current = requestAnimationFrame(flushPending); + } + }, [flushPending]); + + const takeSnapshot = useCallback(async () => { + const ticket = ++snapTicketRef.current; + try { + const snap = await invoke('get_logs_snapshot'); + if (ticket !== snapTicketRef.current) return; // superseded by a newer snapshot + if (snap.epoch !== epochRef.current) return; // raced a clear: discard + epochRef.current = snap.epoch; + sessionRef.current = snap.session; + lastSeqRef.current = snap.entries.reduce((m, e) => Math.max(m, e.seq ?? 0), 0); + pendingRef.current = []; + setLogs(snap.entries); + } catch (error) { + console.error('Failed to take logs snapshot:', error); + } + }, []); + + useEffect(() => { + if (!enabled) return; + let disposed = false; + let unlisten: (() => void) | undefined; + + void takeSnapshot(); + + listen('serial://frames', (event) => { + if (disposed) return; + const b = event.payload; + if (b.session !== sessionRef.current) { + void takeSnapshot(); // session changed: seq restarted, resync instead of append + return; + } + + const fresh: LogEntry[] = []; + if (b.dropped_before > 0) { + fresh.push({ + timestamp: new Date().toISOString(), + direction: 'Received', + data: [], + format: 'Text', + port_name: '', + display_text: t('logViewer.framesDropped').replace('{n}', String(b.dropped_before)), + timestamp_formatted: undefined, + session: b.session, + gap_key: `gap-${b.session}-${b.first_seq}`, + }); + } + for (const f of b.frames) { + // Skip frames already covered by the snapshot/prior batches, and + // frames that predate the last clear. + if (f.seq <= lastSeqRef.current || f.seq <= clearedSeqRef.current) continue; + fresh.push({ + timestamp: f.timestamp, + direction: f.direction, + data: [], + format: 'Text', + port_name: '', + display_text: f.display_text, + timestamp_formatted: f.timestamp_formatted ?? undefined, + seq: f.seq, + session: f.session, + }); + lastSeqRef.current = f.seq; + } + if (fresh.length > 0) { + pendingRef.current.push(...fresh); + scheduleFlush(); // rAF-throttled append + } + }) + .then((u) => { + unlisten = u; + }) + .catch((error) => console.error('Failed to listen serial://frames:', error)); + + return () => { + disposed = true; + unlisten?.(); + if (rafRef.current !== null) cancelAnimationFrame(rafRef.current); + }; + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [enabled, takeSnapshot, scheduleFlush]); + + const clearLogs = useCallback(async () => { + try { + const epoch = await invoke('clear_logs'); + epochRef.current = epoch; + clearedSeqRef.current = lastSeqRef.current; + pendingRef.current = []; + setLogs([]); + } catch (error) { + console.error('Failed to clear logs:', error); + } + }, []); + + return { logs, clearLogs }; +} diff --git a/frontend/src/i18n/translations/en.json b/frontend/src/i18n/translations/en.json index 2bd99a5..348ce3b 100644 --- a/frontend/src/i18n/translations/en.json +++ b/frontend/src/i18n/translations/en.json @@ -77,7 +77,8 @@ "tx": "TX", "rx": "RX", "total": "Total", - "bytes": "Bytes" + "bytes": "Bytes", + "framesDropped": "⋯ dropped {n} frames (channel overloaded) ⋯" }, "sendPanel": { "payload": "Payload", diff --git a/frontend/src/i18n/translations/zh-CN.json b/frontend/src/i18n/translations/zh-CN.json index 0984dc5..830f237 100644 --- a/frontend/src/i18n/translations/zh-CN.json +++ b/frontend/src/i18n/translations/zh-CN.json @@ -77,7 +77,8 @@ "tx": "发送", "rx": "接收", "total": "总计", - "bytes": "字节" + "bytes": "字节", + "framesDropped": "⋯ 丢失 {n} 帧（通道过载） ⋯" }, "sendPanel": { "payload": "发送数据", diff --git a/frontend/src/types.ts b/frontend/src/types.ts index 7c5c635..f55bce5 100644 --- a/frontend/src/types.ts +++ b/frontend/src/types.ts @@ -45,6 +45,12 @@ export interface LogEntry { display_text: string; /** Pre-formatted timestamp string (undefined if timestamps were disabled when entry was created) */ timestamp_formatted?: string; + /** Session-scoped sequence number (RFC #3 Step 4); 0/undefined for legacy entries */ + seq?: number; + /** Owning session id */ + session?: number; + /** Frontend-only marker for synthesized "frames dropped" placeholder rows */ + gap_key?: string; } export interface ConnectionStatus { diff --git a/src-tauri/src/bus.rs b/src-tauri/src/bus.rs new file mode 100644 index 0000000..fece286 --- /dev/null +++ b/src-tauri/src/bus.rs @@ -0,0 +1,345 @@ +//! Frame bus: session/seq allocation, bounded per-subscriber queues with +//! drop-oldest backpressure, and Nagle-style batching (RFC #3 Step 4). +//! +//! Contract pinned here: +//! - `seq` is strictly increasing from 1 within a session; TX and RX share +//! one sequence (transcript interleaving preserved). +//! - A slow subscriber loses the OLDEST frames and every batch reports +//! `dropped_before`, so gaps are always detectable (prefix gap + count). +//! - The publisher (read thread / send path) never blocks: queue push is a +//! µs-scale lock, never I/O. + +use crate::types::Direction; +use chrono::{DateTime, Utc}; +use std::collections::VecDeque; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Condvar, Mutex, OnceLock, Weak}; +use std::time::{Duration, Instant}; + +pub type SessionId = u64; +pub type Seq = u64; + +/// Process-level monotonic clock base (reserved for cross-port time alignment). +fn mono_base() -> &'static Instant { + static BASE: OnceLock = OnceLock::new(); + BASE.get_or_init(Instant::now) +} + +fn mono_now_ns() -> u64 { + mono_base().elapsed().as_nanos() as u64 +} + +#[derive(Clone, Debug)] +pub struct Frame { + pub session: SessionId, + pub seq: Seq, + pub dir: Direction, + /// Process-level monotonic timestamp, reserved for future cross-port + /// time alignment (RFC #3: must land now; adding it later would break + /// the transcript format). + #[allow(dead_code)] + pub t_mono_ns: u64, + pub t_wall: DateTime, + /// Arc-shared so fan-out to N subscribers is zero-copy. + pub data: Arc<[u8]>, +} + +#[derive(Debug)] +pub struct FrameBatch { + /// Session of the batch's LAST frame. A batch never mixes sessions in + /// practice (a session change requires a reconnect, which implies a + /// quiet port), but consumers should resync on session change anyway. + pub session: SessionId, + pub first_seq: Seq, + /// Frames dropped for THIS subscriber since the previous batch + /// (drop-oldest => the gap is always a prefix of the sequence). + pub dropped_before: u64, + pub frames: Vec, +} + +#[derive(Debug, Clone, Copy)] +pub struct BatchPolicy { + /// Max time the oldest pending frame may wait before flushing. + pub max_delay: Duration, + /// Minimum spacing between flushes; a lone frame for an idle consumer + /// pushes immediately only if the last flush is at least this old. + pub min_interval: Duration, + /// Flush when pending payload reaches this many bytes. + pub max_bytes: usize, + /// Flush when pending reaches this many frames. + pub max_frames: usize, + /// Bounded queue capacity per subscriber, in frames. + pub queue_frames: usize, +} + +/// GUI default: ~60 fps worst case, IPC batches bounded by the frame cap. +pub const GUI_DEFAULT: BatchPolicy = BatchPolicy { + max_delay: Duration::from_millis(16), + min_interval: Duration::from_millis(4), + max_bytes: 64 * 1024, + max_frames: 512, + queue_frames: 4096, +}; + +struct SubscriberState { + queue: VecDeque, + dropped: u64, +} + +struct SubscriberShared { + state: Mutex, + cond: Condvar, + /// Bounded queue capacity in frames, from the subscriber's policy. + queue_frames: usize, +} + +/// Receiving end of a subscription. The bridge contract is a pump thread +/// calling `recv_batch` in a loop (bounded blocking, never async). +pub struct Subscription { + shared: Arc, + policy: BatchPolicy, + last_flush: Mutex, +} + +impl Subscription { + /// Block (bounded) until a batch is ready per the Nagle rules: + /// 1. idle consumer + lone frame + min_interval elapsed => push immediately + /// 2. otherwise accumulate until max_delay / max_bytes / max_frames + /// Returns `None` on timeout with no frames (lets the pump react to + /// external state); the subscription lives for the app's lifetime. + pub fn recv_batch(&self) -> Option { + let policy = &self.policy; + let mut st = self.shared.state.lock().unwrap(); + + // Wait for the first frame, bounded by max_delay. + while st.queue.is_empty() { + let (g, timed_out) = self + .shared + .cond + .wait_timeout(st, policy.max_delay) + .unwrap(); + st = g; + if st.queue.is_empty() && timed_out.timed_out() { + return None; + } + } + + let now = Instant::now(); + let mut last_flush = self.last_flush.lock().unwrap(); + + // Rule 1: idle consumer, single frame, quiet period honored. + if st.queue.len() == 1 && now.duration_since(*last_flush) >= policy.min_interval { + let frame = st.queue.pop_front().unwrap(); + let dropped_before = std::mem::take(&mut st.dropped); + *last_flush = Instant::now(); + return Some(FrameBatch { + session: frame.session, + first_seq: frame.seq, + dropped_before, + frames: vec![frame], + }); + } + + // Rule 2: collect until a threshold or the max_delay deadline. + let mut frames: Vec = st.queue.drain(..).collect(); + let mut bytes: usize = frames.iter().map(|f| f.data.len()).sum(); + let deadline = Instant::now() + policy.max_delay; + loop { + if frames.len() >= policy.max_frames || bytes >= policy.max_bytes { + break; + } + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + break; + } + let (g, _) = self.shared.cond.wait_timeout(st, remaining).unwrap(); + st = g; + while let Some(f) = st.queue.pop_front() { + bytes += f.data.len(); + frames.push(f); + } + if Instant::now() >= deadline { + break; + } + } + + let dropped_before = std::mem::take(&mut st.dropped); + *last_flush = Instant::now(); + let first = frames.first().expect("non-empty after wait"); + Some(FrameBatch { + session: frames.last().unwrap().session, + first_seq: first.seq, + dropped_before, + frames: std::mem::take(&mut frames), + }) + } +} + +pub struct FrameBus { + session: AtomicU64, + seq: AtomicU64, + session_counter: AtomicU64, + subscribers: Mutex>>, +} + +impl FrameBus { + pub fn new() -> Self { + Self { + session: AtomicU64::new(0), + seq: AtomicU64::new(0), + session_counter: AtomicU64::new(0), + subscribers: Mutex::new(Vec::new()), + } + } + + /// Start a new session: id increments, seq resets so the next allocated + /// frame is seq 1. Called once per connect. + pub fn start_session(&self) -> SessionId { + let id = self.session_counter.fetch_add(1, Ordering::SeqCst) + 1; + self.seq.store(0, Ordering::SeqCst); + self.session.store(id, Ordering::SeqCst); + id + } + + pub fn current_session(&self) -> SessionId { + self.session.load(Ordering::SeqCst) + } + + /// Allocate a frame (assigns session/seq/timestamps, wraps data in Arc). + /// Separate from `publish` so the caller can also record the same + /// seq/session into its own structures (e.g. LogEntry) before fan-out. + pub fn alloc_frame(&self, dir: Direction, data: Vec) -> Frame { + Frame { + session: self.session.load(Ordering::SeqCst), + seq: self.seq.fetch_add(1, Ordering::SeqCst) + 1, + dir, + t_mono_ns: mono_now_ns(), + t_wall: Utc::now(), + data: Arc::from(data.into_boxed_slice()), + } + } + + /// Fan out to all live subscribers. Never blocks on I/O; a full queue + /// drops the OLDEST frame and counts it for `dropped_before`. + pub fn publish(&self, frame: &Frame) { + let mut subs = self.subscribers.lock().unwrap(); + subs.retain(|weak| { + if let Some(shared) = weak.upgrade() { + let mut st = shared.state.lock().unwrap(); + if st.queue.len() >= shared.queue_frames { + st.queue.pop_front(); + st.dropped += 1; + } + st.queue.push_back(frame.clone()); + drop(st); + shared.cond.notify_one(); + true + } else { + false // prune dead subscriptions + } + }); + } + + pub fn subscribe(&self, policy: BatchPolicy) -> Subscription { + let shared = Arc::new(SubscriberShared { + state: Mutex::new(SubscriberState { + queue: VecDeque::new(), + dropped: 0, + }), + cond: Condvar::new(), + queue_frames: policy.queue_frames, + }); + self.subscribers.lock().unwrap().push(Arc::downgrade(&shared)); + Subscription { + shared, + policy, + // Allow the very first frame to push immediately. + last_flush: Mutex::new(Instant::now() - policy.min_interval), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn fast_policy() -> BatchPolicy { + BatchPolicy { + max_delay: Duration::from_millis(5), + min_interval: Duration::from_millis(2), + max_bytes: 1024, + max_frames: 3, + queue_frames: 4, + } + } + + #[test] + fn seq_is_strictly_monotonic_and_resets_per_session() { + let bus = FrameBus::new(); + let s1 = bus.start_session(); + let f1 = bus.alloc_frame(Direction::Received, b"a".to_vec()); + let f2 = bus.alloc_frame(Direction::Sent, b"b".to_vec()); + assert_eq!((f1.session, f1.seq), (s1, 1)); + assert_eq!((f2.session, f2.seq), (s1, 2)); // TX/RX share one sequence + + let s2 = bus.start_session(); + assert!(s2 > s1); + let f3 = bus.alloc_frame(Direction::Received, b"c".to_vec()); + assert_eq!((f3.session, f3.seq), (s2, 1)); + } + + #[test] + fn lone_frame_pushes_immediately_for_idle_consumer() { + let bus = FrameBus::new(); + bus.start_session(); + let sub = bus.subscribe(fast_policy()); + bus.publish(&bus.alloc_frame(Direction::Received, b"hi".to_vec())); + let start = Instant::now(); + let batch = sub.recv_batch().unwrap(); + assert!(start.elapsed() < Duration::from_millis(5)); + assert_eq!(batch.first_seq, 1); + assert_eq!(batch.frames.len(), 1); + assert_eq!(batch.dropped_before, 0); + } + + #[test] + fn queued_frames_collect_into_one_batch_without_waiting() { + let bus = FrameBus::new(); + bus.start_session(); + let sub = bus.subscribe(fast_policy()); + for i in 0..3u8 { + bus.publish(&bus.alloc_frame(Direction::Received, vec![i])); + } + // 3 frames queued > lone-frame case: drained as one batch, no delay. + let batch = sub.recv_batch().unwrap(); + assert_eq!(batch.frames.len(), 3); + assert_eq!(batch.first_seq, 1); + } + + #[test] + fn slow_subscriber_drops_oldest_and_counts_prefix_gap() { + let bus = FrameBus::new(); + bus.start_session(); + let sub = bus.subscribe(fast_policy()); // queue_frames = 4 + // Publish 6 without draining: frames 1-2 are dropped as oldest. + for i in 0..6u32 { + bus.publish(&bus.alloc_frame(Direction::Received, i.to_le_bytes().to_vec())); + } + let batch = sub.recv_batch().unwrap(); + assert_eq!(batch.dropped_before, 2); + assert_eq!(batch.first_seq, 3); + let seqs: Vec = batch.frames.iter().map(|f| f.seq).collect(); + assert_eq!(seqs, vec![3, 4, 5, 6]); + } + + #[test] + fn recv_batch_times_out_when_quiet() { + let bus = FrameBus::new(); + bus.start_session(); + let sub = bus.subscribe(fast_policy()); + let start = Instant::now(); + assert!(sub.recv_batch().is_none()); + let elapsed = start.elapsed(); + assert!(elapsed >= Duration::from_millis(5) && elapsed < Duration::from_millis(50)); + } +} diff --git a/src-tauri/src/main.rs b/src-tauri/src/main.rs index 33efda3..5125deb 100644 --- a/src-tauri/src/main.rs +++ b/src-tauri/src/main.rs @@ -6,6 +6,7 @@ use std::sync::Mutex; use tauri::State; mod serial_manager; +mod bus; mod framing; mod types; mod updater; @@ -14,6 +15,79 @@ use serial_manager::SerialManager; use types::*; use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering}; +// ── RFC #3 Step 4: event push bridge ───────────────────────────────── +/// Wire DTO for `serial://frames`: the hot path carries no raw bytes +/// (seq/dir/len + decorated display text); raw bytes remain available via +/// the snapshot command and export. +#[derive(Clone, serde::Serialize)] +struct FrameDto { + session: u64, + seq: u64, + direction: Direction, + len: usize, + timestamp: chrono::DateTime, + display_text: String, + timestamp_formatted: Option, +} + +#[derive(Clone, serde::Serialize)] +struct FrameBatchDto { + session: u64, + first_seq: u64, + dropped_before: u64, + frames: Vec, +} + +/// Bridge contract: a plain pump thread with bounded blocking recv; the +/// subscription lives for the app's lifetime (sessions come and go). +fn spawn_frame_pump(app: &tauri::App) { + use tauri::{Emitter, Manager}; + let app_handle = app.handle().clone(); + let state = app.state::(); + let (bus, disp, tz) = { + let manager = state.serial_manager.lock().unwrap(); + ( + manager.bus(), + manager.display_settings_handle(), + manager.timezone_offset_handle(), + ) + }; + std::thread::spawn(move || { + let sub = bus.subscribe(bus::GUI_DEFAULT); + loop { + let Some(batch) = sub.recv_batch() else { + continue; // quiet timeout tick + }; + let settings = disp.lock().map(|g| g.clone()).unwrap_or_default(); + let tz_offset = *tz.lock().unwrap_or_else(|e| e.into_inner()); + let frames: Vec = batch + .frames + .iter() + .map(|f| FrameDto { + session: f.session, + seq: f.seq, + direction: f.dir, + len: f.data.len(), + timestamp: f.t_wall, + display_text: serial_manager::format_data_for_display(&f.data, &settings), + timestamp_formatted: if settings.show_timestamps { + Some(serial_manager::format_timestamp_with_offset(tz_offset)) + } else { + None + }, + }) + .collect(); + let dto = FrameBatchDto { + session: batch.session, + first_seq: batch.first_seq, + dropped_before: batch.dropped_before, + frames, + }; + let _ = app_handle.emit("serial://frames", dto); + } + }); +} + // Application state struct AppState { serial_manager: Mutex, @@ -125,11 +199,19 @@ async fn get_logs(state: State<'_, AppState>) -> Result, String> { Ok(manager.get_logs()) } +/// Initial alignment for the event-driven log view (RFC #3 Step 4). #[tauri::command] -async fn clear_logs(state: State<'_, AppState>) -> Result<(), String> { +async fn get_logs_snapshot(state: State<'_, AppState>) -> Result { + let manager = state.serial_manager.lock().unwrap(); + Ok(manager.get_logs_snapshot()) +} + +/// Returns the new log epoch; the frontend uses it to discard snapshots +/// that raced this clear. +#[tauri::command] +async fn clear_logs(state: State<'_, AppState>) -> Result { let mut manager = state.serial_manager.lock().unwrap(); - manager.clear_logs(); - Ok(()) + Ok(manager.clear_logs()) } #[tauri::command] @@ -360,6 +442,10 @@ fn main() { tauri::Builder::default() .manage(AppState::default()) .plugin(tauri_plugin_dialog::init()) + .setup(|app| { + spawn_frame_pump(app); + Ok(()) + }) .invoke_handler(tauri::generate_handler![ list_serial_ports, connect_to_port, @@ -367,6 +453,7 @@ fn main() { send_data, get_connection_status, get_logs, + get_logs_snapshot, clear_logs, export_logs, save_session, diff --git a/src-tauri/src/serial_manager.rs b/src-tauri/src/serial_manager.rs index ea4058f..2dbbd1e 100644 --- a/src-tauri/src/serial_manager.rs +++ b/src-tauri/src/serial_manager.rs @@ -40,6 +40,12 @@ pub struct SerialManager { reader_error: Arc>>, // Surfaced via ConnectionStatus until the next connect connection_error: Option, + // Frame bus for event push (RFC #3 Step 4); read/send paths publish, + // the bridge pump thread consumes and emits `serial://frames`. + bus: Arc, + // Bumped on every clear_logs; snapshots carry it to detect + // clear-during-snapshot resurrection. + log_epoch: u64, } #[derive(Debug, Default)] @@ -126,6 +132,8 @@ impl SerialManager { reader_handle: None, reader_error: Arc::new(Mutex::new(None)), connection_error: None, + bus: Arc::new(crate::bus::FrameBus::new()), + log_epoch: 0, } } @@ -201,6 +209,9 @@ impl SerialManager { let mut port = self.port_opener.open(port_name, &config)?; info!("Successfully opened serial port: {}", port_name); + // New session: seq restarts at 1 (RFC #3 Step 4). + self.bus.start_session(); + // Reset and start reading thread self.shutdown_flag.store(false, Ordering::Relaxed); let logs = Arc::clone(&self.logs); @@ -214,6 +225,7 @@ impl SerialManager { let port_name_clone = port_name.to_string(); let shutdown_flag = Arc::clone(&self.shutdown_flag); let reader_error = Arc::clone(&self.reader_error); + let bus = Arc::clone(&self.bus); let mut read_port = port.try_clone()?; // Give the write side a longer timeout than the read-friendly 50 ms. @@ -233,7 +245,8 @@ impl SerialManager { let mut segmenter = FrameSegmenter::new(initial_config, Instant::now()); // Single frame-emission path (replaces the four duplicated - // blocks): text recording -> display formatting -> log buffer -> stats. + // blocks): text recording -> display formatting -> log buffer + // -> stats -> bus publish (dual-write 存续期, RFC #3 Step 4). let emit_frame = |frame_data: Vec, disp_settings: &DisplaySettings| { // Write to text recording file with timestamp and RX label if let Ok(mut guard) = text_file.lock() { @@ -245,6 +258,10 @@ impl SerialManager { } } + // Allocate the bus frame first so the LogEntry carries the + // same seq/session the event subscribers see. + let frame = bus.alloc_frame(Direction::Received, frame_data.clone()); + // Format display text and timestamp based on current settings let tz_offset = *timezone_offset.lock().unwrap_or_else(|e| e.into_inner()); let display_text = format_data_for_display(&frame_data, disp_settings); @@ -263,6 +280,8 @@ impl SerialManager { port_name: port_name_clone.clone(), display_text, timestamp_formatted, + seq: frame.seq, + session: frame.session, }; if let Ok(mut logs_guard) = logs.lock() { @@ -276,6 +295,11 @@ impl SerialManager { if let Ok(mut stats_guard) = stats.lock() { stats_guard.bytes_received += data_len; } + + // Event fan-out comes last: the buffer write must land first + // so a snapshot racing this batch never misses the entry + // (subscribers dedupe by seq anyway). + bus.publish(&frame); }; loop { @@ -435,6 +459,10 @@ impl SerialManager { stats_guard.bytes_sent += data.len() as u64; } + // TX shares the session's single seq sequence (transcript + // interleaving preserved, RFC #3 Step 4). + let frame = self.bus.alloc_frame(Direction::Sent, data.clone()); + // Get current display settings for formatting let disp_settings = self.get_display_settings(); let tz_offset = *self.timezone_offset_minutes.lock().unwrap_or_else(|e| e.into_inner()); @@ -454,8 +482,13 @@ impl SerialManager { port_name: self.port_name.clone().unwrap_or_default(), display_text, timestamp_formatted, + seq: frame.seq, + session: frame.session, }); + // Buffer-first, then fan out (see emit_frame). + self.bus.publish(&frame); + Ok(()) } else { Err(anyhow!("No port available")) @@ -507,10 +540,38 @@ impl SerialManager { } } - pub fn clear_logs(&mut self) { + pub fn clear_logs(&mut self) -> u64 { if let Ok(mut logs) = self.logs.lock() { logs.clear(); } + // Bump the epoch so any snapshot started before this clear is + // recognized as stale by the frontend (no resurrection). + self.log_epoch += 1; + self.log_epoch + } + + /// Initial-alignment snapshot for the event-driven log view. + pub fn get_logs_snapshot(&self) -> LogsSnapshot { + LogsSnapshot { + epoch: self.log_epoch, + session: self.bus.current_session(), + entries: self.get_logs(), + } + } + + /// Event bus handle for the bridge pump (RFC #3 Step 4). + pub fn bus(&self) -> Arc { + Arc::clone(&self.bus) + } + + /// Shared display settings, for the bridge pump's decoration pass. + pub fn display_settings_handle(&self) -> Arc> { + Arc::clone(&self.display_settings) + } + + /// Shared timezone offset, for the bridge pump's decoration pass. + pub fn timezone_offset_handle(&self) -> Arc> { + Arc::clone(&self.timezone_offset_minutes) } pub fn export_logs(&self, file_path: &str, format: ExportFormat, timezone_offset_minutes: i32) -> Result<()> { @@ -857,7 +918,7 @@ impl SerialManager { } /// Format current UTC time with timezone offset applied -fn format_timestamp_with_offset(offset_minutes: i32) -> String { +pub(crate) fn format_timestamp_with_offset(offset_minutes: i32) -> String { use chrono::FixedOffset; let offset_seconds = offset_minutes * 60; let tz_offset = FixedOffset::east_opt(offset_seconds).unwrap_or_else(|| FixedOffset::east_opt(0).unwrap()); @@ -1021,7 +1082,7 @@ fn sort_usb_ports_first(mut ports: Vec) -> Vec { } /// Format data based on display settings -fn format_data_for_display(data: &[u8], settings: &DisplaySettings) -> String { +pub(crate) fn format_data_for_display(data: &[u8], settings: &DisplaySettings) -> String { match settings.format { ReceiveDisplayFormat::Hex => format_bytes_as_hex(data), ReceiveDisplayFormat::Txt => format_bytes_as_text(data, &settings.encoding, &settings.special_char_config), @@ -1427,8 +1488,36 @@ mod tests { } #[test] - fn disconnect_returns_joinable_handle_and_reconnect_is_immediate() { - // RFC #3 Step 3: disconnect hands the reader handle to the caller; + fn bus_events_carry_contiguous_seq_across_rx_and_tx() { + // RFC #3 Step 4 acceptance: frames reach subscribers with contiguous + // seqs, TX and RX sharing one sequence, no loss, no duplication. + let port = ScriptedPort::new("SCRIPT", vec![ScriptEvent::Bytes(b"hello".to_vec())]); + let mut manager = SerialManager::new() + .with_port_opener(Arc::new(ScriptedOpener { port })); + manager.set_frame_segmentation_config(seg_timeout()); + let sub = manager.bus().subscribe(crate::bus::GUI_DEFAULT); + manager.connect("SCRIPT", SerialConfig::default()).unwrap(); + + thread::sleep(Duration::from_millis(150)); // RX arrives + idle flush + manager.send_data(b"ping".to_vec()).unwrap(); + + let mut seen: Vec<(u64, Direction)> = Vec::new(); + let deadline = Instant::now() + Duration::from_secs(2); + while Instant::now() < deadline && seen.len() < 2 { + if let Some(batch) = sub.recv_batch() { + seen.extend(batch.frames.iter().map(|f| (f.seq, f.dir))); + } + } + manager.disconnect().unwrap(); + + assert_eq!( + seen, + vec![(1, Direction::Received), (2, Direction::Sent)] + ); + } + + #[test] + fn disconnect_returns_joinable_handle_and_reconnect_is_immediate() { // RFC #3 Step 3: disconnect hands the reader handle to the caller; // a bounded join then guarantees the device is free for reopen. let port = ScriptedPort::new("SCRIPT", vec![]); let mut manager = SerialManager::new() diff --git a/src-tauri/src/types.rs b/src-tauri/src/types.rs index 48ee5e3..ba20b88 100644 --- a/src-tauri/src/types.rs +++ b/src-tauri/src/types.rs @@ -93,6 +93,23 @@ pub struct LogEntry { pub display_text: String, /// Pre-formatted timestamp string (None if timestamps were disabled when entry was created) pub timestamp_formatted: Option, + /// Session-scoped sequence number, strictly increasing from 1 per + /// session, TX and RX sharing one sequence (RFC #3 Step 4). + #[serde(default)] + pub seq: u64, + /// Session this entry belongs to (0 = pre-event-model legacy entries). + #[serde(default)] + pub session: u64, +} + +/// Initial-alignment snapshot for the event-driven log view (RFC #3 Step 4). +/// `epoch` guards against clear-during-snapshot resurrection: a snapshot +/// taken before a `clear_logs` carries a stale epoch and must be discarded. +#[derive(Debug, Clone, Serialize)] +pub struct LogsSnapshot { + pub epoch: u64, + pub session: u64, + pub entries: Vec, } #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]