| 1 | import CoreVideo |
| 2 | import Foundation |
| 3 | |
| 4 | /// The app's half: listens for the extension for as long as a call is up, |
| 5 | /// and hands each connection to the call. |
| 6 | /// |
| 7 | /// One broadcast at a time is all iOS allows, so one connection at a time |
| 8 | /// is all this takes; a second, which would be a bug, is closed at once. |
| 9 | public final class BroadcastInbox: @unchecked Sendable { |
| 10 | private let path: String |
| 11 | private let queue = DispatchQueue(label: "calls.broadcast.inbox") |
| 12 | private var listener: DispatchSourceRead? |
| 13 | private var current: BroadcastConnection? |
| 14 | |
| 15 | /// Called on the main queue for each extension that connects. |
| 16 | public var onConnection: ((BroadcastConnection) -> Void)? |
| 17 | |
| 18 | public init(path: String) { self.path = path } |
| 19 | |
| 20 | public func start() throws { |
| 21 | let fd = try BroadcastSocket.listen(path: path) |
| 22 | BroadcastSocket.nonBlocking(fd) |
| 23 | let source = DispatchSource.makeReadSource(fileDescriptor: fd, queue: queue) |
| 24 | source.setEventHandler { [weak self] in self?.accept(fd) } |
| 25 | source.setCancelHandler { Darwin.close(fd) } |
| 26 | listener = source |
| 27 | source.resume() |
| 28 | } |
| 29 | |
| 30 | /// Stops listening. A connection already handed out is the call's to |
| 31 | /// close, not this. |
| 32 | public func stop() { |
| 33 | queue.sync { |
| 34 | listener?.cancel() |
| 35 | listener = nil |
| 36 | current = nil |
| 37 | } |
| 38 | unlink(path) |
| 39 | } |
| 40 | |
| 41 | private func accept(_ fd: Int32) { |
| 42 | let client = Darwin.accept(fd, nil, nil) |
| 43 | guard client >= 0 else { return } |
| 44 | if let current, !current.isClosed { |
| 45 | Darwin.close(client) |
| 46 | return |
| 47 | } |
| 48 | var on: Int32 = 1 |
| 49 | setsockopt(client, SOL_SOCKET, SO_NOSIGPIPE, &on, socklen_t(MemoryLayout<Int32>.size)) |
| 50 | let connection = BroadcastConnection(fd: client) |
| 51 | current = connection |
| 52 | DispatchQueue.main.async { [weak self] in |
| 53 | guard let self, let onConnection = self.onConnection else { |
| 54 | connection.close() |
| 55 | return |
| 56 | } |
| 57 | onConnection(connection) |
| 58 | } |
| 59 | } |
| 60 | } |
| 61 | |
| 62 | /// One broadcast: frames in, until either side closes. |
| 63 | public final class BroadcastConnection: @unchecked Sendable { |
| 64 | private let fd: Int32 |
| 65 | private let queue = DispatchQueue(label: "calls.broadcast.read", qos: .userInteractive) |
| 66 | private var source: DispatchSourceRead? |
| 67 | private let lock = NSLock() |
| 68 | private var closed = false |
| 69 | private var header = [UInt8](repeating: 0, count: BroadcastFrameHeader.size) |
| 70 | private var headerRead = 0 |
| 71 | private var frame: BroadcastFrameHeader? |
| 72 | private var payload: UnsafeMutableRawPointer? |
| 73 | private var payloadCapacity = 0 |
| 74 | private var payloadRead = 0 |
| 75 | private var pool: CVPixelBufferPool? |
| 76 | private var poolShape: (Int, Int, Bool) = (0, 0, false) |
| 77 | |
| 78 | /// Each complete frame, as an NV12 pixel buffer the receiver may keep, |
| 79 | /// with its CGImagePropertyOrientation. Called on the reading queue. |
| 80 | public var onFrame: ((CVPixelBuffer, UInt16) -> Void)? |
| 81 | /// Called once, on the reading queue, when the extension has gone — the |
| 82 | /// broadcast stopped from Control Center or the status bar, or the |
| 83 | /// extension ended. Not called for `close()`. |
| 84 | public var onClose: (() -> Void)? |
| 85 | |
| 86 | init(fd: Int32) { self.fd = fd } |
| 87 | |
| 88 | deinit { payload?.deallocate() } |
| 89 | |
| 90 | var isClosed: Bool { |
| 91 | lock.lock() |
| 92 | defer { lock.unlock() } |
| 93 | return closed |
| 94 | } |
| 95 | |
| 96 | /// Starts reading. Separate from creation so the reader is only live |
| 97 | /// once somebody is listening for its frames. |
| 98 | public func resume() { |
| 99 | BroadcastSocket.nonBlocking(fd) |
| 100 | let source = DispatchSource.makeReadSource(fileDescriptor: fd, queue: queue) |
| 101 | source.setEventHandler { [weak self] in self?.readable() } |
| 102 | source.setCancelHandler { [fd] in Darwin.close(fd) } |
| 103 | self.source = source |
| 104 | source.resume() |
| 105 | } |
| 106 | |
| 107 | /// Closes from this side. The extension's next write fails, and it ends |
| 108 | /// the broadcast — which is how "stop sharing" in the app reaches the |
| 109 | /// system's recording indicator. |
| 110 | public func close() { |
| 111 | lock.lock() |
| 112 | let first = !closed |
| 113 | closed = true |
| 114 | lock.unlock() |
| 115 | guard first else { return } |
| 116 | if let source { |
| 117 | source.cancel() |
| 118 | } else { |
| 119 | Darwin.close(fd) |
| 120 | } |
| 121 | } |
| 122 | |
| 123 | private func ended() { |
| 124 | lock.lock() |
| 125 | let first = !closed |
| 126 | closed = true |
| 127 | lock.unlock() |
| 128 | guard first else { return } |
| 129 | source?.cancel() |
| 130 | onClose?() |
| 131 | } |
| 132 | |
| 133 | private func readable() { |
| 134 | while !isClosed { |
| 135 | if frame == nil { |
| 136 | let n = header.withUnsafeMutableBytes { |
| 137 | Darwin.read(fd, $0.baseAddress!.advanced(by: headerRead), BroadcastFrameHeader.size - headerRead) |
| 138 | } |
| 139 | guard let n = settle(n) else { return } |
| 140 | headerRead += n |
| 141 | guard headerRead == BroadcastFrameHeader.size else { continue } |
| 142 | guard let next = header.withUnsafeBytes({ BroadcastFrameHeader.read($0.baseAddress!) }) else { |
| 143 | ended() |
| 144 | return |
| 145 | } |
| 146 | frame = next |
| 147 | payloadRead = 0 |
| 148 | if payloadCapacity < next.payloadLength { |
| 149 | payload?.deallocate() |
| 150 | payload = UnsafeMutableRawPointer.allocate(byteCount: next.payloadLength, alignment: 16) |
| 151 | payloadCapacity = next.payloadLength |
| 152 | } |
| 153 | } |
| 154 | guard let frame, let payload else { return } |
| 155 | let n = Darwin.read(fd, payload.advanced(by: payloadRead), frame.payloadLength - payloadRead) |
| 156 | guard let n = settle(n) else { return } |
| 157 | payloadRead += n |
| 158 | guard payloadRead == frame.payloadLength else { continue } |
| 159 | self.frame = nil |
| 160 | headerRead = 0 |
| 161 | if let pixels = picture(frame, payload) { onFrame?(pixels, frame.orientation) } |
| 162 | } |
| 163 | } |
| 164 | |
| 165 | /// A read's answer: bytes to count, nil to stop reading for now, and the |
| 166 | /// connection ended on end-of-file or a real error. |
| 167 | private func settle(_ n: Int) -> Int? { |
| 168 | if n > 0 { return n } |
| 169 | if n < 0 && (errno == EAGAIN || errno == EINTR) { return nil } |
| 170 | ended() |
| 171 | return nil |
| 172 | } |
| 173 | |
| 174 | /// The packed frame, copied into a pixel buffer from a pool, row by row: |
| 175 | /// the pool's rows are padded and the wire's are not. |
| 176 | private func picture(_ frame: BroadcastFrameHeader, _ bytes: UnsafeMutableRawPointer) |
| 177 | -> CVPixelBuffer? |
| 178 | { |
| 179 | let shape = (frame.width, frame.height, frame.videoRange) |
| 180 | if pool == nil || poolShape != shape { |
| 181 | let format = |
| 182 | frame.videoRange |
| 183 | ? kCVPixelFormatType_420YpCbCr8BiPlanarVideoRange |
| 184 | : kCVPixelFormatType_420YpCbCr8BiPlanarFullRange |
| 185 | let attributes: [CFString: Any] = [ |
| 186 | kCVPixelBufferPixelFormatTypeKey: format, |
| 187 | kCVPixelBufferWidthKey: frame.width, |
| 188 | kCVPixelBufferHeightKey: frame.height, |
| 189 | kCVPixelBufferIOSurfacePropertiesKey: [:] as [CFString: Any], |
| 190 | ] |
| 191 | pool = nil |
| 192 | CVPixelBufferPoolCreate(nil, nil, attributes as CFDictionary, &pool) |
| 193 | poolShape = shape |
| 194 | } |
| 195 | guard let pool else { return nil } |
| 196 | var made: CVPixelBuffer? |
| 197 | CVPixelBufferPoolCreatePixelBuffer(nil, pool, &made) |
| 198 | guard let pixels = made else { return nil } |
| 199 | CVPixelBufferLockBaseAddress(pixels, []) |
| 200 | defer { CVPixelBufferUnlockBaseAddress(pixels, []) } |
| 201 | let planes = [ |
| 202 | (bytes, frame.height), |
| 203 | (bytes.advanced(by: frame.width * frame.height), frame.height / 2), |
| 204 | ] |
| 205 | for (plane, (source, rows)) in planes.enumerated() { |
| 206 | guard let target = CVPixelBufferGetBaseAddressOfPlane(pixels, plane) else { return nil } |
| 207 | let stride = CVPixelBufferGetBytesPerRowOfPlane(pixels, plane) |
| 208 | for row in 0..<rows { |
| 209 | target.advanced(by: row * stride).copyMemory( |
| 210 | from: source.advanced(by: row * frame.width), byteCount: frame.width) |
| 211 | } |
| 212 | } |
| 213 | return pixels |
| 214 | } |
| 215 | } |
| 216 | |
amadan