screen_cast/ios/Receiver/Signaling/SignalingServer.swift

import Foundation
#if canImport(Darwin)
import Darwin
#endif

/// Small socket helpers shared by the signaling server and the UDP transport.
enum SocketUtils {
    /// Creates a bound, listening TCP socket. Prefers a dual-stack IPv6
    /// listener (both families), falls back to IPv4-only. Returns -1 on failure.
    static func makeStreamListener(port: UInt16) -> Int32 {
        for family in [AF_INET6, AF_INET] {
            let fd = socket(family, SOCK_STREAM, 0)
            guard fd >= 0 else { continue }
            var yes: Int32 = 1
            setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &yes, socklen_t(MemoryLayout<Int32>.size))
            if family == AF_INET6 {
                var no: Int32 = 0
                setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, &no, socklen_t(MemoryLayout<Int32>.size))
            }

            let bound: Int32
            if family == AF_INET6 {
                var a = sockaddr_in6()
                a.sin6_family = sa_family_t(AF_INET6)
                a.sin6_port = port.bigEndian
                a.sin6_addr = in6addr_any
                bound = withUnsafePointer(to: &a) { p in
                    p.withMemoryRebound(to: sockaddr.self, capacity: 1) { bind(fd, $0, socklen_t(MemoryLayout<sockaddr_in6>.size)) }
                }
            } else {
                var a = sockaddr_in()
                a.sin_family = sa_family_t(AF_INET)
                a.sin_port = port.bigEndian
                a.sin_addr = in_addr(s_addr: INADDR_ANY)
                bound = withUnsafePointer(to: &a) { p in
                    p.withMemoryRebound(to: sockaddr.self, capacity: 1) { bind(fd, $0, socklen_t(MemoryLayout<sockaddr_in>.size)) }
                }
            }
            if bound != 0 { close(fd); continue }
            if listen(fd, 16) != 0 { close(fd); continue }
            return fd
        }
        return -1
    }

    /// The local port of a bound socket (the port is at byte offset 2 for both
    /// IPv4 and IPv6 sockets).
    static func boundPort(_ fd: Int32) -> UInt16 {
        var a = sockaddr_storage()
        var len = socklen_t(MemoryLayout<sockaddr_storage>.size)
        guard getsockname(fd, &a, &len) == 0 else { return 0 }
        return withUnsafeBytes(of: &a) { raw in
            raw.load(fromByteOffset: 2, as: UInt16.self).bigEndian
        }
    }

    /// A UDP socket bound to [port] (or 0 for ephemeral) for receiving, with a
    /// generous receive buffer (the C++ sender's VBV bounds bursts, but a larger
    /// buffer absorbs a burst on a lossy link).
    static func makeUdpReceiver(port: UInt16, receiveBufferSize: Int32) -> Int32 {
        for family in [AF_INET6, AF_INET] {
            let fd = socket(family, SOCK_DGRAM, 0)
            guard fd >= 0 else { continue }
            var yes: Int32 = 1
            setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &yes, socklen_t(MemoryLayout<Int32>.size))
            if family == AF_INET6 {
                var no: Int32 = 0
                setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, &no, socklen_t(MemoryLayout<Int32>.size))
            }
            let bound: Int32
            if family == AF_INET6 {
                var a = sockaddr_in6()
                a.sin6_family = sa_family_t(AF_INET6)
                a.sin6_port = port.bigEndian
                a.sin6_addr = in6addr_any
                bound = withUnsafePointer(to: &a) { p in
                    p.withMemoryRebound(to: sockaddr.self, capacity: 1) { bind(fd, $0, socklen_t(MemoryLayout<sockaddr_in6>.size)) }
                }
            } else {
                var a = sockaddr_in()
                a.sin_family = sa_family_t(AF_INET)
                a.sin_port = port.bigEndian
                a.sin_addr = in_addr(s_addr: INADDR_ANY)
                bound = withUnsafePointer(to: &a) { p in
                    p.withMemoryRebound(to: sockaddr.self, capacity: 1) { bind(fd, $0, socklen_t(MemoryLayout<sockaddr_in>.size)) }
                }
            }
            if bound != 0 { close(fd); continue }
            setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &receiveBufferSize, socklen_t(MemoryLayout<Int32>.size))
            return fd
        }
        return -1
    }
}

