# TCPStreamer / TCPStreamerClient — design

Date: 2026-11-05

## Goal

A pair of MARTe2 DataSources, `TCPStreamer` (output) and `TCPStreamerClient`
(input), equivalent to `UDPStreamer` / `UDPStreamerClient` but carrying data
over TCP. Plus a feasibility study (section 7) of a single in-process service
that streams data from several DataSources over multiple threads.

## Decisions (agreed)

1. **Wire format**: the existing UDPS framing (17-byte header + CONFIG/DATA
   payload) over the TCP byte stream, `totalFragments` always 1. No new
   protocol; no fragmentation or reassembly.
2. **Slow consumers**: drop for the slow client only. Bounded per-client queue
   (`ClientQueueDepth` frames); a client whose queue is full loses frames
   (counted per client, logged at most once per second); a client stalled
   longer than `ClientStallTimeout` is disconnected. Other clients and the RT
   thread are never affected.
3. **Service**: an in-process MARTe2 `Interface` object (like `DebugService`),
   not a separate daemon. Investigated here, not built in v1.

Decisions made without asking (flagged, easy to reverse):

- v1 scope is the two DataSources plus their transport classes. Consumers
  other than `TCPStreamerClient` (StreamHub, Go decoder, C client, UDPScope)
  keep speaking UDP; adding a TCP ingest to them is a follow-up that needs no
  protocol change because of decision 1.
- The service is a design study only (section 7). v1 is built so the service
  can later reuse the same transport classes.

## 1. Architecture: pluggable transport, shared serialisation

`UDPStreamer` already isolates its transport behind seven calls on `UDPSServer`
(`Initialise`, `Start`, `Stop`, `ServiceClients`, `HasClients`, `SendConfig`,
`SendData`); `UDPStreamerClient` uses `UDPSClient` through four
(`Initialise`, `SetListener`, `Start`, `Stop`). Everything else in those files
(signal parsing, quantisation, Strict/Accumulate/Decimate, double buffering,
drop accounting, DATA decode) is transport-independent and ~1000 lines.

Forking the files would duplicate that. Instead:

- Add two small abstract interfaces next to the existing classes in
  `Source/Components/Interfaces/UDPStream/`:
  - `UDPSServerI` — the seven server calls above.
  - `UDPSClientI` — the four client calls above.
- `UDPSServer` / `UDPSClient` implement them (signature-only change; behaviour
  untouched).
- `UDPStreamer` / `UDPStreamerClient` hold a pointer to the interface, created
  by a protected virtual factory (`CreateServer()` / `CreateClient()`) that
  defaults to the UDP implementation.
- `TCPStreamer : UDPStreamer` and `TCPStreamerClient : UDPStreamerClient`
  override the factory, register as their own MARTe2 classes, and override
  `Initialise` only to parse the TCP-specific keys and reject the UDP-only ones.
- New transport classes `TCPSServer` and `TCPSClient` in the same
  `Interfaces/UDPStream` library.

Risk: this touches working RT code. Mitigation: the change to the UDP classes
is mechanical (pointer instead of member), and the existing GTests and the
E2E suite cover them; run both before and after.

## 2. TCPSServer

Single-threaded, driven from the owner's `Execute()` loop exactly like
`UDPSServer`; no thread per client.

- Listening socket, `SO_REUSEADDR`, non-blocking accept in `ServiceClients()`.
- Per client: non-blocking socket, `TCP_NODELAY`, ring of `ClientQueueDepth`
  frames (pre-allocated at `Start()`, each sized `UDPS_HEADER_SIZE +
  MaxPayloadSize`), partial-write cursor, drop counter, last-progress
  timestamp.
- On accept: enqueue the cached CONFIG first, so a client always sees CONFIG
  before any DATA. `SendConfig()` while clients are connected enqueues the new
  CONFIG on every client (same as UDP) and refreshes the cache.
- CONFIG never goes through the DATA ring: each client has a separate
  pending-CONFIG slot that is always flushed before any queued DATA, so
  CONFIG can't be dropped and a CONFIG change can't be reordered behind
  stale DATA.
- `SendData()`: builds one frame, copies it into each client's DATA ring
  (when full, the oldest queued DATA frame is discarded and counted, so a
  recovering client resumes near live), then attempts a non-blocking flush
  on each client. `ServiceClients()` also flushes, so a client that
  became writable drains without waiting for the next DATA.
- Stall: no bytes written for `ClientStallTimeout` while the queue is
  non-empty → close and evict. Peer close / error on write → evict.
