feat: xserial-client with session management, mlua integration, and 238 tests

- Session: background I/O task with SessionHandle for send/read/close/reconfigure
- SessionManager: multi-session collection with aggregated event broadcast
- Config: serializable SessionConfig/PipelineConfig/FramerConfig/DecoderConfig
- Lua: xserial global (open/list_ports/sleep/log) + session userdata API
- RingBuffer: bounded history buffer with pipeline filtering
- Core: added Serialize/Deserialize+Default to Endian/TextEncoding/PlotFormat
- Core: added Send bound to Framer trait for tokio task compatibility
- 238 tests (193 core + 10 config + 7 Lua + 9 session + 18 pipeline + 1 doctest)

crates/xserial-client/src/cmd.rs

@@ -0,0 +1,7 @@
+use crate::config::SessionConfig;
+
+pub enum SessionCmd {
+    Send(Vec<u8>),
+    Close,
+    Reconfigure(SessionConfig),
+}

crates/xserial-client/src/config.rs

@@ -0,0 +1,240 @@
+use serde::{Deserialize, Serialize};
+use xserial_core::frame::{
+    cobs::CobsFramer, fixed::FixedLengthFramer,
+    length::{LengthConfig, LengthPrefixedFramer},
+    line::{LineConfig, LineFramer}, Endian, Framer,
+};
+use xserial_core::protocol::{
+    hex::{HexConfig, HexDecoder},
+    plot::{PlotConfig, PlotDecoder, PlotFormat, SampleType},
+    text::{TextDecoder, TextEncoding}, ProtocolDecoder,
+};
+use xserial_core::transport::TransportConfig;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum FramerConfig {
+    Line {
+        #[serde(default = "default_true")] strip_cr: bool,
+        #[serde(default = "default_max_line")] max_line_len: usize,
+    },
+    Fixed { frame_len: usize },
+    Length {
+        #[serde(default = "default_len_bytes")] len_bytes: usize,
+        #[serde(default)] endian: Endian,
+        #[serde(default)] length_includes_self: bool,
+        #[serde(default = "default_max_payload")] max_payload: usize,
+    },
+    Cobs {
+        #[serde(default = "default_max_frame")] max_frame: usize,
+    },
+}
+
+fn default_true() -> bool { true }
+fn default_max_line() -> usize { 1024 * 1024 }
+fn default_len_bytes() -> usize { 2 }
+fn default_max_payload() -> usize { 1024 * 1024 }
+fn default_max_frame() -> usize { 1024 * 1024 }
+
+impl FramerConfig {
+    pub fn build(self) -> Box<dyn Framer> {
+        match self {
+            FramerConfig::Line { strip_cr, max_line_len } =>
+                Box::new(LineFramer::new(LineConfig { strip_cr, max_line_len })),
+            FramerConfig::Fixed { frame_len } =>
+                Box::new(FixedLengthFramer::new(frame_len)),
+            FramerConfig::Length { len_bytes, endian, length_includes_self, max_payload } =>
+                Box::new(LengthPrefixedFramer::new(LengthConfig { len_bytes, endian, length_includes_self, max_payload })),
+            FramerConfig::Cobs { max_frame } =>
+                Box::new(CobsFramer::new(max_frame)),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum DecoderConfig {
+    Text { #[serde(default)] encoding: TextEncoding },
+    Hex {
+        #[serde(default)] uppercase: bool,
+        #[serde(default = "default_sep")] separator: String,
+        #[serde(default = "default_one")] bytes_per_group: usize,
+        #[serde(default)] endian: Endian,
+    },
+    Plot {
+        sample_type: SampleType,
+        #[serde(default)] endian: Endian,
+        #[serde(default = "default_one")] channels: usize,
+        #[serde(default)] format: PlotFormat,
+    },
+}
+
+fn default_sep() -> String { String::from(" ") }
+fn default_one() -> usize { 1 }
+
+impl DecoderConfig {
+    pub fn build(self) -> Box<dyn ProtocolDecoder> {
+        match self {
+            DecoderConfig::Text { encoding } => Box::new(TextDecoder::new(encoding)),
+            DecoderConfig::Hex { uppercase, separator, bytes_per_group, endian } =>
+                Box::new(HexDecoder::new(HexConfig { uppercase, separator, bytes_per_group, endian })),
+            DecoderConfig::Plot { sample_type, endian, channels, format } =>
+                Box::new(PlotDecoder::new(PlotConfig { sample_type, endian, channels, format })),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct PipelineConfig {
+    pub name: String,
+    pub framer: FramerConfig,
+    pub decoder: DecoderConfig,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct SessionConfig {
+    pub transport: TransportConfig,
+    #[serde(default)] pub pipelines: Vec<PipelineConfig>,
+    #[serde(default = "default_history")] pub history_limit: usize,
+    #[serde(default)] pub auto_reconnect: bool,
+}
+
+fn default_history() -> usize { 10_000 }
+
+impl Default for SessionConfig {
+    fn default() -> Self {
+        Self {
+            transport: TransportConfig::Tcp { addr: "127.0.0.1:8080".into() },
+            pipelines: Vec::new(),
+            history_limit: default_history(),
+            auto_reconnect: false,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use xserial_core::protocol::text::TextEncoding;
+    use xserial_core::protocol::plot::{PlotFormat, SampleType};
+    use xserial_core::protocol::Endian;
+    use xserial_core::transport::TransportConfig;
+
+    #[test]
+    fn framer_line_serde_roundtrip() {
+        let json = r#"{"Line":{"strip_cr":true,"max_line_len":4096}}"#;
+        let cfg: FramerConfig = serde_json::from_str(json).unwrap();
+        match &cfg {
+            FramerConfig::Line { strip_cr, max_line_len } => {
+                assert!(*strip_cr);
+                assert_eq!(*max_line_len, 4096);
+            }
+            _ => panic!("expected Line"),
+        }
+        let back = serde_json::to_string(&cfg).unwrap();
+        let _: FramerConfig = serde_json::from_str(&back).unwrap();
+    }
+
+    #[test]
+    fn framer_fixed_serde() {
+        let cfg: FramerConfig = serde_json::from_str(r#"{"Fixed":{"frame_len":128}}"#).unwrap();
+        assert!(matches!(cfg, FramerConfig::Fixed { frame_len: 128 }));
+    }
+
+    #[test]
+    fn framer_length_serde() {
+        let json = r#"{"Length":{"len_bytes":4,"endian":"Little","length_includes_self":true}}"#;
+        let cfg: FramerConfig = serde_json::from_str(json).unwrap();
+        match cfg {
+            FramerConfig::Length { len_bytes, endian, length_includes_self, .. } => {
+                assert_eq!(len_bytes, 4);
+                assert_eq!(endian, Endian::Little);
+                assert!(length_includes_self);
+            }
+            _ => panic!("expected Length"),
+        }
+    }
+
+    #[test]
+    fn framer_cobs_serde() {
+        let cfg: FramerConfig = serde_json::from_str(r#"{"Cobs":{"max_frame":8192}}"#).unwrap();
+        assert!(matches!(cfg, FramerConfig::Cobs { max_frame: 8192 }));
+    }
+
+    #[test]
+    fn decoder_text_serde() {
+        let cfg: DecoderConfig = serde_json::from_str(r#"{"Text":{"encoding":"Latin1"}}"#).unwrap();
+        match cfg {
+            DecoderConfig::Text { encoding } => assert_eq!(encoding, TextEncoding::Latin1),
+            _ => panic!("expected Text"),
+        }
+    }
+
+    #[test]
+    fn decoder_hex_serde() {
+        let json = r#"{"Hex":{"uppercase":true,"separator":":","bytes_per_group":2,"endian":"Little"}}"#;
+        let cfg: DecoderConfig = serde_json::from_str(json).unwrap();
+        match cfg {
+            DecoderConfig::Hex { uppercase, separator, bytes_per_group, endian } => {
+                assert!(uppercase);
+                assert_eq!(separator, ":");
+                assert_eq!(bytes_per_group, 2);
+                assert_eq!(endian, Endian::Little);
+            }
+            _ => panic!("expected Hex"),
+        }
+    }
+
+    #[test]
+    fn decoder_plot_serde() {
+        let json = r#"{"Plot":{"sample_type":"F32","endian":"Little","channels":2,"format":"Interleaved"}}"#;
+        let cfg: DecoderConfig = serde_json::from_str(json).unwrap();
+        match cfg {
+            DecoderConfig::Plot { sample_type, endian, channels, format } => {
+                assert_eq!(sample_type, SampleType::F32);
+                assert_eq!(channels, 2);
+                assert_eq!(format, PlotFormat::Interleaved);
+                assert_eq!(endian, Endian::Little);
+            }
+            _ => panic!("expected Plot"),
+        }
+    }
+
+    #[test]
+    fn session_config_full_roundtrip() {
+        let json = r#"{
+            "transport":{"Tcp":{"addr":"192.168.1.1:9999"}},
+            "pipelines":[
+                {"name":"t","framer":{"Line":{}},"decoder":{"Text":{}}},
+                {"name":"h","framer":{"Fixed":{"frame_len":16}},"decoder":{"Hex":{}}}
+            ],
+            "history_limit":5000,
+            "auto_reconnect":true
+        }"#;
+        let cfg: SessionConfig = serde_json::from_str(json).unwrap();
+        assert!(cfg.auto_reconnect);
+        assert_eq!(cfg.history_limit, 5000);
+        assert_eq!(cfg.pipelines.len(), 2);
+        assert!(matches!(&cfg.transport, TransportConfig::Tcp { addr } if addr == "192.168.1.1:9999"));
+        let back = serde_json::to_string_pretty(&cfg).unwrap();
+        let _: SessionConfig = serde_json::from_str(&back).unwrap();
+    }
+
+    #[test]
+    fn framer_build_does_not_panic() {
+        for cfg in [
+            FramerConfig::Line { strip_cr: true, max_line_len: 256 },
+            FramerConfig::Fixed { frame_len: 8 },
+            FramerConfig::Length { len_bytes: 2, endian: Endian::Big, length_includes_self: false, max_payload: 65536 },
+            FramerConfig::Cobs { max_frame: 1024 },
+        ] {
+            let f = cfg.build();
+            assert_eq!(f.pending_len(), 0);
+        }
+    }
+
+    #[test]
+    fn decoder_build_returns_correct_name() {
+        assert_eq!(DecoderConfig::Text { encoding: TextEncoding::Utf8 }.build().name(), "Text");
+        assert_eq!(DecoderConfig::Hex { uppercase: false, separator: " ".into(), bytes_per_group: 1, endian: Endian::Big }.build().name(), "Hex");
+        assert_eq!(DecoderConfig::Plot { sample_type: SampleType::I16, endian: Endian::Big, channels: 2, format: PlotFormat::XY }.build().name(), "Plot");
+    }
+}

crates/xserial-client/src/error.rs

@@ -0,0 +1,17 @@
+use thiserror::Error;
+
+#[derive(Error, Debug)]
+pub enum Error {
+    #[error("Connection failed: {0}")]
+    ConnectionFailed(String),
+    #[error("Session error: {0}")]
+    Session(String),
+    #[error("Lua error: {0}")]
+    Lua(#[from] mlua::Error),
+    #[error("I/O error: {0}")]
+    Io(#[from] std::io::Error),
+    #[error("{0}")]
+    Other(String),
+}
+
+pub type Result<T> = std::result::Result<T, Error>;

crates/xserial-client/src/event.rs

@@ -0,0 +1,7 @@
+use xserial_core::protocol::DecodedData;
+
+#[derive(Debug, Clone)]
+pub struct DecodedEntry {
+    pub pipeline_name: String,
+    pub data: DecodedData,
+}

crates/xserial-client/src/history.rs

@@ -0,0 +1,31 @@
+use std::collections::VecDeque;
+use crate::event::DecodedEntry;
+
+pub struct RingBuffer<T> {
+    buf: VecDeque<T>,
+    limit: usize,
+}
+
+impl<T> RingBuffer<T> {
+    pub fn new(limit: usize) -> Self {
+        Self { buf: VecDeque::with_capacity(limit.min(1024)), limit }
+    }
+    pub fn push(&mut self, item: T) {
+        if self.buf.len() >= self.limit { self.buf.pop_front(); }
+        self.buf.push_back(item);
+    }
+    pub fn len(&self) -> usize { self.buf.len() }
+    pub fn is_empty(&self) -> bool { self.buf.is_empty() }
+    pub fn iter(&self) -> impl Iterator<Item = &T> { self.buf.iter() }
+    pub fn clear(&mut self) { self.buf.clear(); }
+    pub fn drain_recent(&self, count: usize) -> Vec<T> where T: Clone {
+        let start = self.buf.len().saturating_sub(count);
+        self.buf.iter().skip(start).cloned().collect()
+    }
+}
+
+impl RingBuffer<DecodedEntry> {
+    pub fn entries_for_pipeline(&self, name: &str) -> Vec<&DecodedEntry> {
+        self.buf.iter().filter(|e| e.pipeline_name == name).collect()
+    }
+}

crates/xserial-client/src/lib.rs

@@ -0,0 +1,15 @@
+pub mod config;
+pub mod event;
+pub mod cmd;
+pub mod error;
+pub mod history;
+pub mod session;
+pub mod manager;
+pub mod lua;
+
+pub use config::{DecoderConfig, FramerConfig, PipelineConfig, SessionConfig};
+pub use error::{Error, Result};
+pub use event::DecodedEntry;
+pub use history::RingBuffer;
+pub use manager::SessionManager;
+pub use session::{Session, SessionEvent, SessionHandle, SessionId};

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

@@ -0,0 +1,38 @@
+use std::sync::atomic::{AtomicU64, Ordering};
+use mlua::{Lua, Result as LuaResult, Table, Value, LuaSerdeExt};
+use crate::config::SessionConfig;
+use crate::session::Session;
+mod session_api;
+
+static NEXT_ID: AtomicU64 = AtomicU64::new(1);
+
+pub fn register(lua: &Lua) -> LuaResult<()> {
+    let xserial = lua.create_table()?;
+
+    xserial.set("open", lua.create_async_function(move |lua, config: Table| {
+        let id = NEXT_ID.fetch_add(1, Ordering::Relaxed);
+        async move {
+            let cfg: SessionConfig = lua.from_value(Value::Table(config))?;
+            let handle = Session::spawn(id, cfg);
+            lua.create_userdata(session_api::LuaSessionHandle::new(handle))
+        }
+    })?)?;
+
+    xserial.set("list_ports", lua.create_function(|_, ()| {
+        let ports = xserial_core::transport::serial::SerialTransport::list_ports();
+        Ok(ports.into_iter().map(|p| p.port_name).collect::<Vec<_>>())
+    })?)?;
+
+    xserial.set("sleep", lua.create_async_function(|_, ms: u64| async move {
+        tokio::time::sleep(std::time::Duration::from_millis(ms)).await;
+        Ok(())
+    })?)?;
+
+    xserial.set("log", lua.create_function(|_, msg: String| {
+        tracing::info!("[lua] {}", msg);
+        Ok(())
+    })?)?;
+
+    lua.globals().set("xserial", xserial)?;
+    Ok(())
+}

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

@@ -0,0 +1,78 @@
+use mlua::{UserData, UserDataMethods, Variadic, LuaSerdeExt};
+use xserial_core::protocol::DecodedData;
+use crate::session::SessionHandle;
+
+pub struct LuaSessionHandle { pub handle: SessionHandle }
+
+impl LuaSessionHandle {
+    pub fn new(handle: SessionHandle) -> Self { Self { handle } }
+}
+
+impl UserData for LuaSessionHandle {
+    fn add_methods<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
+                .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) => {
+                    let t = lua.create_table()?;
+                    t.set("pipeline", entry.pipeline_name)?;
+                    match entry.data {
+                        DecodedData::Text(s) => { t.set("kind", "text")?; t.set("data", s)?; }
+                        DecodedData::Hex(s) => { t.set("kind", "hex")?; t.set("data", s)?; }
+                        DecodedData::Binary(v) => { t.set("kind", "binary")?; t.set("data", lua.create_string(&v)?)?; }
+                        DecodedData::Plot(frame) => {
+                            t.set("kind", "plot")?;
+                            let chs = lua.create_table()?;
+                            for (i, ch) in frame.channels.iter().enumerate() {
+                                chs.set(i + 1, lua.create_sequence_from(ch.iter().copied())?)?;
+                            }
+                            t.set("channels", chs)?;
+                            t.set("sample_count", frame.sample_count())?;
+                        }
+                    }
+                    Ok(Some(t))
+                }
+                None => Ok(None),
+            }
+        });
+
+        methods.add_async_method("close", |_, this, (): ()| async move {
+            this.handle.close().await.map_err(mlua::Error::RuntimeError)
+        });
+
+        methods.add_async_method("reconfigure", |lua, this, config: mlua::Table| async move {
+            let cfg: crate::config::SessionConfig = lua.from_value(mlua::Value::Table(config))?;
+            this.handle.reconfigure(cfg).await.map_err(mlua::Error::RuntimeError)
+        });
+
+        methods.add_async_method("on_data", |_, this, (callback, pipelines): (mlua::Function, Variadic<String>)| async move {
+            let filter: Vec<String> = 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(())
+        });
+    }
+}

crates/xserial-client/src/manager.rs

@@ -0,0 +1,39 @@
+use std::collections::HashMap;
+use crate::config::SessionConfig;
+use crate::session::{Session, SessionEvent, SessionHandle, SessionId};
+
+pub struct SessionManager {
+    sessions: HashMap<SessionId, SessionHandle>,
+    event_tx: tokio::sync::broadcast::Sender<SessionEvent>,
+}
+
+impl SessionManager {
+    pub fn new() -> Self {
+        let (event_tx, _) = tokio::sync::broadcast::channel(256);
+        Self { sessions: HashMap::new(), event_tx }
+    }
+
+    pub fn create(&mut self, config: SessionConfig) -> SessionHandle {
+        let id = self.sessions.len() as u64 + 1;
+        let handle = Session::spawn(id, config);
+        let event_relay = self.event_tx.clone();
+        let mut event_rx = handle.subscribe();
+        tokio::spawn(async move {
+            while let Ok(event) = event_rx.recv().await { let _ = event_relay.send(event); }
+        });
+        self.sessions.insert(id, SessionHandle {
+            id: handle.id, cmd_tx: handle.cmd_tx.clone(), event_tx: handle.event_tx.clone(), _task: None,
+        });
+        handle
+    }
+
+    pub fn remove(&mut self, id: SessionId) { self.sessions.remove(&id); }
+    pub fn get(&self, id: SessionId) -> Option<&SessionHandle> { self.sessions.get(&id) }
+    pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver<SessionEvent> { self.event_tx.subscribe() }
+    pub fn list_ids(&self) -> Vec<SessionId> { self.sessions.keys().copied().collect() }
+    pub fn count(&self) -> usize { self.sessions.len() }
+}
+
+impl Default for SessionManager {
+    fn default() -> Self { Self::new() }
+}

crates/xserial-client/src/session.rs

@@ -0,0 +1,229 @@
+use tokio::io::AsyncReadExt;
+use tokio::select;
+use tokio::sync::{broadcast, mpsc};
+use tokio::task::JoinHandle;
+use tracing::{error, info, warn};
+
+use xserial_core::pipeline::{MultiPipeline, Pipeline};
+use xserial_core::transport::Connection;
+
+use crate::cmd::SessionCmd;
+use crate::config::SessionConfig;
+use crate::event::DecodedEntry;
+use crate::history::RingBuffer;
+
+pub type SessionId = u64;
+
+#[derive(Debug, Clone)]
+pub enum SessionEvent {
+    Connected(SessionId),
+    Disconnected(SessionId),
+    Data(SessionId, DecodedEntry),
+    Error(SessionId, String),
+    Closed(SessionId),
+}
+
+pub struct SessionHandle {
+    pub id: SessionId,
+    pub(crate) cmd_tx: mpsc::Sender<SessionCmd>,
+    pub(crate) event_tx: broadcast::Sender<SessionEvent>,
+    pub(crate) _task: Option<JoinHandle<()>>,
+}
+
+impl SessionHandle {
+    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())
+    }
+
+    pub async fn close(&self) -> Result<(), String> {
+        self.cmd_tx.send(SessionCmd::Close).await.map_err(|_| "session closed".into())
+    }
+
+    pub async fn reconfigure(&self, config: SessionConfig) -> Result<(), String> {
+        self.cmd_tx.send(SessionCmd::Reconfigure(config)).await.map_err(|_| "session closed".into())
+    }
+
+    pub async fn read(&self, timeout_ms: u64) -> Option<DecodedEntry> {
+        let mut rx = self.event_tx.subscribe();
+        let deadline = std::time::Duration::from_millis(timeout_ms);
+        tokio::time::timeout(deadline, async {
+            loop {
+                match rx.recv().await {
+                    Ok(SessionEvent::Data(_, e)) => return Some(e),
+                    Ok(SessionEvent::Error(_, _) | SessionEvent::Closed(_)) => return None,
+                    Err(broadcast::error::RecvError::Lagged(_)) => continue,
+                    Err(broadcast::error::RecvError::Closed) => return None,
+                    _ => continue,
+                }
+            }
+        }).await.ok().flatten()
+    }
+
+    pub fn subscribe(&self) -> broadcast::Receiver<SessionEvent> {
+        self.event_tx.subscribe()
+    }
+}
+
+impl Drop for SessionHandle {
+    fn drop(&mut self) {
+        let _ = self.cmd_tx.try_send(SessionCmd::Close);
+        if let Some(t) = self._task.take() { t.abort(); }
+    }
+}
+
+pub struct Session {
+    id: SessionId,
+    config: SessionConfig,
+    conn: Option<Connection>,
+    pipeline: MultiPipeline,
+    history: RingBuffer<DecodedEntry>,
+    cmd_rx: mpsc::Receiver<SessionCmd>,
+    event_tx: broadcast::Sender<SessionEvent>,
+}
+
+impl Session {
+    pub fn spawn(id: SessionId, config: SessionConfig) -> SessionHandle {
+        let (cmd_tx, cmd_rx) = mpsc::channel(32);
+        let (event_tx, _) = broadcast::channel(256);
+        let mut session = Self {
+            id, conn: None,
+            pipeline: Self::build_pipeline(&config),
+            history: RingBuffer::new(config.history_limit),
+            config, cmd_rx, event_tx: event_tx.clone(),
+        };
+        let task = tokio::spawn(async move { session.run().await });
+        SessionHandle { id, cmd_tx, event_tx, _task: Some(task) }
+    }
+
+    fn build_pipeline(config: &SessionConfig) -> MultiPipeline {
+        let mut mp = MultiPipeline::new();
+        for pc in &config.pipelines {
+            mp.add(Pipeline::new(
+                pc.name.clone(),
+                pc.framer.clone().build(),
+                pc.decoder.clone().build(),
+            ));
+        }
+        mp
+    }
+
+    async fn run(&mut self) {
+        info!(session_id = self.id, "session started");
+
+        // Initial connect
+        let conn = &mut self.conn;
+        if let Err(e) = do_connect(conn, &self.config, &self.event_tx, self.id).await {
+            let _ = self.event_tx.send(SessionEvent::Error(self.id, format!("connect failed: {}", e)));
+            let _ = self.event_tx.send(SessionEvent::Closed(self.id));
+            return;
+        }
+
+        loop {
+            let cmd_rx = &mut self.cmd_rx;
+            let conn = &mut self.conn;
+            let pipeline = &mut self.pipeline;
+            let history = &mut self.history;
+            let event_tx = &self.event_tx;
+            let id = self.id;
+
+            select! {
+                cmd = cmd_rx.recv() => {
+                    match cmd {
+                        Some(SessionCmd::Close) => {
+                            do_disconnect(conn, pipeline, event_tx, id).await;
+                            let _ = event_tx.send(SessionEvent::Closed(id));
+                            break;
+                        }
+                        Some(SessionCmd::Send(data)) => {
+                            if let Some(c) = conn {
+                                use tokio::io::AsyncWriteExt;
+                                if let Err(e) = c.write_all(&data).await {
+                                    let _ = event_tx.send(SessionEvent::Error(id, format!("write: {}", e)));
+                                }
+                            }
+                        }
+                        Some(SessionCmd::Reconfigure(new_cfg)) => {
+                            do_disconnect(conn, pipeline, event_tx, id).await;
+                            self.config = new_cfg;
+                            *pipeline = Self::build_pipeline(&self.config);
+                            *history = RingBuffer::new(self.config.history_limit);
+                            if let Err(e) = do_connect(conn, &self.config, event_tx, id).await {
+                                let _ = event_tx.send(SessionEvent::Error(id, format!("reconnect: {}", e)));
+                            }
+                        }
+                        None => break,
+                    }
+                }
+                result = try_read(conn, pipeline) => {
+                    match result {
+                        Ok(entries) => {
+                            for entry in entries {
+                                history.push(entry.clone());
+                                let _ = event_tx.send(SessionEvent::Data(id, entry));
+                            }
+                        }
+                        Err(e) => {
+                            error!(session_id = id, error = %e, "read error");
+                            let _ = event_tx.send(SessionEvent::Error(id, e.to_string()));
+                            do_disconnect(conn, pipeline, event_tx, id).await;
+                            if self.config.auto_reconnect {
+                                warn!(session_id = id, "auto-reconnecting");
+                                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+                                if let Err(e) = do_connect(conn, &self.config, event_tx, id).await {
+                                    let _ = event_tx.send(SessionEvent::Error(id, format!("reconnect: {}", e)));
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+        }
+        info!(session_id = self.id, "session ended");
+    }
+}
+
+async fn try_read(
+    conn: &mut Option<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 mut buf = [0u8; 4096];
+    match c.read(&mut buf).await {
+        Ok(0) => Ok(Vec::new()),
+        Ok(n) => {
+            Ok(pipeline.feed(&buf[..n])
+                .into_iter()
+                .map(|r| DecodedEntry { pipeline_name: r.pipeline_name, data: r.data })
+                .collect())
+        }
+        Err(e) => Err(e),
+    }
+}
+
+async fn do_connect(
+    conn: &mut Option<Connection>,
+    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(
+    conn: &mut Option<Connection>,
+    pipeline: &mut MultiPipeline,
+    event_tx: &broadcast::Sender<SessionEvent>,
+    id: SessionId,
+) {
+    if let Some(mut c) = conn.take() {
+        let _ = c.disconnect().await;
+        pipeline.reset();
+        let _ = event_tx.send(SessionEvent::Disconnected(id));
+    }
+}