Transport Layer

A transport is the component that moves bytes (or already-framed messages) between CycBox and a remote endpoint. Serial ports, TCP/UDP sockets, WebSockets, MQTT brokers, BLE peripherals, and peer-to-peer links are all exposed to the engine through a single, uniform set of traits.

This page describes the transport contract as defined in the SDK (cycbox-sdk/src/transport.rs). Each concrete transport is documented separately under this section.

Two kinds of transport

CycBox distinguishes between byte-stream transports and message transports:

• Byte-stream transports (serial, TCP, …) deliver an unstructured stream of bytes. They have no concept of a "message" — framing is the codec's job. These implement TransportIO and are wrapped by the engine in a CodecTransport to turn the byte stream into messages.
• Message transports (MQTT, …) already operate on discrete, self-delimiting units. A single MQTT publish is a message; there is no stream to frame. These implement MessageTransport directly.

Regardless of category, every transport ultimately presents the same MessageTransport interface to the engine, so the rest of the pipeline does not care how bytes arrived.

┌──────────────────────────────────────────────────────────┐
│                        Engine pipeline                     │
│                  (works only with messages)                │
└───────────────────────────┬────────────────────────────────┘
                            │ MessageTransport (recv / send)
            ┌────────────────┴───────────────────┐
            │                                     │
  ┌──────────▼──────────┐              ┌───────────▼──────────┐
  │   CodecTransport     │              │  Native message      │
  │  (adapter)           │              │  transport           │
  │  TransportIO + Codec │              │  (e.g. MQTT)         │
  └──────────┬──────────┘              └──────────────────────┘
            │ TransportIO (AsyncRead + AsyncWrite)
  ┌──────────▼──────────┐
  │  Byte stream         │
  │  (serial, TCP, …)    │
  └─────────────────────┘

The Transport trait

Transport is the factory entry point. The engine calls connect to obtain a live MessageTransport from a configuration and a codec.

#[async_trait]
pub trait Transport: Manifestable + Send + Sync {
    async fn connect(
        &self,
        configs: &[FormGroup],
        codec: Box<dyn Codec>,
        timeout: Duration,
    ) -> Result<Box<dyn MessageTransport>, CycBoxError>;
}

• configs — the user-supplied configuration values, as form groups produced from the transport's manifest. A TCP client reads its host/port here; a serial port reads its device path and baud rate.
• codec — the codec selected for this connection. Byte-stream transports hand this to a CodecTransport; message transports may use it to encode/decode payloads or ignore it.
• timeout — the read/idle timeout to apply (see timeout handling).

connect is responsible for establishing the link (opening the port, dialing the socket, subscribing to topics) and returning a ready-to-use transport, or a CycBoxError if the connection fails.

Because Transport requires Manifestable, every transport also advertises its configuration schema to the UI — see Manifest and configuration.

TransportIO

Byte-stream transports implement TransportIO, which is AsyncRead + AsyncWrite plus a few CycBox-specific hooks. Implementors do not implement MessageTransport themselves — they let CodecTransport do it.

#[async_trait]
pub trait TransportIO: AsyncRead + AsyncWrite + Send + Unpin {
    async fn handle_command(&mut self, _command: &Message) -> Option<Message> { None }
    fn take_session_boundary(&mut self) -> bool { false }
    async fn close(&mut self) {}
}

handle_command

Lets a transport answer control commands routed to it (for example a transport-specific action exposed in the UI). Return Some(response) to handle the command, or None to let the default handling apply. See command routing.

take_session_boundary

Returns true once (and clears its internal flag) when the underlying transport has just rotated to a new logical session without surfacing EOF to the upper layer — for example a server-style transport that handed off from one accepted client to the next on the same listening socket.

When this returns true, CodecTransport flushes in-flight decode state before consuming further bytes:

1. calls Codec::decode_eof to drain a possible final frame from the previous session,
2. calls Codec::reset,
3. clears the decode buffer.

This guarantees that a stateful codec (Modbus framing, AT line parsing, …) never carries one session's parser state into the next. Transports without session boundaries inherit the default false.

