Revamp broadcast IPC

Sources/LiveKit/Broadcast/BroadcastBundleInfo.swift

@@ -29,7 +29,7 @@ final class BroadcastBundleInfo {
     static var screenSharingExtension: String?
 
     /// Path to the socket file used for interprocess communication.
-    static var socketPath: String? {
+    static var socketPath: SocketPath? {
         guard let groupIdentifier else { return nil }
         return Self.socketPath(for: groupIdentifier)
     }
@@ -41,11 +41,12 @@ final class BroadcastBundleInfo {
 
     private static let socketFileDescriptor = "rtc_SSFD"
 
-    private static func socketPath(for groupIdentifier: String) -> String? {
+    private static func socketPath(for groupIdentifier: String) -> SocketPath? {
         guard let sharedContainer = FileManager.default
             .containerURL(forSecurityApplicationGroupIdentifier: groupIdentifier)
         else { return nil }
-        return sharedContainer.appendingPathComponent(Self.socketFileDescriptor).path
+        let path = sharedContainer.appendingPathComponent(Self.socketFileDescriptor).path
+        return SocketPath(path)
     }
 }
 

Sources/LiveKit/Broadcast/BroadcastScreenCapturer.swift

@@ -29,18 +29,13 @@ internal import LiveKitWebRTC
 #endif
 
 class BroadcastScreenCapturer: BufferCapturer {
-    var frameReader: SocketConnectionFrameReader?
+    private var receiver: BroadcastReceiver?
 
     override func startCapture() async throws -> Bool {
         let didStart = try await super.startCapture()
 
         guard didStart else { return false }
 
-        guard let socketPath = BroadcastBundleInfo.socketPath else {
-            logger.error("Bundle settings improperly configured for screen capture")
-            return false
-        }
-
         let bounds = await UIScreen.main.bounds
         let width = bounds.size.width
         let height = bounds.size.height
@@ -53,22 +48,32 @@ class BroadcastScreenCapturer: BufferCapturer {
             .toEncodeSafeDimensions()
 
         set(dimensions: targetDimensions)
+        return createReceiver()
+    }
 
-        let frameReader = SocketConnectionFrameReader()
-        guard let socketConnection = BroadcastServerSocketConnection(filePath: socketPath, streamDelegate: frameReader)
-        else { return false }
-        frameReader.didCapture = { pixelBuffer, rotation in
-            self.capture(pixelBuffer, rotation: rotation.toLKType())
+    private func createReceiver() -> Bool {
+        guard let socketPath = BroadcastBundleInfo.socketPath else {
+            logger.error("Bundle settings improperly configured for screen capture")
+            return false
         }
-        frameReader.didEnd = { [weak self] in
-            guard let self else { return }
-            Task {
-                try await self.stopCapture()
+        Task { [weak self] in
+            do {
+                let receiver = try await BroadcastReceiver(socketPath: socketPath)
+                logger.debug("Broadcast receiver connected")
+                self?.receiver = receiver
+
+                for try await sample in receiver.incomingSamples {
+                    switch sample {
+                    case let .image(imageBuffer, rotation):
+                        self?.capture(imageBuffer, rotation: rotation)
+                    }
+                }
+                logger.debug("Broadcast receiver closed")
+            } catch {
+                logger.error("Broadcast receiver error: \(error)")
             }
+            _ = try? await self?.stopCapture()
         }
-        frameReader.startCapture(with: socketConnection)
-        self.frameReader = frameReader
-
         return true
     }
 
@@ -77,9 +82,7 @@ class BroadcastScreenCapturer: BufferCapturer {
 
         // Already stopped
         guard didStop else { return false }
-
-        frameReader?.stopCapture()
-        frameReader = nil
+        receiver?.close()
         return true
     }
 }

Sources/LiveKit/Broadcast/IPC/BroadcastIPCHeader.swift

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2025 LiveKit
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#if os(iOS)
+
+/// Message header for communication between uploader and receiver.
+enum BroadcastIPCHeader: Codable {
+    /// Image sample sent by uploader.
+    case image(BroadcastImageCodec.Metadata, VideoRotation)
+}
+
+#endif