0 Fork 0

Files

T

Martino Ferrari 21b1945769 docs: TCPStreamer source kind spec and plan

2026-11-06 13:41:31 +01:00

32 KiB

Raw Blame History

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:

    /**
     * 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->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):

/** 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:

    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):

#!/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

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:

  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:

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:

  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:

// 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):

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

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:

    } 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:

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:

  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':

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.
Service: mock --port 44500 --stream fast; app additionally --source-opt stream=fast. Same expectations.
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

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

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.

32 KiB Raw Blame History

TCPStreamer Source Kind Implementation Plan

Global Constraints

File Structure

Task 1: TCP transport in deps/udpsclient, and the mock server

Task 2: The tcps source kind

Task 3: Probes

Task 4: Documentation

Self-review (against the spec)

32 KiB

Raw Blame History

Task 1: TCP transport in `deps/udpsclient`, and the mock server

Task 2: The `tcps` source kind