# TCPStreamer / TCPStreamerClient Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Add `TCPStreamer` (output) and `TCPStreamerClient` (input) MARTe2 DataSources that carry the existing UDPS framing over TCP, reusing `UDPStreamer`/`UDPStreamerClient` by making their transport pluggable.

**Architecture:** Two abstract interfaces (`UDPSServerI`, `UDPSClientI`) wrap the seven server calls and four client calls the datasources already use. The base datasources own the transport through a heap pointer that a subclass can swap (`SetServer`/`SetClient`) and expose two virtual hooks (`DefaultMaxPayloadSize`, `InitialiseTransport`). `TCPStreamer : UDPStreamer` and `TCPStreamerClient : UDPStreamerClient` plug in the new `TCPSServer` / `TCPSClient`.

**Tech Stack:** C++ (MARTe2 conventions, no STL in `Source/`), POSIX sockets, GTest (STL allowed in tests), MARTe2 make system.

Spec: `docs/superpowers/specs/2026-11-05-tcp-streamer-design.md`.

## Global Constraints

- No STL, no OS mutexes in `Source/Components/` (spec: "No STL, no OS mutexes; allocation only in `Start()`" — relaxed to "never on the RT thread"; slots grow lazily on the sender thread).
- Wire format unchanged: UDPS 17-byte header + CONFIG/DATA payload, `totalFragments` always 1.
- RT path (`Synchronise()`) must stay "spinlock + memcpy".
- Config keys: `Port` (44500), `MaxPayloadSize` (60000), `MaxClients` (16), `ClientQueueDepth` (8), `ClientStallTimeout` (5 s, 0 disables); client: `ServerAddress`, `Port`, `SilenceTimeout`, `ReconnectDelay`, `CPUMask`, `StackSize`.
- `TCPStreamer` rejects `MulticastGroup`, `Interface`, `DataPort`; `TCPStreamerClient` rejects the same.
- Do not `git add` unrelated working-tree changes (`Client/debugger/go.*`, `run_*.sh`, `Common/Client/c/Makefile`, ...). Add files by name. Commit only when the user asks.
- Build: `source env.sh`, then `make -f Makefile.gcc core test`. Run from repo root. `make clean` first in Task 1 (class layouts change; `Interfaces/UDPStream` has no depends file).

## File Structure

| File | Responsibility |
|---|---|
| `Source/Components/Interfaces/UDPStream/UDPSTransportI.h` (new) | `UDPSServerI`, `UDPSClientI` |
| `.../UDPStream/UDPSServer.h`, `UDPSClient.h` (modify) | implement the interfaces |
| `.../UDPStream/TCPSServer.{h,cpp}` (new) | multi-client non-blocking TCP server, per-client bounded queue |
| `.../UDPStream/TCPSClient.{h,cpp}` (new) | reconnecting TCP frame reader on its own thread |
| `.../UDPStream/Makefile.inc` (modify) | add `TCPSServer.x TCPSClient.x` |
| `Source/Components/DataSources/UDPStreamer/UDPStreamer.{h,cpp}` (modify) | transport via `UDPSServerI*`, hooks |
| `Source/Components/DataSources/UDPStreamerClient/UDPStreamerClient.{h,cpp}` (modify) | transport via `UDPSClientI*`, hooks |
| `Source/Components/DataSources/TCPStreamer/` (new) | `TCPStreamer`, Makefiles |
| `Source/Components/DataSources/TCPStreamerClient/` (new) | `TCPStreamerClient`, Makefiles |
| `Makefile.gcc` (modify) | build/clean the two new components |
| `Test/GTest/TCPTestUtil.h` (new) | raw-socket helpers shared by the new tests |
| `Test/GTest/TCPSServerGTest.cpp`, `TCPSClientGTest.cpp`, `TCPStreamerGTest.cpp` (new) | tests |
| `Test/GTest/Makefile.inc` (modify) | register tests, include paths |
| `Docs/TCPStreamer.md`, `CLAUDE.md`, spec (modify/new) | docs |

---

### Task 1: Pluggable transport seam (pure refactor, UDP behaviour unchanged)

**Files:** create `UDPSTransportI.h`; modify `UDPSServer.h`, `UDPSClient.h`, `UDPStreamer.{h,cpp}`, `UDPStreamerClient.{h,cpp}`.

