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.size)) if family == AF_INET6 { var no: Int32 = 0 setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, &no, socklen_t(MemoryLayout.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.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.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.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.size)) if family == AF_INET6 { var no: Int32 = 0 setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, &no, socklen_t(MemoryLayout.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.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.size)) } } } if bound != 0 { close(fd); continue } setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &receiveBufferSize, socklen_t(MemoryLayout.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.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 } } }