// SPDX-License-Identifier: MIT // SPDX-FileCopyrightText: 2026 Martino Ferrari #pragma once #include "source/data_source.hpp" #include "source/flatudp/flatudp_signals.hpp" #include "source/source_config.hpp" #include #include #include #include #include #include #include namespace udpscope::flatudp { constexpr int kDefaultPort = 44488; ///< `UDPSender`'s own default bool valid_ipv4(const std::string& s); /// An IPv4 address in 224.0.0.0/4. bool valid_multicast(const std::string& s); /// Everything a configured flat UDP source needs, parsed once from the flat `SourceConfig`. struct FlatUdpSettings { int port = kDefaultPort; std::string bind; ///< local address; empty is any std::string multicast; ///< group to join; empty is plain unicast std::string iface; ///< local interface address for the join; empty is the system's choice double rate = 0.0; ///< the sender's cycle rate in Hz; 0 means not given bool big_endian = false; bool endian_ok = true; ///< false when `endian` is neither "little" nor "big" std::vector rows; std::string signals_error; ///< why the `signals` text did not parse; `rows` is empty then static FlatUdpSettings from(const SourceConfig& cfg); /// The first reason this cannot connect, in words for the user; empty when it can. std::string problem() const; /// `[multicast:]port`; names the session file. std::string identity() const; /// Datagrams per frame: 1 for a list with a waveform row, else `clamp(round(rate / 100), 1, 1024)`. std::size_t block() const; }; class FlatUdpSource final : public DataSource { public: explicit FlatUdpSource(const SourceConfig& cfg); ~FlatUdpSource() override; void open(SourceSink& sink) override; void poll(int timeout_ms) override; void close() override; SourceStats stats() const override; bool connected() const override { return connected_.load(std::memory_order_relaxed); } void set_datagram_tap(DatagramTap tap) override { tap_ = std::move(tap); } const FlatUdpSettings& settings() const { return cfg_; } /// Datagrams of the expected size / of any other size; for the probe, readable from any thread. std::uint64_t valid_datagrams() const { return valid_.load(std::memory_order_relaxed); } std::uint64_t wrong_size_datagrams() const { return wrong_.load(std::memory_order_relaxed); } std::string last_message() const; private: using Clock = std::chrono::steady_clock; void close_fd(); void drain_socket(); void accept(const std::uint8_t* data); void reject(std::size_t len); void flush(); void note(const std::string& msg); FlatUdpSettings cfg_; SourceSink* sink_ = nullptr; DatagramTap tap_; int fd_ = -1; bool configured_ = false; bool ever_connected_ = false; bool recv_error_noted_ = false; ///< a failing `recv` is logged once per open std::size_t expected_ = 0; ///< bytes of a valid datagram std::size_t block_ = 1; std::vector signals_; std::vector rx_; std::vector> pending_; ///< decoded, not yet framed: [signal][value] std::vector views_; std::size_t pending_n_ = 0; ///< datagrams in `pending_` Clock::time_point batch_start_; Clock::time_point last_valid_; std::set reported_sizes_; std::atomic connected_{false}; std::atomic valid_{0}; std::atomic wrong_{0}; mutable std::mutex msg_mu_; std::string last_message_; std::uint32_t frame_counter_ = 0; std::uint64_t frames_ = 0; std::uint64_t reconnects_ = 0; }; } // namespace udpscope::flatudp