close

Cooperatively tears the transport down: release OS handles, terminate background bridge tasks, and await until cleanup completes. After close returns, further read/write calls may fail. The default is a no-op, which is correct for transports whose Drop impl already does everything synchronously (e.g. one owning a single OS handle).

MessageTransport

This is the interface the engine actually drives. Byte-stream transports reach it through CodecTransport; message transports implement it directly.

#[async_trait]
pub trait MessageTransport: Send + Unpin {
    async fn recv(&mut self) -> Result<Option<Message>, CycBoxError>;
    async fn send(&mut self, message: &mut Message) -> Result<(), CycBoxError>;
    async fn handle_command(&mut self, _command: &Message) -> Option<Message> { None }
    fn set_raw_observer(&mut self, _observer: Option<RawByteObserver>) {}
    async fn close(&mut self) {}
}

recv

Receives the next complete message. The return value encodes connection state:

Return	Meaning
Ok(Some(msg))	A message was received.
Ok(None)	The connection closed gracefully (EOF).
Err(e)	A connection error occurred.

A Ok(None) is the signal to the engine that the link is finished; an Err typically triggers the connection task's reconnect path.

send

Sends a complete message. The message is passed by &mut because the codec may populate frame/payload during encoding. Returns Ok(()) on success or Err on connection failure.

Discarded vs. failed writes

Server-style transports with no active peer signal "drop these bytes but stay alive" by returning a write error of kind NotConnected, which CodecTransport maps to CycBoxError::Discarded. Every other IO error becomes CycBoxError::Connection, which the engine treats as a real failure and reconnects.

handle_command

Handles a control command and optionally returns a response. The default returns None. See command routing for the full dispatch order.

set_raw_observer

Installs (or, with None, detaches) a RawByteObserver. Only stream-based transports backed by CodecTransport honour this; native message transports keep the default no-op.

close

Cooperatively tears the transport down, awaiting in-flight cleanup (background bridge tasks, remote-disconnect handshakes, …). The engine calls this before dropping the transport on a cooperative shutdown; an abort-based drop is the fallback if close takes too long. The default is a no-op for transports owning only synchronous OS handles (TCP, serial).

CodecTransport

CodecTransport is the adapter that turns a TransportIO byte stream plus a Codec into a full MessageTransport. Stream-based transports never implement message semantics themselves — they rely on this adapter, so framing logic lives in exactly one place.

pub struct CodecTransport {
    transport: Box<dyn TransportIO>,
    codec: Box<dyn Codec>,
    buffer: BytesMut,            // accumulated, not-yet-framed bytes
    read_buf: Vec<u8>,           // scratch buffer for each read
    timeout_duration: Duration,
    pending_messages: VecDeque<Message>,
    raw_observer: Option<RawByteObserver>,
}

It handles:

• reading bytes from the transport with a timeout,
• decoding bytes into messages via the codec,
• buffering multiple messages decoded from a single read,
• encoding outgoing messages via the codec.

Receive loop

The recv loop is the heart of the adapter. On each iteration:

1. Drain pending. If pending_messages is non-empty, pop and return the front one. A single read can yield several frames; they are returned one call at a time.
2. Read with timeout. Otherwise read from the transport, bounded by timeout_duration. The outcome drives one of the branches below.
3. EOF (Ok(0)). Try Codec::decode_eof to drain a trailing frame from the buffer, then reset the codec and clear the buffer. Return the drained message if any, else Ok(None).
4. Data (Ok(n)).
   • Surface the raw bytes to the observer before any decode work, so the UI sees them even if the codec mishandles the buffer.
   • If take_session_boundary reports a rotation, flush the previous session (decode_eof → reset → clear).
   • Append the new bytes to buffer (guarding against the 1 GiB MAX_BUFFER_SIZE ceiling), then call Codec::decode repeatedly until it returns Ok(None), pushing every decoded message into pending_messages.
   • Return the first message, or loop to read more if none completed yet.