- `Stop()` closes everything and frees buffers.
- No STL, no OS mutexes; allocation only in `Start()`.

Config keys (read by `TCPSServer::Initialise` from the datasource node):

| Key | Default | Meaning |
|---|---|---|
| `Port` | 44500 | TCP listen port |
| `MaxPayloadSize` | 60000 | Max payload bytes per frame; also bounds Accumulate batches |
| `MaxClients` | 16 | Simultaneous clients; extra connections refused |
| `ClientQueueDepth` | 8 | Frames queued per client before dropping |
| `ClientStallTimeout` | 5 | Seconds without progress before disconnect; 0 disables |

`MaxPayloadSize` keeps its UDP meaning only in that Accumulate sizes batches
from it; with no MTU to respect the default is raised, capped at
`UDPS_CLIENT_MAX_PACKET_BYTES` (1 MiB).

## 3. TCPStreamer

Same signal and publishing configuration as `UDPStreamer` (`PublishingMode`,
`MinRefreshRate`, `MaxBatchSize`, per-signal `Unit`/`Range*`/`QuantizedType`/
`TimeMode`/`TimeSignal`/`SamplingRate`, `CPUMask`, `StackSize`). Rejected as
invalid: `MulticastGroup`, `Interface`, `DataPort`. Class name `TCPStreamer`.

The CONFIG payload, including the `hrtFrequency` trailer, is unchanged.
Accessors (`GetPort`, `IsClientConnected`, `GetDroppedPublications`) are
inherited; add `GetClientDrops()` for the per-client queue drops.

## 4. TCPSClient and TCPStreamerClient

`TCPSClient` implements `UDPSClientI` and delivers via the existing
`UDPSClientListener` (`OnUDPSConfig`, `OnUDPSData`, `OnUDPSConnected`,
`OnUDPSDisconnected`), so `UDPStreamerClient`'s decode path is reused as is.

- Own thread (`SingleThreadService`). Connect with a bounded timeout; on
  success `OnUDPSConnected`.
- Frame reader: read the 17-byte header, check magic and
  `payloadBytes <= UDPS_CLIENT_MAX_PACKET_BYTES`, read the payload, dispatch
  by type. Reads are non-blocking with a short `poll`, so `Stop()` is prompt
  and a partial frame survives across iterations.
- Bad magic or oversize length → close and reconnect (stream is
  unrecoverable once out of sync).
- Peer close, read error, or `SilenceTimeout` with no bytes → close,
  `OnUDPSDisconnected`, retry after `ReconnectDelay`.
- No CONNECT/ACK/keepalive traffic: the connection itself is the session, and
  the server sends CONFIG on accept.
- Gaps: TCP is lossless but `TCPSServer` may drop frames; `GetLastDataGap()`
  is computed from the `counter` field exactly as for UDP, so timestamp
  reconstruction keeps working. Stale-packet gating is a no-op (ordered
  stream).

Keys: `ServerAddress` (default `127.0.0.1`), `Port`, `SilenceTimeout`,
`ReconnectDelay`, `CPUMask`, `StackSize`. `TCPStreamerClient` rejects the
multicast keys.

## 5. Error handling

- RT thread path unchanged: spinlock + memcpy in `Synchronise()`, nothing
  network-related.
- All socket failures are per client; a failing client never fails
  `SendData()` for the others, and `SendData()` returns true when at least
  one client accepted the frame or none are connected.
- Drops and stalls are counted and reported rate-limited (1/s), like
  `ReportDroppedPublications`.

## 6. Testing

- GTest for `TCPSServer` + `TCPSClient` over loopback: CONFIG-before-DATA on
  late join; multiple clients; slow client drops while a fast one receives all
  frames; stall disconnect; reconnect after server restart; garbage/oversize
  frame rejection; partial-frame reads.
- Datasource tests under `Test/Components/DataSources/TCPStreamer{,Client}`
  mirroring the UDP ones (Strict, Accumulate, Decimate, quantised signals,
  round trip streamer → client).
- Regression: existing UDP GTests and the E2E suite before and after the
  interface refactor.
- E2E scenario with a TCP chain is deferred with the consumer-side follow-up.

## 7. Single service, many DataSources, many threads (implemented as `TCPStreamService`; feasibility notes below)

**Verdict: feasible, moderate effort, and the v1 design is the right
stepping stone.**

