From 46076312cc00b2b91293132ed33829b3387d6bc9 Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Sat, 31 Oct 2026 19:57:11 +0100 Subject: [PATCH] source: the UDPS client behind DataSource, with its config form --- CMakeLists.txt | 2 + src/source/registry.cpp | 2 +- src/source/udps_kind.cpp | 92 +++++++++++++++++++++++ src/source/udps_source.cpp | 150 +++++++++++++++++++++++++++++++++++++ src/source/udps_source.hpp | 52 +++++++++++++ 5 files changed, 297 insertions(+), 1 deletion(-) create mode 100644 src/source/udps_kind.cpp create mode 100644 src/source/udps_source.cpp create mode 100644 src/source/udps_source.hpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 0a95d3b..90f9cd1 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -239,6 +239,8 @@ add_executable(udpscope src/source/source_config.cpp src/source/data_source.cpp src/source/registry.cpp + src/source/udps_source.cpp + src/source/udps_kind.cpp src/plot_host.cpp src/export.cpp src/search.cpp diff --git a/src/source/registry.cpp b/src/source/registry.cpp index 3c3c88e..2c2f1ac 100644 --- a/src/source/registry.cpp +++ b/src/source/registry.cpp @@ -49,7 +49,7 @@ void register_builtin_sources() { static bool done = false; if (done) return; done = true; - // Task 2: register_kind(make_udps_kind()); + register_kind(make_udps_kind()); // Task 6: register make_probe_kind() when UDPSCOPE_PROBE_SOURCE_KINDS is set } diff --git a/src/source/udps_kind.cpp b/src/source/udps_kind.cpp new file mode 100644 index 0000000..ffe53c7 --- /dev/null +++ b/src/source/udps_kind.cpp @@ -0,0 +1,92 @@ +// SPDX-License-Identifier: MIT +// SPDX-FileCopyrightText: 2026 Martino Ferrari + +#include "source/registry.hpp" +#include "source/udps_source.hpp" + +#include + +#include + +namespace udpscope { +namespace { + +using ChangeCb = std::shared_ptr>; + +GtkWidget* entry_row(const char* title) { + GtkWidget* r = adw_entry_row_new(); + adw_preferences_row_set_title(ADW_PREFERENCES_ROW(r), title); + adw_preferences_row_set_use_underline(ADW_PREFERENCES_ROW(r), TRUE); + return r; +} + +void on_edited(GtkEditable*, gpointer user) { + const auto& cb = *static_cast(user); + if (cb && *cb) (*cb)(); +} + +void hook(GtkWidget* entry, const ChangeCb& cb) { + g_signal_connect_data(entry, "changed", G_CALLBACK(on_edited), new ChangeCb(cb), + +[](gpointer d, GClosure*) { delete static_cast(d); }, + static_cast(0)); +} + +std::string text_of(GtkWidget* w) { return gtk_editable_get_text(GTK_EDITABLE(w)); } + +class UdpsKind final : public SourceKind { + public: + std::string id() const override { return "udps"; } + std::string display_name() const override { return "UDPStreamer"; } + bool has_datagrams() const override { return true; } + + std::string identity(const SourceConfig& cfg) const override { + return cfg.get("server", "127.0.0.1") + ":" + std::to_string(cfg.get_int("port", 8000)); + } + + std::unique_ptr make(const SourceConfig& cfg) const override { + return std::make_unique(cfg); + } + + 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), 8000); + GtkWidget* group = entry_row("Multicast _group"); + GtkWidget* iface = entry_row("_Interface"); + + ChangeCb cb = std::make_shared>(); + hook(server, cb); + hook(group, cb); + hook(iface, cb); + + ConfigForm f; + f.rows = {server, port, group, iface}; + f.description = "Leave the multicast group and the interface empty for unicast on the default route"; + 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(c.get_int("port", 8000))); + gtk_editable_set_text(GTK_EDITABLE(group), c.get("multicast").c_str()); + gtk_editable_set_text(GTK_EDITABLE(iface), c.get("interface").c_str()); + }; + f.read = [=]() { + SourceConfig c; + c.set("server", text_of(server)); + c.set("port", std::to_string(static_cast(adw_spin_row_get_value(ADW_SPIN_ROW(port))))); + if (!text_of(group).empty()) c.set("multicast", text_of(group)); + if (!text_of(iface).empty()) c.set("interface", text_of(iface)); + return c; + }; + f.valid = [=]() { return !text_of(server).empty(); }; + f.on_change = [cb](std::function fn) { *cb = std::move(fn); }; + return f; + } +}; + +} // namespace + +std::unique_ptr make_udps_kind() { return std::make_unique(); } + +} // namespace udpscope diff --git a/src/source/udps_source.cpp b/src/source/udps_source.cpp new file mode 100644 index 0000000..1ae89f8 --- /dev/null +++ b/src/source/udps_source.cpp @@ -0,0 +1,150 @@ +// SPDX-License-Identifier: MIT +// SPDX-FileCopyrightText: 2026 Martino Ferrari + +#include "source/udps_source.hpp" + +#include + +namespace udpscope { +namespace { + +static_assert(kTimePacket == UDPS_TIME_PACKET && kTimeFullArray == UDPS_TIME_FULL_ARRAY && + kTimeFirstSample == UDPS_TIME_FIRST_SAMPLE && + kTimeLastSample == UDPS_TIME_LAST_SAMPLE, + "model.hpp's kTime* must equal the wire protocol's UDPS_TIME_*"); + +SignalInfo to_info(const udps_signal_t& s) { + SignalInfo info; + info.name = s.name; + info.unit = s.unit; + info.type_code = s.type_code; + info.quant_type = s.quant_type; + info.rows = s.num_rows; + info.cols = s.num_cols; + info.range_min = s.range_min; + info.range_max = s.range_max; + info.time_mode = s.time_mode; + info.sampling_rate = s.sampling_rate; + info.time_signal_idx = s.time_signal_idx; + return info; +} + +} // namespace + +UdpsSource::UdpsSource(const SourceConfig& cfg) + : server_(cfg.get("server", "127.0.0.1")), + port_(static_cast(cfg.get_int("port", 8000))), + multicast_(cfg.get("multicast")), + interface_(cfg.get("interface")), + recv_buffer_(static_cast(cfg.get_int("recv_buffer", 8 * 1024 * 1024))) {} + +UdpsSource::~UdpsSource() { close(); } + +void UdpsSource::open(SourceSink& sink) { + close(); + sink_ = &sink; + udps_client_config_t c; + udps_client_config_init(&c); + c.server_addr = server_.c_str(); + c.server_port = port_; + c.multicast_group = multicast_.empty() ? nullptr : multicast_.c_str(); + c.interface_addr = interface_.empty() ? nullptr : interface_.c_str(); + c.recv_buffer_bytes = static_cast(recv_buffer_); + c.silence_timeout_s = 2.0; + c.reconnect_delay_s = 0.5; + c.keepalive_interval_s = 1.0; + + client_ = udps_client_create(&c); + if (client_ == nullptr) { + throw std::runtime_error("could not create the UDPS client for " + server_ + ":" + + std::to_string(port_)); + } + udps_client_set_callbacks(client_, &UdpsSource::config_cb, &UdpsSource::data_cb, + &UdpsSource::event_cb, this); + udps_client_set_datagram_cb(client_, &UdpsSource::datagram_cb, this); +} + +void UdpsSource::poll(int timeout_ms) { + drain_commands(); + if (client_ != nullptr) udps_client_poll(client_, timeout_ms); +} + +void UdpsSource::close() { + if (client_ != nullptr) { + udps_client_destroy(client_); + client_ = nullptr; + } +} + +SourceStats UdpsSource::stats() const { + SourceStats out; + if (client_ == nullptr) return out; + udps_stats_t s; + udps_client_stats(client_, &s); + out.packets = s.packets_received; + out.frames = s.frames_delivered; + out.gaps = s.counter_gaps; + out.late = s.late_delivered; + out.resequenced = s.late_resequenced; + out.stale = s.stale_packets; + out.overflows = s.reorder_overflows; + out.reconnects = s.reconnects; + out.connected = udps_client_is_connected(client_) != 0; + return out; +} + +bool UdpsSource::connected() const { + return client_ != nullptr && udps_client_is_connected(client_) != 0; +} + +void UdpsSource::config_cb(const udps_signal_t* signals, std::uint32_t count, + std::uint8_t publish_mode, void* user) { + auto* self = static_cast(user); + self->signals_.clear(); + self->signals_.reserve(count); + for (std::uint32_t i = 0; i < count; ++i) self->signals_.push_back(to_info(signals[i])); + self->sink_->on_config(self->signals_, publish_mode); +} + +void UdpsSource::data_cb(const udps_frame_t* f, void* user) { + auto* self = static_cast(user); + // The frame's signals are the CONFIG's; the cache is rebuilt only if a frame ever disagrees. + if (self->signals_.size() != f->num_signals) { + self->signals_.clear(); + for (std::uint32_t i = 0; i < f->num_signals; ++i) { + self->signals_.push_back(to_info(f->signals[i])); + } + } + self->values_.resize(f->num_signals); + for (std::uint32_t i = 0; i < f->num_signals; ++i) { + self->values_[i].values = f->values[i].values; + self->values_[i].count = f->values[i].count; + } + Frame fr; + fr.counter = f->counter; + fr.lost = f->lost; + fr.late = f->late != 0; + fr.recv_time = f->recv_time; + fr.publish_mode = f->publish_mode; + fr.num_samples = f->num_samples; + fr.num_signals = f->num_signals; + fr.signals = self->signals_.data(); + fr.values = self->values_.data(); + self->sink_->on_data(fr); +} + +void UdpsSource::event_cb(udps_event_t event, const char* detail, void* user) { + auto* self = static_cast(user); + SourceEvent e = SourceEvent::Error; + if (event == UDPS_EVENT_CONNECTED) e = SourceEvent::Connected; + else if (event == UDPS_EVENT_DISCONNECTED) e = SourceEvent::Disconnected; + self->sink_->on_event(e, detail); +} + +void UdpsSource::datagram_cb(const std::uint8_t* bytes, std::size_t len, double recv_time, + void* user) { + auto* self = static_cast(user); + if (self->tap_) self->tap_(bytes, len, recv_time); +} + +} // namespace udpscope diff --git a/src/source/udps_source.hpp b/src/source/udps_source.hpp new file mode 100644 index 0000000..a44d9a7 --- /dev/null +++ b/src/source/udps_source.hpp @@ -0,0 +1,52 @@ +// SPDX-License-Identifier: MIT +// SPDX-FileCopyrightText: 2026 Martino Ferrari + +#pragma once + +#include "source/data_source.hpp" +#include "source/source_config.hpp" + +#include + +#include +#include + +namespace udpscope { + +/// The UDPStreamer client behind the `DataSource` interface. A thin adapter: it maps +/// `udps_signal_t`/`udps_frame_t` to `SignalInfo`/`Frame` and forwards the datagram observer; +/// all protocol work stays in `deps/udpsclient`. +class UdpsSource final : public DataSource { + public: + explicit UdpsSource(const SourceConfig& cfg); + ~UdpsSource() override; + + void open(SourceSink& sink) override; + void poll(int timeout_ms) override; + void close() override; + SourceStats stats() const override; + bool connected() const override; + void set_datagram_tap(DatagramTap tap) override { tap_ = std::move(tap); } + + private: + static void config_cb(const udps_signal_t* signals, std::uint32_t count, + std::uint8_t publish_mode, void* user); + static void data_cb(const udps_frame_t* frame, void* user); + static void event_cb(udps_event_t event, const char* detail, void* user); + static void datagram_cb(const std::uint8_t* bytes, std::size_t len, double recv_time, + void* user); + + std::string server_; + std::uint16_t port_ = 8000; + std::string multicast_; + std::string interface_; + int recv_buffer_ = 8 * 1024 * 1024; + + udps_client_t* client_ = nullptr; + SourceSink* sink_ = nullptr; + DatagramTap tap_; + std::vector signals_; + std::vector values_; +}; + +} // namespace udpscope