screen_cast/src/app/sender_session.cpp

#include "sender_session.h"

#include "screencast/network/signaling.h"

#include <atomic>
#include <charconv>
#include <chrono>
#include <format>
#include <future>
#include <memory>
#include <string_view>
#include <utility>

namespace sc {

namespace {

bool is_private_ipv4(std::string_view host) {
    if (host.rfind("192.168.", 0) == 0 || host.rfind("10.", 0) == 0) {
        return true;
    }
    if (host.rfind("172.", 0) == 0) {
        const std::size_t second = host.find('.', 5);
        if (second != std::string_view::npos) {
            int octet = 0;
            const auto [pointer, error] = std::from_chars(host.data() + 5, host.data() + second, octet);
            if (error == std::errc{}) {
                return octet >= 16 && octet <= 31;
            }
        }
    }
    return false;
}

} // namespace

int SenderSession::address_preference(std::string_view host) {
    if (host.find(':') == std::string_view::npos) {
        return is_private_ipv4(host) ? 0 : 1;
    }
    if (host.rfind("fd", 0) == 0 || host.rfind("fc", 0) == 0) {
        return 2;
    }
    if (host.rfind("2002:", 0) == 0) {
        return 4;
    }
    if (host.rfind("fe80:", 0) == 0) {
        return 5;
    }
    return 3;
}

std::variant<SenderSession, std::string>
SenderSession::start(const Endpoint& signaling_endpoint, int bitrate_kbps, CaptureTarget target) {
    auto channel_result = SignalingFactory::create_client();
    if (is_network_error(channel_result)) {
        return network_error(channel_result).message;
    }
    auto channel = std::move(network_value(channel_result));
    if (!channel->connect(signaling_endpoint)) {
        return std::format(
            "failed to connect to the receiver at {}:{}", signaling_endpoint.address, signaling_endpoint.port);
    }

    // Offer + answer.
    static std::atomic<std::uint32_t> session_counter{0};
    const std::string session_id = std::format(
        "sc-{:x}-{:x}",
        static_cast<std::uint32_t>(std::chrono::steady_clock::now().time_since_epoch().count()) & 0xffffffffU,
        session_counter.fetch_add(1));

    std::promise<SessionAnswer> answer_promise;
    auto answer_future = answer_promise.get_future();
    std::atomic<bool> answered{false};
    channel->on_message([&](const SignalingMessage& message) {
        if (const SessionAnswer* answer = std::get_if<SessionAnswer>(&message)) {
            if (!answered.exchange(true)) {
                answer_promise.set_value(*answer);
            }
        }
    });

    SessionOffer offer;
    offer.session_id = 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) {
        channel->disconnect();
        return std::string{"the receiver did not answer the session offer"};
    }
    const SessionAnswer answer = answer_future.get();
    if (answer.session_id != session_id) {
        channel->disconnect();
        return std::string{"session mismatch in the receiver's answer"};
    }
    if (answer.rtp_endpoint.port == 0) {
        channel->disconnect();
        return std::string{"the receiver did not provide an RTP port"};
    }

    // Stream to the negotiated endpoint; an empty address means "the
    // address you reached me on".
    const Endpoint rtp_endpoint = answer.rtp_endpoint.address.empty()
                                      ? Endpoint{signaling_endpoint.address, answer.rtp_endpoint.port}
                                      : answer.rtp_endpoint;

    SenderPipelineConfig config;
    config.capture_target = target;
    config.peer_rtp_endpoint = rtp_endpoint;
    config.signaling_server = signaling_endpoint;
    config.encoder.bitrate_kbps = bitrate_kbps;
    config.session_id = session_id;

    auto pipeline = std::make_unique<SenderPipeline>(std::move(config));
    if (!pipeline->start()) {
        channel->disconnect();
        return std::string{"failed to start the sender pipeline"};
    }

    // PLI feedback over the still-open channel.
    channel->on_message([pipeline = pipeline.get(), session_id](const SignalingMessage& message) {
        const SessionPli* pli = std::get_if<SessionPli>(&message);
        if (pli != nullptr && pli->session_id == session_id) {
            pipeline->request_keyframe();
        }
    });

    SenderSession session;
    session.channel_ = std::move(channel);
    session.pipeline_ = std::move(pipeline);
    session.session_id_ = session_id;
    session.receiver_ = std::format("{}:{}", rtp_endpoint.address, rtp_endpoint.port);
    return session;
}

SenderSession::SenderSession(SenderSession&& other) noexcept
    : channel_(std::move(other.channel_)),
      pipeline_(std::move(other.pipeline_)),
      session_id_(std::move(other.session_id_)),
      receiver_(std::move(other.receiver_)),
      stopped_(other.stopped_) {}

SenderSession& SenderSession::operator=(SenderSession&& other) noexcept {
    stop();
    channel_ = std::move(other.channel_);
    pipeline_ = std::move(other.pipeline_);
    session_id_ = std::move(other.session_id_);
    receiver_ = std::move(other.receiver_);
    stopped_ = other.stopped_;
    return *this;
}

SenderSession::~SenderSession() {
    stop();
}

void SenderSession::stop() {
    if (stopped_) {
        return;
    }
    stopped_ = true;
    // The channel first: disconnecting joins its reader threads, so no PLI
    // callback can race the pipeline teardown it points at.
    if (channel_ != nullptr) {
        channel_->disconnect();
        channel_ = nullptr;
    }
    if (pipeline_ != nullptr) {
        pipeline_->stop();
        pipeline_ = nullptr;
    }
}

} // namespace sc