source: a synthetic probe kind, and sessions that remember their source kind

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

src/source/probe_source.cpp

@@ -0,0 +1,137 @@
+// SPDX-License-Identifier: MIT
+// SPDX-FileCopyrightText: 2026 Martino Ferrari
+
+#include "source/probe_source.hpp"
+
+#include "source/registry.hpp"
+
+#include <adwaita.h>
+
+#include <algorithm>
+#include <chrono>
+#include <cmath>
+#include <functional>
+#include <memory>
+#include <thread>
+
+namespace udpscope {
+
+ProbeSource::ProbeSource(const SourceConfig& cfg)
+    : rate_hz_(std::max(1.0, cfg.get_double("rate", 1000.0))), batch_(kBatch, 0.0) {
+  SignalInfo s;
+  s.name = "probe";
+  s.unit = "V";
+  s.type_code = 9;  // float64
+  s.rows = 1;
+  s.cols = 1;
+  s.range_min = -1.0;
+  s.range_max = 1.0;
+  s.time_mode = kTimePacket;
+  s.sampling_rate = rate_hz_;
+  signals_.push_back(s);
+}
+
+void ProbeSource::open(SourceSink& sink) {
+  sink_ = &sink;
+  configured_ = false;
+  open_.store(true, std::memory_order_relaxed);
+}
+
+void ProbeSource::close() { open_.store(false, std::memory_order_relaxed); }
+
+SourceStats ProbeSource::stats() const {
+  SourceStats s;
+  s.packets = frames_;
+  s.frames = frames_;
+  s.connected = open_.load(std::memory_order_relaxed);
+  return s;
+}
+
+void ProbeSource::poll(int timeout_ms) {
+  drain_commands();
+  if (!open_.load(std::memory_order_relaxed) || sink_ == nullptr) return;
+  if (!configured_) {
+    configured_ = true;
+    sink_->on_config(signals_, /*publish_mode=*/1);
+    sink_->on_event(SourceEvent::Connected, "");
+  }
+  // One batch takes kBatch / rate seconds to "arrive".
+  const int period_ms = std::max(1, static_cast<int>(1000.0 * kBatch / rate_hz_));
+  std::this_thread::sleep_for(std::chrono::milliseconds(std::min(period_ms, timeout_ms)));
+  const double step = 2.0 * M_PI * 5.0 / rate_hz_;  // a 5 Hz sine
+  for (std::uint32_t k = 0; k < kBatch; ++k) {
+    batch_[k] = std::sin(phase_);
+    phase_ += step;
+  }
+  values_.values = batch_.data();
+  values_.count = kBatch;
+  Frame f;
+  f.counter = counter_++;
+  f.lost = 0;
+  f.late = false;
+  f.recv_time = std::chrono::duration<double>(std::chrono::system_clock::now().time_since_epoch())
+                    .count();
+  f.publish_mode = 1;
+  f.num_samples = kBatch;
+  f.num_signals = 1;
+  f.signals = signals_.data();
+  f.values = &values_;
+  ++frames_;
+  sink_->on_data(f);
+}
+
+namespace {
+
+void on_rate_changed(GObject*, GParamSpec*, gpointer user) {
+  const auto& cb = *static_cast<std::shared_ptr<std::function<void()>>*>(user);
+  if (cb && *cb) (*cb)();
+}
+
+class ProbeKind final : public SourceKind {
+ public:
+  std::string id() const override { return "probe"; }
+  std::string display_name() const override { return "Probe (test)"; }
+  std::string identity(const SourceConfig& cfg) const override {
+    return "rate" + std::to_string(cfg.get_int("rate", 1000));
+  }
+  std::unique_ptr<DataSource> make(const SourceConfig& cfg) const override {
+    return std::make_unique<ProbeSource>(cfg);
+  }
+
+  ConfigForm make_config_form() const override {
+    GtkWidget* rate = GTK_WIDGET(adw_spin_row_new_with_range(1, 100000, 1));
+    adw_preferences_row_set_title(ADW_PREFERENCES_ROW(rate), "_Rate (Hz)");
+    adw_preferences_row_set_use_underline(ADW_PREFERENCES_ROW(rate), TRUE);
+    adw_spin_row_set_value(ADW_SPIN_ROW(rate), 1000);
+
+    auto cb = std::make_shared<std::function<void()>>();
+    g_signal_connect_data(rate, "notify::value", G_CALLBACK(on_rate_changed),
+                          new std::shared_ptr<std::function<void()>>(cb),
+                          +[](gpointer d, GClosure*) {
+                            delete static_cast<std::shared_ptr<std::function<void()>>*>(d);
+                          },
+                          static_cast<GConnectFlags>(0));
+
+    ConfigForm f;
+    f.rows = {rate};
+    f.description = "A synthetic 5 Hz sine, for the self-check";
+    f.focus = rate;
+    f.write = [=](const SourceConfig& c) {
+      adw_spin_row_set_value(ADW_SPIN_ROW(rate), static_cast<double>(c.get_int("rate", 1000)));
+    };
+    f.read = [=]() {
+      SourceConfig c;
+      c.set("rate", std::to_string(static_cast<long>(adw_spin_row_get_value(ADW_SPIN_ROW(rate)))));
+      return c;
+    };
+    f.valid = []() { return true; };
+    f.on_change = [cb](std::function<void()> fn) { *cb = std::move(fn); };
+    return f;
+  }
+};
+
+}  // namespace
+
+std::unique_ptr<SourceKind> make_probe_kind() { return std::make_unique<ProbeKind>(); }
+
+}  // namespace udpscope