5. Read error (Ok(Err)). Map to CycBoxError::Connection.
6. Timeout (Err(_)). Honour a session boundary if one appeared during the idle window, then give the codec a chance via Codec::decode_timeout (used by timeout-delimited protocols). If still nothing, loop and read again.

The loop is iterative, not recursive — repeated reads never grow the stack.

Send path

send calls Codec::encode, falls back to copying payload into frame when the codec produced no explicit frame, and writes the resulting bytes with write_all. Empty frames are skipped. The NotConnected → Discarded mapping described above is applied here.

Raw byte observation

Stream transports can expose the bytes they read from the wire to an external sink. The UI's Terminal tab uses this to show device output exactly as received, independent of how (or whether) the codec frames it.

pub struct RawBytes { pub timestamp: u64, pub bytes: Vec<u8> }   // timestamp = µs since UNIX_EPOCH

pub struct RawByteObserver { pub rx: mpsc::Sender<RawBytes> }

CodecTransport emits to the observer on the RX path only, with each chunk returned by a read, before decoding — so the observed bytes are exactly what was read from the medium, even if the codec later rejects or mishandles them. Bytes written on the TX path are not observed. Delivery uses try_send — if the channel is full the chunk is dropped rather than stalling the transport. Install or detach the observer with set_raw_observer.

The engine installs the observer only on connection 0, and only stream transports backed by CodecTransport honour it; native message transports (e.g. MQTT) produce no raw bytes. Each observed chunk is broadcast as a MESSAGE_TYPE_RAW_RX ("raw_rx") message with the bytes in frame, which the UI routes to the terminal rather than the message list.

The outgoing counterpart is the COMMAND_ID_SEND_RAW command, which writes bytes straight to the transport without passing through the codec.

Command handling and routing

Control commands (MESSAGE_TYPE_REQUEST) flow through handle_command. In a CodecTransport the dispatch order is:

1. COMMAND_ID_SEND_RAW — handled directly: the bytes parameter is written straight to the transport, bypassing the codec. Missing bytes yields an error response; a write failure yields an error response; otherwise a success response.
2. Codec — Codec::handle_command gets the next chance (e.g. a Modbus codec exposing a "read register" command).
3. Transport — finally TransportIO::handle_command is consulted.

The first layer to return Some(response) wins.

Timeout handling

The timeout passed to connect becomes the per-read idle timeout in CodecTransport. It serves two purposes:

• It bounds how long recv blocks waiting for bytes, keeping the connection task responsive.
• It drives timeout-delimited framing: when a read times out with a non-empty buffer, CodecTransport calls Codec::decode_timeout, letting protocols that delimit messages by an inter-byte gap (classic Modbus RTU, some AT flows) emit the buffered frame.

Manifest and configuration

Because Transport: Manifestable, each transport advertises a Manifest describing its configuration schema as FormGroups. The UI renders these into a dynamic form; the resulting values are passed back as the configs slice to connect. The manifest's category is PluginCategory::Transport.

Lifecycle summary

connect(configs, codec, timeout)
   │
   ├─ stream transport ─► wrap in CodecTransport(TransportIO + Codec)
   └─ message transport ─► return directly
   │
   ▼
loop:
   recv()  ──► Ok(Some(msg)) | Ok(None=EOF) | Err(connection)
   send(&mut msg)
   handle_command(cmd)  ──► SEND_RAW → codec → transport
   (set_raw_observer on connection 0 → raw_rx for the terminal)
   │
   ▼
close()   ──► cooperative teardown, then drop

Available transports

Each transport has its own page with configuration details:

Transport	Category	Notes
Serial port	byte stream	Device path, baud rate, parity, flow control.
TCP (client / server)	byte stream	Client dials; server accepts and rotates sessions.
UDP	byte stream	Datagram-oriented.
WebSocket (client / server)	byte stream	Text/binary frames over HTTP upgrade.
MQTT (client / server)	message	Publish/subscribe; topic & QoS carried as metadata.
BLE	byte stream	Bluetooth Low Energy characteristics.
P2P (client / server)	byte stream	Peer-to-peer links over an iroh relay.

Detailed per-transport documentation (configuration fields, behaviour, examples) is added in the pages that follow.