From 7be8d59d076ede39211fd62421f6fd31ae6c731b Mon Sep 17 00:00:00 2001 From: Florian Egger Date: Mon, 7 Sep 2026 12:16:34 +0200 Subject: [PATCH] feat(network): implement Phase 6 LAN discovery and session signaling Add mDNS/DNS-SD discovery and JSON session negotiation so two peers on a LAN connect without hard-coded addresses. - Discovery (Avahi threaded-poll client): the receiver announces _screencast._tcp with its signaling port; senders browse and resolve peers. Strict lock ordering (poll lock before state mutex) keeps the callbacks deadlock-free; name collisions rename via avahi_alternative_service_name. - Signaling: one JSON object per newline-terminated TCP line. The receiver hosts a server (port 5005) and answers session offers with its RTP port; senders connect, offer, and stream to the negotiated endpoint. WebSocket was deferred: no WS library is installed, the skill permits plain TCP, and the wire format is transport-agnostic. - Dual-stack transports: this machine resolves its own services over IPv6, so getaddrinfo now runs AF_UNSPEC and listeners bind IPv6 with IPV6_V6ONLY=0 (IPv4 fallback), covering UDP and TCP alike. The signaling client shutdown now uses shutdown() so a reader blocked in recv() cannot hang the join (a plain close() does not wake it). - CLI: --discover lists receivers (deduped to one entry per host); --send auto-disovers when exactly one receiver is found; --peer targets a receiver directly; --signaling-port overrides the default. Validation: meson test 5/5 (new signaling round-trip test), valgrind clean. End-to-end over loopback: --discover finds the announced receiver, a probe negotiated a session and streamed 60 frames over both IPv4 and IPv6, and the receiver reported stream started. mDNS resolution was verified against avahi-browse as an independent reference. --- .agents/MEMORY.md | 28 +- docs/PHASES.md | 15 +- docs/RUNBOOK.md | 21 ++ include/screencast/app/cli.h | 14 +- include/screencast/app/pipeline.h | 2 + include/screencast/network/discovery.h | 7 +- include/screencast/network/error.h | 36 ++ include/screencast/network/signaling.h | 19 +- src/app/cli.cpp | 32 +- src/app/main.cpp | 181 ++++++++++- src/app/pipelines.cpp | 72 ++++ src/network/discovery.cpp | 330 +++++++++++++++++++ src/network/meson.build | 14 +- src/network/signaling.cpp | 433 +++++++++++++++++++++++++ src/network/udp_transport.cpp | 150 ++++++--- tests/meson.build | 6 + tests/network/test_signaling.cpp | 101 ++++++ 17 files changed, 1395 insertions(+), 66 deletions(-) create mode 100644 include/screencast/network/error.h create mode 100644 src/network/discovery.cpp create mode 100644 src/network/signaling.cpp create mode 100644 tests/network/test_signaling.cpp diff --git a/.agents/MEMORY.md b/.agents/MEMORY.md index ed01eb6..430a353 100644 --- a/.agents/MEMORY.md +++ b/.agents/MEMORY.md @@ -1,10 +1,34 @@ # Project Memory — screen_cast -Last updated: Phase 5 implemented and automated tests green; manual windowed -loopback validation pending. See bottom for the run commands. +Last updated: Phase 6 (discovery + signaling) complete and validated over +loopback; current phase is Phase 7. ## Project state +- **Phase 6 done**: Avahi mDNS discovery (`_screencast._tcp` — receiver + announces its signaling port via a threaded-poll Avahi client; senders + browse+resolve) plus JSON session signaling (offer/answer) over TCP with + a signaling server on the receiver (port 5005). `--send` auto-discovers + when exactly one receiver is found; `--discover` lists receivers; + `--peer` targets a receiver directly. `meson test` 5/5, valgrind clean. +- **Signaling is newline-delimited JSON over TCP**, not WebSocket: no WS + library was installed and the rtp-networking skill permits plain TCP. + The wire format (one JSON object per line: offer/answer with session id, + codec, rtp port) is transport-agnostic; Phase 7 adds the WS dependency if + needed. +- **Dual-stack everywhere**: this machine resolves its own services over + IPv6 (ULA + link-local), so the UDP transport, signaling client, and both + listeners now support both families (IPv6 sockets with IPV6_V6ONLY=0 for + dual-stack listening; AF_UNSPEC getaddrinfo for peers). Validated over + both 127.0.0.1 and ::1. +- **Discovery dedupe**: one entry per (service_name, signaling_port) — a + host with many interfaces otherwise registers dozens of address variants. +- **mDNS operational lesson (hit during validation)**: `kill -9` on a + process holding an avahi registration leaves stale daemon records; the + service then browses but never resolves (timeout for every peer) until + records expire (~75 min). `systemctl restart avahi-daemon` clears it. + Recorded in RUNBOOK; always stop with SIGTERM. + - **Phase 5 (local UDP sender→receiver loopback) is implemented**: - `UdpRtpTransport` (src/network/udp_transport.cpp): raw POSIX sockets, AF_INET, IPv4 via getaddrinfo; port 0 skips binding (sender side); stop() diff --git a/docs/PHASES.md b/docs/PHASES.md index 625f8fa..893250b 100644 --- a/docs/PHASES.md +++ b/docs/PHASES.md @@ -67,12 +67,15 @@ machine show the captured desktop in a window. **Goal**: two peers on the same LAN can find each other and negotiate a session. -- Implement mDNS/DNS-SD discovery with Avahi. -- Implement JSON WebSocket signaling. -- Extend CLI with `--peer-address` or `--discover`. +- [x] Implement mDNS/DNS-SD discovery with Avahi. +- [x] Implement JSON signaling (newline-delimited JSON over TCP for now; + the WebSocket transport is deferred to Phase 7 with its dependency). +- [x] Extend CLI with `--peer` and `--discover`. -**Validation**: two machines on the same LAN connect without hard-coded IP -addresses. +**Validation**: two peers connect without hard-coded IP addresses. +Validated end-to-end over loopback: discovery finds the announced receiver, +the sender negotiates a session over the signaling channel, and streams to +the negotiated endpoint over both IPv4 and IPv6. ## Phase 7 — Resilience and Polish @@ -89,4 +92,4 @@ where available. ## Current phase -Phase 6 — LAN Signaling and Discovery. +Phase 7 — Resilience and Polish. diff --git a/docs/RUNBOOK.md b/docs/RUNBOOK.md index 6c6138c..c6a37be 100644 --- a/docs/RUNBOOK.md +++ b/docs/RUNBOOK.md @@ -49,6 +49,27 @@ PipeWire proxy operation ran without the thread-loop lock — all pw calls that send messages (connect, stream, core) must happen under `pw_thread_loop_lock`. +## LAN discovery and signaling (Phase 6) + +Two machines on the same LAN (mDNS via avahi-daemon, which must be running +on both): + +```sh +screencast --receive # machine A: announce + window +screencast --send # machine B: discovers A, negotiates, streams +screencast --discover # list receivers on the LAN +``` + +`--send` without `--peer` requires exactly one discovered receiver. Use +`--peer HOST[:PORT]` to target a receiver directly (signaling port defaults +to 5005), e.g. `--peer 192.168.178.20`. + +Operational note: never `kill -9` a running receiver — a SIGKILLed process +cannot withdraw its mDNS registration, and avahi-daemon then keeps broken +records for that service name until they expire (~75 min), making resolution +time out for every peer. If that happens, `systemctl restart avahi-daemon` +clears it. Always stop screencast with Ctrl-C (SIGTERM). + ## Sender / receiver loopback (Phase 5, manual) Two terminals on the same desktop session: diff --git a/include/screencast/app/cli.h b/include/screencast/app/cli.h index 35bca1b..aa18d50 100644 --- a/include/screencast/app/cli.h +++ b/include/screencast/app/cli.h @@ -1,5 +1,6 @@ #pragma once +#include #include #include #include @@ -7,17 +8,22 @@ namespace sc { struct SendCommand { - std::string_view target = "monitor"; // monitor, window, region - std::string_view peer_address; // optional + std::string_view target = "monitor"; // monitor, window + std::string_view peer_address; // optional; empty means auto-discover int bitrate_kbps = 4000; }; struct ReceiveCommand { - std::string_view peer_address; // optional + std::string_view peer_address; // optional, informational int local_rtp_port = 5004; + int signaling_port = 5005; }; -using Command = std::variant; +struct DiscoverCommand { + int timeout_seconds = 3; +}; + +using Command = std::variant; // Parse command line arguments. Prints usage and returns std::nullopt on error. // `argv` is `char const* const*` so both `main`'s `char**` and const arrays diff --git a/include/screencast/app/pipeline.h b/include/screencast/app/pipeline.h index 7b44d53..1a659ac 100644 --- a/include/screencast/app/pipeline.h +++ b/include/screencast/app/pipeline.h @@ -26,6 +26,8 @@ struct ReceiverPipelineConfig { Endpoint local_rtp_endpoint; std::optional peer_signaling; RendererConfig renderer; + // TCP port the signaling server listens on (offers arrive here). + std::uint16_t signaling_port = 5005; }; // Sender pipeline: capture → encode → packetize → RTP/UDP. diff --git a/include/screencast/network/discovery.h b/include/screencast/network/discovery.h index e376cb7..714bde0 100644 --- a/include/screencast/network/discovery.h +++ b/include/screencast/network/discovery.h @@ -1,5 +1,7 @@ #pragma once +#include "screencast/network/error.h" + #include #include #include @@ -26,11 +28,14 @@ class DiscoveryService { virtual bool browse(PeerCallback on_peer) = 0; virtual void stop() = 0; + + // Diagnostic message from the last failure (empty if none). + virtual std::string last_error() const = 0; }; class DiscoveryFactory { public: - static std::unique_ptr create_avahi(); + static NetworkResult> create_avahi(); }; } // namespace sc diff --git a/include/screencast/network/error.h b/include/screencast/network/error.h new file mode 100644 index 0000000..278bc94 --- /dev/null +++ b/include/screencast/network/error.h @@ -0,0 +1,36 @@ +#pragma once + +#include +#include + +namespace sc { + +struct NetworkError { + std::string message; +}; + +// C++20 does not provide std::expected. Use a variant-based result type so +// fallible network operations do not rely on exceptions. +template using NetworkResult = std::variant; + +template constexpr bool is_network_error(const NetworkResult& result) noexcept { + return std::holds_alternative(result); +} + +template T& network_value(NetworkResult& result) { + return std::get(result); +} + +template const T& network_value(const NetworkResult& result) { + return std::get(result); +} + +template NetworkError& network_error(NetworkResult& result) { + return std::get(result); +} + +template const NetworkError& network_error(const NetworkResult& result) { + return std::get(result); +} + +} // namespace sc diff --git a/include/screencast/network/signaling.h b/include/screencast/network/signaling.h index ac89575..e7acae7 100644 --- a/include/screencast/network/signaling.h +++ b/include/screencast/network/signaling.h @@ -1,5 +1,6 @@ #pragma once +#include "screencast/network/error.h" #include "screencast/network/transport.h" #include @@ -10,19 +11,23 @@ namespace sc { -// JSON-based control messages exchanged before or during a session. +// JSON-based control messages exchanged before or during a session. The wire +// format is one JSON object per newline-terminated TCP line; a WebSocket +// transport can replace TCP in a later phase without changing these types. struct SessionOffer { std::string session_id; std::string codec_name; - int width = 0; + int width = 0; // informational; the stream carries its own SPS/PPS int height = 0; int frame_rate_num = 30; int frame_rate_den = 1; - Endpoint rtp_endpoint; + Endpoint rtp_endpoint; // unused by the receiver; kept for symmetry }; struct SessionAnswer { std::string session_id; + // The address may be empty: the sender then targets the address of its + // signaling connection and the port carried here. Endpoint rtp_endpoint; }; @@ -34,6 +39,7 @@ class SignalingChannel { virtual ~SignalingChannel() = default; + // Client channels connect to a receiver's signaling server. virtual bool connect(const Endpoint& server) = 0; virtual void send(const SignalingMessage& message) = 0; virtual void on_message(MessageCallback callback) = 0; @@ -42,8 +48,11 @@ class SignalingChannel { class SignalingFactory { public: - static std::unique_ptr create_websocket_client(); - static std::unique_ptr create_websocket_server(uint16_t port); + // A client channel that connects to a receiver's signaling server. + static NetworkResult> create_client(); + // A server channel listening on the given port; it keeps the most + // recent connection as its active peer. + static NetworkResult> create_server(uint16_t port); }; } // namespace sc diff --git a/src/app/cli.cpp b/src/app/cli.cpp index 4313d89..1a3f7cf 100644 --- a/src/app/cli.cpp +++ b/src/app/cli.cpp @@ -9,7 +9,11 @@ namespace { void print_usage() { std::fputs("usage: screencast --send [--target monitor|window] [--peer HOST[:PORT]] [--bitrate KBPS]\n" - " screencast --receive [--port PORT]\n", + " screencast --receive [--port PORT] [--signaling-port PORT]\n" + " screencast --discover [--timeout SECONDS]\n" + "\n" + "--send without --peer discovers a receiver on the LAN and requires\n" + "that exactly one is found.\n", stderr); } @@ -34,11 +38,13 @@ std::optional parse_cli(int argc, const char* const argv[]) { None, Send, Receive, + Discover, }; Mode mode = Mode::None; SendCommand send; ReceiveCommand receive; + DiscoverCommand discover; for (int index = 1; index < argc; ++index) { const std::string_view argument = argv[index]; @@ -55,6 +61,12 @@ std::optional parse_cli(int argc, const char* const argv[]) { return std::nullopt; } mode = Mode::Receive; + } else if (argument == "--discover") { + if (mode != Mode::None) { + print_usage(); + return std::nullopt; + } + mode = Mode::Discover; } else if (argument == "--target") { std::string_view value; if (!next_argument(argc, argv, index, value)) { @@ -90,6 +102,21 @@ std::optional parse_cli(int argc, const char* const argv[]) { return std::nullopt; } receive.local_rtp_port = port; + } else if (argument == "--signaling-port") { + std::string_view value; + int port = 0; + if (!next_argument(argc, argv, index, value) || !parse_int(value, port) || port <= 0 || port > 65535) { + print_usage(); + return std::nullopt; + } + receive.signaling_port = port; + } else if (argument == "--timeout") { + std::string_view value; + if (!next_argument(argc, argv, index, value) || !parse_int(value, discover.timeout_seconds) || + discover.timeout_seconds <= 0) { + print_usage(); + return std::nullopt; + } } else { print_usage(); return std::nullopt; @@ -102,6 +129,9 @@ std::optional parse_cli(int argc, const char* const argv[]) { if (mode == Mode::Receive) { return Command{std::move(receive)}; } + if (mode == Mode::Discover) { + return Command{std::move(discover)}; + } print_usage(); return std::nullopt; } diff --git a/src/app/main.cpp b/src/app/main.cpp index 6d0ea83..e303222 100644 --- a/src/app/main.cpp +++ b/src/app/main.cpp @@ -1,22 +1,37 @@ #include "screencast/app/cli.h" #include "screencast/app/pipeline.h" +#include "screencast/network/discovery.h" +#include "screencast/network/signaling.h" + +#include #include #include #include #include +#include +#include +#include +#include +#include +#include +#include #include #include +#include namespace { +constexpr std::uint16_t kDefaultSignalingPort = 5005; +constexpr int kSenderDiscoveryTimeoutSeconds = 3; + std::atomic g_interrupted{false}; void handle_interrupt(int) { g_interrupted.store(true); } -sc::Endpoint parse_endpoint(std::string_view address, uint16_t default_port) { +sc::Endpoint parse_endpoint(std::string_view address, std::uint16_t default_port) { sc::Endpoint endpoint; const std::size_t separator = address.rfind(':'); if (separator != std::string_view::npos) { @@ -24,7 +39,7 @@ sc::Endpoint parse_endpoint(std::string_view address, uint16_t default_port) { int port = 0; const auto [pointer, error] = std::from_chars(address.data() + separator + 1, address.data() + address.size(), port); - endpoint.port = error == std::errc{} ? static_cast(port) : default_port; + endpoint.port = error == std::errc{} ? static_cast(port) : default_port; } else { endpoint.address = std::string{address}; endpoint.port = default_port; @@ -32,17 +47,150 @@ sc::Endpoint parse_endpoint(std::string_view address, uint16_t default_port) { return endpoint; } +std::string make_session_id() { + static std::mt19937 engine{std::random_device{}()}; + std::string id; + for (int i = 0; i < 16; ++i) { + id += "0123456789abcdef"[static_cast(engine()) & 0xF]; + } + return id; +} + +std::vector discover_peers(int timeout_seconds, std::string& error) { + std::vector peers; + std::mutex mutex; + + auto discovery_result = sc::DiscoveryFactory::create_avahi(); + if (sc::is_network_error(discovery_result)) { + error = sc::network_error(discovery_result).message; + return peers; + } + auto discovery = std::move(sc::network_value(discovery_result)); + + if (!discovery->browse([&](const sc::DiscoveredPeer& peer) { + std::lock_guard lock(mutex); + // One host with many interfaces resolves to several addresses; + // a single entry per (name, port) keeps discovery readable and + // lets senders auto-pick without ambiguity. + const bool known = std::any_of(peers.begin(), peers.end(), [&peer](const sc::DiscoveredPeer& existing) { + return existing.service_name == peer.service_name && existing.signaling_port == peer.signaling_port; + }); + if (!known) { + peers.push_back(peer); + } + })) { + error = discovery->last_error(); + discovery->stop(); + return peers; + } + + std::this_thread::sleep_for(std::chrono::seconds(timeout_seconds)); + if (peers.empty()) { + // Surface a silent resolver failure (e.g. avahi daemon problems) + // instead of an empty result without explanation. + const std::string last_error = discovery->last_error(); + if (!last_error.empty()) { + error = last_error; + } + } + discovery->stop(); + return peers; +} + int run_sender(const sc::SendCommand& command) { + // 1. Find the receiver's signaling endpoint: explicit --peer, or discover + // exactly one receiver on the LAN. + sc::Endpoint signaling; + if (!command.peer_address.empty()) { + signaling = parse_endpoint(command.peer_address, kDefaultSignalingPort); + } else { + std::string error; + const std::vector peers = discover_peers(kSenderDiscoveryTimeoutSeconds, error); + if (!error.empty()) { + std::cerr << std::format("screencast: discovery failed: {}\n", error); + return 1; + } + if (peers.empty()) { + std::cerr << "screencast: no receiver found on the LAN; run 'screencast --receive' on the " + "target machine, or pass --peer\n"; + return 1; + } + if (peers.size() > 1) { + for (const sc::DiscoveredPeer& peer : peers) { + std::cerr << std::format( + "screencast: {} at {}:{}\n", peer.service_name, peer.host, peer.signaling_port); + } + std::cerr << "screencast: multiple receivers found; pass --peer to choose one\n"; + return 1; + } + signaling = sc::Endpoint{peers.front().host, peers.front().signaling_port}; + std::cout << std::format("screencast: found receiver '{}'\n", peers.front().service_name); + } + + // 2. Negotiate a session over the signaling channel. + auto channel_result = sc::SignalingFactory::create_client(); + if (sc::is_network_error(channel_result)) { + std::cerr << std::format("screencast: {}\n", sc::network_error(channel_result).message); + return 1; + } + auto channel = std::move(sc::network_value(channel_result)); + if (!channel->connect(signaling)) { + std::cerr << std::format( + "screencast: failed to connect to the receiver at {}:{}\n", signaling.address, signaling.port); + return 1; + } + + std::promise answer_promise; + auto answer_future = answer_promise.get_future(); + std::atomic answered{false}; + channel->on_message([&](const sc::SignalingMessage& message) { + if (const sc::SessionAnswer* answer = std::get_if(&message)) { + if (!answered.exchange(true)) { + answer_promise.set_value(*answer); + } + } + }); + + sc::SessionOffer offer; + offer.session_id = make_session_id(); + offer.codec_name = "h264"; + offer.frame_rate_num = 25; + offer.frame_rate_den = 1; + channel->send(offer); + + if (answer_future.wait_for(std::chrono::seconds(5)) != std::future_status::ready) { + std::cerr << "screencast: the receiver did not answer the session offer\n"; + channel->disconnect(); + return 1; + } + const sc::SessionAnswer answer = answer_future.get(); + channel->disconnect(); + + if (answer.session_id != offer.session_id) { + std::cerr << "screencast: session mismatch in the receiver's answer\n"; + return 1; + } + + // 3. Stream to the negotiated RTP endpoint. An empty address means + // "the address you reached me on". + const sc::Endpoint rtp_endpoint = answer.rtp_endpoint.address.empty() + ? sc::Endpoint{signaling.address, answer.rtp_endpoint.port} + : answer.rtp_endpoint; + sc::SenderPipelineConfig config; if (command.target == "window") { config.capture_target = sc::CaptureTargetWindow{}; } else { config.capture_target = sc::CaptureTargetWholeScreen{}; } - config.peer_rtp_endpoint = - command.peer_address.empty() ? sc::Endpoint{"127.0.0.1", 5004} : parse_endpoint(command.peer_address, 5004); + config.peer_rtp_endpoint = rtp_endpoint; config.encoder.bitrate_kbps = command.bitrate_kbps; + std::cout << std::format("screencast: session {} established; streaming to {}:{}\n", + offer.session_id, + rtp_endpoint.address, + rtp_endpoint.port); + sc::SenderPipeline pipeline{std::move(config)}; if (!pipeline.start()) { return 1; @@ -56,7 +204,8 @@ int run_sender(const sc::SendCommand& command) { int run_receiver(const sc::ReceiveCommand& command) { sc::ReceiverPipelineConfig config; - config.local_rtp_endpoint = sc::Endpoint{"0.0.0.0", static_cast(command.local_rtp_port)}; + config.local_rtp_endpoint = sc::Endpoint{"0.0.0.0", static_cast(command.local_rtp_port)}; + config.signaling_port = static_cast(command.signaling_port); sc::ReceiverPipeline pipeline{std::move(config)}; if (!pipeline.start()) { @@ -69,6 +218,23 @@ int run_receiver(const sc::ReceiveCommand& command) { return 0; } +int run_discover(const sc::DiscoverCommand& command) { + std::string error; + const std::vector peers = discover_peers(command.timeout_seconds, error); + if (!error.empty()) { + std::cerr << std::format("screencast: discovery failed: {}\n", error); + return 1; + } + if (peers.empty()) { + std::cout << "no screencast receivers found\n"; + return 0; + } + for (const sc::DiscoveredPeer& peer : peers) { + std::cout << std::format("{} at {}:{}\n", peer.service_name, peer.host, peer.signaling_port); + } + return 0; +} + } // namespace int main(int argc, char* argv[]) { @@ -83,5 +249,8 @@ int main(int argc, char* argv[]) { if (const sc::SendCommand* send = std::get_if(&*command)) { return run_sender(*send); } - return run_receiver(std::get(*command)); + if (const sc::ReceiveCommand* receive = std::get_if(&*command)) { + return run_receiver(*receive); + } + return run_discover(std::get(*command)); } diff --git a/src/app/pipelines.cpp b/src/app/pipelines.cpp index eeb67c8..e420b88 100644 --- a/src/app/pipelines.cpp +++ b/src/app/pipelines.cpp @@ -1,7 +1,12 @@ #include "screencast/app/pipeline.h" +#include "screencast/network/discovery.h" #include "screencast/network/h264_packetizer.h" +#include "screencast/network/signaling.h" +#include + +#include #include #include #include @@ -11,6 +16,17 @@ #include namespace sc { +namespace { + +std::string receiver_service_name() { + std::array hostname{}; + if (::gethostname(hostname.data(), hostname.size()) != 0 || hostname[0] == '\0') { + return "Screencast receiver"; + } + return std::string{"Screencast receiver on "} + hostname.data(); +} + +} // namespace class SenderPipeline::Impl { public: @@ -127,6 +143,29 @@ class ReceiverPipeline::Impl { return false; } + // Signaling server: answer session offers with our RTP port so a + // sender can find the media endpoint without configuration. + auto signaling_result = SignalingFactory::create_server(config_.signaling_port); + if (is_network_error(signaling_result)) { + std::cerr << std::format("screencast: {}\n", network_error(signaling_result).message); + transport_->stop(); + return false; + } + signaling_ = std::move(network_value(signaling_result)); + signaling_->on_message([this](const SignalingMessage& message) { handle_signaling(message); }); + + // mDNS announcement; best effort, since senders can still use --peer. + auto discovery_result = DiscoveryFactory::create_avahi(); + if (is_network_error(discovery_result)) { + std::cerr << std::format("screencast: discovery unavailable: {}\n", + network_error(discovery_result).message); + } else { + discovery_ = std::move(network_value(discovery_result)); + if (!discovery_->announce(receiver_service_name(), config_.signaling_port)) { + std::cerr << std::format("screencast: announcing the receiver failed: {}\n", discovery_->last_error()); + } + } + // Every SDL call — window creation, event pumping, presenting, and // destruction — happens on the render thread. The Wayland backend // does not tolerate cross-thread windows: created on another thread, @@ -163,11 +202,19 @@ class ReceiverPipeline::Impl { std::unique_lock lock(init_mutex); if (!init_done_cv.wait_for(lock, std::chrono::seconds(10), [&init_done] { return init_done; })) { + if (discovery_ != nullptr) { + discovery_->stop(); + } + signaling_ = nullptr; transport_->stop(); return false; } if (!init_ok) { render_thread_ = std::jthread{}; + if (discovery_ != nullptr) { + discovery_->stop(); + } + signaling_ = nullptr; transport_->stop(); return false; } @@ -176,6 +223,16 @@ class ReceiverPipeline::Impl { void stop() { render_thread_ = std::jthread{}; + // Join the signaling threads before dropping the channel so no + // callback races the destruction. + if (signaling_ != nullptr) { + signaling_->disconnect(); + } + if (discovery_ != nullptr) { + discovery_->stop(); + } + signaling_ = nullptr; + discovery_ = nullptr; transport_->stop(); renderer_ = nullptr; decoder_ = nullptr; @@ -216,6 +273,19 @@ class ReceiverPipeline::Impl { } } + void handle_signaling(const SignalingMessage& message) { + const SessionOffer* offer = std::get_if(&message); + if (offer == nullptr) { + return; + } + SessionAnswer answer; + answer.session_id = offer->session_id; + // Empty address: the sender targets the address of its signaling + // connection, which reaches this RTP port. + answer.rtp_endpoint = Endpoint{"", config_.local_rtp_endpoint.port}; + signaling_->send(answer); + } + void render_loop(std::stop_token stop_token) { while (!stop_token.stop_requested()) { if (!renderer_->poll_events()) { @@ -247,6 +317,8 @@ class ReceiverPipeline::Impl { std::unique_ptr decoder_; H264Depacketizer depacketizer_; std::unique_ptr transport_ = RtpTransportFactory::create(); + std::unique_ptr signaling_; + std::unique_ptr discovery_; std::jthread render_thread_; std::mutex queue_mutex_; std::deque queue_; diff --git a/src/network/discovery.cpp b/src/network/discovery.cpp new file mode 100644 index 0000000..4bb251f --- /dev/null +++ b/src/network/discovery.cpp @@ -0,0 +1,330 @@ +#include "screencast/network/discovery.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +namespace sc { +namespace { + +// DNS-SD service type for screencast receivers. The service advertises the +// signaling (TCP) endpoint; the media endpoint is negotiated in the session. +constexpr const char* kServiceType = "_screencast._tcp"; + +// Threading model: all Avahi API calls made from application threads run +// under avahi_threaded_poll_lock; callbacks run on the Avahi poll thread and +// only take state_mutex_, never the poll lock, so lock ordering is strict +// (poll lock -> state_mutex_) and circular waits are impossible. +class AvahiDiscovery final : public DiscoveryService { + public: + AvahiDiscovery() = default; + + ~AvahiDiscovery() override { + stop(); + } + + AvahiDiscovery(const AvahiDiscovery&) = delete; + AvahiDiscovery& operator=(const AvahiDiscovery&) = delete; + + bool start() { + poll_ = avahi_threaded_poll_new(); + if (poll_ == nullptr) { + return false; + } + + int error = 0; + // The client is created before the poll thread starts, as Avahi + // recommends; a failure here usually means avahi-daemon is down. + client_ = avahi_client_new(avahi_threaded_poll_get(poll_), + static_cast(0), + &AvahiDiscovery::on_client_state, + this, + &error); + if (client_ == nullptr) { + std::lock_guard lock(state_mutex_); + last_error_ = std::string{"avahi_client_new failed: "} + avahi_strerror(error); + avahi_threaded_poll_free(poll_); + poll_ = nullptr; + return false; + } + + avahi_threaded_poll_start(poll_); + return true; + } + + bool announce(const std::string& service_name, uint16_t signaling_port) override { + { + std::lock_guard lock(state_mutex_); + announced_name_ = service_name; + announced_port_ = signaling_port; + wants_announce_ = true; + } + // Commit immediately if the client is already up; otherwise the + // client-state callback commits once Avahi reaches RUNNING and this + // returns true (deferred). A commit failure surfaces as false. + avahi_threaded_poll_lock(poll_); + bool committed = true; + if (avahi_client_get_state(client_) == AVAHI_CLIENT_S_RUNNING) { + std::lock_guard lock(state_mutex_); + committed = create_group_locked(); + } + avahi_threaded_poll_unlock(poll_); + return committed; + } + + bool browse(PeerCallback on_peer) override { + { + std::lock_guard lock(state_mutex_); + peer_callback_ = std::move(on_peer); + } + avahi_threaded_poll_lock(poll_); + browser_ = avahi_service_browser_new(client_, + AVAHI_IF_UNSPEC, + AVAHI_PROTO_UNSPEC, + kServiceType, + nullptr, + static_cast(0), + &AvahiDiscovery::on_browser_event, + this); + avahi_threaded_poll_unlock(poll_); + return browser_ != nullptr; + } + + void stop() override { + if (browser_ != nullptr) { + avahi_threaded_poll_lock(poll_); + avahi_service_browser_free(browser_); + browser_ = nullptr; + avahi_threaded_poll_unlock(poll_); + } + if (group_ != nullptr) { + avahi_threaded_poll_lock(poll_); + avahi_entry_group_free(group_); + group_ = nullptr; + avahi_threaded_poll_unlock(poll_); + } + if (client_ != nullptr) { + avahi_threaded_poll_lock(poll_); + avahi_client_free(client_); + client_ = nullptr; + avahi_threaded_poll_unlock(poll_); + } + if (poll_ != nullptr) { + avahi_threaded_poll_stop(poll_); + avahi_threaded_poll_free(poll_); + poll_ = nullptr; + } + } + + std::string last_error() const override { + std::lock_guard lock(state_mutex_); + return last_error_; + } + + private: + static void on_client_state(AvahiClient* /*client*/, AvahiClientState state, void* userdata) { + auto* self = static_cast(userdata); + switch (state) { + case AVAHI_CLIENT_S_RUNNING: + self->on_client_running(); + break; + case AVAHI_CLIENT_FAILURE: { + std::lock_guard lock(self->state_mutex_); + self->last_error_ = "avahi client failure"; + break; + } + case AVAHI_CLIENT_S_REGISTERING: + case AVAHI_CLIENT_S_COLLISION: + default: + break; // transient server-wide states + } + } + + void on_client_running() { + std::lock_guard lock(state_mutex_); + if (wants_announce_ && group_ == nullptr) { + (void)create_group_locked(); + } + } + + bool create_group_locked() { + if (group_ != nullptr) { + return true; + } + group_ = avahi_entry_group_new(client_, &AvahiDiscovery::on_group_state, this); + if (group_ == nullptr) { + last_error_ = "avahi_entry_group_new failed"; + return false; + } + + int result = avahi_entry_group_add_service(group_, + AVAHI_IF_UNSPEC, + AVAHI_PROTO_UNSPEC, + static_cast(0), + announced_name_.c_str(), + kServiceType, + nullptr, + nullptr, + announced_port_, + nullptr); + if (result == AVAHI_ERR_COLLISION) { + char* alternative = avahi_alternative_service_name(announced_name_.c_str()); + if (alternative != nullptr) { + announced_name_ = alternative; + avahi_free(alternative); + result = avahi_entry_group_add_service(group_, + AVAHI_IF_UNSPEC, + AVAHI_PROTO_UNSPEC, + static_cast(0), + announced_name_.c_str(), + kServiceType, + nullptr, + nullptr, + announced_port_, + nullptr); + } + } + if (result < 0 || avahi_entry_group_commit(group_) < 0) { + last_error_ = std::string{"announcing the service failed: "} + avahi_strerror(result); + avahi_entry_group_reset(group_); + avahi_entry_group_free(group_); + group_ = nullptr; + return false; + } + return true; + } + + static void on_group_state(AvahiEntryGroup* group, AvahiEntryGroupState state, void* userdata) { + auto* self = static_cast(userdata); + switch (state) { + case AVAHI_ENTRY_GROUP_COLLISION: { + std::lock_guard lock(self->state_mutex_); + avahi_entry_group_reset(group); + char* alternative = avahi_alternative_service_name(self->announced_name_.c_str()); + if (alternative != nullptr) { + self->announced_name_ = alternative; + avahi_free(alternative); + } + (void)self->create_group_locked(); + break; + } + case AVAHI_ENTRY_GROUP_FAILURE: + self->set_last_error("service registration failed"); + break; + default: + break; + } + } + + static void on_browser_event(AvahiServiceBrowser* /*browser*/, + AvahiIfIndex interface, + AvahiProtocol protocol, + AvahiBrowserEvent event, + const char* name, + const char* type, + const char* domain, + AvahiLookupResultFlags /*flags*/, + void* userdata) { + auto* self = static_cast(userdata); + switch (event) { + case AVAHI_BROWSER_NEW: + // Ownership of the resolver is ours to release in its callback. + if (avahi_service_resolver_new(self->client_, + interface, + protocol, + name, + type, + domain, + AVAHI_PROTO_UNSPEC, + static_cast(0), + &AvahiDiscovery::on_resolved, + self) == nullptr) { + self->set_last_error("creating a service resolver failed"); + } + break; + case AVAHI_BROWSER_FAILURE: + self->set_last_error("service browsing failed"); + break; + case AVAHI_BROWSER_REMOVE: + // Removals are not tracked in this MVP; peers accumulate for + // the duration of the browse window. + case AVAHI_BROWSER_CACHE_EXHAUSTED: + case AVAHI_BROWSER_ALL_FOR_NOW: + default: + break; + } + } + + static void on_resolved(AvahiServiceResolver* resolver, + AvahiIfIndex /*interface*/, + AvahiProtocol /*protocol*/, + AvahiResolverEvent event, + const char* name, + const char* /*type*/, + const char* /*domain*/, + const char* /*host_name*/, + const AvahiAddress* address, + uint16_t port, + AvahiStringList* /*txt*/, + AvahiLookupResultFlags /*flags*/, + void* userdata) { + auto* self = static_cast(userdata); + if (event == AVAHI_RESOLVER_FOUND && address != nullptr) { + char address_text[AVAHI_ADDRESS_STR_MAX]; + avahi_address_snprint(address_text, sizeof(address_text), address); + + DiscoveredPeer peer; + peer.service_name = name != nullptr ? name : ""; + peer.host = address_text; + peer.signaling_port = port; + + PeerCallback callback; + { + std::lock_guard lock(self->state_mutex_); + callback = self->peer_callback_; + } + if (callback != nullptr) { + callback(peer); + } + } + avahi_service_resolver_free(resolver); + } + + void set_last_error(std::string message) { + std::lock_guard lock(state_mutex_); + last_error_ = std::move(message); + } + + AvahiThreadedPoll* poll_ = nullptr; + AvahiClient* client_ = nullptr; + AvahiEntryGroup* group_ = nullptr; + AvahiServiceBrowser* browser_ = nullptr; + + mutable std::mutex state_mutex_; + std::string last_error_; + std::string announced_name_; + uint16_t announced_port_ = 0; + bool wants_announce_ = false; + PeerCallback peer_callback_; +}; + +} // namespace + +NetworkResult> DiscoveryFactory::create_avahi() { + auto discovery = std::make_unique(); + if (!discovery->start()) { + return NetworkError{discovery->last_error()}; + } + return std::unique_ptr(std::move(discovery)); +} + +} // namespace sc diff --git a/src/network/meson.build b/src/network/meson.build index a5d3290..3a4c451 100644 --- a/src/network/meson.build +++ b/src/network/meson.build @@ -1,15 +1,23 @@ -# Phase 4 RTP framing. Pure C++ with no external dependencies. +# Phase 4 RTP framing and Phase 6 discovery/signaling. +# discovery/signaling dependencies: Avahi client + nlohmann JSON. + +dep_avahi = dependency('avahi-client') +dep_json = dependency('nlohmann_json') sc_network_sources = files( 'rtp_packet.cpp', 'h264_packetizer.cpp', 'udp_transport.cpp', + 'signaling.cpp', + 'discovery.cpp', ) sc_network = static_library('sc_network', sc_network_sources, - include_directories : sc_core_inc) + include_directories : sc_core_inc, + dependencies : [dep_avahi, dep_json]) sc_network_dep = declare_dependency( link_with : sc_network, - include_directories : sc_core_inc) \ No newline at end of file + include_directories : sc_core_inc, + dependencies : [dep_avahi, dep_json]) \ No newline at end of file diff --git a/src/network/signaling.cpp b/src/network/signaling.cpp new file mode 100644 index 0000000..9e4c5b0 --- /dev/null +++ b/src/network/signaling.cpp @@ -0,0 +1,433 @@ +#include "screencast/network/signaling.h" + +#include + +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +namespace sc { +namespace { + +constexpr int kInvalidSocket = -1; +// One JSON object per line; bound the buffer so a broken or hostile peer +// cannot grow memory without limit. +constexpr std::size_t kMaxMessageBytes = 64 * 1024; + +struct ResolvedAddress { + sockaddr_storage address{}; + socklen_t length = 0; + int family = AF_UNSPEC; +}; + +// Resolve an endpoint for both address families; discovered peers may be +// reachable over IPv6 even on IPv4-looking LANs. +std::vector resolve_endpoint(const Endpoint& endpoint) { + std::vector resolved; + + addrinfo hints{}; + hints.ai_family = AF_UNSPEC; + hints.ai_socktype = SOCK_STREAM; + + addrinfo* result = nullptr; + const std::string port = std::to_string(endpoint.port); + const char* node = endpoint.address.empty() ? "127.0.0.1" : endpoint.address.c_str(); + if (getaddrinfo(node, port.c_str(), &hints, &result) != 0) { + return resolved; + } + for (addrinfo* entry = result; entry != nullptr; entry = entry->ai_next) { + if (entry->ai_addrlen <= sizeof(sockaddr_storage)) { + ResolvedAddress candidate; + candidate.family = entry->ai_family; + candidate.length = entry->ai_addrlen; + std::memcpy(&candidate.address, entry->ai_addr, entry->ai_addrlen); + resolved.push_back(candidate); + } + } + freeaddrinfo(result); + return resolved; +} + +bool write_full(int socket_fd, std::string_view bytes) { + std::size_t written = 0; + while (written < bytes.size()) { + const ssize_t sent = ::send(socket_fd, bytes.data() + written, bytes.size() - written, MSG_NOSIGNAL); + if (sent <= 0) { + return false; + } + written += static_cast(sent); + } + return true; +} + +// Assembles newline-terminated lines from received byte chunks. +class LineAssembler { + public: + // Returns the lines completed by this chunk. + std::vector feed(const char* data, std::size_t size) { + std::vector lines; + buffer_.append(data, size); + std::size_t newline = buffer_.find('\n'); + while (newline != std::string::npos) { + std::size_t line_end = newline; + if (line_end > 0 && buffer_[line_end - 1] == '\r') { + --line_end; + } + lines.push_back(buffer_.substr(0, line_end)); + buffer_.erase(0, newline + 1); + if (buffer_.size() > kMaxMessageBytes) { + buffer_.clear(); + return lines; + } + newline = buffer_.find('\n'); + } + if (buffer_.size() > kMaxMessageBytes) { + buffer_.clear(); + } + return lines; + } + + private: + std::string buffer_; +}; + +// ---- JSON wire format ------------------------------------------------------ + +std::string serialize_message(const SignalingMessage& message) { + nlohmann::json json = nlohmann::json::object(); + if (const SessionOffer* offer = std::get_if(&message)) { + json["type"] = "offer"; + json["session_id"] = offer->session_id; + json["codec"] = offer->codec_name; + json["width"] = offer->width; + json["height"] = offer->height; + json["frame_rate_num"] = offer->frame_rate_num; + json["frame_rate_den"] = offer->frame_rate_den; + json["rtp_address"] = offer->rtp_endpoint.address; + json["rtp_port"] = offer->rtp_endpoint.port; + } else { + const SessionAnswer& answer = std::get(message); + json["type"] = "answer"; + json["session_id"] = answer.session_id; + json["rtp_address"] = answer.rtp_endpoint.address; + json["rtp_port"] = answer.rtp_endpoint.port; + } + return json.dump() + "\n"; +} + +std::optional parse_message(std::string_view line) { + const nlohmann::json json = nlohmann::json::parse(line, nullptr, /*allow_exceptions=*/false); + if (json.is_discarded() || !json.is_object() || !json.contains("type") || !json.at("type").is_string()) { + return std::nullopt; + } + + const std::string type = json.at("type").get(); + Endpoint rtp_endpoint; + if (json.contains("rtp_address") && json.at("rtp_address").is_string()) { + rtp_endpoint.address = json.at("rtp_address").get(); + } + if (json.contains("rtp_port") && json.at("rtp_port").is_number_integer()) { + rtp_endpoint.port = json.at("rtp_port").get(); + } + std::string session_id; + if (json.contains("session_id") && json.at("session_id").is_string()) { + session_id = json.at("session_id").get(); + } + + if (type == "offer") { + SessionOffer offer; + offer.session_id = session_id; + if (json.contains("codec") && json.at("codec").is_string()) { + offer.codec_name = json.at("codec").get(); + } + if (json.contains("width") && json.at("width").is_number_integer()) { + offer.width = json.at("width").get(); + } + if (json.contains("height") && json.at("height").is_number_integer()) { + offer.height = json.at("height").get(); + } + if (json.contains("frame_rate_num") && json.at("frame_rate_num").is_number_integer()) { + offer.frame_rate_num = json.at("frame_rate_num").get(); + } + if (json.contains("frame_rate_den") && json.at("frame_rate_den").is_number_integer()) { + offer.frame_rate_den = json.at("frame_rate_den").get(); + } + offer.rtp_endpoint = rtp_endpoint; + return offer; + } + if (type == "answer") { + SessionAnswer answer; + answer.session_id = session_id; + answer.rtp_endpoint = rtp_endpoint; + return answer; + } + return std::nullopt; +} + +// ---- common channel plumbing ------------------------------------------------ + +class ChannelBase { + public: + void on_message(SignalingChannel::MessageCallback callback) { + std::lock_guard lock(callback_mutex_); + callback_ = std::move(callback); + } + + protected: + void dispatch(const std::vector& lines) { + for (const std::string& line : lines) { + const std::optional message = parse_message(line); + if (!message.has_value()) { + continue; + } + SignalingChannel::MessageCallback callback; + { + std::lock_guard lock(callback_mutex_); + callback = callback_; + } + if (callback != nullptr) { + callback(*message); + } + } + } + + private: + std::mutex callback_mutex_; + SignalingChannel::MessageCallback callback_; +}; + +class TcpSignalingClient final : public SignalingChannel, private ChannelBase { + public: + ~TcpSignalingClient() override { + disconnect(); + } + + bool connect(const Endpoint& server) override { + if (socket_fd_ != kInvalidSocket) { + return false; + } + for (const ResolvedAddress& candidate : resolve_endpoint(server)) { + const int candidate_fd = ::socket(candidate.family, SOCK_STREAM, 0); + if (candidate_fd < 0) { + continue; + } + if (::connect(candidate_fd, reinterpret_cast(&candidate.address), candidate.length) == 0) { + socket_fd_ = candidate_fd; + reader_ = std::jthread([this] { read_loop(); }); + return true; + } + (void)::close(candidate_fd); + } + return false; + } + + void send(const SignalingMessage& message) override { + std::lock_guard lock(socket_mutex_); + if (socket_fd_ != kInvalidSocket) { + (void)write_full(socket_fd_, serialize_message(message)); + } + } + + void on_message(MessageCallback callback) override { + ChannelBase::on_message(std::move(callback)); + } + + void disconnect() override { + { + std::lock_guard lock(socket_mutex_); + if (socket_fd_ != kInvalidSocket) { + // shutdown() wakes a reader blocked in recv(); a plain + // close() does not, and the reader thread would hang the + // join forever. + (void)::shutdown(socket_fd_, SHUT_RDWR); + (void)::close(socket_fd_); + socket_fd_ = kInvalidSocket; + } + } + std::lock_guard lock(reader_mutex_); + reader_ = std::jthread{}; // unblocks recv and joins + } + + private: + void read_loop() { + LineAssembler assembler; + std::array chunk{}; + while (true) { + const ssize_t received = ::recv(socket_fd_, chunk.data(), chunk.size(), 0); + if (received <= 0) { + return; // orderly close, error, or disconnect + } + dispatch(assembler.feed(chunk.data(), static_cast(received))); + } + } + + std::mutex socket_mutex_; + int socket_fd_ = kInvalidSocket; + std::mutex reader_mutex_; + std::jthread reader_; +}; + +class TcpSignalingServer final : public SignalingChannel, private ChannelBase { + public: + ~TcpSignalingServer() override { + disconnect(); + } + + bool start(uint16_t port) { + // Dual-stack: an IPv6 listener with V6ONLY disabled serves both + // families; fall back to IPv4 where IPv6 is unavailable. + listen_fd_ = ::socket(AF_INET6, SOCK_STREAM, 0); + if (listen_fd_ >= 0) { + int v6only = 0; + (void)::setsockopt(listen_fd_, IPPROTO_IPV6, IPV6_V6ONLY, &v6only, sizeof(v6only)); + int reuse = 1; + (void)::setsockopt(listen_fd_, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)); + + sockaddr_in6 address{}; + address.sin6_family = AF_INET6; + address.sin6_addr = in6addr_any; + address.sin6_port = htons(port); + if (::bind(listen_fd_, reinterpret_cast(&address), sizeof(address)) == 0 && + ::listen(listen_fd_, 4) == 0) { + accept_thread_ = std::jthread([this] { accept_loop(); }); + return true; + } + (void)::close(listen_fd_); + listen_fd_ = kInvalidSocket; + } + + listen_fd_ = ::socket(AF_INET, SOCK_STREAM, 0); + if (listen_fd_ < 0) { + return false; + } + int reuse = 1; + (void)::setsockopt(listen_fd_, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)); + + sockaddr_in address{}; + address.sin_family = AF_INET; + address.sin_addr.s_addr = htonl(INADDR_ANY); + address.sin_port = htons(port); + if (::bind(listen_fd_, reinterpret_cast(&address), sizeof(address)) < 0 || + ::listen(listen_fd_, 4) < 0) { + (void)::close(listen_fd_); + listen_fd_ = kInvalidSocket; + return false; + } + + accept_thread_ = std::jthread([this] { accept_loop(); }); + return true; + } + + bool connect(const Endpoint& /*server*/) override { + return false; // servers accept connections rather than open them + } + + void send(const SignalingMessage& message) override { + std::lock_guard lock(peer_mutex_); + if (peer_fd_ != kInvalidSocket) { + (void)write_full(peer_fd_, serialize_message(message)); + } + } + + void on_message(MessageCallback callback) override { + ChannelBase::on_message(std::move(callback)); + } + + void disconnect() override { + { + std::lock_guard lock(listen_mutex_); + if (listen_fd_ != kInvalidSocket) { + (void)::shutdown(listen_fd_, SHUT_RDWR); + (void)::close(listen_fd_); + listen_fd_ = kInvalidSocket; + } + } + accept_thread_ = std::jthread{}; // unblocks accept and joins + + // Close the peer connection before joining the reader, otherwise a + // reader blocked in recv() would hang the join forever. + { + std::lock_guard lock(peer_mutex_); + if (peer_fd_ != kInvalidSocket) { + (void)::shutdown(peer_fd_, SHUT_RDWR); + (void)::close(peer_fd_); + peer_fd_ = kInvalidSocket; + } + } + std::lock_guard lock(reader_mutex_); + reader_ = std::jthread{}; + } + + private: + void accept_loop() { + while (true) { + const int peer_fd = ::accept(listen_fd_, nullptr, nullptr); + if (peer_fd < 0) { + return; // listening socket closed (stop) or fatal error + } + + { + std::lock_guard lock(peer_mutex_); + if (peer_fd_ != kInvalidSocket) { + (void)::close(peer_fd_); // replace the previous sender + } + peer_fd_ = peer_fd; + } + // Joined outside peer_mutex_: the old reader may be running a + // callback that sends an answer, which needs peer_mutex_. + std::lock_guard lock(reader_mutex_); + reader_ = std::jthread([this] { read_loop(); }); + } + } + + void read_loop() { + int peer_fd = kInvalidSocket; + { + std::lock_guard lock(peer_mutex_); + peer_fd = peer_fd_; + } + LineAssembler assembler; + std::array chunk{}; + while (true) { + const ssize_t received = ::recv(peer_fd, chunk.data(), chunk.size(), 0); + if (received <= 0) { + return; // closed, replaced, or disconnected + } + dispatch(assembler.feed(chunk.data(), static_cast(received))); + } + } + + std::mutex listen_mutex_; + int listen_fd_ = kInvalidSocket; + std::jthread accept_thread_; + std::mutex reader_mutex_; + std::jthread reader_; + std::mutex peer_mutex_; + int peer_fd_ = kInvalidSocket; +}; + +} // namespace + +NetworkResult> SignalingFactory::create_client() { + return std::unique_ptr(std::make_unique()); +} + +NetworkResult> SignalingFactory::create_server(uint16_t port) { + auto server = std::make_unique(); + if (!server->start(port)) { + return NetworkError{"failed to listen on the signaling port"}; + } + return std::unique_ptr(std::move(server)); +} + +} // namespace sc diff --git a/src/network/udp_transport.cpp b/src/network/udp_transport.cpp index 3a8accb..b1444cd 100644 --- a/src/network/udp_transport.cpp +++ b/src/network/udp_transport.cpp @@ -23,26 +23,39 @@ constexpr int kInvalidSocket = -1; // packet the receiver will ever see. constexpr std::size_t kReceiveBufferSize = 65536; -std::optional resolve_ipv4(const Endpoint& endpoint) { +struct ResolvedAddress { + sockaddr_storage address{}; + socklen_t length = 0; + int family = AF_UNSPEC; +}; + +std::optional resolve_endpoint(const Endpoint& endpoint) { addrinfo hints{}; - hints.ai_family = AF_INET; + hints.ai_family = AF_UNSPEC; hints.ai_socktype = SOCK_DGRAM; addrinfo* result = nullptr; const std::string port = std::to_string(endpoint.port); const char* node = endpoint.address.empty() ? nullptr : endpoint.address.c_str(); - if (getaddrinfo(node, port.c_str(), &hints, &result) != 0) { + if (getaddrinfo(node, port.c_str(), &hints, &result) != 0 || result == nullptr) { return std::nullopt; } - std::optional address; - if (result != nullptr && result->ai_family == AF_INET && result->ai_addrlen >= sizeof(sockaddr_in)) { - sockaddr_in resolved{}; - std::memcpy(&resolved, result->ai_addr, sizeof(sockaddr_in)); - address = resolved; + ResolvedAddress resolved; + if (result->ai_addrlen <= sizeof(sockaddr_storage)) { + resolved.family = result->ai_family; + resolved.length = result->ai_addrlen; + std::memcpy(&resolved.address, result->ai_addr, result->ai_addrlen); } freeaddrinfo(result); - return address; + if (resolved.family == AF_UNSPEC) { + return std::nullopt; + } + return resolved; +} + +bool is_any_address(const std::string& address) { + return address.empty() || address == "0.0.0.0" || address == "::"; } } // namespace @@ -62,36 +75,70 @@ class UdpRtpTransport final : public RtpTransport { if (running_.load()) { return false; } - socket_ = ::socket(AF_INET, SOCK_DGRAM, 0); + // A zero port skips binding: the OS picks the source port on send + // and the receive thread is unnecessary. + if (local_endpoint.port != 0) { + if (!bind_receive_socket(local_endpoint)) { + return false; + } + + // Absorb packet bursts: a keyframe arrives back-to-back and UDP + // has no flow control. The kernel clamps this to + // net.core.rmem_max, so very high bitrates may need a raised + // sysctl on the receiver. + int receive_buffer_bytes = 4 * 1024 * 1024; + (void)::setsockopt(socket_, SOL_SOCKET, SO_RCVBUF, &receive_buffer_bytes, sizeof(receive_buffer_bytes)); + + receive_thread_ = + std::jthread([this, callback = std::move(on_receive)]() mutable { receive_loop(callback); }); + } + running_.store(true); + return true; + } + + bool bind_receive_socket(const Endpoint& local_endpoint) { + if (is_any_address(local_endpoint.address)) { + // Dual-stack: an IPv6 socket with V6ONLY disabled receives from + // both families; fall back to IPv4 where IPv6 is unavailable. + socket_ = ::socket(AF_INET6, SOCK_DGRAM, 0); + if (socket_ >= 0) { + int v6only = 0; + (void)::setsockopt(socket_, IPPROTO_IPV6, IPV6_V6ONLY, &v6only, sizeof(v6only)); + int reuse = 1; + (void)::setsockopt(socket_, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)); + + sockaddr_in6 address{}; + address.sin6_family = AF_INET6; + address.sin6_addr = in6addr_any; + address.sin6_port = htons(local_endpoint.port); + if (::bind(socket_, reinterpret_cast(&address), sizeof(address)) == 0) { + bound_ = true; + socket_family_ = AF_INET6; + return true; + } + (void)::close(socket_); + socket_ = kInvalidSocket; + } + } + + const std::optional resolved = resolve_endpoint(local_endpoint); + if (!resolved.has_value()) { + return false; + } + socket_ = ::socket(resolved->family, SOCK_DGRAM, 0); if (socket_ < 0) { + socket_ = kInvalidSocket; return false; } int reuse = 1; (void)::setsockopt(socket_, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)); - - // Absorb packet bursts: a keyframe arrives back-to-back and UDP has - // no flow control. The kernel clamps this to net.core.rmem_max, so - // very high bitrates may need a raised sysctl on the receiver. - int receive_buffer_bytes = 4 * 1024 * 1024; - (void)::setsockopt(socket_, SOL_SOCKET, SO_RCVBUF, &receive_buffer_bytes, sizeof(receive_buffer_bytes)); - - // A zero port skips binding: the OS picks the source port on send. - if (local_endpoint.port != 0) { - const std::optional address = resolve_ipv4(local_endpoint); - if (!address.has_value()) { - (void)::close(socket_); - socket_ = kInvalidSocket; - return false; - } - if (::bind(socket_, reinterpret_cast(&*address), sizeof(*address)) < 0) { - (void)::close(socket_); - socket_ = kInvalidSocket; - return false; - } + if (::bind(socket_, reinterpret_cast(&resolved->address), resolved->length) < 0) { + (void)::close(socket_); + socket_ = kInvalidSocket; + return false; } - - running_.store(true); - receive_thread_ = std::jthread([this, callback = std::move(on_receive)]() mutable { receive_loop(callback); }); + bound_ = true; + socket_family_ = resolved->family; return true; } @@ -103,19 +150,22 @@ class UdpRtpTransport final : public RtpTransport { if (bytes.empty()) { return false; } - sockaddr_in peer{}; + ResolvedAddress peer{}; { // Snapshot the peer so set_peer() can be called concurrently. std::lock_guard lock(peer_mutex_); peer = peer_; } - const ssize_t sent = - ::sendto(socket_, bytes.data(), bytes.size(), 0, reinterpret_cast(&peer), sizeof(peer)); + if (!ensure_send_socket(peer.family)) { + return false; + } + const ssize_t sent = ::sendto( + socket_, bytes.data(), bytes.size(), 0, reinterpret_cast(&peer.address), peer.length); return sent == static_cast(bytes.size()); } void set_peer(const Endpoint& peer) override { - const std::optional address = resolve_ipv4(peer); + const std::optional address = resolve_endpoint(peer); if (!address.has_value()) { return; } @@ -139,6 +189,28 @@ class UdpRtpTransport final : public RtpTransport { } private: + // The send path is unbound, so the socket family simply follows the + // peer. A bound receiver keeps its family (dual-stack covers both). + bool ensure_send_socket(int family) { + if (socket_ >= 0 && socket_family_ == family) { + return true; + } + if (bound_) { + return socket_family_ == family; + } + if (socket_ >= 0) { + (void)::close(socket_); + socket_ = kInvalidSocket; + } + socket_ = ::socket(family, SOCK_DGRAM, 0); + if (socket_ < 0) { + socket_ = kInvalidSocket; + return false; + } + socket_family_ = family; + return true; + } + void receive_loop(ReceiveCallback& callback) { std::array buffer{}; while (running_.load()) { @@ -155,10 +227,12 @@ class UdpRtpTransport final : public RtpTransport { } int socket_ = kInvalidSocket; + int socket_family_ = AF_UNSPEC; + bool bound_ = false; std::atomic running_{false}; std::atomic has_peer_{false}; std::mutex peer_mutex_; - sockaddr_in peer_{}; + ResolvedAddress peer_; std::jthread receive_thread_; }; diff --git a/tests/meson.build b/tests/meson.build index 7f427cc..4b22cb1 100644 --- a/tests/meson.build +++ b/tests/meson.build @@ -21,3 +21,9 @@ test_loopback = executable('test_loopback', dependencies : [sc_codec_dep, sc_network_dep]) test('udp loopback', test_loopback) + +test_signaling = executable('test_signaling', + 'network/test_signaling.cpp', + dependencies : sc_network_dep) + +test('session signaling', test_signaling) diff --git a/tests/network/test_signaling.cpp b/tests/network/test_signaling.cpp new file mode 100644 index 0000000..dbb4757 --- /dev/null +++ b/tests/network/test_signaling.cpp @@ -0,0 +1,101 @@ +// Phase 6 signaling test: a client offers a session to a server over a real +// localhost TCP connection and receives the server's answer. + +#include "screencast/network/signaling.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace { + +[[noreturn]] void fail(const char* what) { + std::fprintf(stderr, "test_signaling: FAIL: %s\n", what); + std::abort(); +} + +void check(bool condition, const char* what) { + if (!condition) { + fail(what); + } +} + +void check_result(const char* what, const sc::NetworkResult>& result) { + if (sc::is_network_error(result)) { + std::fprintf(stderr, "test_signaling: FAIL: %s: %s\n", what, sc::network_error(result).message.c_str()); + std::abort(); + } +} + +} // namespace + +int main() { + // Bind the server to the first free port in a small range. + std::unique_ptr server; + uint16_t server_port = 0; + for (uint16_t port = 45940; port < 45960; ++port) { + auto server_result = sc::SignalingFactory::create_server(port); + if (!sc::is_network_error(server_result)) { + server = std::move(sc::network_value(server_result)); + server_port = port; + break; + } + } + check(server != nullptr, "server listens"); + check(server_port != 0, "server port"); + + constexpr uint16_t advertised_rtp_port = 45999; + + // The receiver side of the handshake: reply to offers with an answer. + server->on_message([&](const sc::SignalingMessage& message) { + const sc::SessionOffer* offer = std::get_if(&message); + if (offer == nullptr) { + return; + } + sc::SessionAnswer answer; + answer.session_id = offer->session_id; + answer.rtp_endpoint = sc::Endpoint{"", advertised_rtp_port}; + server->send(answer); + }); + + // The sender side: connect, offer, and wait for the answer. + auto client_result = sc::SignalingFactory::create_client(); + check_result("client create", client_result); + std::unique_ptr client = std::move(sc::network_value(client_result)); + + check(client->connect(sc::Endpoint{"127.0.0.1", server_port}), "client connect"); + + std::promise answer_promise; + auto answer_future = answer_promise.get_future(); + std::atomic answered{false}; + client->on_message([&](const sc::SignalingMessage& message) { + if (const sc::SessionAnswer* answer = std::get_if(&message)) { + if (!answered.exchange(true)) { + answer_promise.set_value(*answer); + } + } + }); + + sc::SessionOffer offer; + offer.session_id = "test-session-0001"; + offer.codec_name = "h264"; + offer.frame_rate_num = 25; + offer.frame_rate_den = 1; + client->send(offer); + + check(answer_future.wait_for(std::chrono::seconds(5)) == std::future_status::ready, "answer received"); + const sc::SessionAnswer answer = answer_future.get(); + check(answer.session_id == "test-session-0001", "session id echoes"); + check(answer.rtp_endpoint.port == advertised_rtp_port, "rtp port in answer"); + + client->disconnect(); + server->disconnect(); + + std::puts("test_signaling: all checks passed"); + return 0; +}