src/source/probe_source.hpp

@@ -0,0 +1,50 @@
+// SPDX-License-Identifier: MIT
+// SPDX-FileCopyrightText: 2026 Martino Ferrari
+
+#pragma once
+
+#include "source/data_source.hpp"
+#include "source/source_config.hpp"
+
+#include <atomic>
+#include <cstdint>
+#include <vector>
+
+namespace udpscope {
+
+/// A synthetic source for the headless self-check: one 1 V sine signal, `kBatch` samples per
+/// frame at `rate` Hz, no socket. It exists to prove the framework end to end (open, CONFIG,
+/// data, commands, panels) and is registered only when `UDPSCOPE_PROBE_SOURCE_KINDS` is set.
+class ProbeSource final : public DataSource {
+ public:
+  static constexpr std::uint32_t kBatch = 10;
+
+  explicit ProbeSource(const SourceConfig& cfg);
+
+  void open(SourceSink& sink) override;
+  void poll(int timeout_ms) override;
+  void close() override;
+  SourceStats stats() const override;
+  bool connected() const override { return open_.load(std::memory_order_relaxed); }
+
+  /// What a panel command does, so the probe can see that `post()` reached the producer thread.
+  void note_applied() noexcept { applied_.fetch_add(1, std::memory_order_relaxed); }
+  std::uint64_t commands_applied() const noexcept {
+    return applied_.load(std::memory_order_relaxed);
+  }
+
+ private:
+  double rate_hz_;
+  SourceSink* sink_ = nullptr;
+  std::atomic<bool> open_{false};
+  bool configured_ = false;
+  std::uint32_t counter_ = 0;
+  std::uint64_t frames_ = 0;
+  double phase_ = 0.0;
+  std::vector<SignalInfo> signals_;
+  std::vector<double> batch_;
+  SignalValues values_;
+  std::atomic<std::uint64_t> applied_{0};
+};
+
+}  // namespace udpscope

src/source/registry.cpp

@@ -50,7 +50,7 @@ void register_builtin_sources() {
   if (done) return;
   done = true;
   register_kind(make_udps_kind());
-  // Task 6: register make_probe_kind() when UDPSCOPE_PROBE_SOURCE_KINDS is set
+  if (std::getenv("UDPSCOPE_PROBE_SOURCE_KINDS") != nullptr) register_kind(make_probe_kind());
 }
 
 std::string session_host(const std::string& kind_id, const SourceConfig& cfg) {