MARTe-Integrated-Components/Docs/TCPStreamer.md

# TCPStreamer / TCPStreamerClient

TCP counterparts of `UDPStreamer` / `UDPStreamerClient`. Same signals,
quantisation, time modes, publishing modes and UDPS wire format; only the
transport differs. Use them when you need lossless, ordered delivery to a few
consumers and do not need multicast.

## How it works

- `TCPStreamer` derives from `UDPStreamer`; `TCPStreamerClient` from
  `UDPStreamerClient`. They install a `TCPSServer` / `TCPSClient` through the
  `UDPSServerI` / `UDPSClientI` interfaces (`UDPSTransportI.h`), so the RT path
  (`Synchronise()` = spinlock + memcpy) and serialisation are shared.
- Frames are UDPS frames (17-byte header + payload) on the byte stream, with
  `totalFragments` always 1. The server sends CONFIG on accept; there is no
  CONNECT / ACK / keepalive.
- Each client has its own bounded queue (`ClientQueueDepth` DATA frames plus
  CONFIG slots). If a client cannot keep up, only its own oldest DATA frames
  are dropped (counted by `GetClientDrops()`, logged at most once per second).
  A client with no progress for `ClientStallTimeout` seconds is disconnected.
  The RT thread and other clients are never affected.
- A new CONFIG (reconfiguration) discards the client's stale queued DATA.
- The client reconnects automatically after any disconnect or protocol error
  (bad magic, oversize length, fragmented frame).

## TCPStreamer configuration

All `UDPStreamer` keys apply (see `UDPStreamer.md`) except the UDP-only
`MulticastGroup`, `Interface`, `DataPort`, `ClientTimeout`, which are rejected.

| Key | Default | Meaning |
|---|---|---|
| `Port` | 44500 | TCP listen port (0 = ephemeral, see `GetBoundPort()`) |
| `MaxPayloadSize` | 60000 | Frame budget; sizes Accumulate batches. Frames up to 1 MiB are sent |
| `MaxClients` | 16 | Simultaneous clients (1..1024); extra connections are closed |
| `ClientQueueDepth` | 8 | DATA frames queued per client (1..4096) |
| `ClientStallTimeout` | 5 | Seconds without progress before disconnect; 0 disables |
| `SendBufferSize` | 0 | `SO_SNDBUF` for client sockets; 0 keeps the OS default |

Accessors: `GetBoundPort()`, `GetClientCount()`, `GetClientDrops()`,
`GetStallEvictions()`, plus the inherited `GetDroppedPublications()`.

```
+Streamer = {
    Class = TCPStreamer
    Port = 44500
    ClientQueueDepth = 16
    Signals = {
        Time  = { Type = uint64 }
        Value = { Type = float32 }
    }
}
```

## TCPStreamerClient configuration

| Key | Default | Meaning |
|---|---|---|
| `ServerAddress` | 127.0.0.1 | Server IPv4 address or host name |
| `Port` | 44500 | Server TCP port |
| `SilenceTimeout` | 0 | Seconds without any byte before reconnecting; 0 disables |
| `ReconnectDelay` | 2 | Seconds between connection attempts |
| `CPUMask`, `StackSize` | default | Receiver thread |

`MulticastGroup`, `DataPort`, `Interface`, `KeepAliveInterval` are rejected.
Accessors: `IsConnected()`, `GetConnectionCount()`, `GetProtocolErrors()`.

The client delivers frames as received; it does no gap or stale-packet
handling.

## Service mode: one in-process server, many streams

`TCPStreamService` is a root-level object that owns the listening socket and a
pool of worker threads. Any number of `TCPStreamer` DataSources (each on its own
real-time thread) publish into it; clients choose a stream by name.

```
+Svc = { Class = TCPStreamService Port = 44500 Workers = 2 }
...
+Fast = { Class = TCPStreamer Service = "Svc" StreamName = "fast" Signals = { ... } }
+Slow = { Class = TCPStreamer Service = "Svc" StreamName = "slow" Signals = { ... } }
+Rx   = { Class = TCPStreamerClient Port = 44500 StreamName = "fast" Signals = { ... } }
```

- Wire protocol: plain UDPS over TCP plus one client-to-server frame,
  `SUBSCRIBE` (type 5, payload = stream name, 1..255 bytes, or a comma separated
  list, see "Joining streams"). The service answers with that stream's CONFIG
  and then its DATA frames. A connection that does not subscribe within `HandshakeTimeout`, or
  names an unknown stream, is closed (the client reconnects).
- Threads: one acceptor plus `Workers` I/O threads; each connection lives on one
  worker. Producers only copy frames into per-connection bounded queues
  (drop-oldest) and wake the worker through a self-pipe, so a slow client never
  delays a producer or another client. Lock order: topology, stream, connection.
- A stream registers at `Start()` (looked up with
  `ObjectRegistryDatabase::Find(Service)`) and unregisters on stop.
- `StreamName` defaults to the DataSource name (max 255 chars). In service mode
  `Port`, `MaxClients`, `ClientQueueDepth`, `ClientStallTimeout` and
  `SendBufferSize` belong to the service and are rejected on the streamer.
- `TCPStreamerClient` / `TCPSClient` send SUBSCRIBE after every (re)connect when
  `StreamName` is set. A plain `TCPSServer` ignores it, so the same client works
  against both.

`TCPStreamService` keys:

