From f27bb17669ffc6d8646231bb8849da6a65a4d805 Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Fri, 6 Nov 2026 13:47:23 +0100 Subject: [PATCH] sources: a TCPStreamer kind (standalone or TCPStreamService) on the UDPS client's TCP mode Co-Authored-By: Claude Opus 4.6 --- CMakeLists.txt | 1 + src/source/registry.cpp | 1 + src/source/registry.hpp | 3 +- src/source/tcps_kind.cpp | 107 +++++++++++++++++++++++++++++++++++++ src/source/udps_source.cpp | 17 ++++-- src/source/udps_source.hpp | 7 ++- 6 files changed, 129 insertions(+), 7 deletions(-) create mode 100644 src/source/tcps_kind.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 042baa2..72e623c 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -241,6 +241,7 @@ add_executable(udpscope src/source/registry.cpp src/source/udps_source.cpp src/source/udps_kind.cpp + src/source/tcps_kind.cpp src/source/probe_source.cpp src/source/e1608/e1608_link.cpp src/source/e1608/e1608_device.cpp diff --git a/src/source/registry.cpp b/src/source/registry.cpp index d67308b..28513a6 100644 --- a/src/source/registry.cpp +++ b/src/source/registry.cpp @@ -50,6 +50,7 @@ void register_builtin_sources() { if (done) return; done = true; register_kind(make_udps_kind()); + register_kind(make_tcps_kind()); register_kind(make_e1608_kind()); register_kind(make_flatudp_kind()); if (std::getenv("UDPSCOPE_PROBE_SOURCE_KINDS") != nullptr) register_kind(make_probe_kind()); diff --git a/src/source/registry.hpp b/src/source/registry.hpp index a601930..3621e45 100644 --- a/src/source/registry.hpp +++ b/src/source/registry.hpp @@ -21,8 +21,9 @@ const std::vector& kinds(); /// may drop. A later `dlopen` loader would only call `register_kind()`. void register_builtin_sources(); -/// The built-in kinds, each defined in its own file (`udps_kind.cpp`, `probe_source.cpp`, `e1608/e1608_kind.cpp`, `flatudp/flatudp_kind.cpp`). +/// The built-in kinds, each defined in its own file (`udps_kind.cpp`, `tcps_kind.cpp`, `probe_source.cpp`, `e1608/e1608_kind.cpp`, `flatudp/flatudp_kind.cpp`). std::unique_ptr make_udps_kind(); +std::unique_ptr make_tcps_kind(); std::unique_ptr make_probe_kind(); std::unique_ptr make_e1608_kind(); std::unique_ptr make_flatudp_kind(); diff --git a/src/source/tcps_kind.cpp b/src/source/tcps_kind.cpp new file mode 100644 index 0000000..d5e3743 --- /dev/null +++ b/src/source/tcps_kind.cpp @@ -0,0 +1,107 @@ +// 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 on_value(GObject*, GParamSpec*, gpointer user) { + const auto& cb = *static_cast(user); + if (cb && *cb) (*cb)(); +} + +void hook_value(GtkWidget* spin, const ChangeCb& cb) { + g_signal_connect_data(spin, "notify::value", G_CALLBACK(on_value), new ChangeCb(cb), + +[](gpointer d, GClosure*) { delete static_cast(d); }, + static_cast(0)); +} + +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 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 make(const SourceConfig& cfg) const override { + return std::make_unique(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>(); + 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(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(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 fn) { *cb = std::move(fn); }; + return f; + } +}; + +} // namespace + +std::unique_ptr make_tcps_kind() { return std::make_unique(); } + +} // namespace udpscope diff --git a/src/source/udps_source.cpp b/src/source/udps_source.cpp index e0a1f54..4196ca4 100644 --- a/src/source/udps_source.cpp +++ b/src/source/udps_source.cpp @@ -40,11 +40,14 @@ SignalInfo to_info(const udps_signal_t& s) { } // 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")), +UdpsSource::UdpsSource(const SourceConfig& cfg, Transport transport) + : transport_(transport), + server_(cfg.get("server", "127.0.0.1")), + port_(static_cast( + 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(cfg.get_int("recv_buffer", 8 * 1024 * 1024))) {} UdpsSource::~UdpsSource() { close(); } @@ -62,6 +65,10 @@ void UdpsSource::open(SourceSink& sink) { c.silence_timeout_s = 2.0; c.reconnect_delay_s = 0.5; c.keepalive_interval_s = 1.0; + if (transport_ == Transport::Tcp) { + c.use_tcp = 1; + c.stream_name = stream_.empty() ? nullptr : stream_.c_str(); + } client_ = udps_client_create(&c); if (client_ == nullptr) { diff --git a/src/source/udps_source.hpp b/src/source/udps_source.hpp index a44d9a7..919f524 100644 --- a/src/source/udps_source.hpp +++ b/src/source/udps_source.hpp @@ -18,7 +18,9 @@ namespace udpscope { /// all protocol work stays in `deps/udpsclient`. class UdpsSource final : public DataSource { public: - explicit UdpsSource(const SourceConfig& cfg); + enum class Transport { Udp, Tcp }; + + explicit UdpsSource(const SourceConfig& cfg, Transport transport = Transport::Udp); ~UdpsSource() override; void open(SourceSink& sink) override; @@ -27,6 +29,7 @@ class UdpsSource final : public DataSource { SourceStats stats() const override; bool connected() const override; void set_datagram_tap(DatagramTap tap) override { tap_ = std::move(tap); } + Transport transport() const { return transport_; } private: static void config_cb(const udps_signal_t* signals, std::uint32_t count, @@ -36,10 +39,12 @@ class UdpsSource final : public DataSource { static void datagram_cb(const std::uint8_t* bytes, std::size_t len, double recv_time, void* user); + Transport transport_ = Transport::Udp; std::string server_; std::uint16_t port_ = 8000; std::string multicast_; std::string interface_; + std::string stream_; int recv_buffer_ = 8 * 1024 * 1024; udps_client_t* client_ = nullptr;