docs: TCPStreamer source kind spec and plan

This commit is contained in:
Martino Ferrari committed 2026-11-06 13:41:31 +01:00
1 parent d47704c6ec
commit 21b1945769
2 files changed
+643

No files matched your search

@@ -0,0 +1,592 @@
# TCPStreamer Source Kind 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 a `tcps` source kind that receives a MARTe2 `TCPStreamer` DataSource, standalone or through a `TCPStreamService`.
**Architecture:** `deps/udpsclient` gets an additive TCP transport mode (connect, optional SUBSCRIBE, framed reads reusing `read_tcp_frame`). `UdpsSource` takes a transport argument so the TCP kind reuses its callbacks. A new `tcps_kind.cpp` supplies the form and registration. The headless self-check gets a `[tcps]` probe against a new Python mock server.
**Tech Stack:** C99 (`deps/udpsclient`), C++17 + GTK4/libadwaita (`src/`), Python 3 + numpy (mock), CMake.
Spec: `docs/superpowers/specs/2026-11-06-tcp-streamer-source-design.md`.
## Global Constraints
- There is no unit test suite; verification is the app's own headless self-check under `xvfb-run` against a mock. Read `docs/claude/verification.md` (baselines) before Task 3 and run probes **short and serial**, killing the mock afterwards.
- Baselines for every run: exactly **one** build warning (the pre-existing one; record the count before starting), and no `Gtk-CRITICAL`.
- Every new source file starts with the two-line SPDX header: `// SPDX-License-Identifier: MIT` / `// SPDX-FileCopyrightText: 2026 Martino Ferrari` (`#` comments in Python, after the shebang).
- The GTK thread reaches a source only through `DataSource::post()`; `udps_*` names stay in `src/source/udps_*.cpp` and the new `tcps` files may include `udps_source.hpp` only.
- One connection carries one stream; multiplexed `SUBSCRIBE` lists are out of scope.
- Session naming for a non-UDPS kind is automatic (`session_host` -> `tcps-<identity>`, port 1, `source_kind=tcps`); do not touch `registry.cpp`'s session functions.
- A kind id is `[a-z0-9]+`: `tcps`. Display name: `TCPStreamer`. Default port 44500.
- Do not commit unless the task step says so.
## File Structure
| File | Change | Responsibility |
| --- | --- | --- |
| `deps/udpsclient/udps_client.h` | modify | `use_tcp`, `stream_name` config fields |
| `deps/udpsclient/udps_client.c` | modify | `connect_tcp()`, dispatch, no DISCONNECT frame in TCP mode, reorder window 0 |
| `tools/mock_tcpstreamer.py` | create | Mock TCPStreamer / TCPStreamService server, reuses `mock_udpstreamer.py` frame builders |
| `src/source/udps_source.hpp/.cpp` | modify | `Transport` enum, constructor argument, accessor |
| `src/source/tcps_kind.cpp` | create | `tcps` `SourceKind` and its form |
| `src/source/registry.hpp/.cpp` | modify | `make_tcps_kind()`, registration |
| `CMakeLists.txt` | modify | add `tcps_kind.cpp` |
| `src/ui_probe.cpp`, `src/ui.cpp` | modify | `[source-kinds]` sample, `[tcps]` probes and their env scheduling |
| `README.md`, `docs/claude/*.md` | modify | document the kind, the upstream change, the probes |
---
### Task 1: TCP transport in `deps/udpsclient`, and the mock server
**Files:**
- Modify: `deps/udpsclient/udps_client.h` (struct `udps_client_config_t`, around line 263-282)
- Modify: `deps/udpsclient/udps_client.c` (struct `udps_client` ~line 166, `send_control` ~1021, `do_connect` ~1076, `do_disconnect` ~1093, `udps_client_create` ~1260)
- Create: `tools/mock_tcpstreamer.py`
**Interfaces:**
- Produces: `udps_client_config_t::use_tcp` (`int`, 0 = UDP as before) and `::stream_name` (`const char *`, NULL or empty = no SUBSCRIBE). Both default to zero/NULL through `udps_client_config_init()` (it memsets). `server_port` is the TCP port in TCP mode.
- Produces: `tools/mock_tcpstreamer.py --port P [--stream NAME] [--rate R] [--batch B] [--close-after S] [--drop-every N]`.
- [ ] **Step 1: Add the config fields** at the end of `udps_client_config_t` in `udps_client.h`, after `reorder_window_s`:
```c
/**
* Non-zero selects the TCP transport (a MARTe2 TCPStreamer / TCPStreamService):
* @c server_port is the TCP port, no CONNECT or DISCONNECT frame is sent, the
* server streams CONFIG then DATA on connect, and the reorder window is forced
* to 0 (TCP is ordered, so a counter gap is a server-side drop). @c multicast_group,
* @c interface_addr and @c data_port are ignored.
*/
int use_tcp;
/**
* TCP mode only: when non-empty (1..255 bytes) a SUBSCRIBE frame naming this stream is
* sent right after every connect, as a TCPStreamService requires. A plain
* TCPStreamer ignores it. NULL or "" sends nothing.
*/
const char *stream_name;
```
- [ ] **Step 2: Own a copy of the name.** In `struct udps_client`, after `char iface_addr[128];` add `char stream_name[256];`. In `udps_client_create`, next to the other `copy_str` calls add `copy_str(c->stream_name, sizeof c->stream_name, cfg->stream_name);` and after the `c->cfg.interface_addr = ...` line add:
```c
c->cfg.stream_name = c->stream_name;
if (c->cfg.use_tcp) {
c->cfg.multicast_group = NULL;
c->mcast_group[0] = '\0';
c->cfg.reorder_window_s = 0.0;
}
```
Also reject a name over 255 bytes: before `copy_str`, `if (cfg->use_tcp && cfg->stream_name != NULL && strlen(cfg->stream_name) > 255u) { free(c); return NULL; }` (place right after the `calloc` succeeds).
- [ ] **Step 3: Add `connect_tcp()`** after `connect_multicast` (before `do_connect`):
```c
/** TCP transport: connect, then SUBSCRIBE when a stream name is set. */
static int connect_tcp(udps_client_t *c) {
struct timeval tv;
size_t name_len = strlen(c->stream_name);
c->tcp_fd = socket(AF_INET, SOCK_STREAM, 0);
if (c->tcp_fd < 0) {
return fail(c, "TCP socket() failed: %s", strerror(errno));
}
set_recv_buffer(c->tcp_fd, c->cfg.recv_buffer_bytes);
/* Linux bounds a blocking connect() with SO_SNDTIMEO, so an unreachable host
* cannot hold the polling thread for minutes. */
tv.tv_sec = 3;
tv.tv_usec = 0;
(void)setsockopt(c->tcp_fd, SOL_SOCKET, SO_SNDTIMEO, &tv, (socklen_t)sizeof tv);
if (connect(c->tcp_fd, (struct sockaddr *)&c->server_sa,
(socklen_t)sizeof c->server_sa) != 0) {
int e = errno;
close_sockets(c);
return fail(c, "TCP connect to %s:%u failed: %s", c->server_addr,
(unsigned)c->cfg.server_port, strerror(e));
}
/* A half-received frame must not wedge the poll loop (read_exact_tcp blocks). */
tv.tv_sec = 1;
(void)setsockopt(c->tcp_fd, SOL_SOCKET, SO_RCVTIMEO, &tv, (socklen_t)sizeof tv);
if (name_len > 0u) {
uint8_t pkt[UDPS_HEADER_SIZE + 255u];
build_header(pkt, UDPS_PKT_SUBSCRIBE, 0u, 0u, 1u, (uint32_t)name_len);
memcpy(pkt + UDPS_HEADER_SIZE, c->stream_name, name_len);
if (send(c->tcp_fd, pkt, UDPS_HEADER_SIZE + name_len, MSG_NOSIGNAL) !=
(ssize_t)(UDPS_HEADER_SIZE + name_len)) {
int e = errno;
close_sockets(c);
return fail(c, "SUBSCRIBE failed: %s", strerror(e));
}
}
return 0;
}
```
Add `UDPS_PKT_SUBSCRIBE = 5 /**< Client -> server (TCP stream service): payload = stream name. */` to the packet-type enum in `udps_client.h` (add a comma after the `UDPS_PKT_DISCONNECT = 4` line).
- [ ] **Step 4: Dispatch and disconnect.** Replace the first lines of `do_connect`:
```c
int rc = c->cfg.use_tcp ? connect_tcp(c)
: (c->mcast_group[0] != '\0') ? connect_multicast(c)
: connect_unicast(c);
```
In `do_disconnect`, replace `(void)send_control(c, UDPS_PKT_DISCONNECT);` with `if (!c->cfg.use_tcp) { (void)send_control(c, UDPS_PKT_DISCONNECT); }` (a write to a peer that closed would raise SIGPIPE and kill the application; the server detects the close itself).
- [ ] **Step 5: Confirm the other TCP-path assumptions by reading, not guessing.** Read `deliver_payload` (~line 848) and confirm that with `reorder_window_s == 0` a counter jump delivers the frame with `lost > 0` and `stats.counter_gaps` incremented, and that `read_tcp_frame` -> `handle_packet` needs no change. If either is false, stop and report; do not patch around it.
- [ ] **Step 6: Write the mock** `tools/mock_tcpstreamer.py` (`chmod +x`):
```python
#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
# SPDX-FileCopyrightText: 2026 Martino Ferrari
"""A minimal MARTe2 TCPStreamer / TCPStreamService producer, for exercising udpscope.
Listens on `--port`, serves one client at a time with the same frames as
mock_udpstreamer.py (CONFIG, then Accumulate DATA packets of sine/square/ramp/noise), each
framed UDPS-over-TCP. Without `--stream` it behaves like a standalone TCPStreamer: the
stream starts at once and anything the client sends is ignored. With `--stream NAME` it
behaves like a TCPStreamService: the client must open with a SUBSCRIBE frame naming NAME
within 5 s, otherwise the connection is closed.
./tools/mock_tcpstreamer.py --port 44500 --stream fast --close-after 6
"""
import argparse
import os
import select
import socket
import struct
import sys
import time
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import mock_udpstreamer as m # noqa: E402
PKT_SUBSCRIBE = 5
HEADER = 17
def read_exact(conn, n, deadline):
buf = b""
while len(buf) < n:
left = deadline - time.monotonic()
if left <= 0:
return None
r, _, _ = select.select([conn], [], [], left)
if not r:
return None
chunk = conn.recv(n - len(buf))
if not chunk:
return None
buf += chunk
return buf
def subscribed_to(conn, want):
"""True once a SUBSCRIBE frame naming `want` arrived within 5 s."""
deadline = time.monotonic() + 5.0
head = read_exact(conn, HEADER, deadline)
if head is None:
return False
magic, ptype, _, _, _, plen = struct.unpack("<IBIHHI", head)
if magic != m.MAGIC or ptype != PKT_SUBSCRIBE or not 0 < plen <= 4096:
return False
name = read_exact(conn, plen, deadline)
return name is not None and name.decode(errors="replace") == want
def serve(conn, args):
dt = 1.0 / args.rate
period = args.batch / args.rate
conn.sendall(m.build_config(args.rate, m.TIME_FULL_ARRAY, "V"))
counter, t_stream = 1, 0.0
started = time.monotonic()
next_send = started
while True:
now = time.monotonic()
if args.close_after > 0 and now - started >= args.close_after:
print("mock tcpstreamer: closing the connection (--close-after)", flush=True)
return
if now < next_send:
time.sleep(min(next_send - now, 0.002))
continue
pkt = m.build_data(counter, t_stream, dt, args.batch, args.freq)
if not (args.drop_every > 0 and counter % args.drop_every == 0):
conn.sendall(pkt)
counter += 1
t_stream += args.batch * dt
next_send += period
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--port", type=int, default=44500)
ap.add_argument("--stream", default="", help="TCPStreamService mode: require SUBSCRIBE of this name")
ap.add_argument("--rate", type=float, default=10000.0, help="samples per second")
ap.add_argument("--batch", type=int, default=100, help="samples per packet")
ap.add_argument("--freq", type=float, default=5.0, help="waveform frequency, Hz")
ap.add_argument("--close-after", type=float, default=0.0, metavar="S",
help="close each connection after S seconds (the client must reconnect); 0 = never")
ap.add_argument("--drop-every", type=int, default=0, metavar="N",
help="skip sending every Nth packet but advance the counter (a server-side drop); 0 = never")
args = ap.parse_args()
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.bind(("0.0.0.0", args.port))
srv.listen(4)
print("mock tcpstreamer: listening on :%d (%s)" % (
args.port, "service stream '%s'" % args.stream if args.stream else "standalone"), flush=True)
while True:
conn, addr = srv.accept()
print("mock tcpstreamer: client %s:%d" % addr, flush=True)
try:
if args.stream and not subscribed_to(conn, args.stream):
print("mock tcpstreamer: no valid SUBSCRIBE for '%s'; closing" % args.stream, flush=True)
continue
if args.stream:
print("mock tcpstreamer: SUBSCRIBE '%s'" % args.stream, flush=True)
serve(conn, args)
except (BrokenPipeError, ConnectionResetError):
print("mock tcpstreamer: client went away", flush=True)
finally:
conn.close()
if __name__ == "__main__":
main()
```
- [ ] **Step 7: Build the dependency and the app** with the project's usual command (see `docs/claude/build-and-run.md`; normally `cmake --build build -j`). Expected: it builds; the warning count equals the count recorded before the task. Run `python3 tools/mock_tcpstreamer.py --help` and `python3 -c "import ast,sys;ast.parse(open('tools/mock_tcpstreamer.py').read())"`: both succeed.
- [ ] **Step 8: Commit**
```bash
git add deps/udpsclient/udps_client.h deps/udpsclient/udps_client.c tools/mock_tcpstreamer.py
git commit -m "udpsclient: a TCP transport mode (SUBSCRIBE, no CONNECT) and a TCP mock server"
```
---
### Task 2: The `tcps` source kind
**Files:**
- Modify: `src/source/udps_source.hpp`, `src/source/udps_source.cpp`
- Create: `src/source/tcps_kind.cpp`
- Modify: `src/source/registry.hpp`, `src/source/registry.cpp`, `CMakeLists.txt` (source list near line 242-243)
**Interfaces:**
- Consumes: `udps_client_config_t::use_tcp` / `::stream_name` from Task 1.
- Produces: `enum class UdpsSource::Transport { Udp, Tcp }`; `UdpsSource(const SourceConfig&, Transport = Transport::Udp)`; `Transport UdpsSource::transport() const`; `std::unique_ptr<SourceKind> make_tcps_kind()`; kind id `tcps` with config keys `server` (default `127.0.0.1`), `port` (default `44500`), `stream` (optional).
- [ ] **Step 1: `udps_source.hpp`.** Inside `class UdpsSource`, add before the constructor:
```cpp
enum class Transport { Udp, Tcp };
```
change the constructor to `explicit UdpsSource(const SourceConfig& cfg, Transport transport = Transport::Udp);`, add `Transport transport() const { return transport_; }` in the public section, and members `Transport transport_ = Transport::Udp;` and `std::string stream_;`.
- [ ] **Step 2: `udps_source.cpp`.** Replace the constructor:
```cpp
UdpsSource::UdpsSource(const SourceConfig& cfg, Transport transport)
: transport_(transport),
server_(cfg.get("server", "127.0.0.1")),
port_(static_cast<std::uint16_t>(
cfg.get_int("port", transport == Transport::Tcp ? 44500 : 8000))),
multicast_(transport == Transport::Tcp ? std::string() : cfg.get("multicast")),
interface_(transport == Transport::Tcp ? std::string() : cfg.get("interface")),
stream_(transport == Transport::Tcp ? cfg.get("stream") : std::string()),
recv_buffer_(static_cast<int>(cfg.get_int("recv_buffer", 8 * 1024 * 1024))) {}
```
Member declaration order must match the initialiser order: in the header declare `transport_` before `server_`, and `stream_` after `interface_` and before `recv_buffer_` (an out-of-order initialiser is a `-Wreorder` warning, which breaks the one-warning baseline). In `open()`, after `c.keepalive_interval_s = 1.0;` add:
```cpp
if (transport_ == Transport::Tcp) {
c.use_tcp = 1;
c.stream_name = stream_.empty() ? nullptr : stream_.c_str();
}
```
`datagram_cb` registration stays; in TCP mode the client never fires it. Leave `set_datagram_tap` as is.
- [ ] **Step 3: Create `src/source/tcps_kind.cpp`.** Copy the helper block of `src/source/udps_kind.cpp` (`ChangeCb`, `entry_row`, `on_edited`, `on_value`, `hook_value`, `hook`, `text_of`; they are in an anonymous namespace, so the copy is local, as three kinds already do) and add the kind:
```cpp
// SPDX-License-Identifier: MIT
// SPDX-FileCopyrightText: 2026 Martino Ferrari
#include "source/registry.hpp"
#include "source/udps_source.hpp"
#include <adwaita.h>
#include <memory>
namespace udpscope {
namespace {
/* ChangeCb, entry_row, on_edited, on_value, hook_value, hook, text_of:
the same six helpers as udps_kind.cpp, verbatim. */
class TcpsKind final : public SourceKind {
public:
std::string id() const override { return "tcps"; }
std::string display_name() const override { return "TCPStreamer"; }
std::string identity(const SourceConfig& cfg) const override {
std::string s = cfg.get("server", "127.0.0.1") + ":" + std::to_string(cfg.get_int("port", 44500));
if (!cfg.get("stream").empty()) s += "/" + cfg.get("stream");
return s;
}
std::unique_ptr<DataSource> make(const SourceConfig& cfg) const override {
return std::make_unique<UdpsSource>(cfg, UdpsSource::Transport::Tcp);
}
ConfigForm make_config_form() const override {
GtkWidget* server = entry_row("_Server");
GtkWidget* port = GTK_WIDGET(adw_spin_row_new_with_range(1, 65535, 1));
adw_preferences_row_set_title(ADW_PREFERENCES_ROW(port), "_Port");
adw_preferences_row_set_use_underline(ADW_PREFERENCES_ROW(port), TRUE);
adw_spin_row_set_value(ADW_SPIN_ROW(port), 44500);
GtkWidget* stream = entry_row("S_tream name");
ChangeCb cb = std::make_shared<std::function<void()>>();
hook(server, cb);
hook_value(port, cb);
hook(stream, cb);
ConfigForm f;
f.rows = {server, port, stream};
f.description =
"Leave the stream name empty for a standalone TCPStreamer; name the stream for a TCPStreamService";
f.focus = server;
f.write = [=](const SourceConfig& c) {
gtk_editable_set_text(GTK_EDITABLE(server), c.get("server", "127.0.0.1").c_str());
adw_spin_row_set_value(ADW_SPIN_ROW(port), static_cast<double>(c.get_int("port", 44500)));
gtk_editable_set_text(GTK_EDITABLE(stream), c.get("stream").c_str());
};
f.read = [=]() {
SourceConfig c;
c.set("server", text_of(server));
c.set("port", std::to_string(static_cast<long>(adw_spin_row_get_value(ADW_SPIN_ROW(port)))));
if (!text_of(stream).empty()) c.set("stream", text_of(stream));
return c;
};
// A stream name is 1..255 bytes and, being the SUBSCRIBE payload, may not hold a comma
// (that would select a multiplexed list, which this kind does not support).
f.valid = [=]() {
const std::string s = text_of(stream);
return !text_of(server).empty() && s.size() <= 255 && s.find(',') == std::string::npos;
};
f.on_change = [cb](std::function<void()> fn) { *cb = std::move(fn); };
return f;
}
};
} // namespace
std::unique_ptr<SourceKind> make_tcps_kind() { return std::make_unique<TcpsKind>(); }
} // namespace udpscope
```
When writing the file, paste the six real helper definitions in place of the comment block. Check `udps_kind.cpp` includes `<functional>` transitively (it does through `source_kind.hpp`).
- [ ] **Step 4: Register.** `registry.hpp`: add `std::unique_ptr<SourceKind> make_tcps_kind();` after `make_udps_kind()` and add `tcps_kind.cpp` to the comment's file list. `registry.cpp`: in `register_builtin_sources()` add `register_kind(make_tcps_kind());` right after `make_udps_kind()` (so the chooser lists the two network kinds together). `CMakeLists.txt`: add `src/source/tcps_kind.cpp` after `src/source/udps_kind.cpp`.
- [ ] **Step 5: Build.** `cmake --build build -j`. Expected: builds, warning count unchanged from the baseline.
- [ ] **Step 6: Smoke-run against the mock** (serial, kill the mock after):
```bash
python3 tools/mock_tcpstreamer.py --port 44500 &
MOCK=$!
UDPSCOPE_EXIT_AFTER_MS=4000 xvfb-run -a build/udpscope --source tcps --source-opt server=127.0.0.1 --source-opt port=44500 2>&1 | tail -20
kill $MOCK
```
Expected: no `Gtk-CRITICAL`; the mock prints `client 127.0.0.1:...`. (If the app's CLI spells the options differently, take the spelling from `docs/claude/build-and-run.md`.) Then the same with `--stream fast` on the mock and `--source-opt stream=fast` on the app: the mock prints `SUBSCRIBE 'fast'`. Then once with the wrong name (`stream=slow`): the mock prints `no valid SUBSCRIBE`, and the app keeps retrying without a stderr flood.
- [ ] **Step 7: Commit**
```bash
git add src/source/udps_source.hpp src/source/udps_source.cpp src/source/tcps_kind.cpp src/source/registry.hpp src/source/registry.cpp CMakeLists.txt
git commit -m "sources: a TCPStreamer kind (standalone or TCPStreamService) on the UDPS client's TCP mode"
```
---
### Task 3: Probes
**Files:**
- Modify: `src/ui_probe.cpp` (sample at ~line 2683 `const auto sample`; new `[tcps]` functions after `flatudp_probe`, ~line 3703), `src/ui.cpp` (env scheduling after the `UDPSCOPE_PROBE_FLATUDP_GAP` block, ~line 7657)
- Declarations: wherever `flatudp_probe` / `flatudp_probe_gap` are declared (grep `flatudp_probe_gap` in `src/*.hpp`; add the two new functions beside them)
**Interfaces:**
- Consumes: `UdpsSource::transport()`, kind `tcps`, `a->model->stats()` fields `connected`, `frames`, `reconnects`, `a->model->signals()`, `a->capture_button`.
- Produces: env vars `UDPSCOPE_PROBE_TCPS=<ms>` and `UDPSCOPE_PROBE_TCPS_RECONNECT=<ms>`; stderr lines `[tcps] stream:`, `[tcps] capture button:`, `[tcps] reconnect:`; a `[source-kinds] form tcps:` line.
- [ ] **Step 1: Form round trip.** In `source_kinds_probe`'s `sample` lambda add a branch:
```cpp
} else if (id == "tcps") {
c.set("server", "10.0.0.7");
c.set("port", "44501");
c.set("stream", "fast");
```
Build and run `UDPSCOPE_PROBE_SOURCE_KINDS=2000 UDPSCOPE_EXIT_AFTER_MS=5000 xvfb-run -a build/udpscope --source probe --source-opt rate=500 2>&1 | grep source-kinds`. Expected: `kinds=udps,tcps,e1608,flatudp,probe` and `form tcps: round_trip=1 valid=1`. The `kinds=` line changes for every run from now on, so update `docs/claude/probe-readings.md` (~line 2120-2123, 2135) and `docs/claude/verification.md` (~line 579, "Nine `[source-kinds]` lines" -> ten, "one `form` line per registered kind: `udps`, `tcps`, ...") in Task 4.
- [ ] **Step 2: Add the source accessor and stream probe** in `ui_probe.cpp`, after `flatudp_probe`:
```cpp
namespace {
UdpsSource* tcps_source(App* a) {
UdpsSource* s = a->acq != nullptr ? dynamic_cast<UdpsSource*>(a->acq->source()) : nullptr;
return s != nullptr && s->transport() == UdpsSource::Transport::Tcp ? s : nullptr;
}
std::uint64_t g_tcps_frames_before = 0;
std::uint64_t g_tcps_reconnects_before = 0;
} // namespace
/// Phase 2, 1 s after the signals were enabled: connected, the mock's four channels, frames
/// flowing, and the capture button present but disabled (the capture is of the UDP socket).
gboolean tcps_probe_report(gpointer user) {
App* a = static_cast<App*>(user);
if (a->window == nullptr) return G_SOURCE_REMOVE;
const bool up = a->model->stats().connected.load(std::memory_order_relaxed);
std::string names;
for (const SignalInfo& s : a->model->signals()) names += (names.empty() ? "" : ",") + s.name;
const std::uint64_t frames = a->model->stats().frames.load(std::memory_order_relaxed);
const bool ok = up && names == "sine,square,ramp,noise" && frames > 0;
std::fprintf(stderr,
"[tcps] stream: connected=%d signals=%zu names=%s frames=%" G_GUINT64_FORMAT " ok=%d\n",
up ? 1 : 0, a->model->signals().size(), names.c_str(), frames, ok ? 1 : 0);
std::fprintf(stderr, "[tcps] capture button: present=%d sensitive=%d\n",
a->capture_button != nullptr ? 1 : 0,
a->capture_button != nullptr && gtk_widget_get_sensitive(a->capture_button) ? 1 : 0);
std::fflush(stderr);
return G_SOURCE_REMOVE;
}
/// Phase 1: enable every signal in one structural change, report a second later.
gboolean tcps_probe(gpointer user) {
App* a = static_cast<App*>(user);
if (a->window == nullptr) return G_SOURCE_REMOVE;
if (tcps_source(a) == nullptr) {
std::fprintf(stderr, "[tcps] no tcps source (run with --source tcps)\n");
std::fflush(stderr);
return G_SOURCE_REMOVE;
}
std::vector<std::size_t> all(a->model->signals().size());
for (std::size_t i = 0; i < all.size(); ++i) all[i] = i;
if (!all.empty()) a->model->set_signals_enabled(all, true);
g_timeout_add(1000, tcps_probe_report, a);
return G_SOURCE_REMOVE;
}
/// After the mock closed the connection (`--close-after`): the client must have reconnected and
/// the frame count moved on. Phase 2 reads 1.5 s after phase 1 recorded the baseline.
gboolean tcps_probe_reconnect_report(gpointer user) {
App* a = static_cast<App*>(user);
if (a->window == nullptr) return G_SOURCE_REMOVE;
const bool up = a->model->stats().connected.load(std::memory_order_relaxed);
const std::uint64_t frames = a->model->stats().frames.load(std::memory_order_relaxed);
const std::uint64_t rc = a->model->stats().reconnects.load(std::memory_order_relaxed);
const bool ok = up && frames > g_tcps_frames_before && rc > g_tcps_reconnects_before;
std::fprintf(stderr,
"[tcps] reconnect: connected=%d frames_before=%" G_GUINT64_FORMAT " after=%" G_GUINT64_FORMAT
" reconnects_before=%" G_GUINT64_FORMAT " after=%" G_GUINT64_FORMAT " ok=%d\n",
up ? 1 : 0, g_tcps_frames_before, frames, g_tcps_reconnects_before, rc, ok ? 1 : 0);
std::fflush(stderr);
return G_SOURCE_REMOVE;
}
gboolean tcps_probe_reconnect(gpointer user) {
App* a = static_cast<App*>(user);
if (a->window == nullptr) return G_SOURCE_REMOVE;
g_tcps_frames_before = a->model->stats().frames.load(std::memory_order_relaxed);
g_tcps_reconnects_before = a->model->stats().reconnects.load(std::memory_order_relaxed);
g_timeout_add(1500, tcps_probe_reconnect_report, a);
return G_SOURCE_REMOVE;
}
```
Make sure `ui_probe.cpp` includes `source/udps_source.hpp` (add it beside the flatudp includes at line 11-12 if absent). The `reconnects` stat is fed from `SourceStats::reconnects` by `Acquisition`; if after Task 3 Step 4 it reads 0 while the client demonstrably reconnected, `Acquisition` does not map it for this path: report that rather than weakening the check.
- [ ] **Step 3: Schedule and declare.** Add the two declarations next to `flatudp_probe_gap`'s, and in `ui.cpp` after the `UDPSCOPE_PROBE_FLATUDP_GAP` block:
```cpp
if (const char* env = g_getenv("UDPSCOPE_PROBE_TCPS")) {
g_timeout_add(static_cast<guint>(std::atoi(env)), tcps_probe, &app);
}
if (const char* env = g_getenv("UDPSCOPE_PROBE_TCPS_RECONNECT")) {
g_timeout_add(static_cast<guint>(std::atoi(env)), tcps_probe_reconnect, &app);
}
```
- [ ] **Step 4: Run the three scenarios**, each short and serial, mock killed after each, stderr filtered with `grep -E '^\[tcps\]|Gtk-CRITICAL|warning'`:
1. Standalone: mock `--port 44500`; app `--source tcps --source-opt port=44500`, `UDPSCOPE_PROBE_TCPS=2500 UDPSCOPE_EXIT_AFTER_MS=5000`. Expected `[tcps] stream: connected=1 signals=4 names=sine,square,ramp,noise frames=<n> ok=1` (n > 0) and `[tcps] capture button: present=1 sensitive=0`.
2. Service: mock `--port 44500 --stream fast`; app additionally `--source-opt stream=fast`. Same expectations.
3. Reconnect: mock `--port 44500 --close-after 3`; app `UDPSCOPE_PROBE_TCPS=2500 UDPSCOPE_PROBE_TCPS_RECONNECT=4500 UDPSCOPE_EXIT_AFTER_MS=8000`. Expected `[tcps] reconnect: connected=1 ... ok=1`.
Also check the no-server case: no mock, `UDPSCOPE_PROBE_TCPS=2500`: `connected=0 ... ok=0` is a reading, not a crash; no stderr flood (a retry every 0.5 s is acceptable, more than a few lines per second is not).
- [ ] **Step 5: Baselines.** A full rebuild shows the same warning count as before the plan, and none of the runs printed `Gtk-CRITICAL`. Then re-run the source-kinds probe (Step 1) and confirm the existing `udps`/`e1608`/`flatudp`/`probe` form lines still read `round_trip=1 valid=1`.
- [ ] **Step 6: Commit**
```bash
git add src/ui_probe.cpp src/ui.cpp
git commit -m "probes: the TCPStreamer kind end to end (standalone, service, reconnect)"
```
(Include the header that carries the probe declarations in `git add` if Step 3 touched one.)
---
### Task 4: Documentation
**Files:**
- Modify: `README.md`, `docs/claude/architecture.md`, `docs/claude/build-and-run.md`, `docs/claude/verification.md`, `docs/claude/probe-readings.md`, `CLAUDE.md` (only its intro sentence if it enumerates the source kinds)
- [ ] **Step 1: `docs/claude/build-and-run.md`.** The `deps/udpsclient` section says "Five things were changed". Make it six and add item 6: "**A TCP transport mode.** `udps_client_config_t::use_tcp` / `stream_name` select a TCP connection to a MARTe2 `TCPStreamer` or `TCPStreamService`: `connect_tcp()` (3 s connect bound via `SO_SNDTIMEO`, 1 s receive bound, an optional `SUBSCRIBE` frame), the existing `read_tcp_frame()` for the frames, the reorder window forced to 0, no DISCONNECT frame on close (a write to a closed peer would raise SIGPIPE), and no datagram tap. A frame must fit `UDPS_MAX_DATAGRAM_BYTES` (the server's `MaxPayloadSize` default of 60000 does)." Also update the earlier sentence "`deps/udpsclient` gained one additive entry point" to mention the TCP mode.
- [ ] **Step 2: `docs/claude/architecture.md` (Sources section).** Add a bullet: "**The TCPStreamer kind (`tcps`) is `UdpsSource` with `Transport::Tcp`** ... `src/source/tcps_kind.cpp` is only the form and identity (`host:port` or `host:port/stream`); there is no new `DataSource`. One stream per connection (no multiplexed `SUBSCRIBE` list: a source is single-timebase; merge on the server with `TCPStreamJoin`). `has_datagrams()` is false, so the capture button stays disabled. `Disconnected` follows the client's 2 s silence timeout or a closed connection, and it reconnects every 0.5 s. A stream whose frames are further apart than 2 s therefore reconnects continually, as UDPS does." Fix the bullet that says `UdpsSource` "wraps `deps/udpsclient` unchanged".
- [ ] **Step 3: `docs/claude/verification.md` and `probe-readings.md`.** Verification: update "Nine `[source-kinds]` lines (... `udps`, `e1608`, `flatudp`, `probe`)" to include `tcps`; add `UDPSCOPE_PROBE_TCPS` and `UDPSCOPE_PROBE_TCPS_RECONNECT` entries beside the FLATUDP ones, with the three scenarios from Task 3 Step 4 (commands, mock options, exit times). Probe readings: update `kinds=` (line ~2120 and ~2135) to `udps,tcps,e1608,flatudp,probe`, add `[source-kinds] form tcps: round_trip=1 valid=1` (and the form sample's keys: server, port, stream), and add a `## [tcps]` section after `[flatudp]` with the sample output of the three lines and what a wrong reading means (`connected=0` with a mock running: SUBSCRIBE or connect failing; `names` wrong: CONFIG not decoded; `sensitive=1`: the capture button was enabled for a kind with no datagrams; `reconnect ... ok=0`: the client did not come back after the close, or `Acquisition` does not map `reconnects`).
- [ ] **Step 4: `README.md`.** Wherever it lists the source kinds or the command-line `--source` values, add `tcps` with its options (`server`, `port` default 44500, `stream` optional), one line saying it receives a MARTe2 `TCPStreamer`/`TCPStreamService`, and under known limitations: single stream per connection, no packet capture over TCP, frames up to 65 KB (the server's `MaxPayloadSize`). If `CLAUDE.md`'s intro sentence enumerates the kinds, add `tcps` there too.
- [ ] **Step 5: Read-through.** `grep -rn "Nine \`\[source-kinds\]\|Five things\|unchanged" README.md CLAUDE.md docs/claude` and fix every statement the change made false.
- [ ] **Step 6: Commit**
```bash
git add README.md CLAUDE.md docs/claude
git commit -m "docs: the TCPStreamer source kind, the client's TCP mode, and its probes"
```
---
## Self-review (against the spec)
- Wire/scope (single stream, SUBSCRIBE only with a name): Task 1 Steps 1-4, Task 2 Step 3 (`valid` rejects a comma).
- `udpsclient` TCP mode (use_tcp, stream_name, connect_tcp, reorder 0, no tap, additive, recorded as a sixth change): Task 1; recording: Task 4 Step 1.
- Source adapter reuse via transport argument; form rows, description, identity, session naming (automatic), registration and CMake: Task 2.
- Verification (fake server, probe lines for signals/frames/reconnect, baselines, real smoke test): Task 3 uses a Python mock rather than the in-process fake server the spec named, because the existing probes all use external mocks under `tools/`. A smoke test against a real MARTe `TCPStreamer` is not in the plan; do it only if one is runnable.
- Docs: Task 4.
- Types: `UdpsSource::Transport`, `transport()`, `make_tcps_kind()`, `tcps_source()`, env names are used identically across tasks.
@@ -0,0 +1,51 @@
# TCPStreamer source kind
A new source kind `tcps` ("TCPStreamer") that receives a MARTe2 `TCPStreamer` DataSource, either
standalone or through a shared `TCPStreamService`.
## Wire and scope
- Both transports are plain UDPS framing over TCP. A standalone `TCPStreamer` (TCPSServer) sends
CONFIG then DATA on connect. A `TCPStreamService` additionally requires a first client frame,
`SUBSCRIBE` (type 5, payload = stream name, 1..255 bytes).
- A plain server ignores `SUBSCRIBE`, so the client sends it only when a stream name is set.
- One connection carries one stream. Multiplexed lists (`"a,b,c"`, stream index in `fragmentIdx`)
are out of scope: a udpscope source is single-timebase. Several streams are merged server-side
with `TCPStreamJoin`, which publishes an ordinary named stream.
## `deps/udpsclient`: a TCP transport mode
- `udps_client_config_t` gains `use_tcp` and `stream_name`.
- New `connect_tcp()`: open the TCP socket, send `SUBSCRIBE` if a stream name is set, send no
`CONNECT`. No UDP socket, no keepalive ACK.
- `udps_client_poll` reuses `read_tcp_frame` / `handle_packet`.
- `reorder_window_s` is forced to 0 in TCP mode: TCP is ordered, so a counter gap is a server-side
drop and is reported at once.
- Silence timeout and reconnect delay behave as in UDP (dead server, or a stalled client the server
evicted, reconnects).
- The datagram tap does not fire; the packet capture stays UDP-only (`has_datagrams()` false).
- Additive. Recorded in `docs/claude/build-and-run.md` as a sixth intentional upstream change.
## Source adapter and kind
- `UdpsSource` takes a transport argument (`UdpsSource(cfg, Transport::Tcp)`) instead of being
copied: its callbacks are identical. The TCP variant reads `server`, `port` (default 44500) and
`stream`.
- `src/source/tcps_kind.cpp`: kind `tcps`, display name "TCPStreamer". Form rows: Server, Port,
Stream name (optional). Description: "Leave empty for a standalone TCPStreamer; name the stream
for a TCPStreamService".
- Identity: `host:port` or `host:port/stream`. Session naming follows the existing non-UDPS rule
(`tcps-<identity>`, port 1, `source_kind=tcps`, `source_opt_*`).
- Registered in `register_builtin_sources()` and `CMakeLists.txt`.
## Verification
No unit suite. Extend the self-check in `src/ui_probe.cpp` with an in-process fake TCP server that
sends CONFIG then DATA, and requires `SUBSCRIBE` in one mode. Probe lines assert the signals, the
frame count, and the reconnect after the server closes. Baselines: exactly one build warning, no
`Gtk-CRITICAL`. A smoke test against a real MARTe `TCPStreamer` if one is runnable.
## Docs
README and `docs/claude/architecture.md` (Sources) describe the kind; `docs/claude/probe-readings.md`
gets the new probe lines.