diff --git a/crates/xserial-client/src/lua/mod.rs b/crates/xserial-client/src/lua/mod.rs index d493ea6..76b0d94 100644 --- a/crates/xserial-client/src/lua/mod.rs +++ b/crates/xserial-client/src/lua/mod.rs @@ -1,38 +1,169 @@ -use std::sync::atomic::{AtomicU64, Ordering}; -use mlua::{Lua, Result as LuaResult, Table, Value, LuaSerdeExt}; +use std::collections::HashMap; +use std::sync::{Arc, Mutex, Weak}; + +use mlua::{Lua, LuaSerdeExt, Result as LuaResult, Table, Value}; + use crate::config::SessionConfig; -use crate::session::Session; +use crate::manager::SessionManager; + mod session_api; -static NEXT_ID: AtomicU64 = AtomicU64::new(1); +#[derive(Clone)] +pub struct LuaRuntime { + manager: SessionManager, + callback_states: Arc>>>, +} + +impl LuaRuntime { + pub fn new(manager: SessionManager) -> Self { + Self { + manager, + callback_states: Arc::new(Mutex::new(HashMap::new())), + } + } + + pub fn manager(&self) -> &SessionManager { + &self.manager + } + + pub fn register_callback_state( + &self, + session_id: u64, + state: &Arc, + ) { + self.callback_states + .lock() + .expect("lua callback state map poisoned") + .insert(session_id, Arc::downgrade(state)); + } + + pub fn unregister_callback_state(&self, session_id: u64) { + self.callback_states + .lock() + .expect("lua callback state map poisoned") + .remove(&session_id); + } + + pub async fn pump_callbacks( + &self, + lua: &Lua, + limit_per_session: Option, + ) -> LuaResult { + let states: Vec<(u64, Arc)> = { + let mut states = self + .callback_states + .lock() + .expect("lua callback state map poisoned"); + let mut live = Vec::with_capacity(states.len()); + states.retain(|session_id, weak| { + if let Some(state) = weak.upgrade() { + live.push((*session_id, state)); + true + } else { + false + } + }); + live + }; + + let mut pumped = 0; + for (_session_id, state) in states { + pumped += state.pump(lua, limit_per_session).await?; + } + Ok(pumped) + } +} pub fn register(lua: &Lua) -> LuaResult<()> { + register_with_manager(lua, SessionManager::new()) +} + +pub fn register_with_manager(lua: &Lua, manager: SessionManager) -> LuaResult<()> { + let runtime = LuaRuntime::new(manager); 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("open", { + let runtime = runtime.clone(); + lua.create_async_function(move |lua, config: Table| { + let runtime = runtime.clone(); + async move { + let cfg: SessionConfig = lua.from_value(Value::Table(config))?; + let handle = runtime.manager().create(cfg); + let userdata = session_api::LuaSessionHandle::new(handle, runtime); + lua.create_userdata(userdata) + } + })? + })?; - 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( + "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("sleep", { + let runtime = runtime.clone(); + lua.create_async_function(move |lua, ms: u64| { + let runtime = runtime.clone(); + async move { + tokio::time::sleep(std::time::Duration::from_millis(ms)).await; + let _ = runtime.pump_callbacks(&lua, None).await?; + Ok(()) + } + })? + })?; - xserial.set("log", lua.create_function(|_, msg: String| { - tracing::info!("[lua] {}", msg); - Ok(()) - })?)?; + xserial.set("poll", { + let runtime = runtime.clone(); + lua.create_async_function(move |lua, limit_per_session: Option| { + let runtime = runtime.clone(); + async move { runtime.pump_callbacks(&lua, limit_per_session).await } + })? + })?; + + xserial.set( + "log", + lua.create_function(|_, msg: String| { + tracing::info!("[lua] {}", msg); + Ok(()) + })?, + )?; lua.globals().set("xserial", xserial)?; Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + use mlua::Value; + + #[test] + fn register_exposes_expected_api() { + let lua = Lua::new(); + register(&lua).unwrap(); + + let xserial: Table = lua.globals().get("xserial").unwrap(); + for name in ["open", "list_ports", "sleep", "poll", "log"] { + let value: Value = xserial.get(name).unwrap(); + assert!( + matches!(value, Value::Function(_)), + "{name} should be a function" + ); + } + } + + #[test] + fn register_with_manager_uses_supplied_manager() { + let lua = Lua::new(); + let manager = SessionManager::new(); + register_with_manager(&lua, manager.clone()).unwrap(); + + let xserial: Table = lua.globals().get("xserial").unwrap(); + let open: Value = xserial.get("open").unwrap(); + assert!(matches!(open, Value::Function(_))); + assert_eq!(manager.count(), 0); + } +} diff --git a/crates/xserial-client/src/lua/session_api.rs b/crates/xserial-client/src/lua/session_api.rs index 11c9862..d01e5d2 100644 --- a/crates/xserial-client/src/lua/session_api.rs +++ b/crates/xserial-client/src/lua/session_api.rs @@ -1,78 +1,535 @@ -use mlua::{UserData, UserDataMethods, Variadic, LuaSerdeExt}; -use xserial_core::protocol::DecodedData; -use crate::session::SessionHandle; +use std::collections::HashMap; +use std::sync::{ + Arc, Mutex, + atomic::{AtomicU64, Ordering}, +}; +use std::time::Duration; -pub struct LuaSessionHandle { pub handle: SessionHandle } +use mlua::{Function, Lua, LuaSerdeExt, RegistryKey, UserData, UserDataMethods, Value, Variadic}; +use tokio::sync::mpsc; +use tokio::task::JoinHandle; +use xserial_core::protocol::DecodedData; + +use crate::config::SessionConfig; +use crate::lua::LuaRuntime; +use crate::session::{SessionEvent, SessionHandle}; + +pub struct LuaSessionHandle { + handle: SessionHandle, + runtime: LuaRuntime, + callbacks: Arc, +} + +pub struct CallbackState { + queue_tx: mpsc::UnboundedSender, + queue_rx: Mutex>, + subscriptions: Mutex>, + next_subscription_id: AtomicU64, + relay_task: Mutex>>, +} + +struct Subscription { + callback: Arc, + filters: Vec, +} + +#[derive(Clone)] +enum QueuedEvent { + Data { + pipeline_name: String, + data: QueuedData, + }, +} + +#[derive(Clone)] +enum QueuedData { + Text(String), + Hex(String), + Binary(Vec), + Plot { + channels: Vec>, + sample_count: usize, + }, +} impl LuaSessionHandle { - pub fn new(handle: SessionHandle) -> Self { Self { handle } } + pub fn new(handle: SessionHandle, runtime: LuaRuntime) -> Self { + let callbacks = Arc::new(CallbackState::new(handle.clone())); + runtime.register_callback_state(handle.id(), &callbacks); + Self { + handle, + runtime, + callbacks, + } + } +} + +impl Drop for LuaSessionHandle { + fn drop(&mut self) { + self.runtime.unregister_callback_state(self.handle.id()); + } +} + +impl CallbackState { + pub fn new(handle: SessionHandle) -> Self { + let (queue_tx, queue_rx) = mpsc::unbounded_channel(); + let state = Self { + queue_tx, + queue_rx: Mutex::new(queue_rx), + subscriptions: Mutex::new(HashMap::new()), + next_subscription_id: AtomicU64::new(1), + relay_task: Mutex::new(None), + }; + state.ensure_relay(handle); + state + } + + fn ensure_relay(&self, handle: SessionHandle) { + let mut relay_task = self + .relay_task + .lock() + .expect("lua relay task mutex poisoned"); + if relay_task.is_some() { + return; + } + + let tx = self.queue_tx.clone(); + let mut rx = handle.subscribe(); + *relay_task = Some(tokio::spawn(async move { + loop { + match rx.recv().await { + Ok(SessionEvent::Data(_, entry)) => { + let queued = match entry.data { + DecodedData::Text(text) => QueuedEvent::Data { + pipeline_name: entry.pipeline_name, + data: QueuedData::Text(text), + }, + DecodedData::Hex(hex) => QueuedEvent::Data { + pipeline_name: entry.pipeline_name, + data: QueuedData::Hex(hex), + }, + DecodedData::Binary(bytes) => QueuedEvent::Data { + pipeline_name: entry.pipeline_name, + data: QueuedData::Binary(bytes), + }, + DecodedData::Plot(frame) => QueuedEvent::Data { + pipeline_name: entry.pipeline_name, + data: QueuedData::Plot { + sample_count: frame.sample_count(), + channels: frame.channels, + }, + }, + }; + if tx.send(queued).is_err() { + break; + } + } + Ok(SessionEvent::Closed(_)) => break, + Ok(_) => {} + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue, + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + } + } + })); + } + + pub fn add_subscription(&self, callback: RegistryKey, filters: Vec) -> u64 { + let id = self.next_subscription_id.fetch_add(1, Ordering::SeqCst); + self.subscriptions + .lock() + .expect("lua callback subscriptions mutex poisoned") + .insert( + id, + Subscription { + callback: Arc::new(callback), + filters, + }, + ); + id + } + + pub fn remove_subscription(&self, lua: &Lua, id: u64) -> mlua::Result { + let removed = self + .subscriptions + .lock() + .expect("lua callback subscriptions mutex poisoned") + .remove(&id); + if let Some(subscription) = removed { + if let Ok(key) = Arc::try_unwrap(subscription.callback) { + lua.remove_registry_value(key)?; + } + Ok(true) + } else { + Ok(false) + } + } + + pub async fn pump(&self, lua: &Lua, limit: Option) -> mlua::Result { + let max = limit.unwrap_or(usize::MAX); + let mut drained = Vec::new(); + { + let mut rx = self + .queue_rx + .lock() + .expect("lua callback queue mutex poisoned"); + while drained.len() < max { + match rx.try_recv() { + Ok(event) => drained.push(event), + Err(tokio::sync::mpsc::error::TryRecvError::Empty) => break, + Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => break, + } + } + } + + let mut called = 0; + for event in drained { + let callbacks = self.matching_callbacks(&event); + for callback in callbacks { + let function: Function = lua.registry_value(&callback)?; + match &event { + QueuedEvent::Data { + pipeline_name, + data, + } => { + let payload = queued_data_to_value(lua, data)?; + function + .call_async::<()>((pipeline_name.clone(), payload)) + .await?; + called += 1; + } + } + } + } + + Ok(called) + } + + fn matching_callbacks(&self, event: &QueuedEvent) -> Vec> { + let subscriptions = self + .subscriptions + .lock() + .expect("lua callback subscriptions mutex poisoned"); + subscriptions + .values() + .filter(|subscription| match event { + QueuedEvent::Data { pipeline_name, .. } => { + subscription.filters.is_empty() + || subscription.filters.iter().any(|f| f == pipeline_name) + } + }) + .map(|subscription| Arc::clone(&subscription.callback)) + .collect() + } +} + +impl Drop for CallbackState { + fn drop(&mut self) { + if let Ok(mut relay_task) = self.relay_task.lock() { + if let Some(task) = relay_task.take() { + task.abort(); + } + } + } } 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 + 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)) - } + Some(entry) => Ok(Some(decoded_entry_to_table( + &lua, + entry.pipeline_name, + entry.data, + )?)), None => Ok(None), } }); + methods.add_async_method( + "next_event", + |lua, this, timeout_ms: Option| async move { + let mut rx = this.handle.subscribe(); + let timeout = Duration::from_millis(timeout_ms.unwrap_or(1000)); + let next = tokio::time::timeout(timeout, async move { + loop { + match rx.recv().await { + Ok(event) => return Some(event), + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue, + Err(tokio::sync::broadcast::error::RecvError::Closed) => return None, + } + } + }) + .await + .ok() + .flatten(); + + match next { + Some(event) => Ok(Some(session_event_to_table(&lua, event)?)), + 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) + let cfg: SessionConfig = lua.from_value(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(()) + methods.add_async_method( + "on_data", + |lua, this, (callback, pipelines): (Function, Variadic)| async move { + let filters: Vec = pipelines.into_iter().collect(); + let key = lua.create_registry_value(callback)?; + Ok(this.callbacks.add_subscription(key, filters)) + }, + ); + + methods.add_method("off", |lua, this, token: u64| { + this.callbacks.remove_subscription(lua, token) }); + + methods.add_async_method( + "pump_events", + |lua, this, limit: Option| async move { this.callbacks.pump(&lua, limit).await }, + ); + } +} + +fn decoded_entry_to_table( + lua: &Lua, + pipeline_name: String, + data: DecodedData, +) -> mlua::Result { + let table = lua.create_table()?; + table.set("type", "data")?; + table.set("pipeline", pipeline_name)?; + match data { + DecodedData::Text(text) => { + table.set("kind", "text")?; + table.set("data", text)?; + } + DecodedData::Hex(hex) => { + table.set("kind", "hex")?; + table.set("data", hex)?; + } + DecodedData::Binary(bytes) => { + table.set("kind", "binary")?; + table.set("data", lua.create_string(&bytes)?)?; + } + DecodedData::Plot(frame) => { + table.set("kind", "plot")?; + table.set("channels", channels_to_table(lua, &frame.channels)?)?; + table.set("sample_count", frame.sample_count())?; + } + } + Ok(table) +} + +fn session_event_to_table(lua: &Lua, event: SessionEvent) -> mlua::Result { + match event { + SessionEvent::Connected(session_id) => { + let table = lua.create_table()?; + table.set("type", "connected")?; + table.set("session_id", session_id)?; + Ok(table) + } + SessionEvent::Disconnected(session_id) => { + let table = lua.create_table()?; + table.set("type", "disconnected")?; + table.set("session_id", session_id)?; + Ok(table) + } + SessionEvent::Closed(session_id) => { + let table = lua.create_table()?; + table.set("type", "closed")?; + table.set("session_id", session_id)?; + Ok(table) + } + SessionEvent::Error(session_id, error) => { + let table = lua.create_table()?; + table.set("type", "error")?; + table.set("session_id", session_id)?; + table.set("error", error)?; + Ok(table) + } + SessionEvent::Data(session_id, entry) => { + let table = decoded_entry_to_table(lua, entry.pipeline_name, entry.data)?; + table.set("session_id", session_id)?; + Ok(table) + } + } +} + +fn queued_data_to_value(lua: &Lua, data: &QueuedData) -> mlua::Result { + match data { + QueuedData::Text(text) => Ok(Value::String(lua.create_string(text)?)), + QueuedData::Hex(hex) => Ok(Value::String(lua.create_string(hex)?)), + QueuedData::Binary(bytes) => Ok(Value::String(lua.create_string(bytes)?)), + QueuedData::Plot { + channels, + sample_count, + } => { + let table = lua.create_table()?; + table.set("kind", "plot")?; + table.set("channels", channels_to_table(lua, channels)?)?; + table.set("sample_count", *sample_count)?; + Ok(Value::Table(table)) + } + } +} + +fn channels_to_table(lua: &Lua, channels: &[Vec]) -> mlua::Result { + let table = lua.create_table()?; + for (index, channel) in channels.iter().enumerate() { + table.set( + index + 1, + lua.create_sequence_from(channel.iter().copied())?, + )?; + } + Ok(table) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::event::DecodedEntry; + use crate::session::SessionId; + + fn test_callback_state() -> Arc { + let (queue_tx, queue_rx) = mpsc::unbounded_channel(); + Arc::new(CallbackState { + queue_tx, + queue_rx: Mutex::new(queue_rx), + subscriptions: Mutex::new(HashMap::new()), + next_subscription_id: AtomicU64::new(1), + relay_task: Mutex::new(None), + }) + } + + fn push_data_event(state: &CallbackState, pipeline_name: &str, data: QueuedData) { + state + .queue_tx + .send(QueuedEvent::Data { + pipeline_name: pipeline_name.to_string(), + data, + }) + .unwrap(); + } + + #[tokio::test] + async fn callback_pump_invokes_matching_subscribers() { + let lua = Lua::new(); + let received = Arc::new(Mutex::new(Vec::<(String, String)>::new())); + lua.set_app_data(received.clone()); + + let callback = lua + .create_function(|lua, (name, data): (String, String)| { + let store = lua + .app_data_ref::>>>() + .unwrap(); + store.lock().unwrap().push((name, data)); + Ok(()) + }) + .unwrap(); + + let state = test_callback_state(); + let token = state.add_subscription(lua.create_registry_value(callback).unwrap(), vec![]); + assert_eq!(token, 1); + + push_data_event(&state, "text", QueuedData::Text("hello".into())); + let called = state.pump(&lua, None).await.unwrap(); + + assert_eq!(called, 1); + assert_eq!( + received.lock().unwrap().as_slice(), + &[("text".to_string(), "hello".to_string())] + ); + } + + #[tokio::test] + async fn callback_pump_respects_pipeline_filters_and_off() { + let lua = Lua::new(); + let calls = Arc::new(Mutex::new(0usize)); + lua.set_app_data(calls.clone()); + + let callback = lua + .create_function(|lua, (_name, _data): (String, String)| { + let calls = lua.app_data_ref::>>().unwrap(); + *calls.lock().unwrap() += 1; + Ok(()) + }) + .unwrap(); + + let state = test_callback_state(); + let token = state.add_subscription( + lua.create_registry_value(callback).unwrap(), + vec!["hex".into()], + ); + + push_data_event(&state, "text", QueuedData::Text("ignored".into())); + assert_eq!(state.pump(&lua, None).await.unwrap(), 0); + assert_eq!(*calls.lock().unwrap(), 0); + + assert!(state.remove_subscription(&lua, token).unwrap()); + push_data_event(&state, "hex", QueuedData::Hex("41".into())); + assert_eq!(state.pump(&lua, None).await.unwrap(), 0); + assert_eq!(*calls.lock().unwrap(), 0); + } + + #[test] + fn decoded_entry_to_table_has_expected_shape() { + let lua = Lua::new(); + let table = + decoded_entry_to_table(&lua, "text".into(), DecodedData::Text("payload".into())) + .unwrap(); + + assert_eq!(table.get::("type").unwrap(), "data"); + assert_eq!(table.get::("pipeline").unwrap(), "text"); + assert_eq!(table.get::("kind").unwrap(), "text"); + assert_eq!(table.get::("data").unwrap(), "payload"); + } + + #[test] + fn session_event_to_table_maps_variants() { + let lua = Lua::new(); + + let error_table = + session_event_to_table(&lua, SessionEvent::Error(7 as SessionId, "boom".into())) + .unwrap(); + assert_eq!(error_table.get::("type").unwrap(), "error"); + assert_eq!(error_table.get::("session_id").unwrap(), 7); + assert_eq!(error_table.get::("error").unwrap(), "boom"); + + let data_table = session_event_to_table( + &lua, + SessionEvent::Data( + 9, + DecodedEntry { + pipeline_name: "hex".into(), + data: DecodedData::Hex("41 42".into()), + }, + ), + ) + .unwrap(); + assert_eq!(data_table.get::("type").unwrap(), "data"); + assert_eq!(data_table.get::("session_id").unwrap(), 9); + assert_eq!(data_table.get::("pipeline").unwrap(), "hex"); + assert_eq!(data_table.get::("kind").unwrap(), "hex"); + assert_eq!(data_table.get::("data").unwrap(), "41 42"); } } diff --git a/crates/xserial-client/src/manager.rs b/crates/xserial-client/src/manager.rs index 2acbbd6..834a07a 100644 --- a/crates/xserial-client/src/manager.rs +++ b/crates/xserial-client/src/manager.rs @@ -1,39 +1,220 @@ use std::collections::HashMap; +use std::sync::{ + Arc, RwLock, + atomic::{AtomicU64, Ordering}, +}; + +use tokio::sync::broadcast; +use tracing::info; + use crate::config::SessionConfig; use crate::session::{Session, SessionEvent, SessionHandle, SessionId}; +#[derive(Clone)] pub struct SessionManager { - sessions: HashMap, - event_tx: tokio::sync::broadcast::Sender, + inner: Arc, +} + +struct SessionManagerInner { + sessions: RwLock>, + event_tx: broadcast::Sender, + next_session_id: AtomicU64, } impl SessionManager { pub fn new() -> Self { - let (event_tx, _) = tokio::sync::broadcast::channel(256); - Self { sessions: HashMap::new(), event_tx } + let (event_tx, _) = broadcast::channel(256); + Self { + inner: Arc::new(SessionManagerInner { + sessions: RwLock::new(HashMap::new()), + event_tx, + next_session_id: AtomicU64::new(1), + }), + } } - pub fn create(&mut self, config: SessionConfig) -> SessionHandle { - let id = self.sessions.len() as u64 + 1; + pub fn create(&self, config: SessionConfig) -> SessionHandle { + let id = self.inner.next_session_id.fetch_add(1, Ordering::SeqCst); + + let pipeline_names: Vec<&str> = config.pipelines.iter().map(|p| p.name.as_str()).collect(); + info!( + session_id = id, + transport = ?config.transport, + pipelines = ?pipeline_names, + history_limit = config.history_limit, + auto_reconnect = config.auto_reconnect, + "session created" + ); + let handle = Session::spawn(id, config); - let event_relay = self.event_tx.clone(); - let mut event_rx = handle.subscribe(); + let stored = handle.clone(); + let mut session_events = handle.subscribe(); + let relay = self.inner.event_tx.clone(); + 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, + loop { + match session_events.recv().await { + Ok(event) => { + let _ = relay.send(event); + } + Err(broadcast::error::RecvError::Lagged(_)) => continue, + Err(broadcast::error::RecvError::Closed) => break, + } + } }); + + self.inner + .sessions + .write() + .expect("session store poisoned") + .insert(id, stored); + 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() } + pub fn remove(&self, id: SessionId) { + let removed = self + .inner + .sessions + .write() + .expect("session store poisoned") + .remove(&id); + + if let Some(handle) = removed { + info!(session_id = id, "session removed"); + tokio::spawn(async move { + let _ = handle.close().await; + handle.join().await; + }); + } + } + + pub fn get(&self, id: SessionId) -> Option { + self.inner + .sessions + .read() + .expect("session store poisoned") + .get(&id) + .cloned() + } + + pub fn subscribe(&self) -> broadcast::Receiver { + self.inner.event_tx.subscribe() + } + + pub fn list_ids(&self) -> Vec { + self.inner + .sessions + .read() + .expect("session store poisoned") + .keys() + .copied() + .collect() + } + + pub fn count(&self) -> usize { + self.inner + .sessions + .read() + .expect("session store poisoned") + .len() + } } impl Default for SessionManager { - fn default() -> Self { Self::new() } + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn create_keeps_monotonic_ids_after_remove() { + let manager = SessionManager::new(); + let first = manager.create(SessionConfig::default()); + assert_eq!(first.id(), 1); + manager.remove(first.id()); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + let second = manager.create(SessionConfig::default()); + assert_eq!(second.id(), 2); + } + + #[tokio::test] + async fn create_multiple_sessions() { + let manager = SessionManager::new(); + let h1 = manager.create(SessionConfig::default()); + let h2 = manager.create(SessionConfig::default()); + let h3 = manager.create(SessionConfig::default()); + assert_eq!(h1.id(), 1); + assert_eq!(h2.id(), 2); + assert_eq!(h3.id(), 3); + } + + #[tokio::test] + async fn list_ids_returns_all() { + let manager = SessionManager::new(); + manager.create(SessionConfig::default()); + manager.create(SessionConfig::default()); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + let ids = manager.list_ids(); + assert_eq!(ids.len(), 2); + assert!(ids.contains(&1)); + assert!(ids.contains(&2)); + } + + #[tokio::test] + async fn get_returns_correct_session() { + let manager = SessionManager::new(); + let h = manager.create(SessionConfig::default()); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + let found = manager.get(h.id()); + assert!(found.is_some()); + assert_eq!(found.unwrap().id(), h.id()); + } + + #[tokio::test] + async fn get_returns_none_for_missing() { + let manager = SessionManager::new(); + assert!(manager.get(999).is_none()); + } + + #[tokio::test] + async fn remove_decrements_count() { + let manager = SessionManager::new(); + let h = manager.create(SessionConfig::default()); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + manager.remove(h.id()); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + assert!(manager.get(h.id()).is_none()); + } + + #[tokio::test] + async fn subscribe_receives_events() { + let manager = SessionManager::new(); + let mut rx = manager.subscribe(); + manager.create(SessionConfig::default()); + let event = tokio::time::timeout(std::time::Duration::from_secs(10), rx.recv()).await; + if let Ok(Ok(event)) = event { + assert!(matches!( + event, + SessionEvent::Connected(_) | SessionEvent::Error(_, _) + )); + } + } + + #[tokio::test] + async fn count_starts_at_zero() { + let manager = SessionManager::new(); + assert_eq!(manager.count(), 0); + assert!(manager.list_ids().is_empty()); + } + + #[tokio::test] + async fn default_creates_empty_manager() { + let manager = SessionManager::default(); + assert_eq!(manager.count(), 0); + } } diff --git a/crates/xserial-client/src/session.rs b/crates/xserial-client/src/session.rs index 3390531..da7bc93 100644 --- a/crates/xserial-client/src/session.rs +++ b/crates/xserial-client/src/session.rs @@ -1,8 +1,14 @@ -use tokio::io::AsyncReadExt; +use std::sync::{ + Arc, + atomic::{AtomicBool, Ordering}, +}; +use std::time::Duration; + +use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::select; -use tokio::sync::{broadcast, mpsc}; +use tokio::sync::{Mutex, broadcast, mpsc}; use tokio::task::JoinHandle; -use tracing::{error, info, warn}; +use tracing::{debug, error, info, warn}; use xserial_core::pipeline::{MultiPipeline, Pipeline}; use xserial_core::transport::Connection; @@ -23,53 +29,126 @@ pub enum SessionEvent { Closed(SessionId), } +struct SessionHandleInner { + id: SessionId, + cmd_tx: mpsc::Sender, + event_tx: broadcast::Sender, + task: Mutex>>, + close_requested: AtomicBool, +} + +impl SessionHandleInner { + fn new( + id: SessionId, + cmd_tx: mpsc::Sender, + event_tx: broadcast::Sender, + task: JoinHandle<()>, + ) -> Self { + Self { + id, + cmd_tx, + event_tx, + task: Mutex::new(Some(task)), + close_requested: AtomicBool::new(false), + } + } + + async fn request_close(&self) -> Result<(), String> { + self.close_requested.store(true, Ordering::SeqCst); + self.cmd_tx + .send(SessionCmd::Close) + .await + .map_err(|_| String::from("session closed")) + } + + fn request_close_nonblocking(&self) { + self.close_requested.store(true, Ordering::SeqCst); + let _ = self.cmd_tx.try_send(SessionCmd::Close); + } +} + +#[derive(Clone)] pub struct SessionHandle { pub id: SessionId, - pub(crate) cmd_tx: mpsc::Sender, - pub(crate) event_tx: broadcast::Sender, - pub(crate) _task: Option>, + inner: Arc, } impl SessionHandle { - pub fn id(&self) -> SessionId { self.id } + fn new(inner: Arc) -> Self { + Self { + id: inner.id, + inner, + } + } + + 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()) + self.inner + .cmd_tx + .send(SessionCmd::Send(data)) + .await + .map_err(|_| String::from("session closed")) } pub async fn close(&self) -> Result<(), String> { - self.cmd_tx.send(SessionCmd::Close).await.map_err(|_| "session closed".into()) + self.inner.request_close().await } pub async fn reconfigure(&self, config: SessionConfig) -> Result<(), String> { - self.cmd_tx.send(SessionCmd::Reconfigure(config)).await.map_err(|_| "session closed".into()) + self.inner + .cmd_tx + .send(SessionCmd::Reconfigure(config)) + .await + .map_err(|_| String::from("session closed")) } 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); + let mut rx = self.inner.event_tx.subscribe(); + let deadline = Duration::from_millis(timeout_ms); tokio::time::timeout(deadline, async { loop { match rx.recv().await { - Ok(SessionEvent::Data(_, e)) => return Some(e), + Ok(SessionEvent::Data(_, entry)) => return Some(entry), Ok(SessionEvent::Error(_, _) | SessionEvent::Closed(_)) => return None, + Ok(SessionEvent::Connected(_) | SessionEvent::Disconnected(_)) => {} Err(broadcast::error::RecvError::Lagged(_)) => continue, Err(broadcast::error::RecvError::Closed) => return None, - _ => continue, } } - }).await.ok().flatten() + }) + .await + .ok() + .flatten() } pub fn subscribe(&self) -> broadcast::Receiver { - self.event_tx.subscribe() + self.inner.event_tx.subscribe() + } + + pub(crate) async fn join(&self) { + let task = { + let mut slot = self.inner.task.lock().await; + slot.take() + }; + if let Some(task) = task { + let _ = task.await; + } } } 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(); } + if Arc::strong_count(&self.inner) == 1 { + self.inner.request_close_nonblocking(); + if let Ok(mut task) = self.inner.task.try_lock() { + if let Some(task) = task.take() { + task.abort(); + } + } + } } } @@ -81,6 +160,7 @@ pub struct Session { history: RingBuffer, cmd_rx: mpsc::Receiver, event_tx: broadcast::Sender, + close_requested: bool, } impl Session { @@ -88,142 +168,380 @@ impl Session { let (cmd_tx, cmd_rx) = mpsc::channel(32); let (event_tx, _) = broadcast::channel(256); let mut session = Self { - id, conn: None, + id, pipeline: Self::build_pipeline(&config), history: RingBuffer::new(config.history_limit), - config, cmd_rx, event_tx: event_tx.clone(), + config, + conn: None, + cmd_rx, + event_tx: event_tx.clone(), + close_requested: false, }; let task = tokio::spawn(async move { session.run().await }); - SessionHandle { id, cmd_tx, event_tx, _task: Some(task) } + let inner = Arc::new(SessionHandleInner::new(id, cmd_tx, event_tx, task)); + SessionHandle::new(inner) } 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(), + let mut pipelines = MultiPipeline::new(); + for pipeline in &config.pipelines { + pipelines.add(Pipeline::new( + pipeline.name.clone(), + pipeline.framer.clone().build(), + pipeline.decoder.clone().build(), )); } - mp + pipelines } 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)); + if let Err(err) = self.connect().await { + warn!(session_id = self.id, error = %err, "initial connect failed"); + self.emit_error(format!("connect failed: {err}")); + self.emit_closed(); + info!(session_id = self.id, "session ended"); 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, + cmd = self.cmd_rx.recv() => { + if !self.handle_command(cmd).await { + 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))); - } - } - } + read = try_read(&mut self.conn, &mut self.pipeline) => { + if !self.handle_read_result(read).await { + break; } } } } + + self.disconnect(false).await; + self.emit_closed(); info!(session_id = self.id, "session ended"); } + + async fn handle_command(&mut self, cmd: Option) -> bool { + match cmd { + Some(SessionCmd::Close) => { + self.close_requested = true; + false + } + Some(SessionCmd::Send(data)) => { + self.handle_send(data).await; + true + } + Some(SessionCmd::Reconfigure(config)) => { + self.handle_reconfigure(config).await; + true + } + None => false, + } + } + + async fn handle_send(&mut self, data: Vec) { + debug!(session_id = self.id, len = data.len(), "sending data"); + match self.conn.as_mut() { + Some(conn) => { + if let Err(err) = conn.write_all(&data).await { + self.emit_error(format!("write: {err}")); + } + } + None => self.emit_error(String::from("write: session not connected")), + } + } + + async fn handle_reconfigure(&mut self, config: SessionConfig) { + info!(session_id = self.id, "reconfiguring session"); + self.disconnect(true).await; + self.config = config; + self.pipeline = Self::build_pipeline(&self.config); + self.history = RingBuffer::new(self.config.history_limit); + + if let Err(err) = self.connect().await { + self.emit_error(format!("reconnect: {err}")); + } + } + + async fn handle_read_result( + &mut self, + result: Result, std::io::Error>, + ) -> bool { + match result { + Ok(entries) => { + for entry in entries { + self.history.push(entry.clone()); + let _ = self.event_tx.send(SessionEvent::Data(self.id, entry)); + } + true + } + Err(err) => { + error!(session_id = self.id, error = %err, "read error"); + self.emit_error(err.to_string()); + self.disconnect(true).await; + + if self.config.auto_reconnect && !self.close_requested { + warn!(session_id = self.id, "auto-reconnecting"); + tokio::time::sleep(Duration::from_secs(1)).await; + if let Err(err) = self.connect().await { + self.emit_error(format!("reconnect: {err}")); + } + true + } else { + false + } + } + } + } + + async fn connect(&mut self) -> Result<(), std::io::Error> { + let transport = self.config.transport.clone(); + debug!(session_id = self.id, ?transport, "connecting session"); + let mut conn = Connection::new(transport); + conn.connect().await.map_err(to_io_error)?; + self.conn = Some(conn); + let _ = self.event_tx.send(SessionEvent::Connected(self.id)); + info!(session_id = self.id, "session connected"); + Ok(()) + } + + async fn disconnect(&mut self, emit_disconnected: bool) { + if let Some(mut conn) = self.conn.take() { + let _ = conn.disconnect().await; + if emit_disconnected { + let _ = self.event_tx.send(SessionEvent::Disconnected(self.id)); + } + info!(session_id = self.id, "session disconnected"); + } + self.pipeline.reset(); + } + + fn emit_error(&self, message: String) { + let _ = self.event_tx.send(SessionEvent::Error(self.id, message)); + } + + fn emit_closed(&self) { + let _ = self.event_tx.send(SessionEvent::Closed(self.id)); + } } 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()) + let conn = match conn.as_mut() { + Some(conn) => conn, + None => { + tokio::time::sleep(Duration::from_millis(20)).await; + return Ok(Vec::new()); } - Err(e) => Err(e), + }; + + let mut buf = [0u8; 4096]; + match conn.read(&mut buf).await { + Ok(0) => Err(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + "connection closed", + )), + Ok(n) => Ok(pipeline + .feed(&buf[..n]) + .into_iter() + .map(|result| DecodedEntry { + pipeline_name: result.pipeline_name, + data: result.data, + }) + .collect()), + Err(err) => Err(err), } } -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(()) +fn to_io_error(err: xserial_core::error::Error) -> std::io::Error { + std::io::Error::new(std::io::ErrorKind::Other, err.to_string()) } -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)); +#[cfg(test)] +mod tests { + use super::*; + use crate::config::{DecoderConfig, FramerConfig, PipelineConfig}; + + fn tcp_config() -> SessionConfig { + SessionConfig { + transport: xserial_core::transport::TransportConfig::Tcp { + addr: "127.0.0.1:1".into(), + }, + pipelines: vec![PipelineConfig { + name: "text".into(), + framer: FramerConfig::Line { + strip_cr: true, + max_line_len: 65536, + }, + decoder: DecoderConfig::Text { + encoding: xserial_core::protocol::text::TextEncoding::Utf8, + }, + }], + ..Default::default() + } + } + + #[test] + fn session_id_is_u64() { + let _id: SessionId = 42; + } + + #[test] + fn session_event_debug() { + let e1 = SessionEvent::Connected(1); + let e2 = SessionEvent::Disconnected(1); + let e3 = SessionEvent::Closed(1); + let e4 = SessionEvent::Error(1, "oops".into()); + let _ = format!("{:?}", e1); + let _ = format!("{:?}", e2); + let _ = format!("{:?}", e3); + let _ = format!("{:?}", e4); + } + + #[test] + fn session_event_clone() { + let e1 = SessionEvent::Connected(1); + let e2 = e1.clone(); + match (e1, e2) { + (SessionEvent::Connected(a), SessionEvent::Connected(b)) => assert_eq!(a, b), + _ => panic!(), + } + } + + #[tokio::test] + async fn spawn_and_close_session() { + let handle = Session::spawn(1, tcp_config()); + let mut rx = handle.subscribe(); + let event = tokio::time::timeout(Duration::from_secs(5), rx.recv()) + .await + .expect("timeout"); + if let Ok(event) = event { + assert!(matches!( + event, + SessionEvent::Error(_, _) | SessionEvent::Connected(_) + )); + } + let _ = handle.close().await; + handle.join().await; + } + + #[tokio::test] + async fn handle_send_to_dead_session() { + let handle = Session::spawn(2, tcp_config()); + tokio::time::sleep(Duration::from_millis(200)).await; + let _ = handle.send(b"data".to_vec()).await; + handle.join().await; + } + + #[tokio::test] + async fn handle_close_twice_is_idempotent() { + let handle = Session::spawn(3, tcp_config()); + let _ = handle.close().await; + let _ = handle.close().await; + handle.join().await; + } + + #[tokio::test] + async fn handle_read_after_close() { + let handle = Session::spawn(4, tcp_config()); + let _ = handle.close().await; + let result = handle.read(100).await; + assert!(result.is_none()); + handle.join().await; + } + + #[tokio::test] + async fn handle_read_timeout_returns_none() { + let handle = Session::spawn(5, tcp_config()); + let result = handle.read(100).await; + assert!(result.is_none()); + let _ = handle.close().await; + handle.join().await; + } + + #[tokio::test] + async fn handle_subscribe_receives_events() { + let handle = Session::spawn(6, tcp_config()); + let mut rx = handle.subscribe(); + let event = tokio::time::timeout(Duration::from_secs(5), rx.recv()) + .await + .expect("timeout"); + if let Ok(event) = event { + assert!(matches!( + event, + SessionEvent::Error(_, _) | SessionEvent::Connected(_) + )); + } + let _ = handle.close().await; + handle.join().await; + } + + #[tokio::test] + async fn handle_id_matches() { + let handle = Session::spawn(99, tcp_config()); + assert_eq!(handle.id(), 99); + let _ = handle.close().await; + handle.join().await; + } + + #[tokio::test] + async fn handle_reconfigure_triggers_reconnect() { + let handle = Session::spawn(7, tcp_config()); + let new_cfg = tcp_config(); + let _ = handle.reconfigure(new_cfg).await; + let _ = handle.close().await; + handle.join().await; + } + + #[tokio::test] + async fn handle_drop_closes_session() { + let handle = Session::spawn(8, tcp_config()); + let mut rx = handle.subscribe(); + drop(handle); + let mut closed = false; + while let Ok(Ok(event)) = tokio::time::timeout(Duration::from_millis(200), rx.recv()).await + { + if matches!(event, SessionEvent::Closed(_)) { + closed = true; + break; + } + } + let _ = closed; + } + + #[tokio::test] + async fn multiple_subscribers_all_receive() { + let handle = Session::spawn(9, tcp_config()); + let mut rx1 = handle.subscribe(); + let mut rx2 = handle.subscribe(); + + let e1 = tokio::time::timeout(Duration::from_secs(5), rx1.recv()) + .await + .unwrap(); + if let Ok(e1) = e1 { + assert!(matches!( + e1, + SessionEvent::Connected(9) | SessionEvent::Error(9, _) + )); + } + + let e2 = tokio::time::timeout(Duration::from_secs(1), rx2.recv()) + .await + .unwrap(); + if let Ok(e2) = e2 { + assert!(matches!( + e2, + SessionEvent::Connected(9) | SessionEvent::Error(9, _) + )); + } + + let _ = handle.close().await; + handle.join().await; } } diff --git a/crates/xserial-client/tests/lua_tests.rs b/crates/xserial-client/tests/lua_tests.rs index 6b802c5..a854c50 100644 --- a/crates/xserial-client/tests/lua_tests.rs +++ b/crates/xserial-client/tests/lua_tests.rs @@ -1,4 +1,5 @@ use std::sync::Once; +use std::time::Duration; use mlua::Lua; use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -15,6 +16,26 @@ fn init_tracing() { }); } +fn text_pipeline_lua() -> &'static str { + r#" + { + name = "text", + framer = { Line = {} }, + decoder = { Text = {} } + } + "# +} + +fn hex_pipeline_lua() -> &'static str { + r#" + { + name = "hex", + framer = { Line = {} }, + decoder = { Hex = { uppercase = true } } + } + "# +} + // ── xserial.sleep / xserial.log ────────────────────────────────── #[tokio::test] @@ -27,7 +48,7 @@ async fn lua_sleep_and_log() { xserial.log("test start") xserial.sleep(50) xserial.log("test end") - "# + "#, ) .exec_async() .await @@ -66,6 +87,7 @@ async fn lua_session_open_send_read_close() { let n = stream.read(&mut buf).await.unwrap(); assert_eq!(&buf[..n], b"hello\n"); stream.write_all(b"world\n").await.unwrap(); + tokio::time::sleep(Duration::from_millis(200)).await; }); let lua = Lua::new(); @@ -75,11 +97,7 @@ async fn lua_session_open_send_read_close() { r#" local sess = xserial.open({{ transport = {{ Tcp = {{ addr = "{}" }} }}, - pipelines = {{{{ - name = "text", - framer = {{ Line = {{}} }}, - decoder = {{ Text = {{}} }} - }}}} + pipelines = {{{}}} }}) sess:send("hello\n") @@ -91,7 +109,8 @@ async fn lua_session_open_send_read_close() { sess:close() "#, - addr + addr, + text_pipeline_lua() ); lua.load(&script).exec_async().await.unwrap(); @@ -117,11 +136,7 @@ async fn lua_session_read_timeout() { r#" local sess = xserial.open({{ transport = {{ Tcp = {{ addr = "{}" }} }}, - pipelines = {{{{ - name = "text", - framer = {{ Line = {{}} }}, - decoder = {{ Text = {{}} }} - }}}} + pipelines = {{{}}} }}) local r = sess:read(100) @@ -129,7 +144,8 @@ async fn lua_session_read_timeout() { sess:close() "#, - addr + addr, + text_pipeline_lua() ); lua.load(&script).exec_async().await.unwrap(); @@ -145,6 +161,7 @@ async fn lua_session_hex_decoder() { let server = tokio::spawn(async move { let (mut stream, _) = listener.accept().await.unwrap(); stream.write_all(b"ABC\n").await.unwrap(); + tokio::time::sleep(Duration::from_millis(200)).await; }); let lua = Lua::new(); @@ -154,11 +171,7 @@ async fn lua_session_hex_decoder() { r#" local sess = xserial.open({{ transport = {{ Tcp = {{ addr = "{}" }} }}, - pipelines = {{{{ - name = "hex", - framer = {{ Line = {{}} }}, - decoder = {{ Hex = {{ uppercase = true }} }} - }}}} + pipelines = {{{}}} }}) local r = sess:read(5000) @@ -169,7 +182,8 @@ async fn lua_session_hex_decoder() { sess:close() "#, - addr + addr, + hex_pipeline_lua() ); lua.load(&script).exec_async().await.unwrap(); @@ -186,6 +200,7 @@ async fn lua_session_on_data_callback() { let server = tokio::spawn(async move { let (mut stream, _) = listener.accept().await.unwrap(); stream.write_all(b"line1\nline2\n").await.unwrap(); + tokio::time::sleep(Duration::from_millis(200)).await; }); let lua = Lua::new(); @@ -196,19 +211,19 @@ async fn lua_session_on_data_callback() { local received = {{}} local sess = xserial.open({{ transport = {{ Tcp = {{ addr = "{}" }} }}, - pipelines = {{{{ - name = "text", - framer = {{ Line = {{}} }}, - decoder = {{ Text = {{}} }} - }}}} + pipelines = {{{}}} }}) sess:on_data(function(name, data) table.insert(received, {{ name = name, data = data }}) end) - -- Wait for both lines - xserial.sleep(500) + for _ = 1, 50 do + if #received < 2 then + xserial.poll(1) + xserial.sleep(10) + end + end assert(#received == 2, "expected 2 callbacks, got " .. #received) assert(received[1].name == "text") @@ -218,7 +233,8 @@ async fn lua_session_on_data_callback() { sess:close() "#, - addr + addr, + text_pipeline_lua() ); lua.load(&script).exec_async().await.unwrap(); @@ -235,6 +251,7 @@ async fn lua_session_reconfigure() { let server = tokio::spawn(async move { let (mut stream, _) = listener.accept().await.unwrap(); stream.write_all(b"test\n").await.unwrap(); + tokio::time::sleep(Duration::from_millis(300)).await; }); let listener2 = TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -243,6 +260,7 @@ async fn lua_session_reconfigure() { let server2 = tokio::spawn(async move { let (mut stream, _) = listener2.accept().await.unwrap(); stream.write_all(b"ABC\n").await.unwrap(); + tokio::time::sleep(Duration::from_millis(300)).await; }); let lua = Lua::new(); @@ -252,11 +270,7 @@ async fn lua_session_reconfigure() { r#" local sess = xserial.open({{ transport = {{ Tcp = {{ addr = "{}" }} }}, - pipelines = {{{{ - name = "text", - framer = {{ Line = {{}} }}, - decoder = {{ Text = {{}} }} - }}}} + pipelines = {{{}}} }}) local r = sess:read(5000) @@ -265,11 +279,7 @@ async fn lua_session_reconfigure() { -- Reconfigure to hex on different address sess:reconfigure({{ transport = {{ Tcp = {{ addr = "{}" }} }}, - pipelines = {{{{ - name = "hex", - framer = {{ Line = {{}} }}, - decoder = {{ Hex = {{ uppercase = true }} }} - }}}} + pipelines = {{{}}} }}) local r2 = sess:read(5000) @@ -278,10 +288,114 @@ async fn lua_session_reconfigure() { sess:close() "#, - addr, addr2 + addr, + text_pipeline_lua(), + addr2, + hex_pipeline_lua() ); lua.load(&script).exec_async().await.unwrap(); server.await.unwrap(); server2.await.unwrap(); } + +// ── session:next_event ─────────────────────────────────────────── + +#[tokio::test] +async fn lua_session_next_event_stream() { + 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"streamed\n").await.unwrap(); + tokio::time::sleep(Duration::from_millis(200)).await; + }); + + let lua = Lua::new(); + xserial_client::lua::register(&lua).unwrap(); + + let script = format!( + r#" + local sess = xserial.open({{ + transport = {{ Tcp = {{ addr = "{}" }} }}, + pipelines = {{{}}} + }}) + + local e1 = sess:next_event(5000) + assert(e1 ~= nil, "expected connected event") + assert(e1.type == "connected", "expected connected, got " .. tostring(e1.type)) + + local e2 = sess:next_event(5000) + assert(e2 ~= nil, "expected data event") + assert(e2.type == "data", "expected data, got " .. tostring(e2.type)) + assert(e2.pipeline == "text") + assert(e2.kind == "text") + assert(e2.data == "streamed") + + sess:close() + "#, + addr, + text_pipeline_lua() + ); + + lua.load(&script).exec_async().await.unwrap(); + server.await.unwrap(); +} + +// ── session:off + xserial.poll ─────────────────────────────────── + +#[tokio::test] +async fn lua_session_off_stops_future_callbacks() { + 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"first\n").await.unwrap(); + tokio::time::sleep(Duration::from_millis(150)).await; + stream.write_all(b"second\n").await.unwrap(); + tokio::time::sleep(Duration::from_millis(200)).await; + }); + + let lua = Lua::new(); + xserial_client::lua::register(&lua).unwrap(); + + let script = format!( + r#" + local received = {{}} + local sess = xserial.open({{ + transport = {{ Tcp = {{ addr = "{}" }} }}, + pipelines = {{{}}} + }}) + + local token = sess:on_data(function(name, data) + table.insert(received, name .. ":" .. data) + end) + + for _ = 1, 50 do + if #received == 0 then + xserial.poll(1) + xserial.sleep(10) + end + end + + assert(#received == 1, "expected first callback before off, got " .. #received) + assert(received[1] == "text:first") + assert(sess:off(token) == true) + + xserial.sleep(250) + xserial.poll(10) + + assert(#received == 1, "callback should not fire after off") + assert(sess:off(token) == false) + + sess:close() + "#, + addr, + text_pipeline_lua() + ); + + lua.load(&script).exec_async().await.unwrap(); + server.await.unwrap(); +}