/// Newline-delimited JSON signaling server (the receiver side). Keeps the most
/// recent connection as its active peer, mirroring the C++ server: `onOffer`
/// may answer synchronously (the sender blocks on the answer).
final class SignalingServer {
    private let onOffer: (SignalingMessage) -> Void
    private let onPli: (SignalingMessage) -> Void
    private var listenFd: Int32 = -1
    private(set) var port: UInt16 = 0
    private var peerFd: Int32 = -1
    private let peerLock = NSLock()
    private let assembler = LineAssembler()
    private var acceptThread: Thread?
    private var readerThread: Thread?
    private var running = false

    init(onOffer: @escaping (SignalingMessage) -> Void,
         onPli: @escaping (SignalingMessage) -> Void) {
        self.onOffer = onOffer
        self.onPli = onPli
    }

    /// Binds the port (SO_REUSEADDR) and starts accepting. Returns true on success.
    func start(_ preferredPort: UInt16) -> Bool {
        guard let fd = SocketUtils.makeStreamListener(port: preferredPort), fd >= 0 else { return false }
        listenFd = fd
        port = SocketUtils.boundPort(fd)
        running = true
        let thread = Thread { [weak self] in self?.acceptLoop() }
        thread.name = "signaling-accept"
        acceptThread = thread
        thread.start()
        return true
    }

    /// Sends to the current peer; never throws. A lost control message is
    /// recoverable — the session re-negotiates or the next keyframe arrives —
    /// but an exception here would kill the RTP reader thread that reaches
    /// send() from requestPli(). Mirrors the C++ server, which ignores write
    /// failures.
    func send(_ message: SignalingMessage) {
        guard let bytes = SignalingMessage.serialize(message).data(using: .utf8) else { return }
        peerLock.lock()
        let fd = peerFd
        peerLock.unlock()
        guard fd >= 0 else { return }
        bytes.withUnsafeBytes { raw in
            _ = send(fd, raw.baseAddress, raw.count, Int32(MSG_NOSIGNAL))
        }
    }

    func close() {
        running = false
        peerLock.lock()
        let peer = peerFd
        peerFd = -1
        peerLock.unlock()
        if peer >= 0 { close(peer) }
        if listenFd >= 0 { close(listenFd) }
        listenFd = -1
    }

    private func acceptLoop() {
        while running {
            var addr = sockaddr()
            var len = socklen_t(MemoryLayout<sockaddr>.size)
            let client = accept(listenFd, &addr, &len)
            if client < 0 {
                if !running { break }
                continue
            }
            // Most-recent-connection-wins: close the previous peer.
            peerLock.lock()
            let old = peerFd
            peerFd = client
            peerLock.unlock()
            if old >= 0 { close(old) }

            let thread = Thread { [weak self] in self?.readLoop(fd: client) }
            thread.name = "signaling-reader"
            readerThread = thread
            thread.start()
        }
    }

    private func readLoop(fd: Int32) {
        assembler.reset()
        var buffer = [UInt8](repeating: 0, count: 4096)
        while running {
            let read = buffer.withUnsafeMutableBytes { raw in
                recv(fd, raw.baseAddress, raw.count, 0)
            }
            guard read > 0 else { break }
            for line in assembler.feed(Array(buffer.prefix(read))) {
                dispatch(line)
            }
        }
        // Peer closed; if it is still our active peer, mark it gone.
        peerLock.lock()
        if peerFd == fd {
            peerFd = -1
        }
        peerLock.unlock()
    }

    private func dispatch(_ line: String) {
        guard let message = SignalingMessage.parse(line) else { return }
        // A broken callback must not kill the reader thread.
        switch message {
        case .offer:
            onOffer(message)
        case .pli:
            onPli(message)
        case .answer:
            break // the receiver never receives answers
        }
    }
}