MARTe-Integrated-Components/docs/superpowers/plans/2026-11-05-tcp-streamer.md

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:

  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.