**Interfaces:**
- Produces:
  ```cpp
  class UDPSServerI { virtual ~UDPSServerI(){}
    virtual bool Initialise(StructuredDataI&)=0; virtual bool Start()=0; virtual bool Stop()=0;
    virtual void ServiceClients()=0; virtual bool SendConfig(const uint8*,uint32)=0;
    virtual bool SendData(uint32 counter,const uint8*,uint32)=0;
    virtual bool HasClients() const=0; virtual bool IsMulticast() const=0; virtual uint16 GetPort() const=0; };
  class UDPSClientI { virtual ~UDPSClientI(){}
    virtual bool Initialise(StructuredDataI&)=0; virtual void SetListener(UDPSClientListener*)=0;
    virtual bool Start()=0; virtual bool Stop()=0; };
  ```
  `UDPStreamer` protected: `virtual uint32 DefaultMaxPayloadSize() const`, `virtual bool InitialiseTransport(StructuredDataI&)`, `void SetServer(UDPSServerI*)`, `UDPSServerI *Transport() const`.
  `UDPStreamerClient` protected: `virtual bool InitialiseTransport(StructuredDataI&)`, `void SetClient(UDPSClientI*)`, members `serverAddress, port, silenceTimeout, cpuMask, stackSize`.

- [ ] **Step 1: Baseline.** `make -f Makefile.gcc clean && make -f Makefile.gcc core test`, then run `./Build/x86-linux/GTest/MainGTest.ex` and record pass/fail counts.
- [ ] **Step 2:** Create `UDPSTransportI.h` (code above; forward-declare `class UDPSClientListener;`).
- [ ] **Step 3:** `UDPSServer.h`: `#include "UDPSTransportI.h"`, `class UDPSServer : public UDPSServerI`. `UDPSClient.h`: `class UDPSClient : public UDPSClientI, public EmbeddedServiceMethodBinderI`. Signatures already match; no `override` keyword (MARTe2 style).
- [ ] **Step 4:** `UDPStreamer`: replace `UDPSServer server;` by `UDPSServerI *server;` (private), constructor `server = new UDPSServer();`, destructor stops then deletes it after the executor is stopped, `server.X` -> `server->X` everywhere, `maxPayloadSize = DefaultMaxPayloadSize()` in `Initialise`, move the `serverCfg` block into `UDPStreamer::InitialiseTransport` (default impl, identical code, calling `server->Initialise`). `GetPort()` keeps returning the member.
- [ ] **Step 5:** `UDPStreamerClient`: same pattern with `UDPSClientI *client`; move the "Forward the transport parameters" block into `InitialiseTransport`; destructor `client->Stop()` when `receiverUp`, then delete.
- [ ] **Step 6:** `make clean && make core test`; run `MainGTest.ex`: counts must equal Step 1. Run `./Build/x86-linux/Test/Integration/Integration/IntegrationTests.ex` if it passed at baseline.

### Task 2: `TCPSServer`

**Files:** create `TCPSServer.{h,cpp}`, `Test/GTest/TCPTestUtil.h`, `Test/GTest/TCPSServerGTest.cpp`; modify `UDPStream/Makefile.inc`, `Test/GTest/Makefile.inc`.

**Design (see code for the authoritative version):**
- Raw POSIX sockets. Listener non-blocking, `SO_REUSEADDR`; `Port=0` binds an ephemeral port and `GetPort()` returns the bound port after `Start()`.
- Per client: `Slot[ClientQueueDepth+2]` queue (`buf, cap, len, isConfig`), `frontSent` (bytes of the front frame already written), `dataQueued`, `lastProgressTicks`.
- Queue capacity `depth+2` is exact: at most one data-in-flight/old frame, plus two CONFIG frames, plus `depth` DATA frames.
- `QueueConfig`: drop every queued frame except an in-flight one, append the CONFIG (CONFIG never dropped, never behind stale DATA).
- `QueueData`: while `dataQueued >= depth`, remove the oldest non-in-flight DATA (count a drop); if none is removable drop the new frame.
- `Flush`: `send(MSG_NOSIGNAL|MSG_DONTWAIT)` loop; `EAGAIN` stops; other errors evict.
- `ServiceClients`: accept (non-blocking), refuse beyond `MaxClients`, enqueue cached CONFIG on accept, detect peer close with `recv`, flush, evict clients with no progress for `ClientStallTimeout`, rate-limited (1/s) drop report.
- Extra key `SendBufferSize` (0 = OS default): `SO_SNDBUF` on accepted sockets, which also bounds latency and makes slow-client tests deterministic.
- `SendData` returns true unless the frame exceeds 1 MiB; per-client drops are reported through `GetClientDrops()`.

- [ ] **Step 1:** write `TCPSServerGTest.cpp` first: `Initialise` rejects (`MaxPayloadSize=0`, `ClientQueueDepth=0`, `MaxClients=0`), `ConfigBeforeData_LateJoin`, `MultipleClients_InOrder`, `ReconfigDeliveredToConnectedClient`, `MaxClientsRefused`, `PeerCloseEvicts`, `SlowClientDropsFastClientComplete` (200 x 60000-byte frames, `SendBufferSize=65536`, slow client `SO_RCVBUF=4096` never reads, fast client thread receives all counters 1..200 in order, `GetClientDrops() > 0`), `StallDisconnect` (`ClientStallTimeout=1`).
- [ ] **Step 2:** run -> compile failure (no `TCPSServer.h`).
- [ ] **Step 3:** implement `TCPSServer`; add to `OBJSX`.
- [ ] **Step 4:** `make -f Makefile.gcc core test && ./Build/x86-linux/GTest/MainGTest.ex --gtest_filter='TCPSServer*'` -> all pass.

