| 1 | import AVFoundation |
| 2 | import CoreVideo |
| 3 | import Foundation |
| 4 | import LiveKitWebRTC |
| 5 | |
| 6 | @MainActor final class MediaEngine { |
| 7 | weak var call: Call? |
| 8 | let factory: RTCPeerConnectionFactory |
| 9 | var localVideo: RTCVideoTrack? |
| 10 | var localAudio: RTCAudioTrack? |
| 11 | /// §5.4: the display track while this app is sharing, and nothing |
| 12 | /// otherwise. Every connection carries it on a transceiver of its own. |
| 13 | private(set) var localScreen: RTCVideoTrack? |
| 14 | private var screenCapture: ScreenCapture? |
| 15 | // Bumped by everything that ends a share, so a capture whose picker |
| 16 | // was still open when the share or the call ended is stopped when it |
| 17 | // finally answers, rather than sent. |
| 18 | private var screenAttempt = 0 |
| 19 | private var capture: VideoCapture? |
| 20 | private var peers: [String: MediaPeer] = [:] |
| 21 | /// What the screen is optimised for (ScreenSendPolicy), and the size |
| 22 | /// its source was last seen at, which motion mode scales down from. |
| 23 | var screenMode = ScreenShareMode.balanced { |
| 24 | didSet { applySendPolicy() } |
| 25 | } |
| 26 | private var screenSize = (width: 0, height: 0) |
| 27 | |
| 28 | /// The audio device module and processing behind the factory. See CallAudio. |
| 29 | let audio: CallAudio |
| 30 | |
| 31 | /// Builds on the calling thread: only for a call touched before it |
| 32 | /// joined. A join uses `make`. |
| 33 | convenience init(call: Call) { |
| 34 | RTCInitializeSSL() |
| 35 | #if os(iOS) |
| 36 | self.init(call: call, audio: CallAudio(routes: call.audio, audioDevice: call.customAudioDevice)) |
| 37 | #else |
| 38 | self.init(call: call, audio: CallAudio(routes: call.audio, audioDevice: nil, input: call.audioInput)) |
| 39 | #endif |
| 40 | } |
| 41 | |
| 42 | init(call: Call, audio: CallAudio) { |
| 43 | self.call = call |
| 44 | self.audio = audio |
| 45 | factory = audio.factory |
| 46 | } |
| 47 | |
| 48 | /// The media for a call, with its factory and audio device module built |
| 49 | /// on the audio queue (CallAudio.make): 100–200 ms the main thread no |
| 50 | /// longer spends when someone joins. |
| 51 | static func make(call: Call) async -> MediaEngine { |
| 52 | await AudioWork.run { RTCInitializeSSL() } |
| 53 | #if os(iOS) |
| 54 | let audio = await CallAudio.make(routes: call.audio, audioDevice: call.customAudioDevice) |
| 55 | #else |
| 56 | let audio = await CallAudio.make(routes: call.audio, audioDevice: nil, input: call.audioInput) |
| 57 | #endif |
| 58 | return MediaEngine(call: call, audio: audio) |
| 59 | } |
| 60 | func prepare(capture supplied: VideoCapture?) async throws { |
| 61 | close() |
| 62 | let source = factory.videoSource() |
| 63 | localVideo = factory.videoTrack(with: source, trackId: "camera") |
| 64 | localAudio = audio.makeTrack() |
| 65 | let capture = supplied ?? CameraCapture.make() |
| 66 | self.capture = capture |
| 67 | offersCameras = capture.offersCameras |
| 68 | capture.observe { [weak self] in self?.call?.cameraChanged() } |
| 69 | try await capture.start(source: source) |
| 70 | #if os(iOS) |
| 71 | // A supplied audio device owns its session configuration (SDK contract). |
| 72 | if call?.customAudioDevice == nil { |
| 73 | let audio = RTCAudioSession.sharedInstance() |
| 74 | audio.lockForConfiguration() |
| 75 | defer { audio.unlockForConfiguration() } |
| 76 | // Not .allowBluetoothHFP: see AudioRoutes.options. At join nobody |
| 77 | // has chosen a Bluetooth microphone, so the call starts wideband |
| 78 | // and only drops to the hands-free profile if someone asks for a |
| 79 | // Bluetooth input by name. |
| 80 | try audio.setCategory( |
| 81 | AVAudioSession.Category.playAndRecord, |
| 82 | mode: AVAudioSession.Mode.videoChat, |
| 83 | options: AudioRoutes.options(bluetoothInput: false)) |
| 84 | // WebRTC owns activation and interruption recovery as peers start/stop. |
| 85 | } |
| 86 | #endif |
| 87 | } |
| 88 | /// The cameras the running capture can offer, and the one it is on. |
| 89 | /// Empty before prepare, and empty for the interop executable's |
| 90 | /// synthetic frames, which is how the picker knows not to appear. |
| 91 | /// |
| 92 | /// With real cameras the list stays live after the capture has gone |
| 93 | /// (the call ended under the person), so the panel can still say what |
| 94 | /// is connected and a choice there is remembered for next time. |
| 95 | var cameras: [CameraDevice] { |
| 96 | if let capture { return capture.cameras } |
| 97 | return offersCameras ? CameraWatch.shared.cameras : [] |
| 98 | } |
| 99 | var currentCamera: String? { capture?.currentCamera } |
| 100 | /// Whether this call's camera is a real one, whose row the panel always |
| 101 | /// shows. Kept past the capture, for the same reason as `cameras`. |
| 102 | private(set) var offersCameras = false |
| 103 | var cameraStatus: CameraStatus? { capture?.cameraStatus } |
| 104 | var cameraNote: String { capture?.cameraNote ?? "" } |
| 105 | var cameraToKeep: CameraDevice? { capture?.cameraToKeep } |
| 106 | func use(camera id: String) async throws { |
| 107 | if let capture { |
| 108 | try await capture.use(camera: id) |
| 109 | } else if offersCameras, let device = CameraWatch.shared.cameras.first(where: { $0.id == id }) { |
| 110 | CameraPreference(device).save() |
| 111 | } |
| 112 | } |
| 113 | func retryCamera() async { |
| 114 | CameraWatch.shared.refresh() |
| 115 | await capture?.retryCamera() |
| 116 | } |
| 117 | |
| 118 | func closePeers() { |
| 119 | for (_, p) in peers { p.close() } |
| 120 | peers.removeAll() |
| 121 | } |
| 122 | |
| 123 | /// Starts sending a screen (§5.4). The capture is asked first: a person |
| 124 | /// who changes their mind at the system's picker has shared nothing, |
| 125 | /// and nothing is added to any connection. |
| 126 | /// |
| 127 | /// What gives way under pressure — resolution or frame rate — is the |
| 128 | /// "Optimise for" mode's, through sender parameters (ScreenSendPolicy). |
| 129 | func startScreen(_ capture: ScreenCapture, ended: @escaping @MainActor () -> Void) async throws { |
| 130 | guard localScreen == nil else { return } |
| 131 | screenAttempt += 1 |
| 132 | let attempt = screenAttempt |
| 133 | screenSize = (0, 0) |
| 134 | screenIsScreencast = ScreenSendPolicy.screencast(screenMode) |
| 135 | let source = factory.videoSource(forScreenCast: screenIsScreencast) |
| 136 | try await capture.start(source: source, ended: ended) |
| 137 | guard attempt == screenAttempt, localScreen == nil else { |
| 138 | capture.stop() |
| 139 | throw CancellationError() |
| 140 | } |
| 141 | screenCapture = capture |
| 142 | screenSourceInUse = source |
| 143 | if let size = capture.frameSize, size.width > 0, size.height > 0 { screenSize = size } |
| 144 | localScreen = factory.videoTrack(with: source, trackId: "screen") |
| 145 | for peer in peers.values { peer.addScreen() } |
| 146 | } |
| 147 | /// The source the screen's capture is writing into, for a share again |
| 148 | /// from the same capture. |
| 149 | private var screenSourceInUse: RTCVideoSource? |
| 150 | private var reshares = 0 |
| 151 | |
| 152 | /// Stop and Share, without the picker (spec §15.22): every connection's |
| 153 | /// screen transceiver stopped and renegotiated, a fresh track on the |
| 154 | /// same source, the capture started again on the same content, and each |
| 155 | /// connection given the new track as soon as it settles. New |
| 156 | /// transceivers mean new encoders here and new receivers and decoders at |
| 157 | /// every far end, which is what the hand-made stop and share did. |
| 158 | func reshareScreen() async { |
| 159 | guard let capture = screenCapture, let source = screenSourceInUse, localScreen != nil else { return } |
| 160 | reshares += 1 |
| 161 | for peer in peers.values { peer.removeScreen() } |
| 162 | localScreen = factory.videoTrack(with: source, trackId: "screen-\(reshares)") |
| 163 | if !(await capture.restart()) { ScreenLog.info("share again: the capture carries on as it was") } |
| 164 | guard screenCapture === capture else { return } |
| 165 | // Any connection already stable takes it now; the rest as they settle. |
| 166 | for peer in peers.values { peer.addScreen() } |
| 167 | } |
| 168 | |
| 169 | /// Stops sending the screen: every connection's screen transceiver is |
| 170 | /// stopped and renegotiated, so the receivers' tracks *end*, and the |
| 171 | /// capture goes last. |
| 172 | func stopScreen() { |
| 173 | screenAttempt += 1 |
| 174 | guard let capture = screenCapture else { return } |
| 175 | screenCapture = nil |
| 176 | screenSourceInUse = nil |
| 177 | localScreen = nil |
| 178 | for peer in peers.values { |
| 179 | peer.removeScreen() |
| 180 | peer.screenStopped() |
| 181 | } |
| 182 | capture.stop() |
| 183 | } |
| 184 | |
| 185 | // ------------------------------------------------------------- |
| 186 | // The screen watchdog's hands (ScreenWatchdog, run by Call) |
| 187 | // ------------------------------------------------------------- |
| 188 | |
| 189 | /// What the capture has produced so far; nothing while not sharing. |
| 190 | var screenCaptureCounts: ScreenCaptureCounts? { screenCapture?.counts } |
| 191 | |
| 192 | /// A fresh capture of the same content, into the same source. |
| 193 | func restartScreenCapture() async -> Bool { |
| 194 | guard let capture = screenCapture else { return false } |
| 195 | return await capture.restart() |
| 196 | } |
| 197 | func forceScreenKeyframe(_ peer: String) { peers[peer]?.forceScreenKeyframe() } |
| 198 | func recreateScreenEncoder(_ peer: String) { peers[peer]?.recreateScreenEncoder() } |
| 199 | func reAddScreen(_ peer: String) { peers[peer]?.reAddScreen() } |
| 200 | |
| 201 | /// For the gate: every screen encoding switched off behind the |
| 202 | /// watchdog's back, which is what an encoder that has stopped looks |
| 203 | /// like from outside it — frames go in, and nothing comes out. |
| 204 | func stallScreenEncoderForTest() { |
| 205 | for peer in peers.values { peer.setScreenActive(false) } |
| 206 | } |
| 207 | |
| 208 | /// Each peer's screen as it arrives here, for the peers whose screen is |
| 209 | /// arriving (the receiver's watchdog, and "Copy call diagnostics"). |
| 210 | func remoteScreenReports() async -> [String: ScreenReceiveReport] { |
| 211 | var out: [String: ScreenReceiveReport] = [:] |
| 212 | for (id, peer) in peers { |
| 213 | guard let pc = peer.pc, peer.receivingScreen else { continue } |
| 214 | let report = await PeerCalls.statistics(pc) |
| 215 | var reading = ScreenReceiveReport() |
| 216 | for stat in report.statistics.values where stat.type == "inbound-rtp" { |
| 217 | let mid = stat.values["mid"] as? String ?? "" |
| 218 | guard peer.isScreen(mid: mid) else { continue } |
| 219 | reading.add(stat.values) |
| 220 | } |
| 221 | reading.mid = peer.screenMidHere |
| 222 | out[id] = reading |
| 223 | } |
| 224 | return out |
| 225 | } |
| 226 | |
| 227 | /// Frames decoded from screens here, never going backwards: a rebuilt |
| 228 | /// screen arrives on a new receiver whose own count starts again, and |
| 229 | /// the gate's "decoding again" has to see its frames as more. |
| 230 | private var screenDecodedSeen: [String: UInt64] = [:] |
| 231 | private(set) var screenDecodedTotal: UInt64 = 0 |
| 232 | private func tallyScreenDecoded(_ peer: String, _ now: UInt64) { |
| 233 | let last = screenDecodedSeen[peer] ?? 0 |
| 234 | screenDecodedTotal += now >= last ? now - last : now |
| 235 | screenDecodedSeen[peer] = now |
| 236 | } |
| 237 | |
| 238 | /// Puts every connection's screen and camera senders where the policy |
| 239 | /// wants them; a no-op for any that already are. |
| 240 | func applySendPolicy() { |
| 241 | for peer in peers.values { peer.applySendPolicy() } |
| 242 | } |
| 243 | /// The screen's source size, as its statistics report it. |
| 244 | func screenSource(width: Int, height: Int) { |
| 245 | guard width > 0, height > 0, (width, height) != screenSize else { return } |
| 246 | screenSize = (width, height) |
| 247 | applySendPolicy() |
| 248 | } |
| 249 | var screenPlan: EncodingPlan { |
| 250 | ScreenSendPolicy.screen( |
| 251 | screenMode, width: screenSize.width, height: screenSize.height, screencast: screenIsScreencast) |
| 252 | } |
| 253 | /// Whether this share's source was made as a screencast: fixed when the |
| 254 | /// share starts, whatever the mode is changed to afterwards. |
| 255 | private var screenIsScreencast = true |
| 256 | |
| 257 | func close() { |
| 258 | stopScreen() |
| 259 | closePeers() |
| 260 | capture?.stop() |
| 261 | capture = nil |
| 262 | // On the audio queue: a track's switch waits out an engine start. |
| 263 | let video = localVideo |
| 264 | let audio = localAudio |
| 265 | AudioWork.post { |
| 266 | video?.isEnabled = false |
| 267 | audio?.isEnabled = false |
| 268 | } |
| 269 | localVideo = nil |
| 270 | localAudio = nil |
| 271 | } |
| 272 | func reconcile(state: RoomState, ice: ICEConfig) { |
| 273 | let wanted = Set(state.peers.filter { $0.peer != state.peer }.map(\.peer)) |
| 274 | for id in Array(peers.keys) where !wanted.contains(id) { |
| 275 | peers.removeValue(forKey: id)?.close() |
| 276 | call?.removeRemote(id) |
| 277 | } |
| 278 | for person in state.peers where person.peer != state.peer { |
| 279 | let gen = person.gen ?? 1 |
| 280 | if let old = peers[person.peer], old.generation == gen { continue } |
| 281 | peers.removeValue(forKey: person.peer)?.close() |
| 282 | call?.removeRemote(person.peer) |
| 283 | let peer = MediaPeer( |
| 284 | id: person.peer, generation: gen, polite: state.peer < person.peer, engine: self) |
| 285 | peers[person.peer] = peer |
| 286 | peer.open(ice: ice) |
| 287 | } |
| 288 | } |
| 289 | func receive(_ message: Message) { if let from = message.from { peers[from]?.receive(message) } } |
| 290 | /// §5.4's label. A peer this app has no connection to yet is ignored, |
| 291 | /// as the browser ignores it: its screen will be told apart from its |
| 292 | /// camera by the m-line when the connection comes, and a caption with |
| 293 | /// no connection behind it is a tile nothing will ever fill or clear. |
| 294 | func receiveShare(_ message: Message) { |
| 295 | guard let from = message.from, let peer = peers[from] else { return } |
| 296 | peer.share(on: message.payload?.on == true) |
| 297 | } |
| 298 | /// ICE restarts this app has started for a renewal, for the gate. |
| 299 | var iceRestarts: UInt64 = 0 |
| 300 | func updateICE(_ config: ICEConfig) { |
| 301 | let now = Int64(Date().timeIntervalSince1970) |
| 302 | for p in peers.values { p.updateICE(config, now: now) } |
| 303 | } |
| 304 | /// The five audio-health fields, gathered alongside the byte counts. |
| 305 | /// |
| 306 | /// Bytes say audio arrived; they say nothing about what it sounded like. |
| 307 | /// These five separate the two ways received audio goes wrong, which is |
| 308 | /// the distinction a person reporting "it growled" cannot make for us: |
| 309 | /// |
| 310 | /// - `concealedSamples` against `totalSamplesReceived` is loss. The |
| 311 | /// decoder invented that fraction of the audio because nothing |
| 312 | /// arrived to decode. |
| 313 | /// - `insertedSamplesForDeceleration` and |
| 314 | /// `removedSamplesForAcceleration` are clock drift. The jitter |
| 315 | /// buffer is stretching or compressing the signal to keep up with a |
| 316 | /// sender whose clock runs at a different rate than this device's — |
| 317 | /// and stretching a voice is what a growl *is*. |
| 318 | /// |
| 319 | /// With the device pickers beside it, a suspect device can be ruled out |
| 320 | /// by switching away from it while watching these. |
| 321 | static let audioHealthFields = [ |
| 322 | "concealedSamples", "totalSamplesReceived", |
| 323 | "insertedSamplesForDeceleration", "removedSamplesForAcceleration", |
| 324 | ] |
| 325 | |
| 326 | func stats() async -> [String: UInt64] { |
| 327 | var values: [String: UInt64] = [ |
| 328 | "framesDecoded": 0, "audioBytesReceived": 0, "videoBytesSent": 0, "audioBytesSent": 0, |
| 329 | "screenFramesDecoded": 0, "screenBytesSent": 0, |
| 330 | "peerConnections": UInt64(peers.count), |
| 331 | "concealedSamples": 0, "totalSamplesReceived": 0, |
| 332 | "insertedSamplesForDeceleration": 0, "removedSamplesForAcceleration": 0, |
| 333 | "jitterBufferDelayMs": 0, |
| 334 | ] |
| 335 | for (id, peer) in peers { |
| 336 | guard let pc = peer.pc else { continue } |
| 337 | let ownScreen = await peer.ownScreenMid() |
| 338 | let report = await PeerCalls.statistics(pc) |
| 339 | var screenDecoded: UInt64 = 0 |
| 340 | defer { tallyScreenDecoded(id, screenDecoded) } |
| 341 | for stat in report.statistics.values { |
| 342 | let kind = (stat.values["kind"] as? String) ?? (stat.values["mediaType"] as? String) ?? "" |
| 343 | let mid = stat.values["mid"] as? String ?? "" |
| 344 | if stat.type == "inbound-rtp" { |
| 345 | if kind == "video" { |
| 346 | // A peer sharing its screen sends two video streams, and a |
| 347 | // count that added them together would let a screen stand in |
| 348 | // for a camera that had stopped. Told apart by mid, which is |
| 349 | // what the browser's own gate does. |
| 350 | let field = peer.isScreen(mid: mid) ? "screenFramesDecoded" : "framesDecoded" |
| 351 | let decoded = (stat.values["framesDecoded"] as? NSNumber)?.uint64Value ?? 0 |
| 352 | values[field, default: 0] += decoded |
| 353 | if peer.isScreen(mid: mid) { screenDecoded += decoded } |
| 354 | } |
| 355 | if kind == "audio" { |
| 356 | values["audioBytesReceived", default: 0] += |
| 357 | (stat.values["bytesReceived"] as? NSNumber)?.uint64Value ?? 0 |
| 358 | for field in MediaEngine.audioHealthFields { |
| 359 | values[field, default: 0] += (stat.values[field] as? NSNumber)?.uint64Value ?? 0 |
| 360 | } |
| 361 | // jitterBufferDelay is seconds as a double, and is cumulative |
| 362 | // against jitterBufferEmittedCount rather than a gauge. It is |
| 363 | // carried in milliseconds because everything else here is an |
| 364 | // integer, and a sub-millisecond buffer is not a fact anyone |
| 365 | // reading this is going to act on. |
| 366 | let delay = (stat.values["jitterBufferDelay"] as? NSNumber)?.doubleValue ?? 0 |
| 367 | values["jitterBufferDelayMs", default: 0] += UInt64((delay * 1000).rounded()) |
| 368 | } |
| 369 | } |
| 370 | if stat.type == "outbound-rtp" { |
| 371 | if kind == "video" { |
| 372 | let field = !mid.isEmpty && mid == ownScreen ? "screenBytesSent" : "videoBytesSent" |
| 373 | values[field, default: 0] += |
| 374 | (stat.values["bytesSent"] as? NSNumber)?.uint64Value ?? 0 |
| 375 | } |
| 376 | if kind == "audio" { |
| 377 | values["audioBytesSent", default: 0] += |
| 378 | (stat.values["bytesSent"] as? NSNumber)?.uint64Value ?? 0 |
| 379 | } |
| 380 | } |
| 381 | } |
| 382 | } |
| 383 | values["screenFramesDecodedTotal"] = screenDecodedTotal |
| 384 | return values |
| 385 | } |
| 386 | |
| 387 | /// Each remote person's sound, for the meters (spec §15.20): the level |
| 388 | /// their receive tap has seen since the last look, whether the |
| 389 | /// connection is up, and — when `stats` is set, or for the people in |
| 390 | /// `detail` — the audio bytes and packets received so far and whether |
| 391 | /// the receiver is live, one `statistics(for:)` per audio receiver rather |
| 392 | /// than the whole report. For the people in `acks`, whether their |
| 393 | /// receiver reports say this app's audio reaches them (spec §15.24's |
| 394 | /// fallback for a client that does not say who it hears). |
| 395 | func peerAudio(stats: Bool, detail: Set<String> = [], acks: Set<String> = []) async -> [String: PeerAudioReading] { |
| 396 | var out: [String: PeerAudioReading] = [:] |
| 397 | for (id, peer) in peers { |
| 398 | guard let pc = peer.pc else { continue } |
| 399 | let (peak, buffers) = peer.tap.take() |
| 400 | let deep = stats || detail.contains(id) |
| 401 | let ack = acks.contains(id) |
| 402 | // Read on the audio queue, as every call into the connection is |
| 403 | // (PeerCalls): five times a second, and each one a blocking hop to |
| 404 | // WebRTC's signalling thread. |
| 405 | let (state, receivers, senders) = await PeerCalls.read(pc) { pc in |
| 406 | ( |
| 407 | pc.iceConnectionState, deep ? pc.receivers.filter { $0.track?.kind == "audio" } : [], |
| 408 | ack ? pc.senders.filter { $0.track?.kind == "audio" } : [] |
| 409 | ) |
| 410 | } |
| 411 | var reading = PeerAudioReading( |
| 412 | peak: peak, buffers: buffers, connected: state == .connected || state == .completed, |
| 413 | connectionBad: state == .disconnected || state == .failed) |
| 414 | reading.gone = state == .failed || state == .closed |
| 415 | if deep { |
| 416 | var bytes: UInt64 = 0 |
| 417 | var packets: UInt64 = 0 |
| 418 | var live = false |
| 419 | for receiver in receivers { |
| 420 | if receiver.track?.readyState == .live { live = true } |
| 421 | let report = await PeerCalls.statistics(pc, for: receiver) |
| 422 | for stat in report.statistics.values where stat.type == "inbound-rtp" { |
| 423 | bytes += (stat.values["bytesReceived"] as? NSNumber)?.uint64Value ?? 0 |
| 424 | packets += (stat.values["packetsReceived"] as? NSNumber)?.uint64Value ?? 0 |
| 425 | } |
| 426 | } |
| 427 | reading.bytes = bytes |
| 428 | reading.packets = packets |
| 429 | reading.hasReceiver = !receivers.isEmpty |
| 430 | reading.receiverLive = live |
| 431 | } |
| 432 | if ack { |
| 433 | var sent: UInt64 = 0 |
| 434 | var acked = false |
| 435 | for sender in senders { |
| 436 | let report = await PeerCalls.statistics(pc, for: sender) |
| 437 | for stat in report.statistics.values { |
| 438 | if stat.type == "outbound-rtp" { |
| 439 | sent += (stat.values["packetsSent"] as? NSNumber)?.uint64Value ?? 0 |
| 440 | } else if stat.type == "remote-inbound-rtp" { |
| 441 | // Exists only once a receiver report has come back. |
| 442 | let received = (stat.values["packetsReceived"] as? NSNumber)?.uint64Value ?? 0 |
| 443 | let measured = (stat.values["roundTripTimeMeasurements"] as? NSNumber)?.uint64Value ?? 0 |
| 444 | if received > 0 || measured > 0 || stat.values["roundTripTime"] != nil { acked = true } |
| 445 | } |
| 446 | } |
| 447 | } |
| 448 | reading.acked = sent > 0 && acked |
| 449 | } |
| 450 | out[id] = reading |
| 451 | } |
| 452 | return out |
| 453 | } |
| 454 | |
| 455 | /// Every connection's statistics at one look, for the quality report |
| 456 | /// (spec §15.26): the whole report per connection, with its generation |
| 457 | /// and which mids carry screens, read as the other statistics are. |
| 458 | func qualitySamples() async -> [String: QualityPeerSample] { |
| 459 | var out: [String: QualityPeerSample] = [:] |
| 460 | for (id, peer) in peers { |
| 461 | guard let pc = peer.pc else { continue } |
| 462 | let ownScreen = await peer.ownScreenMid() |
| 463 | let report = await PeerCalls.statistics(pc) |
| 464 | out[id] = QualityPeerSample( |
| 465 | generation: peer.generation, ownScreenMid: ownScreen, screenMid: peer.screenMidHere, |
| 466 | stats: report.statistics.mapValues { QualityStat(type: $0.type, values: $0.values) }) |
| 467 | } |
| 468 | return out |
| 469 | } |
| 470 | |
| 471 | /// Caps every connection's bandwidth estimate, which is what a thin |
| 472 | /// uplink does to it. For the gate, which has no thin uplink to test on; |
| 473 | /// nil takes the cap off. |
| 474 | func limitUplink(_ bps: Int?) { |
| 475 | uplinkLimit = bps |
| 476 | for peer in peers.values { peer.limitUplink(bps) } |
| 477 | } |
| 478 | private(set) var uplinkLimit: Int? |
| 479 | |
| 480 | /// This app's screen on the way out, one report per connection sending |
| 481 | /// it (see ScreenSendReport). |
| 482 | func screenReports() async -> [String: ScreenSendReport] { |
| 483 | var reports: [String: ScreenSendReport] = [:] |
| 484 | for (id, peer) in peers { |
| 485 | guard let pc = peer.pc, let mid = await peer.ownScreenMid() else { continue } |
| 486 | let report = await PeerCalls.statistics(pc) |
| 487 | let flat = report.statistics.mapValues { (type: $0.type, values: $0.values as [String: Any]) } |
| 488 | if var read = ScreenSendReport.read(flat, mid: mid) { |
| 489 | read.timestamp = report.timestamp_us / 1_000_000 |
| 490 | reports[id] = read |
| 491 | } |
| 492 | } |
| 493 | return reports |
| 494 | } |
| 495 | } |
| 496 | |
| 497 | /// A transceiver direction as SDP spells it, which is how the classifier, |
| 498 | /// and the browser it was ported from, compare them. |
| 499 | private func sdpDirection(_ direction: RTCRtpTransceiverDirection) -> String { |
| 500 | switch direction { |
| 501 | case .sendRecv: return "sendrecv" |
| 502 | case .sendOnly: return "sendonly" |
| 503 | case .recvOnly: return "recvonly" |
| 504 | case .inactive: return "inactive" |
| 505 | case .stopped: return "stopped" |
| 506 | @unknown default: return "" |
| 507 | } |
| 508 | } |
| 509 | |
| 510 | /// One look at a remote person's audio (see MediaEngine.peerAudio). |
| 511 | struct PeerAudioReading: Equatable, Sendable { |
| 512 | var peak: Float |
| 513 | var buffers: Int |
| 514 | var connected: Bool |
| 515 | /// ICE `disconnected` or `failed` (spec §15.23's three seconds). |
| 516 | var connectionBad = false |
| 517 | /// ICE `failed` or `closed`: whatever was heard on it is gone (§15.24). |
| 518 | var gone = false |
| 519 | var bytes: UInt64? |
| 520 | /// Audio packets received, the receiver there at all, and its track |
| 521 | /// live (§15.24); read with `bytes`. |
| 522 | var packets: UInt64? |
| 523 | var hasReceiver = false |
| 524 | var receiverLive = false |
| 525 | /// Their receiver reports say this app's audio reaches them; nil when |
| 526 | /// not asked. |
| 527 | var acked: Bool? |
| 528 | } |
| 529 | |
| 530 | @MainActor private final class MediaPeer: NSObject, RTCPeerConnectionDelegate { |
| 531 | let id: String |
| 532 | /// Measures every audio track that arrives on this connection (their |
| 533 | /// microphone; a browser's shared tab audio would add to it). |
| 534 | nonisolated let tap = RemoteAudioTap() |
| 535 | private var tapped: [RTCAudioTrack] = [] |
| 536 | let generation: Int64 |
| 537 | let polite: Bool |
| 538 | nonisolated let playAudio: Bool |
| 539 | weak var engine: MediaEngine? |
| 540 | var pc: RTCPeerConnection? |
| 541 | private var work: Task<Void, Never>? |
| 542 | private var closed = false |
| 543 | private var ignoreOffer = false |
| 544 | private var candidates: [Candidate] = [] |
| 545 | private var lastICE: ICEConfig? |
| 546 | /// The list this connection last gathered on — built with, or last |
| 547 | /// restarted onto — whose `renewAt` says when it needs a restart. |
| 548 | private var gatheredICE: ICEConfig? |
| 549 | // §5.4, receiving. What this peer's camera and microphone came in on, |
| 550 | // and whether it has said it is sharing; the receiver its screen |
| 551 | // arrived on, so that only *that* receiver going away takes the tile |
| 552 | // with it; and the mid, so the statistics can count the two video |
| 553 | // streams apart. |
| 554 | private var clues = ScreenClues() |
| 555 | private var screenReceiver: String? |
| 556 | private var screenMid = "" |
| 557 | // Whether this connection carries a camera of ours: the classifier's |
| 558 | // direction rule turns on it. |
| 559 | private var sentVideo = false |
| 560 | // §5.4, sending: the transceiver this connection carries our screen on, |
| 561 | // while we are sharing. And an offer this connection owes the far side |
| 562 | // but could not make yet, because it was in the middle of another. |
| 563 | private var screenTx: RTCRtpTransceiver? |
| 564 | private var owesOffer = false |
| 565 | |
| 566 | init(id: String, generation: Int64, polite: Bool, engine: MediaEngine) { |
| 567 | self.id = id |
| 568 | self.generation = generation |
| 569 | self.polite = polite |
| 570 | self.engine = engine |
| 571 | self.playAudio = engine.call?.playAudio ?? true |
| 572 | } |
| 573 | private func configuration(_ ice: ICEConfig) -> RTCConfiguration { |
| 574 | let c = RTCConfiguration() |
| 575 | c.sdpSemantics = .unifiedPlan |
| 576 | c.iceServers = ice.iceServers.map { |
| 577 | RTCIceServer(urlStrings: $0.urls, username: $0.username, credential: $0.credential) |
| 578 | } |
| 579 | return c |
| 580 | } |
| 581 | /// Makes the connection on the audio queue, as the first thing this |
| 582 | /// peer's work does: everything that arrives for it waits behind it. |
| 583 | /// Making one waits on WebRTC's worker thread, which can be busy |
| 584 | /// starting the audio engine for someone else (846 ms measured on the |
| 585 | /// main thread with a Bluetooth switch simulated). |
| 586 | func open(ice: ICEConfig) { |
| 587 | guard let engine else { return } |
| 588 | lastICE = ice |
| 589 | gatheredICE = ice |
| 590 | let factory = engine.factory |
| 591 | let config = configuration(ice) |
| 592 | let audio = engine.localAudio |
| 593 | let video = engine.localVideo |
| 594 | let delegate: RTCPeerConnectionDelegate = self |
| 595 | enqueue { [weak self] in |
| 596 | let (made, sent) = await AudioWork.run { () -> (RTCPeerConnection?, Bool) in |
| 597 | guard |
| 598 | let pc = factory.peerConnection( |
| 599 | with: config, constraints: RTCMediaConstraints(mandatoryConstraints: nil, optionalConstraints: nil), |
| 600 | delegate: delegate) |
| 601 | else { return (nil, false) } |
| 602 | if let audio { _ = pc.add(audio, streamIds: ["calls"]) } |
| 603 | let sent = video.map { pc.add($0, streamIds: ["calls"]) != nil } ?? false |
| 604 | return (pc, sent) |
| 605 | } |
| 606 | guard let self, !self.closed, let made else { |
| 607 | if let made { PeerCalls.close(made) } |
| 608 | return |
| 609 | } |
| 610 | self.pc = made |
| 611 | self.sentVideo = sent |
| 612 | if let limit = self.engine?.uplinkLimit { self.limitUplink(limit) } |
| 613 | // A connection built while this app is already sharing gets the |
| 614 | // screen with everything else: someone joining mid-share sees it. |
| 615 | // Last, as in the browser, so the camera and microphone take the |
| 616 | // m-lines the far side built for them and the screen a place of its |
| 617 | // own after them. |
| 618 | self.addScreen() |
| 619 | } |
| 620 | } |
| 621 | |
| 622 | /// Gives this connection the screen on a `sendonly` transceiver in a |
| 623 | /// stream of its own, which is what keeps it clear of the camera at the |
| 624 | /// far end. Adding it asks for negotiation, and perfect negotiation |
| 625 | /// carries the offer with no help from here. |
| 626 | /// |
| 627 | /// Only on a stable connection. A transceiver added mid-negotiation is |
| 628 | /// unassociated when the far side's offer arrives, and m-line matching |
| 629 | /// will hand it that offer's m-line — so two people sharing at once end |
| 630 | /// up each sending a screen in the place the other built for something |
| 631 | /// else. The browser's mutualshare gate found that; the deferred ones |
| 632 | /// are added when the connection settles. |
| 633 | /// |
| 634 | /// Added on the audio queue, as every call into the connection is; a |
| 635 | /// share that ended while it was being added is taken away again. |
| 636 | func addScreen() { |
| 637 | guard let pc, !closed, screenTx == nil, !addingScreen, let track = engine?.localScreen, let engine |
| 638 | else { return } |
| 639 | addingScreen = true |
| 640 | let factory = engine.factory |
| 641 | Task { @MainActor [weak self] in |
| 642 | let added = await PeerCalls.read(pc) { pc -> RTCRtpTransceiver? in |
| 643 | guard pc.signalingState == .stable else { return nil } |
| 644 | let setup = RTCRtpTransceiverInit() |
| 645 | setup.direction = .sendOnly |
| 646 | setup.streamIds = ["screen"] |
| 647 | guard let tx = pc.addTransceiver(with: track, init: setup) else { return nil } |
| 648 | try? tx.setCodecPreferences( |
| 649 | ScreenSendPolicy.preferringH264(factory.rtpSenderCapabilities(forKind: "video").codecs), error: ()) |
| 650 | return tx |
| 651 | } |
| 652 | guard let self else { return } |
| 653 | self.addingScreen = false |
| 654 | guard let added else { return } |
| 655 | self.screenTx = added |
| 656 | self.screenMidCache = nil |
| 657 | guard !self.closed, self.pc === pc, self.engine?.localScreen === track else { |
| 658 | self.removeScreen() |
| 659 | return |
| 660 | } |
| 661 | self.applySendPolicy() |
| 662 | } |
| 663 | } |
| 664 | private var addingScreen = false |
| 665 | |
| 666 | /// The screen at its mode's parameters and the camera capped beside |
| 667 | /// it, or the camera back at its defaults when there is no screen. |
| 668 | /// Sender parameters only: nothing here renegotiates. |
| 669 | /// |
| 670 | /// Applied on the audio queue: a sender's parameters are read and set on |
| 671 | /// WebRTC's worker thread, and this runs each time a negotiation settles |
| 672 | /// — just when the worker may be starting the audio engine. |
| 673 | func applySendPolicy() { |
| 674 | guard let pc, !closed, let engine else { return } |
| 675 | let screen = screenTx?.sender |
| 676 | var screenPlan = engine.screenPlan |
| 677 | if nudge > 1 { screenPlan.scaleResolutionDownBy = (screenPlan.scaleResolutionDownBy ?? 1) * nudge } |
| 678 | let cameraTrack = engine.localVideo |
| 679 | let camera = ScreenSendPolicy.camera(sharing: screenTx != nil) |
| 680 | AudioWork.post { |
| 681 | if let screen { screen.apply(screenPlan) } |
| 682 | let id = cameraTrack?.trackId |
| 683 | if let sender = pc.senders.first(where: { $0.track != nil && $0.track?.trackId == id }) { |
| 684 | sender.apply(camera) |
| 685 | } |
| 686 | } |
| 687 | } |
| 688 | |
| 689 | /// Ends this connection's screen m-line. Stopping the transceiver is |
| 690 | /// §5.4's own instruction and the only thing that makes the receiver's |
| 691 | /// track end rather than freeze on its last frame. |
| 692 | /// |
| 693 | /// The framework's stop is the immediate one: it stops the transceiver |
| 694 | /// and asks for nothing, where the browser's `stop()` also raises |
| 695 | /// negotiation-needed. So the offer that tells the far side is asked |
| 696 | /// for here, by hand. |
| 697 | func removeScreen() { |
| 698 | guard let tx = screenTx else { return } |
| 699 | screenTx = nil |
| 700 | screenMidCache = nil |
| 701 | AudioWork.post { tx.stopInternal() } |
| 702 | applySendPolicy() |
| 703 | renegotiate() |
| 704 | } |
| 705 | |
| 706 | /// An offer this side has to make for a reason negotiation-needed will |
| 707 | /// not announce. Made now if the connection is quiet, and otherwise |
| 708 | /// owed, and made the next time it settles. |
| 709 | private func renegotiate() { |
| 710 | guard let pc, !closed else { return } |
| 711 | Task { @MainActor [weak self] in |
| 712 | guard await PeerCalls.read(pc, { $0.signalingState }) == .stable else { |
| 713 | self?.owesOffer = true |
| 714 | return |
| 715 | } |
| 716 | guard let self else { return } |
| 717 | if self.polite { try? await Task.sleep(nanoseconds: 200_000_000) } |
| 718 | self.enqueue { [weak self] in |
| 719 | guard let self, let pc = self.pc else { return } |
| 720 | if await PeerCalls.read(pc, { $0.signalingState }) != .stable { |
| 721 | self.owesOffer = true |
| 722 | return |
| 723 | } |
| 724 | try await self.offer() |
| 725 | } |
| 726 | } |
| 727 | } |
| 728 | |
| 729 | /// The connection is back at stable: the one moment a deferred screen |
| 730 | /// can be added, and an owed offer made. |
| 731 | private func settled() { |
| 732 | guard let pc, !closed else { return } |
| 733 | Task { @MainActor [weak self] in |
| 734 | // Read on the audio queue, as every call into the connection is. |
| 735 | let stable = await PeerCalls.read(pc) { $0.signalingState } == .stable |
| 736 | guard let self, stable, !self.closed, self.pc === pc else { return } |
| 737 | self.addScreen() |
| 738 | self.applySendPolicy() |
| 739 | if self.owesOffer { |
| 740 | self.owesOffer = false |
| 741 | self.renegotiate() |
| 742 | } |
| 743 | } |
| 744 | } |
| 745 | func isScreen(mid: String) -> Bool { !mid.isEmpty && mid == screenMid } |
| 746 | var screenMidHere: String { screenMid } |
| 747 | /// The mid our screen went out on, once negotiated: read on the audio |
| 748 | /// queue (a transceiver's mid is a hop to WebRTC's signalling thread), |
| 749 | /// and kept once known. |
| 750 | func ownScreenMid() async -> String? { |
| 751 | if let screenMidCache { return screenMidCache } |
| 752 | guard let pc, let tx = screenTx else { return nil } |
| 753 | let mid = await PeerCalls.read(pc) { _ in tx.mid } |
| 754 | guard !mid.isEmpty, screenTx === tx else { return nil } |
| 755 | screenMidCache = mid |
| 756 | return mid |
| 757 | } |
| 758 | private var screenMidCache: String? |
| 759 | func limitUplink(_ bps: Int?) { |
| 760 | guard let pc else { return } |
| 761 | AudioWork.post { |
| 762 | _ = pc.setBweMinBitrateBps(nil, currentBitrateBps: nil, maxBitrateBps: bps.map { NSNumber(value: $0) }) |
| 763 | } |
| 764 | } |
| 765 | |
| 766 | /// The `share` message: on draws the captioned tile ahead of the media, |
| 767 | /// off is one of §5.4's two endings. |
| 768 | func share(on: Bool) { |
| 769 | if on { |
| 770 | clues.sharing = true |
| 771 | engine?.call?.announceScreen(id) |
| 772 | } else { |
| 773 | dropScreen() |
| 774 | } |
| 775 | } |
| 776 | private func dropScreen() { |
| 777 | clues.sharing = false |
| 778 | screenReceiver = nil |
| 779 | screenMid = "" |
| 780 | engine?.call?.dropScreen(id) |
| 781 | } |
| 782 | |
| 783 | /// One track arriving, put in its place. A camera goes on the peer's |
| 784 | /// tile exactly as it always did; a screen gets a tile of its own and |
| 785 | /// leaves the camera alone. |
| 786 | private func arrived( |
| 787 | _ track: RTCMediaStreamTrack, receiver: String, info: ArrivingTrack |
| 788 | ) { |
| 789 | let role = classifyTrack(clues, info, hasVideo: sentVideo) |
| 790 | if role == .screen, let video = track as? RTCVideoTrack { |
| 791 | clues.sharing = true |
| 792 | screenReceiver = receiver |
| 793 | if !info.mid.isEmpty { screenMid = info.mid } |
| 794 | engine?.call?.setScreen(video, peer: id) |
| 795 | return |
| 796 | } |
| 797 | if role == .camera { |
| 798 | if !info.mid.isEmpty { clues.cameraMid = info.mid } |
| 799 | if !info.streamId.isEmpty { clues.cameraStream = info.streamId } |
| 800 | } |
| 801 | // Recorded from the first track of any kind: it is what places a |
| 802 | // video track for a peer that has no camera. |
| 803 | if clues.peerStream.isEmpty { clues.peerStream = info.streamId } |
| 804 | if role == .camera, let video = track as? RTCVideoTrack { |
| 805 | engine?.call?.setRemote(video, peer: id) |
| 806 | } |
| 807 | if let audio = track as? RTCAudioTrack, !tapped.contains(where: { $0 === audio }) { |
| 808 | audio.add(tap) |
| 809 | tapped.append(audio) |
| 810 | } |
| 811 | } |
| 812 | |
| 813 | func close() { |
| 814 | closed = true |
| 815 | for audio in tapped { audio.remove(tap) } |
| 816 | tapped = [] |
| 817 | screenTx = nil |
| 818 | work?.cancel() |
| 819 | work = nil |
| 820 | // Closing can stop the audio engine (the last connection), which |
| 821 | // takes as long as starting it: on the audio queue. |
| 822 | if let pc { PeerCalls.close(pc) } |
| 823 | pc = nil |
| 824 | candidates = [] |
| 825 | } |
| 826 | private func enqueue(_ action: @escaping @MainActor () async throws -> Void) { |
| 827 | let previous = work |
| 828 | work = Task { [weak self] in |
| 829 | await previous?.value |
| 830 | guard let self, !self.closed, !Task.isCancelled else { return } |
| 831 | do { try await action() } catch { if !self.closed { self.engine?.call?.mediaFailed() } } |
| 832 | } |
| 833 | } |
| 834 | private func send(_ type: String, _ payload: [String: Any]) { |
| 835 | guard !closed, let body = json(payload) else { return } |
| 836 | engine?.call?.send(type: type, to: id, gen: generation, payload: body) |
| 837 | } |
| 838 | // Descriptions and candidates are applied from the audio queue |
| 839 | // (PeerCalls). Applying one waits for WebRTC's worker thread, which |
| 840 | // starts the audio engine for the first stream in each direction — a |
| 841 | // Bluetooth headset's profile switch included — and up to 0.3.8 the |
| 842 | // main thread waited with it. |
| 843 | private func setLocal(_ desc: RTCSessionDescription) async throws { |
| 844 | guard let pc, !closed else { return } |
| 845 | try await PeerCalls.set(pc, local: true, desc) |
| 846 | } |
| 847 | private func setRemote(_ desc: RTCSessionDescription) async throws { |
| 848 | guard let pc, !closed else { return } |
| 849 | try await PeerCalls.set(pc, local: false, desc) |
| 850 | } |
| 851 | private func makeDescription(answer: Bool, restart: Bool = false) async throws |
| 852 | -> RTCSessionDescription |
| 853 | { |
| 854 | guard let pc else { throw CancellationError() } |
| 855 | let constraints = RTCMediaConstraints( |
| 856 | mandatoryConstraints: restart ? ["IceRestart": "true"] : nil, optionalConstraints: nil) |
| 857 | return try await PeerCalls.describe(pc, answer: answer, constraints) |
| 858 | } |
| 859 | private func offer(restart: Bool = false) async throws { |
| 860 | guard let pc, !closed, await PeerCalls.read(pc, { $0.signalingState }) == .stable, !closed else { return } |
| 861 | let desc = try await makeDescription(answer: false, restart: restart) |
| 862 | guard !closed else { return } |
| 863 | try await setLocal(desc) |
| 864 | send("offer", ["description": ["type": "offer", "sdp": desc.sdp]]) |
| 865 | } |
| 866 | func receive(_ message: Message) { |
| 867 | guard message.gen == generation else { return } |
| 868 | enqueue { [weak self] in |
| 869 | guard let self, let pc = self.pc, !self.closed else { return } |
| 870 | if message.type == "candidate", let candidate = message.payload?.candidate { |
| 871 | if self.ignoreOffer { return } |
| 872 | if await PeerCalls.read(pc, { $0.remoteDescription == nil }) { |
| 873 | guard self.candidates.count < 256 else { throw CancellationError() } |
| 874 | self.candidates.append(candidate) |
| 875 | } else { |
| 876 | try await self.add(candidate) |
| 877 | } |
| 878 | return |
| 879 | } |
| 880 | guard let desc = message.payload?.description else { return } |
| 881 | let signaling = await PeerCalls.read(pc) { $0.signalingState } |
| 882 | guard !self.closed else { return } |
| 883 | if message.type == "offer", desc.type == "offer" { |
| 884 | let collision = signaling != .stable |
| 885 | self.ignoreOffer = collision && !self.polite |
| 886 | if self.ignoreOffer { return } |
| 887 | if collision { try await self.setLocal(RTCSessionDescription(type: .rollback, sdp: "")) } |
| 888 | try await self.setRemote(RTCSessionDescription(type: .offer, sdp: desc.sdp)) |
| 889 | self.ignoreOffer = false |
| 890 | try await self.drain() |
| 891 | let answer = try await self.makeDescription(answer: true) |
| 892 | try await self.setLocal(answer) |
| 893 | self.send("answer", ["description": ["type": "answer", "sdp": answer.sdp]]) |
| 894 | } else if message.type == "answer", desc.type == "answer", signaling == .haveLocalOffer { |
| 895 | try await self.setRemote(RTCSessionDescription(type: .answer, sdp: desc.sdp)) |
| 896 | self.ignoreOffer = false |
| 897 | try await self.drain() |
| 898 | } |
| 899 | } |
| 900 | } |
| 901 | private func add(_ candidate: Candidate) async throws { |
| 902 | guard let pc, !closed, !candidate.candidate.isEmpty else { return } |
| 903 | let ice = RTCIceCandidate( |
| 904 | sdp: candidate.candidate, sdpMLineIndex: candidate.sdpMLineIndex ?? 0, |
| 905 | sdpMid: candidate.sdpMid) |
| 906 | try await PeerCalls.add(pc, ice) |
| 907 | } |
| 908 | private func drain() async throws { |
| 909 | let saved = candidates |
| 910 | candidates = [] |
| 911 | for c in saved { try await add(c) } |
| 912 | } |
| 913 | /// A fresh `/ice` answer (§5.2, §5.3). The core fetches one on every |
| 914 | /// roster, and a roster answers every stream resume — every five |
| 915 | /// minutes at the stream's own TTL — so most answers are not renewals |
| 916 | /// at all. Up to 0.4.1 every one of them restarted ICE on every |
| 917 | /// connection, twelve times an hour, each one a new path for the |
| 918 | /// screen and camera to find their way back onto. Now only a renewal |
| 919 | /// that is due restarts (ICERenewal.restart); anything else is adopted |
| 920 | /// quietly, for whatever this connection gathers next. |
| 921 | func updateICE(_ ice: ICEConfig, now: Int64) { |
| 922 | guard ice.renewAt != lastICE?.renewAt || ice.iceServers.count != lastICE?.iceServers.count else { return } |
| 923 | lastICE = ice |
| 924 | let restart = ICERenewal.restart(gathered: gatheredICE, next: ice, now: now) |
| 925 | if restart { gatheredICE = ice } |
| 926 | enqueue { [weak self] in |
| 927 | guard let self, let pc = self.pc else { return } |
| 928 | let config = self.configuration(ice) |
| 929 | guard await PeerCalls.read(pc, { $0.setConfiguration(config) }) else { return } |
| 930 | if restart { |
| 931 | ScreenLog.info("ICE restart with \(ScreenWatchdog.short(self.id)): renewal due") |
| 932 | self.engine?.iceRestarts += 1 |
| 933 | try await self.offer(restart: true) |
| 934 | } |
| 935 | } |
| 936 | } |
| 937 | |
| 938 | // ------------------------------------------------------------- |
| 939 | // The screen watchdog's hands, for this connection |
| 940 | // ------------------------------------------------------------- |
| 941 | |
| 942 | /// Whether this peer's screen is arriving here. |
| 943 | var receivingScreen: Bool { screenReceiver != nil && !screenMid.isEmpty } |
| 944 | |
| 945 | /// The screen's encoding switched off and on again. WebRTC starts a |
| 946 | /// stream that comes back on with a keyframe, which is the one thing a |
| 947 | /// receiver stuck on a broken picture needs; there is no API on this |
| 948 | /// build of WebRTC to ask for one directly. |
| 949 | func forceScreenKeyframe() { |
| 950 | setScreenActive(false) |
| 951 | Task { @MainActor [weak self] in |
| 952 | try? await Task.sleep(nanoseconds: 150_000_000) |
| 953 | self?.setScreenActive(true) |
| 954 | } |
| 955 | } |
| 956 | |
| 957 | func setScreenActive(_ active: Bool) { |
| 958 | guard let sender = screenTx?.sender else { return } |
| 959 | AudioWork.post { |
| 960 | let parameters = sender.parameters |
| 961 | guard let encoding = parameters.encodings.first, encoding.isActive != active else { return } |
| 962 | encoding.isActive = active |
| 963 | sender.parameters = parameters |
| 964 | } |
| 965 | } |
| 966 | |
| 967 | /// A fresh encoder: the screen sent smaller for a while. A change of |
| 968 | /// size is what makes WebRTC set its encoder up again from nothing — for |
| 969 | /// H.264 on a Mac, a new VideoToolbox session, which is the part that |
| 970 | /// can go bad under a sleep or a GPU switch and hold the frames it was |
| 971 | /// given. The other way an encoder stops is a frame it cannot take at |
| 972 | /// all (the level's block count, see ScreenSendPolicy), which a smaller |
| 973 | /// size is also the way out of. A third off each side each time, down to |
| 974 | /// a quarter; full size again after thirty seconds, and if that stops it |
| 975 | /// again the watchdog comes back round, after its rest. |
| 976 | func recreateScreenEncoder() { |
| 977 | guard screenTx != nil else { return } |
| 978 | nudge = min(nudge * 1.5, 4) |
| 979 | nudges += 1 |
| 980 | let mine = nudges |
| 981 | ScreenLog.info("screen to \(ScreenWatchdog.short(id)) now sent at 1/\(String(format: "%.2f", nudge)) of its size") |
| 982 | applySendPolicy() |
| 983 | setScreenActive(true) |
| 984 | Task { @MainActor [weak self] in |
| 985 | try? await Task.sleep(nanoseconds: 30_000_000_000) |
| 986 | guard let self, self.nudges == mine, self.nudge > 1 else { return } |
| 987 | self.nudge = 1 |
| 988 | ScreenLog.info("screen to \(ScreenWatchdog.short(self.id)) back to full size") |
| 989 | self.applySendPolicy() |
| 990 | } |
| 991 | } |
| 992 | private var nudges = 0 |
| 993 | private var nudge = 1.0 |
| 994 | |
| 995 | /// What stopping and sharing again did, on this connection alone: its |
| 996 | /// screen transceiver stopped and renegotiated, and a new one added as |
| 997 | /// soon as that settles (`settled` adds the screen to a connection that |
| 998 | /// has none while the app is sharing). |
| 999 | func reAddScreen() { |
| 1000 | guard screenTx != nil else { return } |
| 1001 | removeScreen() |
| 1002 | } |
| 1003 | /// The share has stopped: the next one starts at full size. |
| 1004 | func screenStopped() { nudge = 1 } |
| 1005 | |
| 1006 | nonisolated func peerConnectionShouldNegotiate(_ peerConnection: RTCPeerConnection) { |
| 1007 | Task { @MainActor [weak self] in |
| 1008 | guard let self else { return } |
| 1009 | // Give the higher-id peer first chance, matching the browser's |
| 1010 | // polite yield. Perfect negotiation still handles actual glare. |
| 1011 | if self.polite { try? await Task.sleep(nanoseconds: 200_000_000) } |
| 1012 | self.enqueue { [weak self] in try await self?.offer() } |
| 1013 | } |
| 1014 | } |
| 1015 | nonisolated func peerConnection( |
| 1016 | _ peerConnection: RTCPeerConnection, didGenerate candidate: RTCIceCandidate |
| 1017 | ) { |
| 1018 | Task { @MainActor [weak self] in |
| 1019 | self?.send( |
| 1020 | "candidate", |
| 1021 | [ |
| 1022 | "candidate": [ |
| 1023 | "candidate": candidate.sdp, "sdpMid": candidate.sdpMid as Any? ?? NSNull(), |
| 1024 | "sdpMLineIndex": candidate.sdpMLineIndex, |
| 1025 | ] |
| 1026 | ]) |
| 1027 | } |
| 1028 | } |
| 1029 | nonisolated func peerConnection( |
| 1030 | _ peerConnection: RTCPeerConnection, didAdd rtpReceiver: RTCRtpReceiver, |
| 1031 | streams: [RTCMediaStream] |
| 1032 | ) { |
| 1033 | // Keep decoding and RTP statistics available when a test disables speakers. |
| 1034 | if let audio = rtpReceiver.track as? RTCAudioTrack, !playAudio { |
| 1035 | audio.source.volume = 0 |
| 1036 | } |
| 1037 | guard let track = rtpReceiver.track else { return } |
| 1038 | // Read here, on the signalling thread this callback arrives on, rather |
| 1039 | // than on the main actor: by the time the main actor gets to it the |
| 1040 | // next description can have moved the transceiver on. |
| 1041 | let receiver = rtpReceiver.receiverId |
| 1042 | let transceiver = peerConnection.transceivers.first { $0.receiver.receiverId == receiver } |
| 1043 | var info = ArrivingTrack(kind: track.kind, streamId: streams.first?.streamId ?? "") |
| 1044 | if let transceiver { |
| 1045 | info.mid = transceiver.mid |
| 1046 | // The negotiated direction when there is one. On the side that |
| 1047 | // answers an offer there is not yet, and that is exactly the side a |
| 1048 | // screen arrives at; the direction the transceiver was built with is |
| 1049 | // recvonly for an m-line the far side added, which is the point. |
| 1050 | var current = RTCRtpTransceiverDirection.inactive |
| 1051 | info.direction = sdpDirection( |
| 1052 | transceiver.currentDirection(¤t) ? current : transceiver.direction) |
| 1053 | } |
| 1054 | Task { @MainActor [weak self] in |
| 1055 | guard let self, !self.closed else { return } |
| 1056 | self.arrived(track, receiver: receiver, info: info) |
| 1057 | } |
| 1058 | } |
| 1059 | /// §5.4's other ending: the sender stopped its screen's transceiver and |
| 1060 | /// the renegotiation took the receiver away. Only the receiver the |
| 1061 | /// screen is showing from counts. A screen replaced on its own mid ends |
| 1062 | /// its predecessor afterwards, and without the check that took the live |
| 1063 | /// replacement's tile with it (the browser's Codex S4-15). |
| 1064 | nonisolated func peerConnection( |
| 1065 | _ peerConnection: RTCPeerConnection, didRemove rtpReceiver: RTCRtpReceiver |
| 1066 | ) { |
| 1067 | let receiver = rtpReceiver.receiverId |
| 1068 | Task { @MainActor [weak self] in |
| 1069 | guard let self, !self.closed, self.screenReceiver == receiver else { return } |
| 1070 | self.dropScreen() |
| 1071 | } |
| 1072 | } |
| 1073 | nonisolated func peerConnection( |
| 1074 | _ peerConnection: RTCPeerConnection, didChange newState: RTCIceConnectionState |
| 1075 | ) { |
| 1076 | if newState == .failed { Task { @MainActor [weak self] in self?.engine?.call?.mediaFailed() } } |
| 1077 | } |
| 1078 | nonisolated func peerConnection( |
| 1079 | _ peerConnection: RTCPeerConnection, didChange stateChanged: RTCSignalingState |
| 1080 | ) { |
| 1081 | guard stateChanged == .stable else { return } |
| 1082 | Task { @MainActor [weak self] in self?.settled() } |
| 1083 | } |
| 1084 | nonisolated func peerConnection( |
| 1085 | _ peerConnection: RTCPeerConnection, didAdd stream: RTCMediaStream |
| 1086 | ) {} |
| 1087 | nonisolated func peerConnection( |
| 1088 | _ peerConnection: RTCPeerConnection, didRemove stream: RTCMediaStream |
| 1089 | ) {} |
| 1090 | nonisolated func peerConnection( |
| 1091 | _ peerConnection: RTCPeerConnection, didChange newState: RTCIceGatheringState |
| 1092 | ) {} |
| 1093 | nonisolated func peerConnection( |
| 1094 | _ peerConnection: RTCPeerConnection, didRemove candidates: [RTCIceCandidate] |
| 1095 | ) {} |
| 1096 | nonisolated func peerConnection( |
| 1097 | _ peerConnection: RTCPeerConnection, didOpen dataChannel: RTCDataChannel |
| 1098 | ) {} |
| 1099 | } |
| 1100 | |
amadan