From 530986d06878f9d3a0e5b75455f7d8af19c45d3e Mon Sep 17 00:00:00 2001 From: FallenSigh Date: Mon, 8 Jun 2026 00:25:59 +0800 Subject: [PATCH] feat: xserial-client with session management, mlua integration, and 238 tests - Session: background I/O task with SessionHandle for send/read/close/reconfigure - SessionManager: multi-session collection with aggregated event broadcast - Config: serializable SessionConfig/PipelineConfig/FramerConfig/DecoderConfig - Lua: xserial global (open/list_ports/sleep/log) + session userdata API - RingBuffer: bounded history buffer with pipeline filtering - Core: added Serialize/Deserialize+Default to Endian/TextEncoding/PlotFormat - Core: added Send bound to Framer trait for tokio task compatibility - 238 tests (193 core + 10 config + 7 Lua + 9 session + 18 pipeline + 1 doctest) --- Cargo.lock | 151 ++++++++ Cargo.toml | 4 +- crates/xserial-client/Cargo.toml | 10 +- crates/xserial-client/src/cmd.rs | 7 + crates/xserial-client/src/config.rs | 240 +++++++++++++ crates/xserial-client/src/error.rs | 17 + crates/xserial-client/src/event.rs | 7 + crates/xserial-client/src/history.rs | 31 ++ crates/xserial-client/src/lib.rs | 15 + crates/xserial-client/src/lua/mod.rs | 38 ++ crates/xserial-client/src/lua/session_api.rs | 78 +++++ crates/xserial-client/src/manager.rs | 39 +++ crates/xserial-client/src/session.rs | 229 ++++++++++++ crates/xserial-client/tests/lua_tests.rs | 287 +++++++++++++++ .../xserial-client/tests/session_lifecycle.rs | 330 ++++++++++++++++++ crates/xserial-core/src/frame/mod.rs | 2 +- crates/xserial-core/src/protocol/mod.rs | 7 +- crates/xserial-core/src/protocol/plot.rs | 9 +- crates/xserial-core/src/protocol/text.rs | 9 +- 19 files changed, 1494 insertions(+), 16 deletions(-) create mode 100644 crates/xserial-client/src/cmd.rs create mode 100644 crates/xserial-client/src/config.rs create mode 100644 crates/xserial-client/src/error.rs create mode 100644 crates/xserial-client/src/event.rs create mode 100644 crates/xserial-client/src/history.rs create mode 100644 crates/xserial-client/src/lua/mod.rs create mode 100644 crates/xserial-client/src/lua/session_api.rs create mode 100644 crates/xserial-client/src/manager.rs create mode 100644 crates/xserial-client/src/session.rs create mode 100644 crates/xserial-client/tests/lua_tests.rs create mode 100644 crates/xserial-client/tests/session_lifecycle.rs diff --git a/Cargo.lock b/Cargo.lock index b231551..85ccd13 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -569,6 +569,16 @@ dependencies = [ "piper", ] +[[package]] +name = "bstr" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63044e1ae8e69f3b5a92c736ca6269b8d12fa7efe39bf34ddb06d102cf0e2cab" +dependencies = [ + "memchr", + "serde", +] + [[package]] name = "bumpalo" version = "3.20.3" @@ -1323,6 +1333,17 @@ version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" +[[package]] +name = "erased-serde" +version = "0.4.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2add8a07dd6a8d93ff627029c51de145e12686fbc36ecb298ac22e74cf02dec" +dependencies = [ + "serde", + "serde_core", + "typeid", +] + [[package]] name = "errno" version = "0.3.14" @@ -2291,6 +2312,25 @@ dependencies = [ "hashbrown 0.16.1", ] +[[package]] +name = "lua-src" +version = "550.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e836dc8ae16806c9bdcf42003a88da27d163433e3f9684c52f0301258004a4fb" +dependencies = [ + "cc", +] + +[[package]] +name = "luajit-src" +version = "210.6.6+707c12b" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a86cc925d4053d0526ae7f5bc765dbd0d7a5d1a63d43974f4966cb349ca63295" +dependencies = [ + "cc", + "which", +] + [[package]] name = "mac_address" version = "1.1.8" @@ -2390,6 +2430,56 @@ dependencies = [ "winapi", ] +[[package]] +name = "mlua" +version = "0.11.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccd36acfa49ce6ee56d1307a061dd302c564eee757e6e4cd67eb4f7204846fab" +dependencies = [ + "bstr", + "either", + "erased-serde", + "futures-util", + "libc", + "mlua-sys", + "mlua_derive", + "num-traits", + "parking_lot", + "rustc-hash 2.1.2", + "rustversion", + "serde", + "serde-value", +] + +[[package]] +name = "mlua-sys" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0f1c3a7fc7580227ece249fd90aa2fa3b39eb2b49d3aec5e103b3e85f2c3dfc8" +dependencies = [ + "cc", + "cfg-if", + "libc", + "lua-src", + "luajit-src", + "pkg-config", +] + +[[package]] +name = "mlua_derive" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "465bddde514c4eb3b50b543250e97c1d4b284fa3ef7dc0ba2992c77545dbceb2" +dependencies = [ + "itertools", + "once_cell", + "proc-macro-error2", + "proc-macro2", + "quote", + "regex", + "syn 2.0.117", +] + [[package]] name = "moxcms" version = "0.8.1" @@ -2903,6 +2993,15 @@ dependencies = [ "libredox", ] +[[package]] +name = "ordered-float" +version = "2.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68f19d67e5a2795c94e73e0bb1cc1a7edeb2e28efd39e2e1c9b7a40c1108b11c" +dependencies = [ + "num-traits", +] + [[package]] name = "ordered-float" version = "4.6.0" @@ -3272,6 +3371,28 @@ dependencies = [ "toml_edit", ] +[[package]] +name = "proc-macro-error-attr2" +version = "2.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96de42df36bb9bba5542fe9f1a054b8cc87e172759a1868aa05c1f3acc89dfc5" +dependencies = [ + "proc-macro2", + "quote", +] + +[[package]] +name = "proc-macro-error2" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11ec05c52be0a07b08061f7dd003e7d7092e0472bc731b4af7bb1ef876109802" +dependencies = [ + "proc-macro-error-attr2", + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "proc-macro2" version = "1.0.106" @@ -3641,6 +3762,16 @@ dependencies = [ "serde_derive", ] +[[package]] +name = "serde-value" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3a1a3341211875ef120e117ea7fd5228530ae7e7036a779fdc9117be6b3282c" +dependencies = [ + "ordered-float 2.10.1", + "serde", +] + [[package]] name = "serde_core" version = "1.0.228" @@ -4373,6 +4504,12 @@ dependencies = [ "rustc-hash 2.1.2", ] +[[package]] +name = "typeid" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc7d623258602320d5c55d1bc22793b57daff0ec7efc270ea7d55ce1d5f5471c" + [[package]] name = "typenum" version = "1.20.1" @@ -5074,6 +5211,15 @@ dependencies = [ "web-sys", ] +[[package]] +name = "which" +version = "8.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "81995fafaaaf6ae47a7d0cc83c67caf92aeb7e5331650ae6ff856f7c0c60c459" +dependencies = [ + "libc", +] + [[package]] name = "winapi" version = "0.3.9" @@ -5608,8 +5754,13 @@ checksum = "3ae8337f8a065cfc972643663ea4279e04e7256de865aa66fe25cec5fb912d3f" name = "xserial-client" version = "0.1.0" dependencies = [ + "mlua", + "serde", + "serde_json", + "thiserror 2.0.18", "tokio", "tracing", + "tracing-subscriber", "xserial-core", ] diff --git a/Cargo.toml b/Cargo.toml index 89f09b1..6ba7d72 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -50,5 +50,5 @@ egui = "0.34" eframe = "0.34" egui_plot = "0.35" -# ── Lua (Step 6 启用) ── -# mlua = { version = "0.11", features = ["luajit", "async", "serde", "send", "macros", "vendored"] } +# ── Lua ── +mlua = { version = "0.11", features = ["luajit", "async", "serde", "send", "macros", "vendored"] } diff --git a/crates/xserial-client/Cargo.toml b/crates/xserial-client/Cargo.toml index c257c37..e29fb84 100644 --- a/crates/xserial-client/Cargo.toml +++ b/crates/xserial-client/Cargo.toml @@ -5,5 +5,13 @@ edition = "2024" [dependencies] xserial-core = { path = "../xserial-core" } -tokio = { workspace = true } +tokio = { workspace = true, features = ["sync", "time"] } tracing = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +mlua = { workspace = true } +thiserror = { workspace = true } + +[dev-dependencies] +tokio = { workspace = true, features = ["full"] } +tracing-subscriber = { workspace = true } diff --git a/crates/xserial-client/src/cmd.rs b/crates/xserial-client/src/cmd.rs new file mode 100644 index 0000000..f34c018 --- /dev/null +++ b/crates/xserial-client/src/cmd.rs @@ -0,0 +1,7 @@ +use crate::config::SessionConfig; + +pub enum SessionCmd { + Send(Vec), + Close, + Reconfigure(SessionConfig), +} diff --git a/crates/xserial-client/src/config.rs b/crates/xserial-client/src/config.rs new file mode 100644 index 0000000..e20f5a5 --- /dev/null +++ b/crates/xserial-client/src/config.rs @@ -0,0 +1,240 @@ +use serde::{Deserialize, Serialize}; +use xserial_core::frame::{ + cobs::CobsFramer, fixed::FixedLengthFramer, + length::{LengthConfig, LengthPrefixedFramer}, + line::{LineConfig, LineFramer}, Endian, Framer, +}; +use xserial_core::protocol::{ + hex::{HexConfig, HexDecoder}, + plot::{PlotConfig, PlotDecoder, PlotFormat, SampleType}, + text::{TextDecoder, TextEncoding}, ProtocolDecoder, +}; +use xserial_core::transport::TransportConfig; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub enum FramerConfig { + Line { + #[serde(default = "default_true")] strip_cr: bool, + #[serde(default = "default_max_line")] max_line_len: usize, + }, + Fixed { frame_len: usize }, + Length { + #[serde(default = "default_len_bytes")] len_bytes: usize, + #[serde(default)] endian: Endian, + #[serde(default)] length_includes_self: bool, + #[serde(default = "default_max_payload")] max_payload: usize, + }, + Cobs { + #[serde(default = "default_max_frame")] max_frame: usize, + }, +} + +fn default_true() -> bool { true } +fn default_max_line() -> usize { 1024 * 1024 } +fn default_len_bytes() -> usize { 2 } +fn default_max_payload() -> usize { 1024 * 1024 } +fn default_max_frame() -> usize { 1024 * 1024 } + +impl FramerConfig { + pub fn build(self) -> Box { + match self { + FramerConfig::Line { strip_cr, max_line_len } => + Box::new(LineFramer::new(LineConfig { strip_cr, max_line_len })), + FramerConfig::Fixed { frame_len } => + Box::new(FixedLengthFramer::new(frame_len)), + FramerConfig::Length { len_bytes, endian, length_includes_self, max_payload } => + Box::new(LengthPrefixedFramer::new(LengthConfig { len_bytes, endian, length_includes_self, max_payload })), + FramerConfig::Cobs { max_frame } => + Box::new(CobsFramer::new(max_frame)), + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub enum DecoderConfig { + Text { #[serde(default)] encoding: TextEncoding }, + Hex { + #[serde(default)] uppercase: bool, + #[serde(default = "default_sep")] separator: String, + #[serde(default = "default_one")] bytes_per_group: usize, + #[serde(default)] endian: Endian, + }, + Plot { + sample_type: SampleType, + #[serde(default)] endian: Endian, + #[serde(default = "default_one")] channels: usize, + #[serde(default)] format: PlotFormat, + }, +} + +fn default_sep() -> String { String::from(" ") } +fn default_one() -> usize { 1 } + +impl DecoderConfig { + pub fn build(self) -> Box { + match self { + DecoderConfig::Text { encoding } => Box::new(TextDecoder::new(encoding)), + DecoderConfig::Hex { uppercase, separator, bytes_per_group, endian } => + Box::new(HexDecoder::new(HexConfig { uppercase, separator, bytes_per_group, endian })), + DecoderConfig::Plot { sample_type, endian, channels, format } => + Box::new(PlotDecoder::new(PlotConfig { sample_type, endian, channels, format })), + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PipelineConfig { + pub name: String, + pub framer: FramerConfig, + pub decoder: DecoderConfig, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SessionConfig { + pub transport: TransportConfig, + #[serde(default)] pub pipelines: Vec, + #[serde(default = "default_history")] pub history_limit: usize, + #[serde(default)] pub auto_reconnect: bool, +} + +fn default_history() -> usize { 10_000 } + +impl Default for SessionConfig { + fn default() -> Self { + Self { + transport: TransportConfig::Tcp { addr: "127.0.0.1:8080".into() }, + pipelines: Vec::new(), + history_limit: default_history(), + auto_reconnect: false, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use xserial_core::protocol::text::TextEncoding; + use xserial_core::protocol::plot::{PlotFormat, SampleType}; + use xserial_core::protocol::Endian; + use xserial_core::transport::TransportConfig; + + #[test] + fn framer_line_serde_roundtrip() { + let json = r#"{"Line":{"strip_cr":true,"max_line_len":4096}}"#; + let cfg: FramerConfig = serde_json::from_str(json).unwrap(); + match &cfg { + FramerConfig::Line { strip_cr, max_line_len } => { + assert!(*strip_cr); + assert_eq!(*max_line_len, 4096); + } + _ => panic!("expected Line"), + } + let back = serde_json::to_string(&cfg).unwrap(); + let _: FramerConfig = serde_json::from_str(&back).unwrap(); + } + + #[test] + fn framer_fixed_serde() { + let cfg: FramerConfig = serde_json::from_str(r#"{"Fixed":{"frame_len":128}}"#).unwrap(); + assert!(matches!(cfg, FramerConfig::Fixed { frame_len: 128 })); + } + + #[test] + fn framer_length_serde() { + let json = r#"{"Length":{"len_bytes":4,"endian":"Little","length_includes_self":true}}"#; + let cfg: FramerConfig = serde_json::from_str(json).unwrap(); + match cfg { + FramerConfig::Length { len_bytes, endian, length_includes_self, .. } => { + assert_eq!(len_bytes, 4); + assert_eq!(endian, Endian::Little); + assert!(length_includes_self); + } + _ => panic!("expected Length"), + } + } + + #[test] + fn framer_cobs_serde() { + let cfg: FramerConfig = serde_json::from_str(r#"{"Cobs":{"max_frame":8192}}"#).unwrap(); + assert!(matches!(cfg, FramerConfig::Cobs { max_frame: 8192 })); + } + + #[test] + fn decoder_text_serde() { + let cfg: DecoderConfig = serde_json::from_str(r#"{"Text":{"encoding":"Latin1"}}"#).unwrap(); + match cfg { + DecoderConfig::Text { encoding } => assert_eq!(encoding, TextEncoding::Latin1), + _ => panic!("expected Text"), + } + } + + #[test] + fn decoder_hex_serde() { + let json = r#"{"Hex":{"uppercase":true,"separator":":","bytes_per_group":2,"endian":"Little"}}"#; + let cfg: DecoderConfig = serde_json::from_str(json).unwrap(); + match cfg { + DecoderConfig::Hex { uppercase, separator, bytes_per_group, endian } => { + assert!(uppercase); + assert_eq!(separator, ":"); + assert_eq!(bytes_per_group, 2); + assert_eq!(endian, Endian::Little); + } + _ => panic!("expected Hex"), + } + } + + #[test] + fn decoder_plot_serde() { + let json = r#"{"Plot":{"sample_type":"F32","endian":"Little","channels":2,"format":"Interleaved"}}"#; + let cfg: DecoderConfig = serde_json::from_str(json).unwrap(); + match cfg { + DecoderConfig::Plot { sample_type, endian, channels, format } => { + assert_eq!(sample_type, SampleType::F32); + assert_eq!(channels, 2); + assert_eq!(format, PlotFormat::Interleaved); + assert_eq!(endian, Endian::Little); + } + _ => panic!("expected Plot"), + } + } + + #[test] + fn session_config_full_roundtrip() { + let json = r#"{ + "transport":{"Tcp":{"addr":"192.168.1.1:9999"}}, + "pipelines":[ + {"name":"t","framer":{"Line":{}},"decoder":{"Text":{}}}, + {"name":"h","framer":{"Fixed":{"frame_len":16}},"decoder":{"Hex":{}}} + ], + "history_limit":5000, + "auto_reconnect":true + }"#; + let cfg: SessionConfig = serde_json::from_str(json).unwrap(); + assert!(cfg.auto_reconnect); + assert_eq!(cfg.history_limit, 5000); + assert_eq!(cfg.pipelines.len(), 2); + assert!(matches!(&cfg.transport, TransportConfig::Tcp { addr } if addr == "192.168.1.1:9999")); + let back = serde_json::to_string_pretty(&cfg).unwrap(); + let _: SessionConfig = serde_json::from_str(&back).unwrap(); + } + + #[test] + fn framer_build_does_not_panic() { + for cfg in [ + FramerConfig::Line { strip_cr: true, max_line_len: 256 }, + FramerConfig::Fixed { frame_len: 8 }, + FramerConfig::Length { len_bytes: 2, endian: Endian::Big, length_includes_self: false, max_payload: 65536 }, + FramerConfig::Cobs { max_frame: 1024 }, + ] { + let f = cfg.build(); + assert_eq!(f.pending_len(), 0); + } + } + + #[test] + fn decoder_build_returns_correct_name() { + assert_eq!(DecoderConfig::Text { encoding: TextEncoding::Utf8 }.build().name(), "Text"); + assert_eq!(DecoderConfig::Hex { uppercase: false, separator: " ".into(), bytes_per_group: 1, endian: Endian::Big }.build().name(), "Hex"); + assert_eq!(DecoderConfig::Plot { sample_type: SampleType::I16, endian: Endian::Big, channels: 2, format: PlotFormat::XY }.build().name(), "Plot"); + } +} diff --git a/crates/xserial-client/src/error.rs b/crates/xserial-client/src/error.rs new file mode 100644 index 0000000..ad65281 --- /dev/null +++ b/crates/xserial-client/src/error.rs @@ -0,0 +1,17 @@ +use thiserror::Error; + +#[derive(Error, Debug)] +pub enum Error { + #[error("Connection failed: {0}")] + ConnectionFailed(String), + #[error("Session error: {0}")] + Session(String), + #[error("Lua error: {0}")] + Lua(#[from] mlua::Error), + #[error("I/O error: {0}")] + Io(#[from] std::io::Error), + #[error("{0}")] + Other(String), +} + +pub type Result = std::result::Result; diff --git a/crates/xserial-client/src/event.rs b/crates/xserial-client/src/event.rs new file mode 100644 index 0000000..85fa07b --- /dev/null +++ b/crates/xserial-client/src/event.rs @@ -0,0 +1,7 @@ +use xserial_core::protocol::DecodedData; + +#[derive(Debug, Clone)] +pub struct DecodedEntry { + pub pipeline_name: String, + pub data: DecodedData, +} diff --git a/crates/xserial-client/src/history.rs b/crates/xserial-client/src/history.rs new file mode 100644 index 0000000..554247a --- /dev/null +++ b/crates/xserial-client/src/history.rs @@ -0,0 +1,31 @@ +use std::collections::VecDeque; +use crate::event::DecodedEntry; + +pub struct RingBuffer { + buf: VecDeque, + limit: usize, +} + +impl RingBuffer { + pub fn new(limit: usize) -> Self { + Self { buf: VecDeque::with_capacity(limit.min(1024)), limit } + } + pub fn push(&mut self, item: T) { + if self.buf.len() >= self.limit { self.buf.pop_front(); } + self.buf.push_back(item); + } + pub fn len(&self) -> usize { self.buf.len() } + pub fn is_empty(&self) -> bool { self.buf.is_empty() } + pub fn iter(&self) -> impl Iterator { self.buf.iter() } + pub fn clear(&mut self) { self.buf.clear(); } + pub fn drain_recent(&self, count: usize) -> Vec where T: Clone { + let start = self.buf.len().saturating_sub(count); + self.buf.iter().skip(start).cloned().collect() + } +} + +impl RingBuffer { + pub fn entries_for_pipeline(&self, name: &str) -> Vec<&DecodedEntry> { + self.buf.iter().filter(|e| e.pipeline_name == name).collect() + } +} diff --git a/crates/xserial-client/src/lib.rs b/crates/xserial-client/src/lib.rs index e69de29..0726ad1 100644 --- a/crates/xserial-client/src/lib.rs +++ b/crates/xserial-client/src/lib.rs @@ -0,0 +1,15 @@ +pub mod config; +pub mod event; +pub mod cmd; +pub mod error; +pub mod history; +pub mod session; +pub mod manager; +pub mod lua; + +pub use config::{DecoderConfig, FramerConfig, PipelineConfig, SessionConfig}; +pub use error::{Error, Result}; +pub use event::DecodedEntry; +pub use history::RingBuffer; +pub use manager::SessionManager; +pub use session::{Session, SessionEvent, SessionHandle, SessionId}; diff --git a/crates/xserial-client/src/lua/mod.rs b/crates/xserial-client/src/lua/mod.rs new file mode 100644 index 0000000..d493ea6 --- /dev/null +++ b/crates/xserial-client/src/lua/mod.rs @@ -0,0 +1,38 @@ +use std::sync::atomic::{AtomicU64, Ordering}; +use mlua::{Lua, Result as LuaResult, Table, Value, LuaSerdeExt}; +use crate::config::SessionConfig; +use crate::session::Session; +mod session_api; + +static NEXT_ID: AtomicU64 = AtomicU64::new(1); + +pub fn register(lua: &Lua) -> LuaResult<()> { + let xserial = lua.create_table()?; + + xserial.set("open", lua.create_async_function(move |lua, config: Table| { + let id = NEXT_ID.fetch_add(1, Ordering::Relaxed); + async move { + let cfg: SessionConfig = lua.from_value(Value::Table(config))?; + let handle = Session::spawn(id, cfg); + lua.create_userdata(session_api::LuaSessionHandle::new(handle)) + } + })?)?; + + xserial.set("list_ports", lua.create_function(|_, ()| { + let ports = xserial_core::transport::serial::SerialTransport::list_ports(); + Ok(ports.into_iter().map(|p| p.port_name).collect::>()) + })?)?; + + xserial.set("sleep", lua.create_async_function(|_, ms: u64| async move { + tokio::time::sleep(std::time::Duration::from_millis(ms)).await; + Ok(()) + })?)?; + + xserial.set("log", lua.create_function(|_, msg: String| { + tracing::info!("[lua] {}", msg); + Ok(()) + })?)?; + + lua.globals().set("xserial", xserial)?; + Ok(()) +} diff --git a/crates/xserial-client/src/lua/session_api.rs b/crates/xserial-client/src/lua/session_api.rs new file mode 100644 index 0000000..11c9862 --- /dev/null +++ b/crates/xserial-client/src/lua/session_api.rs @@ -0,0 +1,78 @@ +use mlua::{UserData, UserDataMethods, Variadic, LuaSerdeExt}; +use xserial_core::protocol::DecodedData; +use crate::session::SessionHandle; + +pub struct LuaSessionHandle { pub handle: SessionHandle } + +impl LuaSessionHandle { + pub fn new(handle: SessionHandle) -> Self { Self { handle } } +} + +impl UserData for LuaSessionHandle { + fn add_methods>(methods: &mut M) { + methods.add_async_method("send", |_, this, data: mlua::String| async move { + this.handle.send(data.as_bytes().to_vec()).await + .map_err(mlua::Error::RuntimeError) + }); + + methods.add_async_method("read", |lua, this, timeout_ms: Option| async move { + let ms = timeout_ms.unwrap_or(1000); + match this.handle.read(ms).await { + Some(entry) => { + let t = lua.create_table()?; + t.set("pipeline", entry.pipeline_name)?; + match entry.data { + DecodedData::Text(s) => { t.set("kind", "text")?; t.set("data", s)?; } + DecodedData::Hex(s) => { t.set("kind", "hex")?; t.set("data", s)?; } + DecodedData::Binary(v) => { t.set("kind", "binary")?; t.set("data", lua.create_string(&v)?)?; } + DecodedData::Plot(frame) => { + t.set("kind", "plot")?; + let chs = lua.create_table()?; + for (i, ch) in frame.channels.iter().enumerate() { + chs.set(i + 1, lua.create_sequence_from(ch.iter().copied())?)?; + } + t.set("channels", chs)?; + t.set("sample_count", frame.sample_count())?; + } + } + Ok(Some(t)) + } + None => Ok(None), + } + }); + + methods.add_async_method("close", |_, this, (): ()| async move { + this.handle.close().await.map_err(mlua::Error::RuntimeError) + }); + + methods.add_async_method("reconfigure", |lua, this, config: mlua::Table| async move { + let cfg: crate::config::SessionConfig = lua.from_value(mlua::Value::Table(config))?; + this.handle.reconfigure(cfg).await.map_err(mlua::Error::RuntimeError) + }); + + methods.add_async_method("on_data", |_, this, (callback, pipelines): (mlua::Function, Variadic)| async move { + let filter: Vec = pipelines.into_iter().collect(); + let mut rx = this.handle.subscribe(); + tokio::spawn(async move { + loop { + match rx.recv().await { + Ok(crate::session::SessionEvent::Data(_, entry)) + if filter.is_empty() || filter.iter().any(|f| f == &entry.pipeline_name) => + { + let data_str = match &entry.data { + DecodedData::Text(s) => s.clone(), + DecodedData::Hex(s) => s.clone(), + _ => continue, + }; + let _ = callback.call_async::<()>((entry.pipeline_name, data_str)).await; + } + Ok(crate::session::SessionEvent::Closed(_)) => break, + Err(_) => break, + _ => {} + } + } + }); + Ok(()) + }); + } +} diff --git a/crates/xserial-client/src/manager.rs b/crates/xserial-client/src/manager.rs new file mode 100644 index 0000000..2acbbd6 --- /dev/null +++ b/crates/xserial-client/src/manager.rs @@ -0,0 +1,39 @@ +use std::collections::HashMap; +use crate::config::SessionConfig; +use crate::session::{Session, SessionEvent, SessionHandle, SessionId}; + +pub struct SessionManager { + sessions: HashMap, + event_tx: tokio::sync::broadcast::Sender, +} + +impl SessionManager { + pub fn new() -> Self { + let (event_tx, _) = tokio::sync::broadcast::channel(256); + Self { sessions: HashMap::new(), event_tx } + } + + pub fn create(&mut self, config: SessionConfig) -> SessionHandle { + let id = self.sessions.len() as u64 + 1; + let handle = Session::spawn(id, config); + let event_relay = self.event_tx.clone(); + let mut event_rx = handle.subscribe(); + tokio::spawn(async move { + while let Ok(event) = event_rx.recv().await { let _ = event_relay.send(event); } + }); + self.sessions.insert(id, SessionHandle { + id: handle.id, cmd_tx: handle.cmd_tx.clone(), event_tx: handle.event_tx.clone(), _task: None, + }); + handle + } + + pub fn remove(&mut self, id: SessionId) { self.sessions.remove(&id); } + pub fn get(&self, id: SessionId) -> Option<&SessionHandle> { self.sessions.get(&id) } + pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver { self.event_tx.subscribe() } + pub fn list_ids(&self) -> Vec { self.sessions.keys().copied().collect() } + pub fn count(&self) -> usize { self.sessions.len() } +} + +impl Default for SessionManager { + fn default() -> Self { Self::new() } +} diff --git a/crates/xserial-client/src/session.rs b/crates/xserial-client/src/session.rs new file mode 100644 index 0000000..3390531 --- /dev/null +++ b/crates/xserial-client/src/session.rs @@ -0,0 +1,229 @@ +use tokio::io::AsyncReadExt; +use tokio::select; +use tokio::sync::{broadcast, mpsc}; +use tokio::task::JoinHandle; +use tracing::{error, info, warn}; + +use xserial_core::pipeline::{MultiPipeline, Pipeline}; +use xserial_core::transport::Connection; + +use crate::cmd::SessionCmd; +use crate::config::SessionConfig; +use crate::event::DecodedEntry; +use crate::history::RingBuffer; + +pub type SessionId = u64; + +#[derive(Debug, Clone)] +pub enum SessionEvent { + Connected(SessionId), + Disconnected(SessionId), + Data(SessionId, DecodedEntry), + Error(SessionId, String), + Closed(SessionId), +} + +pub struct SessionHandle { + pub id: SessionId, + pub(crate) cmd_tx: mpsc::Sender, + pub(crate) event_tx: broadcast::Sender, + pub(crate) _task: Option>, +} + +impl SessionHandle { + pub fn id(&self) -> SessionId { self.id } + + pub async fn send(&self, data: Vec) -> Result<(), String> { + self.cmd_tx.send(SessionCmd::Send(data)).await.map_err(|_| "session closed".into()) + } + + pub async fn close(&self) -> Result<(), String> { + self.cmd_tx.send(SessionCmd::Close).await.map_err(|_| "session closed".into()) + } + + pub async fn reconfigure(&self, config: SessionConfig) -> Result<(), String> { + self.cmd_tx.send(SessionCmd::Reconfigure(config)).await.map_err(|_| "session closed".into()) + } + + pub async fn read(&self, timeout_ms: u64) -> Option { + let mut rx = self.event_tx.subscribe(); + let deadline = std::time::Duration::from_millis(timeout_ms); + tokio::time::timeout(deadline, async { + loop { + match rx.recv().await { + Ok(SessionEvent::Data(_, e)) => return Some(e), + Ok(SessionEvent::Error(_, _) | SessionEvent::Closed(_)) => return None, + Err(broadcast::error::RecvError::Lagged(_)) => continue, + Err(broadcast::error::RecvError::Closed) => return None, + _ => continue, + } + } + }).await.ok().flatten() + } + + pub fn subscribe(&self) -> broadcast::Receiver { + self.event_tx.subscribe() + } +} + +impl Drop for SessionHandle { + fn drop(&mut self) { + let _ = self.cmd_tx.try_send(SessionCmd::Close); + if let Some(t) = self._task.take() { t.abort(); } + } +} + +pub struct Session { + id: SessionId, + config: SessionConfig, + conn: Option, + pipeline: MultiPipeline, + history: RingBuffer, + cmd_rx: mpsc::Receiver, + event_tx: broadcast::Sender, +} + +impl Session { + pub fn spawn(id: SessionId, config: SessionConfig) -> SessionHandle { + let (cmd_tx, cmd_rx) = mpsc::channel(32); + let (event_tx, _) = broadcast::channel(256); + let mut session = Self { + id, conn: None, + pipeline: Self::build_pipeline(&config), + history: RingBuffer::new(config.history_limit), + config, cmd_rx, event_tx: event_tx.clone(), + }; + let task = tokio::spawn(async move { session.run().await }); + SessionHandle { id, cmd_tx, event_tx, _task: Some(task) } + } + + fn build_pipeline(config: &SessionConfig) -> MultiPipeline { + let mut mp = MultiPipeline::new(); + for pc in &config.pipelines { + mp.add(Pipeline::new( + pc.name.clone(), + pc.framer.clone().build(), + pc.decoder.clone().build(), + )); + } + mp + } + + async fn run(&mut self) { + info!(session_id = self.id, "session started"); + + // Initial connect + let conn = &mut self.conn; + if let Err(e) = do_connect(conn, &self.config, &self.event_tx, self.id).await { + let _ = self.event_tx.send(SessionEvent::Error(self.id, format!("connect failed: {}", e))); + let _ = self.event_tx.send(SessionEvent::Closed(self.id)); + return; + } + + loop { + let cmd_rx = &mut self.cmd_rx; + let conn = &mut self.conn; + let pipeline = &mut self.pipeline; + let history = &mut self.history; + let event_tx = &self.event_tx; + let id = self.id; + + select! { + cmd = cmd_rx.recv() => { + match cmd { + Some(SessionCmd::Close) => { + do_disconnect(conn, pipeline, event_tx, id).await; + let _ = event_tx.send(SessionEvent::Closed(id)); + break; + } + Some(SessionCmd::Send(data)) => { + if let Some(c) = conn { + use tokio::io::AsyncWriteExt; + if let Err(e) = c.write_all(&data).await { + let _ = event_tx.send(SessionEvent::Error(id, format!("write: {}", e))); + } + } + } + Some(SessionCmd::Reconfigure(new_cfg)) => { + do_disconnect(conn, pipeline, event_tx, id).await; + self.config = new_cfg; + *pipeline = Self::build_pipeline(&self.config); + *history = RingBuffer::new(self.config.history_limit); + if let Err(e) = do_connect(conn, &self.config, event_tx, id).await { + let _ = event_tx.send(SessionEvent::Error(id, format!("reconnect: {}", e))); + } + } + None => break, + } + } + result = try_read(conn, pipeline) => { + match result { + Ok(entries) => { + for entry in entries { + history.push(entry.clone()); + let _ = event_tx.send(SessionEvent::Data(id, entry)); + } + } + Err(e) => { + error!(session_id = id, error = %e, "read error"); + let _ = event_tx.send(SessionEvent::Error(id, e.to_string())); + do_disconnect(conn, pipeline, event_tx, id).await; + if self.config.auto_reconnect { + warn!(session_id = id, "auto-reconnecting"); + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + if let Err(e) = do_connect(conn, &self.config, event_tx, id).await { + let _ = event_tx.send(SessionEvent::Error(id, format!("reconnect: {}", e))); + } + } + } + } + } + } + } + info!(session_id = self.id, "session ended"); + } +} + +async fn try_read( + conn: &mut Option, + pipeline: &mut MultiPipeline, +) -> Result, std::io::Error> { + let c = match conn.as_mut() { Some(c) => c, None => return Ok(Vec::new()) }; + let mut buf = [0u8; 4096]; + match c.read(&mut buf).await { + Ok(0) => Ok(Vec::new()), + Ok(n) => { + Ok(pipeline.feed(&buf[..n]) + .into_iter() + .map(|r| DecodedEntry { pipeline_name: r.pipeline_name, data: r.data }) + .collect()) + } + Err(e) => Err(e), + } +} + +async fn do_connect( + conn: &mut Option, + config: &SessionConfig, + event_tx: &broadcast::Sender, + id: SessionId, +) -> Result<(), std::io::Error> { + let mut c = Connection::new(config.transport.clone()); + c.connect().await.map_err(|e| std::io::Error::new(std::io::ErrorKind::NotConnected, format!("{}", e)))?; + *conn = Some(c); + let _ = event_tx.send(SessionEvent::Connected(id)); + Ok(()) +} + +async fn do_disconnect( + conn: &mut Option, + pipeline: &mut MultiPipeline, + event_tx: &broadcast::Sender, + id: SessionId, +) { + if let Some(mut c) = conn.take() { + let _ = c.disconnect().await; + pipeline.reset(); + let _ = event_tx.send(SessionEvent::Disconnected(id)); + } +} diff --git a/crates/xserial-client/tests/lua_tests.rs b/crates/xserial-client/tests/lua_tests.rs new file mode 100644 index 0000000..6b802c5 --- /dev/null +++ b/crates/xserial-client/tests/lua_tests.rs @@ -0,0 +1,287 @@ +use std::sync::Once; + +use mlua::Lua; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; + +static INIT: Once = Once::new(); + +fn init_tracing() { + INIT.call_once(|| { + tracing_subscriber::fmt() + .with_env_filter("xserial=trace") + .try_init() + .ok(); + }); +} + +// ── xserial.sleep / xserial.log ────────────────────────────────── + +#[tokio::test] +async fn lua_sleep_and_log() { + let lua = Lua::new(); + xserial_client::lua::register(&lua).unwrap(); + + lua.load( + r#" + xserial.log("test start") + xserial.sleep(50) + xserial.log("test end") + "# + ) + .exec_async() + .await + .unwrap(); +} + +// ── xserial.list_ports ─────────────────────────────────────────── + +#[tokio::test] +async fn lua_list_ports() { + let lua = Lua::new(); + xserial_client::lua::register(&lua).unwrap(); + + let result: Vec = lua + .load(r#"return xserial.list_ports()"#) + .eval_async() + .await + .unwrap(); + + // Should return a table (possibly empty if no serial ports) + // Just verify it doesn't crash + let _ = result; +} + +// ── xserial.open + session:send / session:read / session:close ──── + +#[tokio::test] +async fn lua_session_open_send_read_close() { + init_tracing(); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut buf = [0u8; 256]; + let n = stream.read(&mut buf).await.unwrap(); + assert_eq!(&buf[..n], b"hello\n"); + stream.write_all(b"world\n").await.unwrap(); + }); + + let lua = Lua::new(); + xserial_client::lua::register(&lua).unwrap(); + + let script = format!( + r#" + local sess = xserial.open({{ + transport = {{ Tcp = {{ addr = "{}" }} }}, + pipelines = {{{{ + name = "text", + framer = {{ Line = {{}} }}, + decoder = {{ Text = {{}} }} + }}}} + }}) + + sess:send("hello\n") + + local r = sess:read(5000) + assert(r ~= nil, "expected data") + assert(r.pipeline == "text", "expected text pipeline") + assert(r.data == "world", "expected 'world', got " .. tostring(r.data)) + + sess:close() + "#, + addr + ); + + lua.load(&script).exec_async().await.unwrap(); + server.await.unwrap(); +} + +// ── session:read with timeout ──────────────────────────────────── + +#[tokio::test] +async fn lua_session_read_timeout() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + + let _server = tokio::spawn(async move { + let (_stream, _) = listener.accept().await.unwrap(); + tokio::time::sleep(std::time::Duration::from_secs(10)).await; + }); + + let lua = Lua::new(); + xserial_client::lua::register(&lua).unwrap(); + + let script = format!( + r#" + local sess = xserial.open({{ + transport = {{ Tcp = {{ addr = "{}" }} }}, + pipelines = {{{{ + name = "text", + framer = {{ Line = {{}} }}, + decoder = {{ Text = {{}} }} + }}}} + }}) + + local r = sess:read(100) + assert(r == nil, "read should time out") + + sess:close() + "#, + addr + ); + + lua.load(&script).exec_async().await.unwrap(); +} + +// ── xserial.open with hex decoder ──────────────────────────────── + +#[tokio::test] +async fn lua_session_hex_decoder() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + stream.write_all(b"ABC\n").await.unwrap(); + }); + + let lua = Lua::new(); + xserial_client::lua::register(&lua).unwrap(); + + let script = format!( + r#" + local sess = xserial.open({{ + transport = {{ Tcp = {{ addr = "{}" }} }}, + pipelines = {{{{ + name = "hex", + framer = {{ Line = {{}} }}, + decoder = {{ Hex = {{ uppercase = true }} }} + }}}} + }}) + + local r = sess:read(5000) + assert(r ~= nil) + assert(r.pipeline == "hex") + assert(r.kind == "hex") + assert(r.data == "41 42 43", "expected hex of ABC, got " .. tostring(r.data)) + + sess:close() + "#, + addr + ); + + lua.load(&script).exec_async().await.unwrap(); + server.await.unwrap(); +} + +// ── session:on_data callback ───────────────────────────────────── + +#[tokio::test] +async fn lua_session_on_data_callback() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + stream.write_all(b"line1\nline2\n").await.unwrap(); + }); + + let lua = Lua::new(); + xserial_client::lua::register(&lua).unwrap(); + + let script = format!( + r#" + local received = {{}} + local sess = xserial.open({{ + transport = {{ Tcp = {{ addr = "{}" }} }}, + pipelines = {{{{ + name = "text", + framer = {{ Line = {{}} }}, + decoder = {{ Text = {{}} }} + }}}} + }}) + + sess:on_data(function(name, data) + table.insert(received, {{ name = name, data = data }}) + end) + + -- Wait for both lines + xserial.sleep(500) + + assert(#received == 2, "expected 2 callbacks, got " .. #received) + assert(received[1].name == "text") + assert(received[1].data == "line1") + assert(received[2].name == "text") + assert(received[2].data == "line2") + + sess:close() + "#, + addr + ); + + lua.load(&script).exec_async().await.unwrap(); + server.await.unwrap(); +} + +// ── session:reconfigure from Lua ───────────────────────────────── + +#[tokio::test] +async fn lua_session_reconfigure() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + stream.write_all(b"test\n").await.unwrap(); + }); + + let listener2 = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr2 = listener2.local_addr().unwrap().to_string(); + + let server2 = tokio::spawn(async move { + let (mut stream, _) = listener2.accept().await.unwrap(); + stream.write_all(b"ABC\n").await.unwrap(); + }); + + let lua = Lua::new(); + xserial_client::lua::register(&lua).unwrap(); + + let script = format!( + r#" + local sess = xserial.open({{ + transport = {{ Tcp = {{ addr = "{}" }} }}, + pipelines = {{{{ + name = "text", + framer = {{ Line = {{}} }}, + decoder = {{ Text = {{}} }} + }}}} + }}) + + local r = sess:read(5000) + assert(r.data == "test") + + -- Reconfigure to hex on different address + sess:reconfigure({{ + transport = {{ Tcp = {{ addr = "{}" }} }}, + pipelines = {{{{ + name = "hex", + framer = {{ Line = {{}} }}, + decoder = {{ Hex = {{ uppercase = true }} }} + }}}} + }}) + + local r2 = sess:read(5000) + assert(r2.pipeline == "hex") + assert(r2.data == "41 42 43") + + sess:close() + "#, + addr, addr2 + ); + + lua.load(&script).exec_async().await.unwrap(); + server.await.unwrap(); + server2.await.unwrap(); +} diff --git a/crates/xserial-client/tests/session_lifecycle.rs b/crates/xserial-client/tests/session_lifecycle.rs new file mode 100644 index 0000000..4b42be2 --- /dev/null +++ b/crates/xserial-client/tests/session_lifecycle.rs @@ -0,0 +1,330 @@ +use std::time::Duration; + +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; + +use xserial_client::config::{ + DecoderConfig, FramerConfig, PipelineConfig, SessionConfig, +}; +use xserial_client::session::{Session, SessionEvent}; +use xserial_client::SessionManager; +use xserial_core::protocol::DecodedData; +use xserial_core::transport::TransportConfig; + +fn tcp_config(addr: String) -> SessionConfig { + SessionConfig { + transport: TransportConfig::Tcp { addr }, + pipelines: vec![PipelineConfig { + name: "text".into(), + framer: FramerConfig::Line { + strip_cr: true, + max_line_len: 1024 * 1024, + }, + decoder: DecoderConfig::Text { + encoding: xserial_core::protocol::text::TextEncoding::Utf8, + }, + }], + history_limit: 100, + auto_reconnect: false, + } +} + +// ── Session spawn + basic send/read ────────────────────────────── + +#[tokio::test] +async fn session_echo_send_and_read() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut buf = [0u8; 256]; + let n = stream.read(&mut buf).await.unwrap(); + assert_eq!(&buf[..n], b"ping\n"); + stream.write_all(b"pong\n").await.unwrap(); + }); + + let handle = Session::spawn(1, tcp_config(addr)); + handle.send(b"ping\n".to_vec()).await.unwrap(); + + let entry = handle.read(5000).await.expect("should receive response"); + match entry.data { + DecodedData::Text(s) => assert_eq!(s, "pong"), + other => panic!("expected Text, got {:?}", other), + } + + handle.close().await.unwrap(); + server.await.unwrap(); +} + +#[tokio::test] +async fn session_multiple_reads() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + stream.write_all(b"a\nb\nc\n").await.unwrap(); + }); + + let handle = Session::spawn(2, tcp_config(addr)); + + // Subscribe once and collect all text events + let mut rx = handle.subscribe(); + let mut texts = Vec::new(); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + match rx.recv().await.unwrap() { + SessionEvent::Data(_, entry) => { + if let DecodedData::Text(s) = entry.data { + texts.push(s); + if texts.len() == 3 { break; } + } + } + _ => {} + } + } + }).await.unwrap(); + + assert_eq!(texts, vec!["a", "b", "c"]); + handle.close().await.unwrap(); + server.await.unwrap(); +} + +// ── Session with multiple pipelines ────────────────────────────── + +#[tokio::test] +async fn session_multi_pipeline_text_and_hex() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + stream.write_all(b"hello\n").await.unwrap(); + }); + + let mut config = tcp_config(addr); + config.pipelines.push(PipelineConfig { + name: "hex".into(), + framer: FramerConfig::Line { + strip_cr: true, + max_line_len: 1024 * 1024, + }, + decoder: DecoderConfig::Hex { + uppercase: false, + separator: " ".into(), + bytes_per_group: 1, + endian: xserial_core::protocol::Endian::Big, + }, + }); + + let handle = Session::spawn(3, config); + + // Subscribe and collect both pipeline results + let mut rx = handle.subscribe(); + let mut results = Vec::new(); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + match rx.recv().await.unwrap() { + SessionEvent::Data(_, entry) => { + results.push((entry.pipeline_name.clone(), entry.data.clone())); + if results.len() == 2 { break; } + } + _ => {} + } + } + }).await.unwrap(); + + assert_eq!(results.len(), 2); + assert_eq!(results[0].0, "text"); + assert_eq!(results[1].0, "hex"); + assert!(matches!(results[1].1, DecodedData::Hex(_))); + + handle.close().await.unwrap(); + server.await.unwrap(); +} + +// ── SessionClose + read timeout ────────────────────────────────── + +#[tokio::test] +async fn session_read_timeout_returns_none() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + + let server = tokio::spawn(async move { + let (_stream, _) = listener.accept().await.unwrap(); + // Accept but send nothing — client should time out + tokio::time::sleep(Duration::from_secs(10)).await; + }); + + let handle = Session::spawn(4, tcp_config(addr)); + + // Short timeout — no data sent by server + let result = handle.read(200).await; + assert!(result.is_none(), "read should time out and return None"); + + handle.close().await.unwrap(); + server.abort(); +} + +#[tokio::test] +async fn session_close_stops_events() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + + let _server = tokio::spawn(async move { + let (_stream, _) = listener.accept().await.unwrap(); + tokio::time::sleep(Duration::from_secs(10)).await; + }); + + let handle = Session::spawn(5, tcp_config(addr)); + handle.close().await.unwrap(); + + // After close, read should return None immediately + let result = handle.read(100).await; + assert!(result.is_none()); +} + +// ── SessionManager ─────────────────────────────────────────────── + +#[tokio::test] +async fn manager_create_and_list() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + + let _server = tokio::spawn(async move { + let (_stream, _) = listener.accept().await.unwrap(); + tokio::time::sleep(Duration::from_secs(10)).await; + }); + + let mut mgr = SessionManager::new(); + assert_eq!(mgr.count(), 0); + assert!(mgr.list_ids().is_empty()); + + let h1 = mgr.create(tcp_config(addr.clone())); + assert_eq!(mgr.count(), 1); + assert_eq!(mgr.list_ids(), vec![1]); + + let h2 = mgr.create(tcp_config(addr)); + assert_eq!(mgr.count(), 2); + + assert!(mgr.get(1).is_some()); + assert!(mgr.get(2).is_some()); + assert!(mgr.get(99).is_none()); + + mgr.remove(1); + assert_eq!(mgr.count(), 1); + assert!(mgr.get(1).is_none()); + + h1.close().await.unwrap(); + h2.close().await.unwrap(); +} + +#[tokio::test] +async fn manager_event_subscription() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + stream.write_all(b"data\n").await.unwrap(); + }); + + let mut mgr = SessionManager::new(); + let mut rx = mgr.subscribe(); + + let handle = mgr.create(tcp_config(addr)); + + // Wait for Connected event + let event = tokio::time::timeout(Duration::from_secs(5), rx.recv()) + .await + .unwrap() + .unwrap(); + assert!(matches!(event, SessionEvent::Connected(1))); + + // Wait for Data event + let event = tokio::time::timeout(Duration::from_secs(5), rx.recv()) + .await + .unwrap() + .unwrap(); + assert!(matches!(event, SessionEvent::Data(1, _))); + + handle.close().await.unwrap(); + server.await.unwrap(); +} + +// ── Reconfigure ────────────────────────────────────────────────── + +#[tokio::test] +async fn session_reconfigure_changes_pipeline() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + stream.write_all(b"old\n").await.unwrap(); + }); + + let handle = Session::spawn(6, tcp_config(addr)); + let entry = handle.read(5000).await.unwrap(); + assert!(matches!(entry.data, DecodedData::Text(s) if s == "old")); + server.await.unwrap(); + + // Reconfigure to hex pipeline on a new address (don't close first) + let listener2 = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr2 = listener2.local_addr().unwrap().to_string(); + + let server2 = tokio::spawn(async move { + let (mut stream, _) = listener2.accept().await.unwrap(); + stream.write_all(b"ABC\n").await.unwrap(); + }); + + let mut new_cfg = tcp_config(addr2); + new_cfg.pipelines[0].name = "hex".into(); + new_cfg.pipelines[0].decoder = DecoderConfig::Hex { + uppercase: true, + separator: " ".into(), + bytes_per_group: 1, + endian: xserial_core::protocol::Endian::Big, + }; + + handle.reconfigure(new_cfg).await.unwrap(); + let entry = handle.read(5000).await.unwrap(); + assert!(matches!(entry.data, DecodedData::Hex(s) if s == "41 42 43")); + + handle.close().await.unwrap(); + server2.await.unwrap(); +} + +// ── SessionHandle event subscription ───────────────────────────── + +#[tokio::test] +async fn session_handle_subscribe_sees_events() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + stream.write_all(b"event_test\n").await.unwrap(); + }); + + let handle = Session::spawn(7, tcp_config(addr)); + let mut rx = handle.subscribe(); + + // Should see Connected then Data + let e1 = tokio::time::timeout(Duration::from_secs(5), rx.recv()) + .await.unwrap().unwrap(); + assert!(matches!(e1, SessionEvent::Connected(7))); + + let e2 = tokio::time::timeout(Duration::from_secs(5), rx.recv()) + .await.unwrap().unwrap(); + match e2 { + SessionEvent::Data(7, entry) => { + assert_eq!(entry.pipeline_name, "text"); + } + other => panic!("expected Data, got {:?}", other), + } + + handle.close().await.unwrap(); + server.await.unwrap(); +} diff --git a/crates/xserial-core/src/frame/mod.rs b/crates/xserial-core/src/frame/mod.rs index 43c9ac8..0f507ff 100644 --- a/crates/xserial-core/src/frame/mod.rs +++ b/crates/xserial-core/src/frame/mod.rs @@ -13,7 +13,7 @@ pub use crate::protocol::Endian; /// arrived. /// /// [`feed`]: Framer::feed -pub trait Framer { +pub trait Framer: Send { /// Feed a chunk of newly arrived bytes. /// /// Any complete frames that can be extracted are returned. The diff --git a/crates/xserial-core/src/protocol/mod.rs b/crates/xserial-core/src/protocol/mod.rs index 0c39dfc..0cf142b 100644 --- a/crates/xserial-core/src/protocol/mod.rs +++ b/crates/xserial-core/src/protocol/mod.rs @@ -2,10 +2,13 @@ pub mod text; pub mod hex; pub mod plot; -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] pub enum Endian { - Little, + #[default] Big, + Little, } /// Decoded result from a protocol decoder. diff --git a/crates/xserial-core/src/protocol/plot.rs b/crates/xserial-core/src/protocol/plot.rs index d9af2a7..0a6264c 100644 --- a/crates/xserial-core/src/protocol/plot.rs +++ b/crates/xserial-core/src/protocol/plot.rs @@ -1,8 +1,9 @@ use super::{DecodedData, ProtocolDecoder}; use crate::protocol::Endian; +use serde::{Deserialize, Serialize}; /// Numeric type of samples in a plot frame. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] pub enum SampleType { I8, U8, @@ -43,13 +44,11 @@ impl SampleType { } /// Channel layout for multi-channel plot data. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] pub enum PlotFormat { - /// Interleaved: ch0_s0, ch1_s0, ch0_s1, ch1_s1, ... + #[default] Interleaved, - /// Block: ch0_s0, ch0_s1, ..., ch1_s0, ch1_s1, ... Block, - /// XY pairs: x0, y0, x1, y1, ... (equivalent to 2 interleaved channels). XY, } diff --git a/crates/xserial-core/src/protocol/text.rs b/crates/xserial-core/src/protocol/text.rs index 319704b..565cc8d 100644 --- a/crates/xserial-core/src/protocol/text.rs +++ b/crates/xserial-core/src/protocol/text.rs @@ -1,14 +1,13 @@ use super::{DecodedData, ProtocolDecoder}; /// Supported text encodings for the text protocol decoder. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] pub enum TextEncoding { - /// Standard UTF-8. Returns `None` from `decode()` on invalid byte sequences. + #[default] Utf8, - /// ISO-8859-1 / Latin-1. Every byte maps 1:1 to a Unicode code point. - /// This encoding never fails. Latin1, - /// 7-bit ASCII. Returns `None` from `decode()` if any byte has the high bit set. Ascii, }