import AVFoundation import CoreVideo import Foundation import LiveKitWebRTC @MainActor final class MediaEngine { weak var call: Call? let factory: RTCPeerConnectionFactory var localVideo: RTCVideoTrack? var localAudio: RTCAudioTrack? /// §5.4: the display track while this app is sharing, and nothing /// otherwise. Every connection carries it on a transceiver of its own. private(set) var localScreen: RTCVideoTrack? private var screenCapture: ScreenCapture? // Bumped by everything that ends a share, so a capture whose picker // was still open when the share or the call ended is stopped when it // finally answers, rather than sent. private var screenAttempt = 0 private var capture: VideoCapture? private var peers: [String: MediaPeer] = [:] /// What the screen is optimised for (ScreenSendPolicy), and the size /// its source was last seen at, which motion mode scales down from. var screenMode = ScreenShareMode.balanced { didSet { applySendPolicy() } } private var screenSize = (width: 0, height: 0) /// The audio device module and processing behind the factory. See CallAudio. let audio: CallAudio /// Builds on the calling thread: only for a call touched before it /// joined. A join uses `make`. convenience init(call: Call) { RTCInitializeSSL() #if os(iOS) self.init(call: call, audio: CallAudio(routes: call.audio, audioDevice: call.customAudioDevice)) #else self.init(call: call, audio: CallAudio(routes: call.audio, audioDevice: nil, input: call.audioInput)) #endif } init(call: Call, audio: CallAudio) { self.call = call self.audio = audio factory = audio.factory } /// The media for a call, with its factory and audio device module built /// on the audio queue (CallAudio.make): 100–200 ms the main thread no /// longer spends when someone joins. static func make(call: Call) async -> MediaEngine { await AudioWork.run { RTCInitializeSSL() } #if os(iOS) let audio = await CallAudio.make(routes: call.audio, audioDevice: call.customAudioDevice) #else let audio = await CallAudio.make(routes: call.audio, audioDevice: nil, input: call.audioInput) #endif return MediaEngine(call: call, audio: audio) } func prepare(capture supplied: VideoCapture?) async throws { close() let source = factory.videoSource() localVideo = factory.videoTrack(with: source, trackId: "camera") localAudio = audio.makeTrack() let capture = supplied ?? CameraCapture.make() self.capture = capture offersCameras = capture.offersCameras capture.observe { [weak self] in self?.call?.cameraChanged() } try await capture.start(source: source) #if os(iOS) // A supplied audio device owns its session configuration (SDK contract). if call?.customAudioDevice == nil { let audio = RTCAudioSession.sharedInstance() audio.lockForConfiguration() defer { audio.unlockForConfiguration() } // Not .allowBluetoothHFP: see AudioRoutes.options. At join nobody // has chosen a Bluetooth microphone, so the call starts wideband // and only drops to the hands-free profile if someone asks for a // Bluetooth input by name. try audio.setCategory( AVAudioSession.Category.playAndRecord, mode: AVAudioSession.Mode.videoChat, options: AudioRoutes.options(bluetoothInput: false)) // WebRTC owns activation and interruption recovery as peers start/stop. } #endif } /// The cameras the running capture can offer, and the one it is on. /// Empty before prepare, and empty for the interop executable's /// synthetic frames, which is how the picker knows not to appear. /// /// With real cameras the list stays live after the capture has gone /// (the call ended under the person), so the panel can still say what /// is connected and a choice there is remembered for next time. var cameras: [CameraDevice] { if let capture { return capture.cameras } return offersCameras ? CameraWatch.shared.cameras : [] } var currentCamera: String? { capture?.currentCamera } /// Whether this call's camera is a real one, whose row the panel always /// shows. Kept past the capture, for the same reason as `cameras`. private(set) var offersCameras = false var cameraStatus: CameraStatus? { capture?.cameraStatus } var cameraNote: String { capture?.cameraNote ?? "" } var cameraToKeep: CameraDevice? { capture?.cameraToKeep } func use(camera id: String) async throws { if let capture { try await capture.use(camera: id) } else if offersCameras, let device = CameraWatch.shared.cameras.first(where: { $0.id == id }) { CameraPreference(device).save() } } func retryCamera() async { CameraWatch.shared.refresh() await capture?.retryCamera() } func closePeers() { for (_, p) in peers { p.close() } peers.removeAll() } /// Starts sending a screen (§5.4). The capture is asked first: a person /// who changes their mind at the system's picker has shared nothing, /// and nothing is added to any connection. /// /// What gives way under pressure — resolution or frame rate — is the /// "Optimise for" mode's, through sender parameters (ScreenSendPolicy). func startScreen(_ capture: ScreenCapture, ended: @escaping @MainActor () -> Void) async throws { guard localScreen == nil else { return } screenAttempt += 1 let attempt = screenAttempt screenSize = (0, 0) screenIsScreencast = ScreenSendPolicy.screencast(screenMode) let source = factory.videoSource(forScreenCast: screenIsScreencast) try await capture.start(source: source, ended: ended) guard attempt == screenAttempt, localScreen == nil else { capture.stop() throw CancellationError() } screenCapture = capture screenSourceInUse = source if let size = capture.frameSize, size.width > 0, size.height > 0 { screenSize = size } localScreen = factory.videoTrack(with: source, trackId: "screen") for peer in peers.values { peer.addScreen() } } /// The source the screen's capture is writing into, for a share again /// from the same capture. private var screenSourceInUse: RTCVideoSource? private var reshares = 0 /// Stop and Share, without the picker (spec §15.22): every connection's /// screen transceiver stopped and renegotiated, a fresh track on the /// same source, the capture started again on the same content, and each /// connection given the new track as soon as it settles. New /// transceivers mean new encoders here and new receivers and decoders at /// every far end, which is what the hand-made stop and share did. func reshareScreen() async { guard let capture = screenCapture, let source = screenSourceInUse, localScreen != nil else { return } reshares += 1 for peer in peers.values { peer.removeScreen() } localScreen = factory.videoTrack(with: source, trackId: "screen-\(reshares)") if !(await capture.restart()) { ScreenLog.info("share again: the capture carries on as it was") } guard screenCapture === capture else { return } // Any connection already stable takes it now; the rest as they settle. for peer in peers.values { peer.addScreen() } } /// Stops sending the screen: every connection's screen transceiver is /// stopped and renegotiated, so the receivers' tracks *end*, and the /// capture goes last. func stopScreen() { screenAttempt += 1 guard let capture = screenCapture else { return } screenCapture = nil screenSourceInUse = nil localScreen = nil for peer in peers.values { peer.removeScreen() peer.screenStopped() } capture.stop() } // ------------------------------------------------------------- // The screen watchdog's hands (ScreenWatchdog, run by Call) // ------------------------------------------------------------- /// What the capture has produced so far; nothing while not sharing. var screenCaptureCounts: ScreenCaptureCounts? { screenCapture?.counts } /// A fresh capture of the same content, into the same source. func restartScreenCapture() async -> Bool { guard let capture = screenCapture else { return false } return await capture.restart() } func forceScreenKeyframe(_ peer: String) { peers[peer]?.forceScreenKeyframe() } func recreateScreenEncoder(_ peer: String) { peers[peer]?.recreateScreenEncoder() } func reAddScreen(_ peer: String) { peers[peer]?.reAddScreen() } /// For the gate: every screen encoding switched off behind the /// watchdog's back, which is what an encoder that has stopped looks /// like from outside it — frames go in, and nothing comes out. func stallScreenEncoderForTest() { for peer in peers.values { peer.setScreenActive(false) } } /// Each peer's screen as it arrives here, for the peers whose screen is /// arriving (the receiver's watchdog, and "Copy call diagnostics"). func remoteScreenReports() async -> [String: ScreenReceiveReport] { var out: [String: ScreenReceiveReport] = [:] for (id, peer) in peers { guard let pc = peer.pc, peer.receivingScreen else { continue } let report = await PeerCalls.statistics(pc) var reading = ScreenReceiveReport() for stat in report.statistics.values where stat.type == "inbound-rtp" { let mid = stat.values["mid"] as? String ?? "" guard peer.isScreen(mid: mid) else { continue } reading.add(stat.values) } reading.mid = peer.screenMidHere out[id] = reading } return out } /// Frames decoded from screens here, never going backwards: a rebuilt /// screen arrives on a new receiver whose own count starts again, and /// the gate's "decoding again" has to see its frames as more. private var screenDecodedSeen: [String: UInt64] = [:] private(set) var screenDecodedTotal: UInt64 = 0 private func tallyScreenDecoded(_ peer: String, _ now: UInt64) { let last = screenDecodedSeen[peer] ?? 0 screenDecodedTotal += now >= last ? now - last : now screenDecodedSeen[peer] = now } /// Puts every connection's screen and camera senders where the policy /// wants them; a no-op for any that already are. func applySendPolicy() { for peer in peers.values { peer.applySendPolicy() } } /// The screen's source size, as its statistics report it. func screenSource(width: Int, height: Int) { guard width > 0, height > 0, (width, height) != screenSize else { return } screenSize = (width, height) applySendPolicy() } var screenPlan: EncodingPlan { ScreenSendPolicy.screen( screenMode, width: screenSize.width, height: screenSize.height, screencast: screenIsScreencast) } /// Whether this share's source was made as a screencast: fixed when the /// share starts, whatever the mode is changed to afterwards. private var screenIsScreencast = true func close() { stopScreen() closePeers() capture?.stop() capture = nil // On the audio queue: a track's switch waits out an engine start. let video = localVideo let audio = localAudio AudioWork.post { video?.isEnabled = false audio?.isEnabled = false } localVideo = nil localAudio = nil } func reconcile(state: RoomState, ice: ICEConfig) { let wanted = Set(state.peers.filter { $0.peer != state.peer }.map(\.peer)) for id in Array(peers.keys) where !wanted.contains(id) { peers.removeValue(forKey: id)?.close() call?.removeRemote(id) } for person in state.peers where person.peer != state.peer { let gen = person.gen ?? 1 if let old = peers[person.peer], old.generation == gen { continue } peers.removeValue(forKey: person.peer)?.close() call?.removeRemote(person.peer) let peer = MediaPeer( id: person.peer, generation: gen, polite: state.peer < person.peer, engine: self) peers[person.peer] = peer peer.open(ice: ice) } } func receive(_ message: Message) { if let from = message.from { peers[from]?.receive(message) } } /// §5.4's label. A peer this app has no connection to yet is ignored, /// as the browser ignores it: its screen will be told apart from its /// camera by the m-line when the connection comes, and a caption with /// no connection behind it is a tile nothing will ever fill or clear. func receiveShare(_ message: Message) { guard let from = message.from, let peer = peers[from] else { return } peer.share(on: message.payload?.on == true) } /// ICE restarts this app has started for a renewal, for the gate. var iceRestarts: UInt64 = 0 func updateICE(_ config: ICEConfig) { let now = Int64(Date().timeIntervalSince1970) for p in peers.values { p.updateICE(config, now: now) } } /// The five audio-health fields, gathered alongside the byte counts. /// /// Bytes say audio arrived; they say nothing about what it sounded like. /// These five separate the two ways received audio goes wrong, which is /// the distinction a person reporting "it growled" cannot make for us: /// /// - `concealedSamples` against `totalSamplesReceived` is loss. The /// decoder invented that fraction of the audio because nothing /// arrived to decode. /// - `insertedSamplesForDeceleration` and /// `removedSamplesForAcceleration` are clock drift. The jitter /// buffer is stretching or compressing the signal to keep up with a /// sender whose clock runs at a different rate than this device's — /// and stretching a voice is what a growl *is*. /// /// With the device pickers beside it, a suspect device can be ruled out /// by switching away from it while watching these. static let audioHealthFields = [ "concealedSamples", "totalSamplesReceived", "insertedSamplesForDeceleration", "removedSamplesForAcceleration", ] func stats() async -> [String: UInt64] { var values: [String: UInt64] = [ "framesDecoded": 0, "audioBytesReceived": 0, "videoBytesSent": 0, "audioBytesSent": 0, "screenFramesDecoded": 0, "screenBytesSent": 0, "peerConnections": UInt64(peers.count), "concealedSamples": 0, "totalSamplesReceived": 0, "insertedSamplesForDeceleration": 0, "removedSamplesForAcceleration": 0, "jitterBufferDelayMs": 0, ] for (id, peer) in peers { guard let pc = peer.pc else { continue } let ownScreen = await peer.ownScreenMid() let report = await PeerCalls.statistics(pc) var screenDecoded: UInt64 = 0 defer { tallyScreenDecoded(id, screenDecoded) } for stat in report.statistics.values { let kind = (stat.values["kind"] as? String) ?? (stat.values["mediaType"] as? String) ?? "" let mid = stat.values["mid"] as? String ?? "" if stat.type == "inbound-rtp" { if kind == "video" { // A peer sharing its screen sends two video streams, and a // count that added them together would let a screen stand in // for a camera that had stopped. Told apart by mid, which is // what the browser's own gate does. let field = peer.isScreen(mid: mid) ? "screenFramesDecoded" : "framesDecoded" let decoded = (stat.values["framesDecoded"] as? NSNumber)?.uint64Value ?? 0 values[field, default: 0] += decoded if peer.isScreen(mid: mid) { screenDecoded += decoded } } if kind == "audio" { values["audioBytesReceived", default: 0] += (stat.values["bytesReceived"] as? NSNumber)?.uint64Value ?? 0 for field in MediaEngine.audioHealthFields { values[field, default: 0] += (stat.values[field] as? NSNumber)?.uint64Value ?? 0 } // jitterBufferDelay is seconds as a double, and is cumulative // against jitterBufferEmittedCount rather than a gauge. It is // carried in milliseconds because everything else here is an // integer, and a sub-millisecond buffer is not a fact anyone // reading this is going to act on. let delay = (stat.values["jitterBufferDelay"] as? NSNumber)?.doubleValue ?? 0 values["jitterBufferDelayMs", default: 0] += UInt64((delay * 1000).rounded()) } } if stat.type == "outbound-rtp" { if kind == "video" { let field = !mid.isEmpty && mid == ownScreen ? "screenBytesSent" : "videoBytesSent" values[field, default: 0] += (stat.values["bytesSent"] as? NSNumber)?.uint64Value ?? 0 } if kind == "audio" { values["audioBytesSent", default: 0] += (stat.values["bytesSent"] as? NSNumber)?.uint64Value ?? 0 } } } } values["screenFramesDecodedTotal"] = screenDecodedTotal return values } /// Each remote person's sound, for the meters (spec §15.20): the level /// their receive tap has seen since the last look, whether the /// connection is up, and — when `stats` is set, or for the people in /// `detail` — the audio bytes and packets received so far and whether /// the receiver is live, one `statistics(for:)` per audio receiver rather /// than the whole report. For the people in `acks`, whether their /// receiver reports say this app's audio reaches them (spec §15.24's /// fallback for a client that does not say who it hears). func peerAudio(stats: Bool, detail: Set = [], acks: Set = []) async -> [String: PeerAudioReading] { var out: [String: PeerAudioReading] = [:] for (id, peer) in peers { guard let pc = peer.pc else { continue } let (peak, buffers) = peer.tap.take() let deep = stats || detail.contains(id) let ack = acks.contains(id) // Read on the audio queue, as every call into the connection is // (PeerCalls): five times a second, and each one a blocking hop to // WebRTC's signalling thread. let (state, receivers, senders) = await PeerCalls.read(pc) { pc in ( pc.iceConnectionState, deep ? pc.receivers.filter { $0.track?.kind == "audio" } : [], ack ? pc.senders.filter { $0.track?.kind == "audio" } : [] ) } var reading = PeerAudioReading( peak: peak, buffers: buffers, connected: state == .connected || state == .completed, connectionBad: state == .disconnected || state == .failed) reading.gone = state == .failed || state == .closed if deep { var bytes: UInt64 = 0 var packets: UInt64 = 0 var live = false for receiver in receivers { if receiver.track?.readyState == .live { live = true } let report = await PeerCalls.statistics(pc, for: receiver) for stat in report.statistics.values where stat.type == "inbound-rtp" { bytes += (stat.values["bytesReceived"] as? NSNumber)?.uint64Value ?? 0 packets += (stat.values["packetsReceived"] as? NSNumber)?.uint64Value ?? 0 } } reading.bytes = bytes reading.packets = packets reading.hasReceiver = !receivers.isEmpty reading.receiverLive = live } if ack { var sent: UInt64 = 0 var acked = false for sender in senders { let report = await PeerCalls.statistics(pc, for: sender) for stat in report.statistics.values { if stat.type == "outbound-rtp" { sent += (stat.values["packetsSent"] as? NSNumber)?.uint64Value ?? 0 } else if stat.type == "remote-inbound-rtp" { // Exists only once a receiver report has come back. let received = (stat.values["packetsReceived"] as? NSNumber)?.uint64Value ?? 0 let measured = (stat.values["roundTripTimeMeasurements"] as? NSNumber)?.uint64Value ?? 0 if received > 0 || measured > 0 || stat.values["roundTripTime"] != nil { acked = true } } } } reading.acked = sent > 0 && acked } out[id] = reading } return out } /// Every connection's statistics at one look, for the quality report /// (spec §15.26): the whole report per connection, with its generation /// and which mids carry screens, read as the other statistics are. func qualitySamples() async -> [String: QualityPeerSample] { var out: [String: QualityPeerSample] = [:] for (id, peer) in peers { guard let pc = peer.pc else { continue } let ownScreen = await peer.ownScreenMid() let report = await PeerCalls.statistics(pc) out[id] = QualityPeerSample( generation: peer.generation, ownScreenMid: ownScreen, screenMid: peer.screenMidHere, stats: report.statistics.mapValues { QualityStat(type: $0.type, values: $0.values) }) } return out } /// Caps every connection's bandwidth estimate, which is what a thin /// uplink does to it. For the gate, which has no thin uplink to test on; /// nil takes the cap off. func limitUplink(_ bps: Int?) { uplinkLimit = bps for peer in peers.values { peer.limitUplink(bps) } } private(set) var uplinkLimit: Int? /// This app's screen on the way out, one report per connection sending /// it (see ScreenSendReport). func screenReports() async -> [String: ScreenSendReport] { var reports: [String: ScreenSendReport] = [:] for (id, peer) in peers { guard let pc = peer.pc, let mid = await peer.ownScreenMid() else { continue } let report = await PeerCalls.statistics(pc) let flat = report.statistics.mapValues { (type: $0.type, values: $0.values as [String: Any]) } if var read = ScreenSendReport.read(flat, mid: mid) { read.timestamp = report.timestamp_us / 1_000_000 reports[id] = read } } return reports } } /// A transceiver direction as SDP spells it, which is how the classifier, /// and the browser it was ported from, compare them. private func sdpDirection(_ direction: RTCRtpTransceiverDirection) -> String { switch direction { case .sendRecv: return "sendrecv" case .sendOnly: return "sendonly" case .recvOnly: return "recvonly" case .inactive: return "inactive" case .stopped: return "stopped" @unknown default: return "" } } /// One look at a remote person's audio (see MediaEngine.peerAudio). struct PeerAudioReading: Equatable, Sendable { var peak: Float var buffers: Int var connected: Bool /// ICE `disconnected` or `failed` (spec §15.23's three seconds). var connectionBad = false /// ICE `failed` or `closed`: whatever was heard on it is gone (§15.24). var gone = false var bytes: UInt64? /// Audio packets received, the receiver there at all, and its track /// live (§15.24); read with `bytes`. var packets: UInt64? var hasReceiver = false var receiverLive = false /// Their receiver reports say this app's audio reaches them; nil when /// not asked. var acked: Bool? } @MainActor private final class MediaPeer: NSObject, RTCPeerConnectionDelegate { let id: String /// Measures every audio track that arrives on this connection (their /// microphone; a browser's shared tab audio would add to it). nonisolated let tap = RemoteAudioTap() private var tapped: [RTCAudioTrack] = [] let generation: Int64 let polite: Bool nonisolated let playAudio: Bool weak var engine: MediaEngine? var pc: RTCPeerConnection? private var work: Task? private var closed = false private var ignoreOffer = false private var candidates: [Candidate] = [] private var lastICE: ICEConfig? /// The list this connection last gathered on — built with, or last /// restarted onto — whose `renewAt` says when it needs a restart. private var gatheredICE: ICEConfig? // §5.4, receiving. What this peer's camera and microphone came in on, // and whether it has said it is sharing; the receiver its screen // arrived on, so that only *that* receiver going away takes the tile // with it; and the mid, so the statistics can count the two video // streams apart. private var clues = ScreenClues() private var screenReceiver: String? private var screenMid = "" // Whether this connection carries a camera of ours: the classifier's // direction rule turns on it. private var sentVideo = false // §5.4, sending: the transceiver this connection carries our screen on, // while we are sharing. And an offer this connection owes the far side // but could not make yet, because it was in the middle of another. private var screenTx: RTCRtpTransceiver? private var owesOffer = false init(id: String, generation: Int64, polite: Bool, engine: MediaEngine) { self.id = id self.generation = generation self.polite = polite self.engine = engine self.playAudio = engine.call?.playAudio ?? true } private func configuration(_ ice: ICEConfig) -> RTCConfiguration { let c = RTCConfiguration() c.sdpSemantics = .unifiedPlan c.iceServers = ice.iceServers.map { RTCIceServer(urlStrings: $0.urls, username: $0.username, credential: $0.credential) } return c } /// Makes the connection on the audio queue, as the first thing this /// peer's work does: everything that arrives for it waits behind it. /// Making one waits on WebRTC's worker thread, which can be busy /// starting the audio engine for someone else (846 ms measured on the /// main thread with a Bluetooth switch simulated). func open(ice: ICEConfig) { guard let engine else { return } lastICE = ice gatheredICE = ice let factory = engine.factory let config = configuration(ice) let audio = engine.localAudio let video = engine.localVideo let delegate: RTCPeerConnectionDelegate = self enqueue { [weak self] in let (made, sent) = await AudioWork.run { () -> (RTCPeerConnection?, Bool) in guard let pc = factory.peerConnection( with: config, constraints: RTCMediaConstraints(mandatoryConstraints: nil, optionalConstraints: nil), delegate: delegate) else { return (nil, false) } if let audio { _ = pc.add(audio, streamIds: ["calls"]) } let sent = video.map { pc.add($0, streamIds: ["calls"]) != nil } ?? false return (pc, sent) } guard let self, !self.closed, let made else { if let made { PeerCalls.close(made) } return } self.pc = made self.sentVideo = sent if let limit = self.engine?.uplinkLimit { self.limitUplink(limit) } // A connection built while this app is already sharing gets the // screen with everything else: someone joining mid-share sees it. // Last, as in the browser, so the camera and microphone take the // m-lines the far side built for them and the screen a place of its // own after them. self.addScreen() } } /// Gives this connection the screen on a `sendonly` transceiver in a /// stream of its own, which is what keeps it clear of the camera at the /// far end. Adding it asks for negotiation, and perfect negotiation /// carries the offer with no help from here. /// /// Only on a stable connection. A transceiver added mid-negotiation is /// unassociated when the far side's offer arrives, and m-line matching /// will hand it that offer's m-line — so two people sharing at once end /// up each sending a screen in the place the other built for something /// else. The browser's mutualshare gate found that; the deferred ones /// are added when the connection settles. /// /// Added on the audio queue, as every call into the connection is; a /// share that ended while it was being added is taken away again. func addScreen() { guard let pc, !closed, screenTx == nil, !addingScreen, let track = engine?.localScreen, let engine else { return } addingScreen = true let factory = engine.factory Task { @MainActor [weak self] in let added = await PeerCalls.read(pc) { pc -> RTCRtpTransceiver? in guard pc.signalingState == .stable else { return nil } let setup = RTCRtpTransceiverInit() setup.direction = .sendOnly setup.streamIds = ["screen"] guard let tx = pc.addTransceiver(with: track, init: setup) else { return nil } try? tx.setCodecPreferences( ScreenSendPolicy.preferringH264(factory.rtpSenderCapabilities(forKind: "video").codecs), error: ()) return tx } guard let self else { return } self.addingScreen = false guard let added else { return } self.screenTx = added self.screenMidCache = nil guard !self.closed, self.pc === pc, self.engine?.localScreen === track else { self.removeScreen() return } self.applySendPolicy() } } private var addingScreen = false /// The screen at its mode's parameters and the camera capped beside /// it, or the camera back at its defaults when there is no screen. /// Sender parameters only: nothing here renegotiates. /// /// Applied on the audio queue: a sender's parameters are read and set on /// WebRTC's worker thread, and this runs each time a negotiation settles /// — just when the worker may be starting the audio engine. func applySendPolicy() { guard let pc, !closed, let engine else { return } let screen = screenTx?.sender var screenPlan = engine.screenPlan if nudge > 1 { screenPlan.scaleResolutionDownBy = (screenPlan.scaleResolutionDownBy ?? 1) * nudge } let cameraTrack = engine.localVideo let camera = ScreenSendPolicy.camera(sharing: screenTx != nil) AudioWork.post { if let screen { screen.apply(screenPlan) } let id = cameraTrack?.trackId if let sender = pc.senders.first(where: { $0.track != nil && $0.track?.trackId == id }) { sender.apply(camera) } } } /// Ends this connection's screen m-line. Stopping the transceiver is /// §5.4's own instruction and the only thing that makes the receiver's /// track end rather than freeze on its last frame. /// /// The framework's stop is the immediate one: it stops the transceiver /// and asks for nothing, where the browser's `stop()` also raises /// negotiation-needed. So the offer that tells the far side is asked /// for here, by hand. func removeScreen() { guard let tx = screenTx else { return } screenTx = nil screenMidCache = nil AudioWork.post { tx.stopInternal() } applySendPolicy() renegotiate() } /// An offer this side has to make for a reason negotiation-needed will /// not announce. Made now if the connection is quiet, and otherwise /// owed, and made the next time it settles. private func renegotiate() { guard let pc, !closed else { return } Task { @MainActor [weak self] in guard await PeerCalls.read(pc, { $0.signalingState }) == .stable else { self?.owesOffer = true return } guard let self else { return } if self.polite { try? await Task.sleep(nanoseconds: 200_000_000) } self.enqueue { [weak self] in guard let self, let pc = self.pc else { return } if await PeerCalls.read(pc, { $0.signalingState }) != .stable { self.owesOffer = true return } try await self.offer() } } } /// The connection is back at stable: the one moment a deferred screen /// can be added, and an owed offer made. private func settled() { guard let pc, !closed else { return } Task { @MainActor [weak self] in // Read on the audio queue, as every call into the connection is. let stable = await PeerCalls.read(pc) { $0.signalingState } == .stable guard let self, stable, !self.closed, self.pc === pc else { return } self.addScreen() self.applySendPolicy() if self.owesOffer { self.owesOffer = false self.renegotiate() } } } func isScreen(mid: String) -> Bool { !mid.isEmpty && mid == screenMid } var screenMidHere: String { screenMid } /// The mid our screen went out on, once negotiated: read on the audio /// queue (a transceiver's mid is a hop to WebRTC's signalling thread), /// and kept once known. func ownScreenMid() async -> String? { if let screenMidCache { return screenMidCache } guard let pc, let tx = screenTx else { return nil } let mid = await PeerCalls.read(pc) { _ in tx.mid } guard !mid.isEmpty, screenTx === tx else { return nil } screenMidCache = mid return mid } private var screenMidCache: String? func limitUplink(_ bps: Int?) { guard let pc else { return } AudioWork.post { _ = pc.setBweMinBitrateBps(nil, currentBitrateBps: nil, maxBitrateBps: bps.map { NSNumber(value: $0) }) } } /// The `share` message: on draws the captioned tile ahead of the media, /// off is one of §5.4's two endings. func share(on: Bool) { if on { clues.sharing = true engine?.call?.announceScreen(id) } else { dropScreen() } } private func dropScreen() { clues.sharing = false screenReceiver = nil screenMid = "" engine?.call?.dropScreen(id) } /// One track arriving, put in its place. A camera goes on the peer's /// tile exactly as it always did; a screen gets a tile of its own and /// leaves the camera alone. private func arrived( _ track: RTCMediaStreamTrack, receiver: String, info: ArrivingTrack ) { let role = classifyTrack(clues, info, hasVideo: sentVideo) if role == .screen, let video = track as? RTCVideoTrack { clues.sharing = true screenReceiver = receiver if !info.mid.isEmpty { screenMid = info.mid } engine?.call?.setScreen(video, peer: id) return } if role == .camera { if !info.mid.isEmpty { clues.cameraMid = info.mid } if !info.streamId.isEmpty { clues.cameraStream = info.streamId } } // Recorded from the first track of any kind: it is what places a // video track for a peer that has no camera. if clues.peerStream.isEmpty { clues.peerStream = info.streamId } if role == .camera, let video = track as? RTCVideoTrack { engine?.call?.setRemote(video, peer: id) } if let audio = track as? RTCAudioTrack, !tapped.contains(where: { $0 === audio }) { audio.add(tap) tapped.append(audio) } } func close() { closed = true for audio in tapped { audio.remove(tap) } tapped = [] screenTx = nil work?.cancel() work = nil // Closing can stop the audio engine (the last connection), which // takes as long as starting it: on the audio queue. if let pc { PeerCalls.close(pc) } pc = nil candidates = [] } private func enqueue(_ action: @escaping @MainActor () async throws -> Void) { let previous = work work = Task { [weak self] in await previous?.value guard let self, !self.closed, !Task.isCancelled else { return } do { try await action() } catch { if !self.closed { self.engine?.call?.mediaFailed() } } } } private func send(_ type: String, _ payload: [String: Any]) { guard !closed, let body = json(payload) else { return } engine?.call?.send(type: type, to: id, gen: generation, payload: body) } // Descriptions and candidates are applied from the audio queue // (PeerCalls). Applying one waits for WebRTC's worker thread, which // starts the audio engine for the first stream in each direction — a // Bluetooth headset's profile switch included — and up to 0.3.8 the // main thread waited with it. private func setLocal(_ desc: RTCSessionDescription) async throws { guard let pc, !closed else { return } try await PeerCalls.set(pc, local: true, desc) } private func setRemote(_ desc: RTCSessionDescription) async throws { guard let pc, !closed else { return } try await PeerCalls.set(pc, local: false, desc) } private func makeDescription(answer: Bool, restart: Bool = false) async throws -> RTCSessionDescription { guard let pc else { throw CancellationError() } let constraints = RTCMediaConstraints( mandatoryConstraints: restart ? ["IceRestart": "true"] : nil, optionalConstraints: nil) return try await PeerCalls.describe(pc, answer: answer, constraints) } private func offer(restart: Bool = false) async throws { guard let pc, !closed, await PeerCalls.read(pc, { $0.signalingState }) == .stable, !closed else { return } let desc = try await makeDescription(answer: false, restart: restart) guard !closed else { return } try await setLocal(desc) send("offer", ["description": ["type": "offer", "sdp": desc.sdp]]) } func receive(_ message: Message) { guard message.gen == generation else { return } enqueue { [weak self] in guard let self, let pc = self.pc, !self.closed else { return } if message.type == "candidate", let candidate = message.payload?.candidate { if self.ignoreOffer { return } if await PeerCalls.read(pc, { $0.remoteDescription == nil }) { guard self.candidates.count < 256 else { throw CancellationError() } self.candidates.append(candidate) } else { try await self.add(candidate) } return } guard let desc = message.payload?.description else { return } let signaling = await PeerCalls.read(pc) { $0.signalingState } guard !self.closed else { return } if message.type == "offer", desc.type == "offer" { let collision = signaling != .stable self.ignoreOffer = collision && !self.polite if self.ignoreOffer { return } if collision { try await self.setLocal(RTCSessionDescription(type: .rollback, sdp: "")) } try await self.setRemote(RTCSessionDescription(type: .offer, sdp: desc.sdp)) self.ignoreOffer = false try await self.drain() let answer = try await self.makeDescription(answer: true) try await self.setLocal(answer) self.send("answer", ["description": ["type": "answer", "sdp": answer.sdp]]) } else if message.type == "answer", desc.type == "answer", signaling == .haveLocalOffer { try await self.setRemote(RTCSessionDescription(type: .answer, sdp: desc.sdp)) self.ignoreOffer = false try await self.drain() } } } private func add(_ candidate: Candidate) async throws { guard let pc, !closed, !candidate.candidate.isEmpty else { return } let ice = RTCIceCandidate( sdp: candidate.candidate, sdpMLineIndex: candidate.sdpMLineIndex ?? 0, sdpMid: candidate.sdpMid) try await PeerCalls.add(pc, ice) } private func drain() async throws { let saved = candidates candidates = [] for c in saved { try await add(c) } } /// A fresh `/ice` answer (§5.2, §5.3). The core fetches one on every /// roster, and a roster answers every stream resume — every five /// minutes at the stream's own TTL — so most answers are not renewals /// at all. Up to 0.4.1 every one of them restarted ICE on every /// connection, twelve times an hour, each one a new path for the /// screen and camera to find their way back onto. Now only a renewal /// that is due restarts (ICERenewal.restart); anything else is adopted /// quietly, for whatever this connection gathers next. func updateICE(_ ice: ICEConfig, now: Int64) { guard ice.renewAt != lastICE?.renewAt || ice.iceServers.count != lastICE?.iceServers.count else { return } lastICE = ice let restart = ICERenewal.restart(gathered: gatheredICE, next: ice, now: now) if restart { gatheredICE = ice } enqueue { [weak self] in guard let self, let pc = self.pc else { return } let config = self.configuration(ice) guard await PeerCalls.read(pc, { $0.setConfiguration(config) }) else { return } if restart { ScreenLog.info("ICE restart with \(ScreenWatchdog.short(self.id)): renewal due") self.engine?.iceRestarts += 1 try await self.offer(restart: true) } } } // ------------------------------------------------------------- // The screen watchdog's hands, for this connection // ------------------------------------------------------------- /// Whether this peer's screen is arriving here. var receivingScreen: Bool { screenReceiver != nil && !screenMid.isEmpty } /// The screen's encoding switched off and on again. WebRTC starts a /// stream that comes back on with a keyframe, which is the one thing a /// receiver stuck on a broken picture needs; there is no API on this /// build of WebRTC to ask for one directly. func forceScreenKeyframe() { setScreenActive(false) Task { @MainActor [weak self] in try? await Task.sleep(nanoseconds: 150_000_000) self?.setScreenActive(true) } } func setScreenActive(_ active: Bool) { guard let sender = screenTx?.sender else { return } AudioWork.post { let parameters = sender.parameters guard let encoding = parameters.encodings.first, encoding.isActive != active else { return } encoding.isActive = active sender.parameters = parameters } } /// A fresh encoder: the screen sent smaller for a while. A change of /// size is what makes WebRTC set its encoder up again from nothing — for /// H.264 on a Mac, a new VideoToolbox session, which is the part that /// can go bad under a sleep or a GPU switch and hold the frames it was /// given. The other way an encoder stops is a frame it cannot take at /// all (the level's block count, see ScreenSendPolicy), which a smaller /// size is also the way out of. A third off each side each time, down to /// a quarter; full size again after thirty seconds, and if that stops it /// again the watchdog comes back round, after its rest. func recreateScreenEncoder() { guard screenTx != nil else { return } nudge = min(nudge * 1.5, 4) nudges += 1 let mine = nudges ScreenLog.info("screen to \(ScreenWatchdog.short(id)) now sent at 1/\(String(format: "%.2f", nudge)) of its size") applySendPolicy() setScreenActive(true) Task { @MainActor [weak self] in try? await Task.sleep(nanoseconds: 30_000_000_000) guard let self, self.nudges == mine, self.nudge > 1 else { return } self.nudge = 1 ScreenLog.info("screen to \(ScreenWatchdog.short(self.id)) back to full size") self.applySendPolicy() } } private var nudges = 0 private var nudge = 1.0 /// What stopping and sharing again did, on this connection alone: its /// screen transceiver stopped and renegotiated, and a new one added as /// soon as that settles (`settled` adds the screen to a connection that /// has none while the app is sharing). func reAddScreen() { guard screenTx != nil else { return } removeScreen() } /// The share has stopped: the next one starts at full size. func screenStopped() { nudge = 1 } nonisolated func peerConnectionShouldNegotiate(_ peerConnection: RTCPeerConnection) { Task { @MainActor [weak self] in guard let self else { return } // Give the higher-id peer first chance, matching the browser's // polite yield. Perfect negotiation still handles actual glare. if self.polite { try? await Task.sleep(nanoseconds: 200_000_000) } self.enqueue { [weak self] in try await self?.offer() } } } nonisolated func peerConnection( _ peerConnection: RTCPeerConnection, didGenerate candidate: RTCIceCandidate ) { Task { @MainActor [weak self] in self?.send( "candidate", [ "candidate": [ "candidate": candidate.sdp, "sdpMid": candidate.sdpMid as Any? ?? NSNull(), "sdpMLineIndex": candidate.sdpMLineIndex, ] ]) } } nonisolated func peerConnection( _ peerConnection: RTCPeerConnection, didAdd rtpReceiver: RTCRtpReceiver, streams: [RTCMediaStream] ) { // Keep decoding and RTP statistics available when a test disables speakers. if let audio = rtpReceiver.track as? RTCAudioTrack, !playAudio { audio.source.volume = 0 } guard let track = rtpReceiver.track else { return } // Read here, on the signalling thread this callback arrives on, rather // than on the main actor: by the time the main actor gets to it the // next description can have moved the transceiver on. let receiver = rtpReceiver.receiverId let transceiver = peerConnection.transceivers.first { $0.receiver.receiverId == receiver } var info = ArrivingTrack(kind: track.kind, streamId: streams.first?.streamId ?? "") if let transceiver { info.mid = transceiver.mid // The negotiated direction when there is one. On the side that // answers an offer there is not yet, and that is exactly the side a // screen arrives at; the direction the transceiver was built with is // recvonly for an m-line the far side added, which is the point. var current = RTCRtpTransceiverDirection.inactive info.direction = sdpDirection( transceiver.currentDirection(¤t) ? current : transceiver.direction) } Task { @MainActor [weak self] in guard let self, !self.closed else { return } self.arrived(track, receiver: receiver, info: info) } } /// §5.4's other ending: the sender stopped its screen's transceiver and /// the renegotiation took the receiver away. Only the receiver the /// screen is showing from counts. A screen replaced on its own mid ends /// its predecessor afterwards, and without the check that took the live /// replacement's tile with it (the browser's Codex S4-15). nonisolated func peerConnection( _ peerConnection: RTCPeerConnection, didRemove rtpReceiver: RTCRtpReceiver ) { let receiver = rtpReceiver.receiverId Task { @MainActor [weak self] in guard let self, !self.closed, self.screenReceiver == receiver else { return } self.dropScreen() } } nonisolated func peerConnection( _ peerConnection: RTCPeerConnection, didChange newState: RTCIceConnectionState ) { if newState == .failed { Task { @MainActor [weak self] in self?.engine?.call?.mediaFailed() } } } nonisolated func peerConnection( _ peerConnection: RTCPeerConnection, didChange stateChanged: RTCSignalingState ) { guard stateChanged == .stable else { return } Task { @MainActor [weak self] in self?.settled() } } nonisolated func peerConnection( _ peerConnection: RTCPeerConnection, didAdd stream: RTCMediaStream ) {} nonisolated func peerConnection( _ peerConnection: RTCPeerConnection, didRemove stream: RTCMediaStream ) {} nonisolated func peerConnection( _ peerConnection: RTCPeerConnection, didChange newState: RTCIceGatheringState ) {} nonisolated func peerConnection( _ peerConnection: RTCPeerConnection, didRemove candidates: [RTCIceCandidate] ) {} nonisolated func peerConnection( _ peerConnection: RTCPeerConnection, didOpen dataChannel: RTCDataChannel ) {} }