| Key | Default | Meaning |
|---|---|---|
| `Port` | 44500 | TCP listen port |
| `Workers` | 2 | I/O threads (1..32) |
| `MaxClients` | 64 | Total simultaneous clients (1..1024) |
| `MaxStreams` | 32 | Streams that can register (1..256) |
| `ClientQueueDepth` | 8 | DATA frames queued per client (1..4096) |
| `ClientStallTimeout` | 5 | Seconds without progress before eviction; 0 disables |
| `HandshakeTimeout` | 5 | Seconds a new connection has to SUBSCRIBE (>= 1) |
| `SendBufferSize` | 0 | `SO_SNDBUF`; 0 keeps the OS default |
| `CPUMask`, `StackSize` | default | Worker threads |

## Joining streams

Two independent ways to take several service streams through one cheaper path.

### Multiplexed connection

`SUBSCRIBE` may carry a comma separated list (`"fast,slow"`; at most 64 names,
4096 payload bytes, no duplicates, no `,` in a stream name). One socket then
carries every listed stream, each keeping its own CONFIG, rate and timeline:

- The index of a stream is its position in the list; every server frame carries
  it in `fragmentIdx` (`totalFragments` stays 1). A single name is index 0, so
  existing clients are unchanged. One unknown name closes the connection.
- Queues are accounted per stream: `ClientQueueDepth` DATA frames *per stream*,
  drop-oldest within the same stream only, so a fast stream never evicts a slow
  stream's frames. Queued frames are written with one `sendmsg` (up to 16).
- `TCPSClient` accepts `StreamName = "a,b,c"`, demultiplexes by `fragmentIdx`
  (`SetStreamListener(i, ...)`) and tracks the counter gap per stream.
  `GetStream(i)` returns a `UDPSClientI` facade whose `Start()`/`Stop()` are
  reference counted on the shared connection. `TCPStreamerClient` is
  single-stream and rejects a comma.
- `GetClientCount()` of a service counts (connection, stream) subscriptions.

### Merged stream: `TCPStreamJoin`

```
+Svc  = { Class = TCPStreamService Port = 44500 }
+Fast = { Class = TCPStreamer Service = "Svc" StreamName = "fast" ... }
+Slow = { Class = TCPStreamer Service = "Svc" StreamName = "slow" ... }
+Join = { Class = TCPStreamJoin Service = "Svc" StreamName = "all"
          Sources = "fast,slow" Master = "fast" }
```

A root-level object, declared after the service, that publishes a new stream
`all` holding the signals of several streams of the same service:

- CONFIG: source descriptors concatenated in `Sources` order, named
  `<stream>_<signal>` (at most 63 characters), `timeSignalIdx` rebased, publish
  mode Strict, one HRT frequency (sources with different known frequencies are
  refused; an unknown one matches any).
- DATA: one frame per `Master` frame, `[master timestamp][source0 body][source1
  body]...`, a body being that source's last DATA payload without its 8-byte
  timestamp (sample-and-hold). Nothing is emitted until every source has
  delivered one frame. A held value is stamped with the master's time unless the
  source has a time signal. No thread is added: the frame is built on the master
  producer's thread.
- Only Strict and Decimate sources can be merged; a source whose body size
  changes without a new CONFIG is ignored; the merged frame must fit
  `MaxPayloadSize` (default 60000).
- The merged stream is registered while every source is registered, removed as
  soon as one leaves (its subscribers are dropped and reconnect) and rebuilt when
  a source re-registers or changes its CONFIG. A source with a join attached
  keeps publishing even with no socket subscriber.
- Keys: `Service` (required), `StreamName` (default object name), `Sources`
  (required, 1..16), `Master` (default first source), `MaxPayloadSize`.

Design: `docs/superpowers/specs/2026-11-06-tcp-stream-join-design.md`.

## StreamHub

A StreamHub source can consume over TCP: in the config `Sources` block (and in
`addSource` / the `SourcesFile`) set `Transport = "tcp"` (default `"udp"`) and,
for a service, `StreamName = "<stream>"`. `Addr` is then the TCP server
address. The hub uses `TCPSClient`, which reports the UDPS counter gap so the
accumulate-mode timing estimator tolerates dropped frames.

`StreamName = "a,b,c"` takes several streams over one multiplexed connection and
creates one session per stream, labelled `<Label>_<stream>` (just the stream
name without a label). Session ids are `<node>_<stream>` for a config `Sources`
node and `s<N>` for `addSource`.

## Tests

- GTest: `Test/GTest/TCPSServerGTest.cpp`, `TCPSClientGTest.cpp`,
  `TCPStreamerGTest.cpp` (helpers in `TCPTestUtil.h`) and
  `TCPStreamServiceGTest.cpp` (queue, protocol, multiplexing, joins,
  multi-stream/worker, stalls).
- E2E (`Test/E2E/suite/scenarios.py`): `s58_direct_tcp`, `s59_direct_tcp_service`
  (FileReader to FileWriter round trip) and chain scenarios `s60`-`s65` (plain
  TCP: scalar, full array, accumulate, decimate/quantise, multi-source,
  high rate), `s66`-`s71` (service: same features plus 3- and 6-stream
  fan-in), `s72` (hub on one multiplexed connection) and `s73` (merged stream). Scenario configs are generated by `gen_cfg.py`
  (`network = "tcp"` / `"tcp_service"`).

## Design

`docs/superpowers/specs/2026-11-05-tcp-streamer-design.md` (section 7 covers
the single-service design implemented by `TCPStreamService`).