feat: enable remote access for embedded no_std environments with tokio adapter integration

Author: lxsaah

Cargo.lock

@@ -43,6 +43,11 @@ pub struct SessionConfig {
     pub acks_subscribe: bool,
 }
 
+/// Bound for the per-connection event funnel — caps how many pending outbound
+/// updates a single connection may buffer before pumps start dropping (matches
+/// the hand-rolled WS server's default per-client channel capacity).
+const EVENT_BUFFER: usize = 256;
+
 /// One subscription update on its way back to the connection's send half.
 struct SubEvent {
     sub: String,
@@ -91,8 +96,12 @@ pub async fn run_session<C, D>(
     let mut session = dispatch.open(&ctx);
 
     // Event funnel: every per-subscription pump sends its updates here; the main
-    // loop is the sole writer to the connection.
-    let (event_tx, mut event_rx) = mpsc::unbounded_channel::<SubEvent>();
+    // loop is the sole writer to the connection. **Bounded** so a slow client
+    // (one whose socket is backpressured, stalling the main loop) cannot grow the
+    // funnel without limit — the pumps drop on overflow rather than accumulate
+    // (events carry a monotonic `seq`, so a client can detect the gap). This
+    // restores the bounded-buffer slow-client protection the hand-rolled loops had.
+    let (event_tx, mut event_rx) = mpsc::channel::<SubEvent>(EVENT_BUFFER);
     // Per-connection subscription pumps; the engine future is their sole owner.
     let mut subs: FuturesUnordered<BoxFut<'static, ()>> = FuturesUnordered::new();
     // sub-id → cancel handle (dropping/sending the oneshot cancels the pump,
@@ -248,9 +257,10 @@ async fn send_reply_err<C: EnvelopeCodec + ?Sized>(
 async fn pump_subscription(
     sub_id: String,
     mut stream: BoxStream<'static, Payload>,
-    tx: mpsc::UnboundedSender<SubEvent>,
+    tx: mpsc::Sender<SubEvent>,
     mut cancel: oneshot::Receiver<()>,
 ) {
+    use tokio::sync::mpsc::error::TrySendError;
     let mut seq: u64 = 0;
     loop {
         tokio::select! {
@@ -260,8 +270,12 @@ async fn pump_subscription(
             next = stream.next() => match next {
                 Some(data) => {
                     seq += 1;
-                    if tx.send(SubEvent { sub: sub_id.clone(), seq, data }).is_err() {
-                        break; // funnel closed → connection gone
+                    // `try_send` keeps the pump non-blocking: a backpressured
+                    // funnel drops this update (slow-client protection) rather
+                    // than stalling the bus; only a closed funnel ends the pump.
+                    match tx.try_send(SubEvent { sub: sub_id.clone(), seq, data }) {
+                        Ok(()) | Err(TrySendError::Full(_)) => {}
+                        Err(TrySendError::Closed(_)) => break, // connection gone
                     }
                 }
                 None => break, // stream exhausted

aimdb-core/src/session/server.rs

@@ -71,3 +71,4 @@ tracing = { version = "0.1", optional = true }
 [dev-dependencies]
 tokio = { version = "1", features = ["full", "test-util"] }
 tokio-tungstenite = "0.26"
+aimdb-tokio-adapter = { version = "0.6.0", path = "../aimdb-tokio-adapter" }

aimdb-websocket-connector/Cargo.toml

@@ -215,4 +215,21 @@ mod tests {
         mgr.broadcast("t", b"v").await;
         assert_eq!(mgr.subscription_count(), 0);
     }
+
+    // Layer 2.2 (#2): one broadcast → N subscribers all receive the *same*
+    // pre-serialized bytes (a shared `Arc`), evidencing a single serialization
+    // regardless of subscriber count (O(1) fan-out, not O(N)).
+    #[tokio::test]
+    async fn broadcast_serializes_once_and_shares_to_all() {
+        let mgr = ClientManager::new(false);
+        let mut streams: Vec<_> = (0..8).map(|_| mgr.subscribe("#").1).collect();
+        mgr.broadcast("t", b"123").await;
+        let mut frames = Vec::new();
+        for s in &mut streams {
+            frames.push(s.next().await.unwrap());
+        }
+        // Every subscriber got byte-identical content (the one serialized frame).
+        let first = &frames[0];
+        assert!(frames.iter().all(|f| f.as_ref() == first.as_ref()));
+    }
 }

aimdb-websocket-connector/src/client_manager.rs

@@ -424,6 +424,27 @@ mod tests {
         }
     }
 
+    // Layer 2.3 (#1): decoding many *distinct*-topic Data frames must not
+    // accumulate any process-lifetime state (the old `leak_topic` interner would
+    // have grown one `&'static str` per topic here). The borrow is zero-copy.
+    #[test]
+    fn decode_outbound_high_cardinality_no_static_growth() {
+        let codec = WsCodec::new();
+        for i in 0..10_000 {
+            let topic = format!("sensors/dev-{i}");
+            let frame = serde_json::to_vec(&ServerMessage::Data {
+                topic: topic.clone(),
+                payload: Some(serde_json::json!(i)),
+                ts: 0,
+            })
+            .unwrap();
+            match codec.decode_outbound(&frame).unwrap() {
+                Outbound::Event { sub, .. } => assert_eq!(sub, topic),
+                _ => panic!("expected Event"),
+            }
+        }
+    }
+
     #[test]
     fn write_carries_payload() {
         let codec = WsCodec::new();