From 81879b8b7a81308ba13abe3b70f3824088afcd9f Mon Sep 17 00:00:00 2001
From: Martino Ferrari <martinogiordano.ferrari@iter.org>
Date: Fri, 6 Nov 2026 13:44:19 +0100
Subject: [PATCH] docs: describe the TCP transport, multiplexing and
 TCPStreamJoin in README, architecture and StreamHub docs

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 ARCHITECTURE.md             | 32 +++++++++++++++++++++++++++++++-
 Docs/E2E-Suite.md           |  5 +++++
 Docs/StreamHub-API.md       |  4 ++++
 Docs/StreamHub-UserGuide.md |  4 ++++
 README.md                   | 24 ++++++++++++++++++++++--
 5 files changed, 66 insertions(+), 3 deletions(-)

diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md
index 22c25ca..6957763 100644
--- a/ARCHITECTURE.md
+++ b/ARCHITECTURE.md
@@ -246,6 +246,32 @@ DebugServiceI *svc = DebugServiceI::GetInstance(); // broker wrapper retrieves i
 5. Applies quantization (`QuantizedType`, `RangeMin`, `RangeMax`) per signal
 6. Supports unicast and multicast UDP delivery (configurable per instance)
 
+### TCP variant
+
+`TCPStreamer` / `TCPStreamerClient` subclass the UDP DataSources and replace the transport
+(`TCPSServer` / `TCPSClient`) behind the `UDPSServerI` / `UDPSClientI` seam
+(`UDPSTransportI.h`). Framing is unchanged (`totalFragments` is always 1); each client has
+a bounded queue, so a slow client drops only its own frames.
+
+`TCPStreamService` (`Source/Components/Interfaces/TCPStreamService/`) is an optional
+in-process server: one acceptor plus `Workers` I/O threads serve many `TCPSServiceStream`s
+(one per `TCPStreamer` with `Service=`/`StreamName=`). Producers only copy frames into
+per-connection queues and wake the worker via a self-pipe. Lock order: topology, stream,
+connection. A client picks streams with a `SUBSCRIBE` frame (type 5):
+
+- **Multiplexing**: the payload may be a comma separated list; each server frame carries the
+  stream's list index in `fragmentIdx`; queues are accounted per stream. `TCPSClient`
+  demultiplexes into per-stream listeners (`GetStream(i)` facades), and StreamHub creates
+  one session per stream over one shared client.
+- **Joining**: `TCPStreamJoin` taps several streams (`TCPSStreamTapI`), watches the service
+  for (un)registration (`TCPSStreamObserverI`) and publishes a merged Strict stream: CONFIG =
+  concatenated descriptors named `<stream>_<signal>`; one DATA frame per master frame with
+  the other sources held at their last value. No extra thread. Accumulate sources and
+  differing HRT frequencies are refused. Lock order: source tap, join, output stream,
+  connection queue.
+
+See `Docs/TCPStreamer.md`.
+
 ### Packed Signals (Accumulate mode)
 
 When a signal has `NumberOfElements > 1` and `SamplingRate` is configured:
@@ -387,9 +413,13 @@ Hub = {
 }
 ```
 
+A TCP source sets `Transport = "tcp"` and, for a `TCPStreamService`, `StreamName = "<stream>"`;
+a comma list (`"a,b"`) takes several streams over one connection and creates one session per
+stream (label `<Label>_<stream>`, id `<node>_<stream>`).
+
 Sources added at runtime (`addSource`) get generated ids `s1, s2, …`;
 `saveSources` persists them to `SourcesFile` (JSON array of
-`{label, addr, multicastGroup?, dataPort?}`), reloaded at start-up.
+`{label, addr, multicastGroup?, dataPort?, transport?, streamName?}`), reloaded at start-up.
 
 `AllowedOrigins` is a comma/space-separated allowlist of `scheme://host[:port]`
 values accepted in the WebSocket `Origin` header (max 8 entries, 128 chars
diff --git a/Docs/E2E-Suite.md b/Docs/E2E-Suite.md
index 064c816..4e53c35 100644
--- a/Docs/E2E-Suite.md
+++ b/Docs/E2E-Suite.md
@@ -63,6 +63,11 @@ Full streaming pipeline: MARTe2 (FileReader -> IOGAM -> UDPStreamer) -> StreamHu
 (live, zoom, window, trigger). The validator compares the recording against the analytic
 ground truth (fidelity, sine shape fit, continuity) and optionally a fed-reference tap.
 
+Chain scenarios `s60`-`s73` use TCP transports (`network = "tcp"` or `"tcp_service"`). The
+optional `hub` field adds `{"mode": "mux"}` (the hub takes all sources over one multiplexed
+connection, `s72`) or `{"mode": "join"}` (a `TCPStreamJoin` merges the sources and the hub
+subscribes to the merge, `s73`).
+
 ### direct
 
 MARTe2 FileReader -> UDPStreamer -> UDPStreamerClient -> FileWriter round-trip. Validates that
diff --git a/Docs/StreamHub-API.md b/Docs/StreamHub-API.md
index 4b76410..3aa56cd 100644
--- a/Docs/StreamHub-API.md
+++ b/Docs/StreamHub-API.md
@@ -33,6 +33,9 @@ Reply (unicast): `{"type":"pong"}`.
 - `addr` — `host:port` of the UDPStreamer control port.
 - `multicastGroup` / `dataPort` — optional; if present the hub joins the
   multicast group for DATA and keeps a TCP control connection for CONNECT/CONFIG.
+- `transport` (`"udp"` default, or `"tcp"`) and `streamName` — for a TCP source, the
+  `TCPStreamService` stream to subscribe to. A comma list (`"a,b"`) multiplexes several
+  streams over one connection and adds one source per stream, labelled `<label>_<stream>`.
 - The hub assigns ids `s1`, `s2`, … and broadcasts an updated `sources` event.
 
 ### `removeSource`
@@ -421,6 +424,7 @@ warning.
 [
   {"label": "wave", "addr": "127.0.0.1:44500"},
   {"label": "mc",   "addr": "127.0.0.1:44501", "multicastGroup": "239.0.0.1", "dataPort": 44502},
+  {"label": "svc",  "addr": "127.0.0.1:44500", "transport": "tcp", "streamName": "fast,slow"},
   {"source": "wave", "signal": "Adc", "scale": 0.00030518, "offset": -1.25, "unit": "V"}
 ]
 ```
diff --git a/Docs/StreamHub-UserGuide.md b/Docs/StreamHub-UserGuide.md
index 96cb041..780c366 100644
--- a/Docs/StreamHub-UserGuide.md
+++ b/Docs/StreamHub-UserGuide.md
@@ -58,6 +58,10 @@ Useful flags: `-p <port>` WS port, `-n <pts>` ring size, `-s` skip rebuild,
    ./Client/streamhub/build/StreamHubClient -host <hub-host> -port 8090
    ```
 
+A source can use TCP instead of UDP: add `Transport = "tcp"` (and `StreamName = "<stream>"`
+for a `TCPStreamService`). `StreamName = "a,b"` takes several streams over one connection,
+shown as separate sources `<Label>_a`, `<Label>_b`. See `Docs/TCPStreamer.md`.
+
 Sources can also be added at runtime from either client ("Add source": label +
 `host:port`, optional multicast group/data port). "Save sources" persists the
 runtime list to `streamhub_sources.json` so the hub reloads it on restart.
diff --git a/README.md b/README.md
index ca9d933..a252346 100644
--- a/README.md
+++ b/README.md
@@ -12,6 +12,7 @@ This repository integrates two complementary capabilities:
 | Capability            | Component                | Purpose                                                                                                       |
 | --------------------- | ------------------------ | ------------------------------------------------------------------------------------------------------------- |
 | **Signal streaming**  | `UDPStreamer` DataSource | Continuously stream selected signals to a browser-based oscilloscope over UDP                                 |
+| **Signal streaming (TCP)** | `TCPStreamer` / `TCPStreamerClient` DataSources, `TCPStreamService`, `TCPStreamJoin` | Same UDPS stream over reliable TCP; one in-process server for many streams; multiplexed connections and merged streams |
 | **Signal debugging**  | `DebugService` Interface | On-demand signal tracing, value forcing, and conditional breakpoints — zero application code changes required |
 | **Sine generation**   | `SineArrayGAM`           | Generate continuous sine-wave arrays for testing and simulation                                               |
 | **Time stamping**     | `TimeArrayGAM`           | Provide time-reference arrays aligned to an RT cycle                                                          |
@@ -30,8 +31,11 @@ MARTe_Integrated_components/
 │   └── Client/go/                  Go packages: udpsprotocol, wshub
 ├── Source/Components/
 │   ├── DataSources/UDPStreamer/     Real-time UDP signal streaming DataSource
+│   ├── DataSources/TCPStreamer/     TCP variant of UDPStreamer (plus TCPStreamerClient)
 │   ├── GAMs/SineArrayGAM/          Sine-wave array generator GAM
 │   ├── GAMs/TimeArrayGAM/          Time-reference array GAM
+│   ├── Interfaces/UDPStream/       UDPS server/client transports (UDP and TCP)
+│   ├── Interfaces/TCPStreamService/ In-process multi-stream TCP server + TCPStreamJoin
 │   ├── Interfaces/DebugService/    Signal tracing/forcing/breakpoint Interface
 │   └── Interfaces/TCPLogger/       TCP log-forwarding LoggerConsumerI
 ├── Test/
@@ -59,6 +63,21 @@ sending a `CONNECT` packet; the server then sends `CONFIG` (signal metadata) and
 
 See `Docs/UDPStreamer.md` and `Docs/Protocol.md`.
 
+### TCPStreamer / TCPStreamService / TCPStreamJoin
+
+`TCPStreamer` and `TCPStreamerClient` are the TCP counterparts of the UDP DataSources: same
+UDPS framing, per-client bounded queues (a slow client only drops its own frames).
+`TCPStreamService` is an optional single in-process server (acceptor + worker threads)
+shared by many `TCPStreamer`s; clients pick a stream by name with a `SUBSCRIBE` frame.
+
+- **Multiplexing**: a `SUBSCRIBE` may list several streams (`"fast,slow"`) to carry them over
+  one socket, each with its own CONFIG and rate. StreamHub does this for a comma separated
+  `StreamName`.
+- **Joining**: `TCPStreamJoin` publishes a new stream merging several streams of the service
+  (`<stream>_<signal>`, sample-and-hold, paced by a master source).
+
+See `Docs/TCPStreamer.md`.
+
 ### SineArrayGAM
 
 Generates a continuous float32 sine-wave array every RT cycle. Used as a signal
@@ -119,7 +138,7 @@ clients on a configurable port. Works as a sidecar to `DebugService`.
 ### StreamHub Application
 
 Headless C++ hub (`Source/Applications/StreamHub/`) that aggregates multiple
-UDPStreamer sources and serves them to oscilloscope clients over WebSocket
+UDPStreamer / TCPStreamer sources (`Transport = "tcp"`, optional `StreamName`) and serves them to oscilloscope clients over WebSocket
 (port 8090): ring buffers with wall-clock time calibration, LTTB decimation,
 hub-side trigger engine, per-window zoom. Clients: browser SPA
 (`Client/webui` + `Client/udpstreamer/static`) and native ImGui desktop client
@@ -132,7 +151,7 @@ See `Docs/StreamHub-UserGuide.md`, `Docs/StreamHub-API.md` and
 ### UDPS Protocol
 
 The `Common/UDP/UDPSProtocol.h` header defines the shared binary wire format used by
-both `UDPStreamer` and `DebugService`. It is intentionally free of MARTe2-specific
+`UDPStreamer`, `TCPStreamer` and `DebugService`. It is intentionally free of MARTe2-specific
 dependencies so it can also be used by Go clients (via `Common/Client/go/udpsprotocol`).
 
 See `Docs/Protocol.md`.
@@ -244,6 +263,7 @@ Open `http://localhost:9090`, explore the object tree, trace signals, force valu
 | ----------------------------- | -------------------------------------------------------------- |
 | `Docs/Protocol.md`            | UDPS binary wire protocol specification                        |
 | `Docs/UDPStreamer.md`         | UDPStreamer DataSource configuration reference                 |
+| `Docs/TCPStreamer.md`         | TCPStreamer, TCPStreamService, multiplexing and TCPStreamJoin   |
 | `Docs/UDPS-C-Client.md`       | Standalone C/C++ UDPS receiver library (`Common/Client/c`)     |
 | `Docs/SineArrayGAM.md`        | SineArrayGAM configuration reference                           |
 | `Docs/DebugService.md`        | DebugService TCP API and architecture                          |