Rework client sessions and Lua integration

crates/xserial-client/src/lua/mod.rs

@@ -1,38 +1,169 @@
-use std::sync::atomic::{AtomicU64, Ordering};
+use std::collections::HashMap;
-use mlua::{Lua, Result as LuaResult, Table, Value, LuaSerdeExt};
+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<Mutex<HashMap<u64, Weak<session_api::CallbackState>>>>,
+}
+
+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<session_api::CallbackState>,
+    ) {
+        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<usize>,
+    ) -> LuaResult<usize> {
+        let states: Vec<(u64, Arc<session_api::CallbackState>)> = {
+            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| {
+    xserial.set("open", {
-        let id = NEXT_ID.fetch_add(1, Ordering::Relaxed);
+        let runtime = runtime.clone();
-        async move {
+        lua.create_async_function(move |lua, config: Table| {
-            let cfg: SessionConfig = lua.from_value(Value::Table(config))?;
+            let runtime = runtime.clone();
-            let handle = Session::spawn(id, cfg);
+            async move {
-            lua.create_userdata(session_api::LuaSessionHandle::new(handle))
+                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(|_, ()| {
+    xserial.set(
-        let ports = xserial_core::transport::serial::SerialTransport::list_ports();
+        "list_ports",
-        Ok(ports.into_iter().map(|p| p.port_name).collect::<Vec<_>>())
+        lua.create_function(|_, ()| {
-    })?)?;
+            let ports = xserial_core::transport::serial::SerialTransport::list_ports();
+            Ok(ports.into_iter().map(|p| p.port_name).collect::<Vec<_>>())
+        })?,
+    )?;
 
-    xserial.set("sleep", lua.create_async_function(|_, ms: u64| async move {
+    xserial.set("sleep", {
-        tokio::time::sleep(std::time::Duration::from_millis(ms)).await;
+        let runtime = runtime.clone();
-        Ok(())
+        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| {
+    xserial.set("poll", {
-        tracing::info!("[lua] {}", msg);
+        let runtime = runtime.clone();
-        Ok(())
+        lua.create_async_function(move |lua, limit_per_session: Option<usize>| {
-    })?)?;
+            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);
+    }
+}

crates/xserial-client/src/lua/session_api.rs

@@ -1,78 +1,535 @@
-use mlua::{UserData, UserDataMethods, Variadic, LuaSerdeExt};
+use std::collections::HashMap;
-use xserial_core::protocol::DecodedData;
+use std::sync::{
-use crate::session::SessionHandle;
+    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<CallbackState>,
+}
+
+pub struct CallbackState {
+    queue_tx: mpsc::UnboundedSender<QueuedEvent>,
+    queue_rx: Mutex<mpsc::UnboundedReceiver<QueuedEvent>>,
+    subscriptions: Mutex<HashMap<u64, Subscription>>,
+    next_subscription_id: AtomicU64,
+    relay_task: Mutex<Option<JoinHandle<()>>>,
+}
+
+struct Subscription {
+    callback: Arc<RegistryKey>,
+    filters: Vec<String>,
+}
+
+#[derive(Clone)]
+enum QueuedEvent {
+    Data {
+        pipeline_name: String,
+        data: QueuedData,
+    },
+}
+
+#[derive(Clone)]
+enum QueuedData {
+    Text(String),
+    Hex(String),
+    Binary(Vec<u8>),
+    Plot {
+        channels: Vec<Vec<f64>>,
+        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<String>) -> 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<bool> {
+        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<usize>) -> mlua::Result<usize> {
+        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<Arc<RegistryKey>> {
+        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<M: UserDataMethods<Self>>(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<u64>| async move {
             let ms = timeout_ms.unwrap_or(1000);
             match this.handle.read(ms).await {
-                Some(entry) => {
+                Some(entry) => Ok(Some(decoded_entry_to_table(
-                    let t = lua.create_table()?;
+                    &lua,
-                    t.set("pipeline", entry.pipeline_name)?;
+                    entry.pipeline_name,
-                    match entry.data {
+                    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(
+            "next_event",
+            |lua, this, timeout_ms: Option<u64>| 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))?;
+            let cfg: SessionConfig = lua.from_value(Value::Table(config))?;
-            this.handle.reconfigure(cfg).await.map_err(mlua::Error::RuntimeError)
+            this.handle
+                .reconfigure(cfg)
+                .await
+                .map_err(mlua::Error::RuntimeError)
         });
 
-        methods.add_async_method("on_data", |_, this, (callback, pipelines): (mlua::Function, Variadic<String>)| async move {
+        methods.add_async_method(
-            let filter: Vec<String> = pipelines.into_iter().collect();
+            "on_data",
-            let mut rx = this.handle.subscribe();
+            |lua, this, (callback, pipelines): (Function, Variadic<String>)| async move {
-            tokio::spawn(async move {
+                let filters: Vec<String> = pipelines.into_iter().collect();
-                loop {
+                let key = lua.create_registry_value(callback)?;
-                    match rx.recv().await {
+                Ok(this.callbacks.add_subscription(key, filters))
-                        Ok(crate::session::SessionEvent::Data(_, entry))
+            },
-                            if filter.is_empty() || filter.iter().any(|f| f == &entry.pipeline_name) =>
+        );
-                        {
+
-                                let data_str = match &entry.data {
+        methods.add_method("off", |lua, this, token: u64| {
-                                    DecodedData::Text(s) => s.clone(),
+            this.callbacks.remove_subscription(lua, token)
-                                    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(
+            "pump_events",
+            |lua, this, limit: Option<usize>| async move { this.callbacks.pump(&lua, limit).await },
+        );
+    }
+}
+
+fn decoded_entry_to_table(
+    lua: &Lua,
+    pipeline_name: String,
+    data: DecodedData,
+) -> mlua::Result<mlua::Table> {
+    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<mlua::Table> {
+    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<Value> {
+    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<f64>]) -> mlua::Result<mlua::Table> {
+    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<CallbackState> {
+        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::<Arc<Mutex<Vec<(String, String)>>>>()
+                    .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::<Arc<Mutex<usize>>>().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::<String>("type").unwrap(), "data");
+        assert_eq!(table.get::<String>("pipeline").unwrap(), "text");
+        assert_eq!(table.get::<String>("kind").unwrap(), "text");
+        assert_eq!(table.get::<String>("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::<String>("type").unwrap(), "error");
+        assert_eq!(error_table.get::<u64>("session_id").unwrap(), 7);
+        assert_eq!(error_table.get::<String>("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::<String>("type").unwrap(), "data");
+        assert_eq!(data_table.get::<u64>("session_id").unwrap(), 9);
+        assert_eq!(data_table.get::<String>("pipeline").unwrap(), "hex");
+        assert_eq!(data_table.get::<String>("kind").unwrap(), "hex");
+        assert_eq!(data_table.get::<String>("data").unwrap(), "41 42");
     }
 }

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<SessionId, SessionHandle>,
+    inner: Arc<SessionManagerInner>,
-    event_tx: tokio::sync::broadcast::Sender<SessionEvent>,
+}
+
+struct SessionManagerInner {
+    sessions: RwLock<HashMap<SessionId, SessionHandle>>,
+    event_tx: broadcast::Sender<SessionEvent>,
+    next_session_id: AtomicU64,
 }
 
 impl SessionManager {
     pub fn new() -> Self {
-        let (event_tx, _) = tokio::sync::broadcast::channel(256);
+        let (event_tx, _) = broadcast::channel(256);
-        Self { sessions: HashMap::new(), event_tx }
+        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 {
+    pub fn create(&self, config: SessionConfig) -> SessionHandle {
-        let id = self.sessions.len() as u64 + 1;
+        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 stored = handle.clone();
-        let mut event_rx = handle.subscribe();
+        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); }
+            loop {
-        });
+                match session_events.recv().await {
-        self.sessions.insert(id, SessionHandle {
+                    Ok(event) => {
-            id: handle.id, cmd_tx: handle.cmd_tx.clone(), event_tx: handle.event_tx.clone(), _task: None,
+                        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 remove(&self, id: SessionId) {
-    pub fn get(&self, id: SessionId) -> Option<&SessionHandle> { self.sessions.get(&id) }
+        let removed = self
-    pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver<SessionEvent> { self.event_tx.subscribe() }
+            .inner
-    pub fn list_ids(&self) -> Vec<SessionId> { self.sessions.keys().copied().collect() }
+            .sessions
-    pub fn count(&self) -> usize { self.sessions.len() }
+            .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<SessionHandle> {
+        self.inner
+            .sessions
+            .read()
+            .expect("session store poisoned")
+            .get(&id)
+            .cloned()
+    }
+
+    pub fn subscribe(&self) -> broadcast::Receiver<SessionEvent> {
+        self.inner.event_tx.subscribe()
+    }
+
+    pub fn list_ids(&self) -> Vec<SessionId> {
+        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);
+    }
 }

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<SessionCmd>,
+    event_tx: broadcast::Sender<SessionEvent>,
+    task: Mutex<Option<JoinHandle<()>>>,
+    close_requested: AtomicBool,
+}
+
+impl SessionHandleInner {
+    fn new(
+        id: SessionId,
+        cmd_tx: mpsc::Sender<SessionCmd>,
+        event_tx: broadcast::Sender<SessionEvent>,
+        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<SessionCmd>,
+    inner: Arc<SessionHandleInner>,
-    pub(crate) event_tx: broadcast::Sender<SessionEvent>,
-    pub(crate) _task: Option<JoinHandle<()>>,
 }
 
 impl SessionHandle {
-    pub fn id(&self) -> SessionId { self.id }
+    fn new(inner: Arc<SessionHandleInner>) -> Self {
+        Self {
+            id: inner.id,
+            inner,
+        }
+    }
+
+    pub fn id(&self) -> SessionId {
+        self.id
+    }
 
     pub async fn send(&self, data: Vec<u8>) -> 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<DecodedEntry> {
-        let mut rx = self.event_tx.subscribe();
+        let mut rx = self.inner.event_tx.subscribe();
-        let deadline = std::time::Duration::from_millis(timeout_ms);
+        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<SessionEvent> {
-        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 Arc::strong_count(&self.inner) == 1 {
-        if let Some(t) = self._task.take() { t.abort(); }
+            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<DecodedEntry>,
     cmd_rx: mpsc::Receiver<SessionCmd>,
     event_tx: broadcast::Sender<SessionEvent>,
+    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();
+        let mut pipelines = MultiPipeline::new();
-        for pc in &config.pipelines {
+        for pipeline in &config.pipelines {
-            mp.add(Pipeline::new(
+            pipelines.add(Pipeline::new(
-                pc.name.clone(),
+                pipeline.name.clone(),
-                pc.framer.clone().build(),
+                pipeline.framer.clone().build(),
-                pc.decoder.clone().build(),
+                pipeline.decoder.clone().build(),
             ));
         }
-        mp
+        pipelines
     }
 
     async fn run(&mut self) {
         info!(session_id = self.id, "session started");
 
-        // Initial connect
+        if let Err(err) = self.connect().await {
-        let conn = &mut self.conn;
+            warn!(session_id = self.id, error = %err, "initial connect failed");
-        if let Err(e) = do_connect(conn, &self.config, &self.event_tx, self.id).await {
+            self.emit_error(format!("connect failed: {err}"));
-            let _ = self.event_tx.send(SessionEvent::Error(self.id, format!("connect failed: {}", e)));
+            self.emit_closed();
-            let _ = self.event_tx.send(SessionEvent::Closed(self.id));
+            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() => {
+                cmd = self.cmd_rx.recv() => {
-                    match cmd {
+                    if !self.handle_command(cmd).await {
-                        Some(SessionCmd::Close) => {
+                        break;
-                            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) => {
+                read = try_read(&mut self.conn, &mut self.pipeline) => {
-                    match result {
+                    if !self.handle_read_result(read).await {
-                        Ok(entries) => {
+                        break;
-                            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)));
-                                }
-                            }
-                        }
                     }
                 }
             }
         }
+
+        self.disconnect(false).await;
+        self.emit_closed();
         info!(session_id = self.id, "session ended");
     }
+
+    async fn handle_command(&mut self, cmd: Option<SessionCmd>) -> 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<u8>) {
+        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<Vec<DecodedEntry>, 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<Connection>,
     pipeline: &mut MultiPipeline,
 ) -> Result<Vec<DecodedEntry>, std::io::Error> {
-    let c = match conn.as_mut() { Some(c) => c, None => return Ok(Vec::new()) };
+    let conn = match conn.as_mut() {
-    let mut buf = [0u8; 4096];
+        Some(conn) => conn,
-    match c.read(&mut buf).await {
+        None => {
-        Ok(0) => Ok(Vec::new()),
+            tokio::time::sleep(Duration::from_millis(20)).await;
-        Ok(n) => {
+            return Ok(Vec::new());
-            Ok(pipeline.feed(&buf[..n])
-                .into_iter()
-                .map(|r| DecodedEntry { pipeline_name: r.pipeline_name, data: r.data })
-                .collect())
         }
-        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(
+fn to_io_error(err: xserial_core::error::Error) -> std::io::Error {
-    conn: &mut Option<Connection>,
+    std::io::Error::new(std::io::ErrorKind::Other, err.to_string())
-    config: &SessionConfig,
-    event_tx: &broadcast::Sender<SessionEvent>,
-    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(
+#[cfg(test)]
-    conn: &mut Option<Connection>,
+mod tests {
-    pipeline: &mut MultiPipeline,
+    use super::*;
-    event_tx: &broadcast::Sender<SessionEvent>,
+    use crate::config::{DecoderConfig, FramerConfig, PipelineConfig};
-    id: SessionId,
+
-) {
+    fn tcp_config() -> SessionConfig {
-    if let Some(mut c) = conn.take() {
+        SessionConfig {
-        let _ = c.disconnect().await;
+            transport: xserial_core::transport::TransportConfig::Tcp {
-        pipeline.reset();
+                addr: "127.0.0.1:1".into(),
-        let _ = event_tx.send(SessionEvent::Disconnected(id));
+            },
+            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;
     }
 }

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 = {{{{
+            pipelines = {{{}}}
-                name = "text",
-                framer = {{ Line = {{}} }},
-                decoder = {{ Text = {{}} }}
-            }}}}
         }})
 
         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 = {{{{
+            pipelines = {{{}}}
-                name = "text",
-                framer = {{ Line = {{}} }},
-                decoder = {{ Text = {{}} }}
-            }}}}
         }})
 
         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 = {{{{
+            pipelines = {{{}}}
-                name = "hex",
-                framer = {{ Line = {{}} }},
-                decoder = {{ Hex = {{ uppercase = true }} }}
-            }}}}
         }})
 
         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 = {{{{
+            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
+        for _ = 1, 50 do
-        xserial.sleep(500)
+            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 = {{{{
+            pipelines = {{{}}}
-                name = "text",
-                framer = {{ Line = {{}} }},
-                decoder = {{ Text = {{}} }}
-            }}}}
         }})
 
         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 = {{{{
+            pipelines = {{{}}}
-                name = "hex",
-                framer = {{ Line = {{}} }},
-                decoder = {{ Hex = {{ uppercase = true }} }}
-            }}}}
         }})
 
         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();
+}