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.
This commit is contained in:
2026-09-07 12:16:34 +02:00
parent dcbcde6410
commit 7be8d59d07
17 changed files with 1395 additions and 66 deletions
+31 -1
View File
@@ -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<Command> 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<Command> 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<Command> 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<Command> 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;
}
+175 -6
View File
@@ -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 <algorithm>
#include <atomic>
#include <charconv>
#include <chrono>
#include <csignal>
#include <format>
#include <future>
#include <iostream>
#include <memory>
#include <mutex>
#include <random>
#include <string>
#include <string_view>
#include <thread>
#include <vector>
namespace {
constexpr std::uint16_t kDefaultSignalingPort = 5005;
constexpr int kSenderDiscoveryTimeoutSeconds = 3;
std::atomic<bool> 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<uint16_t>(port) : default_port;
endpoint.port = error == std::errc{} ? static_cast<std::uint16_t>(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<std::size_t>(engine()) & 0xF];
}
return id;
}
std::vector<sc::DiscoveredPeer> discover_peers(int timeout_seconds, std::string& error) {
std::vector<sc::DiscoveredPeer> 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<sc::DiscoveredPeer> 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<sc::SessionAnswer> answer_promise;
auto answer_future = answer_promise.get_future();
std::atomic<bool> answered{false};
channel->on_message([&](const sc::SignalingMessage& message) {
if (const sc::SessionAnswer* answer = std::get_if<sc::SessionAnswer>(&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<uint16_t>(command.local_rtp_port)};
config.local_rtp_endpoint = sc::Endpoint{"0.0.0.0", static_cast<std::uint16_t>(command.local_rtp_port)};
config.signaling_port = static_cast<std::uint16_t>(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<sc::DiscoveredPeer> 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<sc::SendCommand>(&*command)) {
return run_sender(*send);
}
return run_receiver(std::get<sc::ReceiveCommand>(*command));
if (const sc::ReceiveCommand* receive = std::get_if<sc::ReceiveCommand>(&*command)) {
return run_receiver(*receive);
}
return run_discover(std::get<sc::DiscoverCommand>(*command));
}
+72
View File
@@ -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 <unistd.h>
#include <array>
#include <chrono>
#include <condition_variable>
#include <deque>
@@ -11,6 +16,17 @@
#include <thread>
namespace sc {
namespace {
std::string receiver_service_name() {
std::array<char, 256> 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<SessionOffer>(&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> decoder_;
H264Depacketizer depacketizer_;
std::unique_ptr<RtpTransport> transport_ = RtpTransportFactory::create();
std::unique_ptr<SignalingChannel> signaling_;
std::unique_ptr<DiscoveryService> discovery_;
std::jthread render_thread_;
std::mutex queue_mutex_;
std::deque<DecodedFrame> queue_;
+330
View File
@@ -0,0 +1,330 @@
#include "screencast/network/discovery.h"
#include <avahi-client/client.h>
#include <avahi-client/lookup.h>
#include <avahi-client/publish.h>
#include <avahi-common/address.h>
#include <avahi-common/alternative.h>
#include <avahi-common/error.h>
#include <avahi-common/malloc.h>
#include <avahi-common/thread-watch.h>
#include <mutex>
#include <string>
#include <utility>
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<AvahiClientFlags>(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<AvahiLookupFlags>(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<AvahiDiscovery*>(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<AvahiPublishFlags>(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<AvahiPublishFlags>(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<AvahiDiscovery*>(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<AvahiDiscovery*>(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<AvahiLookupFlags>(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<AvahiDiscovery*>(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<std::unique_ptr<DiscoveryService>> DiscoveryFactory::create_avahi() {
auto discovery = std::make_unique<AvahiDiscovery>();
if (!discovery->start()) {
return NetworkError{discovery->last_error()};
}
return std::unique_ptr<DiscoveryService>(std::move(discovery));
}
} // namespace sc
+11 -3
View File
@@ -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)
include_directories : sc_core_inc,
dependencies : [dep_avahi, dep_json])
+433
View File
@@ -0,0 +1,433 @@
#include "screencast/network/signaling.h"
#include <nlohmann/json.hpp>
#include <arpa/inet.h>
#include <netdb.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <unistd.h>
#include <array>
#include <cstring>
#include <mutex>
#include <optional>
#include <string>
#include <thread>
#include <vector>
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<ResolvedAddress> resolve_endpoint(const Endpoint& endpoint) {
std::vector<ResolvedAddress> 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<std::size_t>(sent);
}
return true;
}
// Assembles newline-terminated lines from received byte chunks.
class LineAssembler {
public:
// Returns the lines completed by this chunk.
std::vector<std::string> feed(const char* data, std::size_t size) {
std::vector<std::string> 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<SessionOffer>(&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<SessionAnswer>(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<SignalingMessage> 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<std::string>();
Endpoint rtp_endpoint;
if (json.contains("rtp_address") && json.at("rtp_address").is_string()) {
rtp_endpoint.address = json.at("rtp_address").get<std::string>();
}
if (json.contains("rtp_port") && json.at("rtp_port").is_number_integer()) {
rtp_endpoint.port = json.at("rtp_port").get<uint16_t>();
}
std::string session_id;
if (json.contains("session_id") && json.at("session_id").is_string()) {
session_id = json.at("session_id").get<std::string>();
}
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<std::string>();
}
if (json.contains("width") && json.at("width").is_number_integer()) {
offer.width = json.at("width").get<int>();
}
if (json.contains("height") && json.at("height").is_number_integer()) {
offer.height = json.at("height").get<int>();
}
if (json.contains("frame_rate_num") && json.at("frame_rate_num").is_number_integer()) {
offer.frame_rate_num = json.at("frame_rate_num").get<int>();
}
if (json.contains("frame_rate_den") && json.at("frame_rate_den").is_number_integer()) {
offer.frame_rate_den = json.at("frame_rate_den").get<int>();
}
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<std::string>& lines) {
for (const std::string& line : lines) {
const std::optional<SignalingMessage> 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<const sockaddr*>(&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<char, 4096> 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<std::size_t>(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<const sockaddr*>(&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<const sockaddr*>(&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<char, 4096> 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<std::size_t>(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<std::unique_ptr<SignalingChannel>> SignalingFactory::create_client() {
return std::unique_ptr<SignalingChannel>(std::make_unique<TcpSignalingClient>());
}
NetworkResult<std::unique_ptr<SignalingChannel>> SignalingFactory::create_server(uint16_t port) {
auto server = std::make_unique<TcpSignalingServer>();
if (!server->start(port)) {
return NetworkError{"failed to listen on the signaling port"};
}
return std::unique_ptr<SignalingChannel>(std::move(server));
}
} // namespace sc
+112 -38
View File
@@ -23,26 +23,39 @@ constexpr int kInvalidSocket = -1;
// packet the receiver will ever see.
constexpr std::size_t kReceiveBufferSize = 65536;
std::optional<sockaddr_in> resolve_ipv4(const Endpoint& endpoint) {
struct ResolvedAddress {
sockaddr_storage address{};
socklen_t length = 0;
int family = AF_UNSPEC;
};
std::optional<ResolvedAddress> 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<sockaddr_in> 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<const sockaddr*>(&address), sizeof(address)) == 0) {
bound_ = true;
socket_family_ = AF_INET6;
return true;
}
(void)::close(socket_);
socket_ = kInvalidSocket;
}
}
const std::optional<ResolvedAddress> 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<sockaddr_in> address = resolve_ipv4(local_endpoint);
if (!address.has_value()) {
(void)::close(socket_);
socket_ = kInvalidSocket;
return false;
}
if (::bind(socket_, reinterpret_cast<const sockaddr*>(&*address), sizeof(*address)) < 0) {
(void)::close(socket_);
socket_ = kInvalidSocket;
return false;
}
if (::bind(socket_, reinterpret_cast<const sockaddr*>(&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<const sockaddr*>(&peer), sizeof(peer));
if (!ensure_send_socket(peer.family)) {
return false;
}
const ssize_t sent = ::sendto(
socket_, bytes.data(), bytes.size(), 0, reinterpret_cast<const sockaddr*>(&peer.address), peer.length);
return sent == static_cast<ssize_t>(bytes.size());
}
void set_peer(const Endpoint& peer) override {
const std::optional<sockaddr_in> address = resolve_ipv4(peer);
const std::optional<ResolvedAddress> 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<std::byte, kReceiveBufferSize> 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<bool> running_{false};
std::atomic<bool> has_peer_{false};
std::mutex peer_mutex_;
sockaddr_in peer_{};
ResolvedAddress peer_;
std::jthread receive_thread_;
};