### Task 3: `TCPSClient`

**Files:** create `TCPSClient.{h,cpp}`, `Test/GTest/TCPSClientGTest.cpp`; modify `UDPStream/Makefile.inc`, `Test/GTest/Makefile.inc` (already shared).

**Design:** `TCPSClient : UDPSClientI, EmbeddedServiceMethodBinderI`; `SingleThreadService`; `Execute` per iteration: if disconnected honour `ReconnectDelay`, then non-blocking `connect` + `poll` (bounded 500 ms) -> `OnUDPSConnected`; if connected `poll(50 ms)`, `recv` into a `HEADER+1 MiB` buffer, parse complete frames, bad magic / `payloadBytes > 1 MiB` / `totalFragments != 1` -> close + `OnUDPSDisconnected`; peer close/read error/`SilenceTimeout` the same. Keys: `ServerAddr`, `Port`, `SilenceTimeout` (float32 s, default 0 = off), `ReconnectDelay` (float32 s, default 2), `CPUMask`, `StackSize`. `Stop()` closes silently (no callbacks).

- [ ] **Step 1:** tests (`RecordingListener` with `std::mutex`): `ReceivesConfigThenData` (real `TCPSServer`), `ReconnectAfterServerRestart` (fixed port, `ReconnectDelay=0.2`), `RejectsBadMagic`, `RejectsOversizeLength`, `PartialFrameReads` (raw server writing a frame in 3 chunks), `SilenceTimeoutReconnects`, `StopIsPrompt`.
- [ ] **Step 2:** run -> compile failure. **Step 3:** implement. **Step 4:** `--gtest_filter='TCPSClient*'` passes.

### Task 4: `TCPStreamer`, `TCPStreamerClient`, build wiring

**Files:** create `Source/Components/DataSources/TCPStreamer/{TCPStreamer.h,TCPStreamer.cpp,Makefile.gcc,Makefile.inc}`, same for `TCPStreamerClient`, `Test/GTest/TCPStreamerGTest.cpp`; modify root `Makefile.gcc`, `Test/GTest/Makefile.inc`.

- [ ] **Step 1:** `TCPStreamerGTest.cpp`: `Constructor_Defaults` (`GetMaxPayloadSize()==60000`, `GetPort()==44500`), `Initialise_RejectsUdpOnlyKeys` (both classes), `Initialise_TcpKeys`, `ClassRegistered`, `StreamerToRawClient` (RealTimeApplication with `TCPStreamer`, raw client sees CONFIG then DATA while the test calls `Synchronise()`), `RoundTripStreamerToClient` (`TCPSServer`-driven client DS: send CONFIG+DATA, `Synchronise()` on a helper thread, check decoded value).
- [ ] **Step 2:** run -> compile failure. **Step 3:** implement both classes (each overrides `Initialise` to reject keys, `DefaultMaxPayloadSize`/`InitialiseTransport`, constructor swaps in the TCP transport). Makefile.inc copies UDPStreamer's, with `-lUDPStreamer -lUDPStream` (client: `-lUDPStreamerClient`), packages `Components/DataSources`.
- [ ] **Step 4:** add to root `Makefile.gcc` `core` and `clean`; if the `MainGTest` link reports undefined `UDPStreamer` symbols wrap `LIBRARIES_STATIC` in `-Wl,--start-group ... -Wl,--end-group`.
- [ ] **Step 5:** full `make core test`, whole `MainGTest.ex` and UDP datasource tests pass.

### Task 5: Docs and spec sync

- [ ] `Docs/TCPStreamer.md` (config tables, behaviour, service feasibility pointer), `CLAUDE.md` one-line mention, and update the spec for decisions taken while building: lazy slot growth, `SendBufferSize`, `GetPort()` returns the bound port, `SendData` true unless oversize, client `SilenceTimeout` default 0, no client-side gap handling, Strict frames may exceed `MaxPayloadSize` up to 1 MiB.

## Self-review

Spec coverage: pluggable transport (T1), `TCPSServer` incl. CONFIG-first, drop, stall (T2), `TCPSClient` (T3), both datasources + rejection of UDP-only keys + tests + regression (T4), docs/service study already in spec section 7 (T5). Naming checked: `UDPSServerI`, `UDPSClientI`, `SetServer`, `SetClient`, `Transport()`, `DefaultMaxPayloadSize`, `InitialiseTransport` are used identically across tasks.