probes: the TCPStreamer kind end to end (standalone, service, reconnect)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

src/ui.cpp

@@ -7655,6 +7655,12 @@ int run_ui(int argc, char** argv, const Config& cfg, ScopeModel& model, Acquisit
   if (const char* env = g_getenv("UDPSCOPE_PROBE_FLATUDP_GAP")) {
     g_timeout_add(static_cast<guint>(std::atoi(env)), flatudp_probe_gap, &app);
   }
+  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);
+  }
 
   // Our own options were consumed by parse_args(); GTK would reject them, so only
   // the program name is handed to g_application_run().

src/ui_app.hpp

@@ -2308,6 +2308,13 @@ gboolean flatudp_probe(gpointer user);
 gboolean flatudp_probe_report(gpointer user);
 gboolean flatudp_probe_gap(gpointer user);
 gboolean flatudp_probe_resumed(gpointer user);
+/// The TCPStreamer kind: `tcps_probe` enables every signal, `tcps_probe_report` reads the stream
+/// and the capture button a second later; `tcps_probe_reconnect` records a baseline and
+/// `tcps_probe_reconnect_report` checks a reconnect moved the counters on.
+gboolean tcps_probe(gpointer user);
+gboolean tcps_probe_report(gpointer user);
+gboolean tcps_probe_reconnect(gpointer user);
+gboolean tcps_probe_reconnect_report(gpointer user);
 gboolean pan_probe(gpointer user);
 gboolean arm_probe(gpointer user);
 gboolean trigrenum_probe_step3(gpointer user);

src/ui_probe.cpp

@@ -11,6 +11,7 @@
 #include "source/flatudp/flatudp_signals.hpp"
 #include "source/flatudp/flatudp_source.hpp"
 #include "source/registry.hpp"
+#include "source/udps_source.hpp"
 #include "ui_app.hpp"
 
 #include <sys/stat.h>
@@ -2711,6 +2712,10 @@ gboolean source_kinds_probe(gpointer user) {
       c.set("rate", "2500");
       c.set("endian", "big");
       c.set("signals", "Trigger:uint8,Ia:float32,Wave:int16[4]");
+    } else if (id == "tcps") {
+      c.set("server", "10.0.0.7");
+      c.set("port", "44501");
+      c.set("stream", "fast");
     }
     return c;
   };
@@ -3696,6 +3701,84 @@ gboolean flatudp_probe(gpointer user) {
   return G_SOURCE_REMOVE;
 }
 
+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 time master and 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 == "time,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 1 records the baseline; phase 2 polls every 100 ms until both
+/// have happened (the close falls whenever the mock's timer says, so a fixed delay would race it)
+/// and reports the last reading after 6 s.
+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;
+  static int ticks = 0;
+  if (!ok && ++ticks < 60) return G_SOURCE_CONTINUE;
+  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(100, tcps_probe_reconnect_report, a);
+  return G_SOURCE_REMOVE;
+}
+
 gboolean pan_probe(gpointer user) {
   App* a = static_cast<App*>(user);
   if (a->window == nullptr || a->rows->row_count() <= 0) return G_SOURCE_REMOVE;