Shape: `TCPStreamService` as a MARTe2 `Interface` singleton (registered like
`DebugService`). Config declares the port and the worker count. `TCPStreamer`
instances gain `Service = <name>` and, instead of owning a `TCPSServer`, call
`service->Publish(streamId, frame)` and `service->PublishConfig(...)`.

Threading:
- One accept thread. N worker threads (config `Workers`), each owning a shard
  of client connections and running the same flush loop as `TCPSServer`
  (poll-driven, non-blocking). Clients are assigned least-loaded at accept.
- Producers (the per-DataSource sender threads from `UDPStreamer::Execute`, not
  the RT threads) publish into a per-stream buffer. Each worker reads those
  buffers; handoff by a per-stream `FastPollingMutexSem`-guarded latest-frame
  slot plus a monotonically increasing sequence number, so a worker that falls
  behind skips to the newest frame rather than queueing — consistent with the
  drop policy. This keeps RT paths free of cross-thread contention beyond what
  exists today.
- Per-client queues stay private to the owning worker, so no locking on the
  send path.

Protocol gap: the UDPS header has no stream identifier. Two ways to carry
several DataSources on one connection:
1. **One port per stream** — no protocol change, but then there is no
   multiplexing and the service only saves threads.
2. **Add a stream id** — a new packet type `UDPS_TYPE_SUBSCRIBE`
   (client → server, lists stream names) and a 2-byte stream-id prefix on
   DATA/CONFIG payloads of multiplexed connections, negotiated by the
   SUBSCRIBE exchange so legacy single-stream clients are unaffected. This is
   a protocol change and must be mirrored across the C++, Go, C and JS
   parsers (see `CLAUDE.md`).

Recommendation: build v1 now; if the service is wanted, do it as a separate
spec choosing option 2, reusing `TCPSServer`'s per-client queue/flush code as
the worker's inner loop. A separate-process daemon was considered and
rejected: it overlaps StreamHub, adds IPC to the RT path and a deployment
unit, with the only gain being survival across application restarts.

Open for that follow-up spec: stream naming and collision rules across
DataSources, per-stream vs per-client queue accounting, and whether
subscribe-time CONFIG filtering is needed.

## 8. As implemented (deviations from sections 2-6)

Decisions taken while building v1; where they differ from the text above,
this section wins.

- Per-client queue: `ClientQueueDepth + 2` slots (up to `depth` DATA, one
  in-flight CONFIG, one pending CONFIG), grown lazily on the sender thread
  rather than pre-allocated at `Start()`. A new CONFIG discards queued frames
  that are not in flight (stale DATA) instead of being "flushed before DATA".
  DATA with a full queue drops the oldest non-in-flight DATA frame, or the
  incoming frame if only the in-flight one is queued.
- Extra key `SendBufferSize` (SO_SNDBUF for client sockets, 0 = OS default).
- `Port = 0` binds an ephemeral port; `TCPStreamer::GetBoundPort()` /
  `TCPSServer::GetPort()` return the bound port once started.
- `SendData()` returns false only for frames larger than 1 MiB
  (`TCPS_MAX_FRAME_PAYLOAD`). Frames may exceed `MaxPayloadSize`; it only
  sizes Accumulate batches.
- Extra accessors: `GetClientCount()`, `GetClientDrops()` (uint64),
  `GetStallEvictions()`.
- UDP-only keys rejected by `TCPStreamer`: `MulticastGroup`, `Interface`,
  `DataPort`, `ClientTimeout`. By `TCPStreamerClient`: `MulticastGroup`,
  `DataPort`, `Interface`, `KeepAliveInterval`.
- `TCPSClient` keys: `ServerAddr`, `Port`, `SilenceTimeout` (default 0 =
  disabled, TCP reports dead peers itself), `ReconnectDelay` (default 2 s),
  `CPUMask`, `StackSize`. `TCPStreamerClient` takes `ServerAddress` as the
  UDP one does.
- No client-side gap or stale-packet handling: the client delivers frames
  exactly as received. A bad magic, oversize length or fragmented frame drops
  the connection and counts a protocol error.
- Tests: `Test/GTest/TCPSServerGTest.cpp`, `TCPSClientGTest.cpp`,
  `TCPStreamerGTest.cpp` (shared helpers in `TCPTestUtil.h`), built into
  `MainGTest.ex`. Not yet covered: Accumulate/Decimate/quantised signals over
  TCP (serialisation is inherited unchanged from `UDPStreamer`), and the E2E
  suite.

## 9. Known gap

- Default `MaxPayloadSize` (60000) is a first guess; confirm against
  Accumulate batch latency during testing.