Data streams

LiveKitClient.podspec

@@ -19,6 +19,7 @@ Pod::Spec.new do |spec|
   spec.dependency("LiveKitWebRTC", "= 125.6422.19")
   spec.dependency("SwiftProtobuf")
   spec.dependency("Logging")
+  spec.dependency("DequeModule", "= 1.1.4")
 
   spec.resource_bundles = {"Privacy" => ["Sources/LiveKit/PrivacyInfo.xcprivacy"]}
 

Package.swift

@@ -21,6 +21,7 @@ let package = Package(
         .package(url: "https://github.com/livekit/webrtc-xcframework.git", exact: "125.6422.19"),
         .package(url: "https://github.com/apple/swift-protobuf.git", from: "1.26.0"),
         .package(url: "https://github.com/apple/swift-log.git", from: "1.5.4"),
+        .package(url: "https://github.com/apple/swift-collections.git", from: "1.1.0"),
         // Only used for DocC generation
         .package(url: "https://github.com/apple/swift-docc-plugin.git", from: "1.3.0"),
         // Only used for Testing
@@ -36,6 +37,7 @@ let package = Package(
             dependencies: [
                 .product(name: "LiveKitWebRTC", package: "webrtc-xcframework"),
                 .product(name: "SwiftProtobuf", package: "swift-protobuf"),
+                .product(name: "DequeModule", package: "swift-collections"),
                 .product(name: "Logging", package: "swift-log"),
                 "LKObjCHelpers",
             ],

Package@swift-5.9.swift

@@ -23,6 +23,7 @@ let package = Package(
         .package(url: "https://github.com/livekit/webrtc-xcframework.git", exact: "125.6422.19"),
         .package(url: "https://github.com/apple/swift-protobuf.git", from: "1.26.0"),
         .package(url: "https://github.com/apple/swift-log.git", from: "1.5.4"),
+        .package(url: "https://github.com/apple/swift-collections.git", from: "1.1.0"),
         // Only used for DocC generation
         .package(url: "https://github.com/apple/swift-docc-plugin.git", from: "1.3.0"),
         // Only used for Testing
@@ -38,6 +39,7 @@ let package = Package(
             dependencies: [
                 .product(name: "LiveKitWebRTC", package: "webrtc-xcframework"),
                 .product(name: "SwiftProtobuf", package: "swift-protobuf"),
+                .product(name: "DequeModule", package: "swift-collections"),
                 .product(name: "Logging", package: "swift-log"),
                 "LKObjCHelpers",
             ],

Sources/LiveKit/Core/DataChannelPair.swift

@@ -14,6 +14,7 @@
  * limitations under the License.
  */
 
+import DequeModule
 import Foundation
 
 #if swift(>=5.9)
@@ -51,6 +52,115 @@
 
     private let _state: StateSync<State>
 
+    fileprivate enum ChannelKind {
+        case lossy, reliable
+    }
+
+    private struct BufferingState {
+        var queue: Deque<PublishDataRequest> = []
+        var amount: UInt64 = 0
+    }
+
+    private struct PublishDataRequest {
+        let data: LKRTCDataBuffer
+        let continuation: CheckedContinuation<Void, any Error>?
+    }
+
+    private struct ChannelEvent {
+        let channelKind: ChannelKind
+        let detail: Detail
+
+        enum Detail {
+            case publishData(PublishDataRequest)
+            case bufferedAmountChanged(UInt64)
+        }
+    }
+
+    private var eventContinuation: AsyncStream<ChannelEvent>.Continuation?
+
+    @Sendable private func handleEvents(
+        events: AsyncStream<ChannelEvent>
+    ) async {
+        var lossyBuffering = BufferingState()
+        var reliableBuffering = BufferingState()
+
+        for await event in events {
+            switch event.detail {
+            case let .publishData(request):
+                switch event.channelKind {
+                case .lossy: lossyBuffering.queue.append(request)
+                case .reliable: reliableBuffering.queue.append(request)
+                }
+            case let .bufferedAmountChanged(amount):
+                switch event.channelKind {
+                case .lossy: updateBufferingState(state: &lossyBuffering, newAmount: amount)
+                case .reliable: updateBufferingState(state: &reliableBuffering, newAmount: amount)
+                }
+            }
+
+            switch event.channelKind {
+            case .lossy:
+                processSendQueue(
+                    threshold: Self.lossyLowThreshold,
+                    state: &lossyBuffering,
+                    kind: .lossy
+                )
+            case .reliable:
+                processSendQueue(
+                    threshold: Self.reliableLowThreshold,
+                    state: &reliableBuffering,
+                    kind: .reliable
+                )
+            }
+        }
+    }
+
+    private func channel(for kind: ChannelKind) -> LKRTCDataChannel? {
+        _state.read {
+            guard let lossy = $0.lossy, let reliable = $0.reliable, $0.isOpen else { return nil }
+            return kind == .reliable ? reliable : lossy
+        }
+    }
+
+    private func processSendQueue(
+        threshold: UInt64,
+        state: inout BufferingState,
+        kind: ChannelKind
+    ) {
+        while state.amount <= threshold {
+            guard !state.queue.isEmpty else { break }
+            let request = state.queue.removeFirst()
+
+            state.amount += UInt64(request.data.data.count)
+
+            guard let channel = channel(for: kind) else {
+                request.continuation?.resume(
+                    throwing: LiveKitError(.invalidState, message: "Data channel is not open")
+                )
+                return
+            }
+            guard channel.sendData(request.data) else {
+                request.continuation?.resume(
+                    throwing: LiveKitError(.invalidState, message: "sendData failed")
+                )
+                return
+            }
+            request.continuation?.resume()
+        }
+    }
+
+    private func updateBufferingState(
+        state: inout BufferingState,
+        newAmount: UInt64
+    ) {
+        guard state.amount >= newAmount else {
+            log("Unexpected buffer size detected", .error)
+            state.amount = 0
+            return
+        }
+        state.amount -= newAmount
+    }
+
     public init(delegate: DataChannelDelegate? = nil,
                 lossyChannel: LKRTCDataChannel? = nil,
                 reliableChannel: LKRTCDataChannel? = nil)
@@ -61,6 +171,14 @@
         if let delegate {
             delegates.add(delegate: delegate)
         }
+        super.init()
+
+        Task {
+            let eventStream = AsyncStream<ChannelEvent> { continuation in
+                self.eventContinuation = continuation
+            }
+            await handleEvents(events: eventStream)
+        }
     }
 
     public func set(reliable channel: LKRTCDataChannel?) {
@@ -103,24 +221,27 @@
         openCompleter.reset()
     }
 
-    public func send(userPacket: Livekit_UserPacket, kind: Livekit_DataPacket.Kind) throws {
-        try send(dataPacket: .with {
-            $0.kind = kind
+    public func send(userPacket: Livekit_UserPacket, kind: Livekit_DataPacket.Kind) async throws {
+        try await send(dataPacket: .with {
+            $0.kind = kind // TODO: field is deprecated
             $0.user = userPacket
         })
     }
 
-    public func send(dataPacket packet: Livekit_DataPacket) throws {
-        guard isOpen else {
-            throw LiveKitError(.invalidState, message: "Data channel is not open")
-        }
-
+    public func send(dataPacket packet: Livekit_DataPacket) async throws {
         let serializedData = try packet.serializedData()
         let rtcData = RTC.createDataBuffer(data: serializedData)
 
-        let channel = _state.read { packet.kind == .reliable ? $0.reliable : $0.lossy }
-        guard let sendDataResult = channel?.sendData(rtcData), sendDataResult else {
-            throw LiveKitError(.invalidState, message: "sendData failed")
+        try await withCheckedThrowingContinuation { continuation in
+            let request = PublishDataRequest(
+                data: rtcData,
+                continuation: continuation
+            )
+            let event = ChannelEvent(
+                channelKind: ChannelKind(packet.kind), // TODO: field is deprecated
+                detail: .publishData(request)
+            )
+            eventContinuation?.yield(event)
         }
     }
 
@@ -129,11 +250,26 @@
             .compactMap { $0 }
             .map { $0.toLKInfoType() }
     }
+
+    private static let reliableLowThreshold: UInt64 = 2 * 1024 * 1024 // 2 MB
+    private static let lossyLowThreshold: UInt64 = reliableLowThreshold
+
+    deinit {
+        eventContinuation?.finish()
+    }
 }
 
 // MARK: - RTCDataChannelDelegate
 
 extension DataChannelPair: LKRTCDataChannelDelegate {
+    func dataChannel(_ dataChannel: LKRTCDataChannel, didChangeBufferedAmount amount: UInt64) {
+        let event = ChannelEvent(
+            channelKind: dataChannel.kind,
+            detail: .bufferedAmountChanged(amount)
+        )
+        eventContinuation?.yield(event)
+    }
+
     func dataChannelDidChangeState(_: LKRTCDataChannel) {
         if isOpen {
             openCompleter.resume(returning: ())
@@ -141,7 +277,7 @@
     }
 
     func dataChannel(_: LKRTCDataChannel, didReceiveMessageWith buffer: LKRTCDataBuffer) {
         guard let dataPacket = try? Livekit_DataPacket(serializedData: buffer.data) else {
             log("Could not decode data message", .error)
             return
         }
@@ -151,3 +287,22 @@
         }
     }
 }
+
+private extension DataChannelPair.ChannelKind {
+    init(_ packetKind: Livekit_DataPacket.Kind) {
+        guard case .lossy = packetKind else {
+            self = .reliable
+            return
+        }
+        self = .lossy
+    }
+}
+
+private extension LKRTCDataChannel {
+    var kind: DataChannelPair.ChannelKind {
+        guard label == LKRTCDataChannel.labels.lossy else {
+            return .reliable
+        }
+        return .lossy
+    }
+}

Sources/LiveKit/Core/RPC.swift

@@ -158,7 +158,7 @@ actor RpcStateManager: Loggable {
             log("No handler registered for RPC method '\(method)'", .warning)
         }
     }
-    
+
     func isRpcMethodRegistered(_ method: String) -> Bool {
         handlers[method] != nil
     }

Sources/LiveKit/Core/Room+DataStream.swift

@@ -0,0 +1,89 @@
+/*
+ * 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.
+ */
+
+import Foundation
+
+public extension Room {
+    /// Registers a handler for incoming byte streams matching the given topic.
+    ///
+    /// - Parameters:
+    ///   - topic: Topic identifier that filters which streams will be handled.
+    ///     Only streams with a matching topic will trigger the handler.
+    ///   - onNewStream: Handler that is invoked whenever a remote participant
+    ///     opens a new stream with the matching topic. The handler receives a
+    ///     ``ByteStreamReader`` for consuming the stream data and the identity of
+    ///     the remote participant who initiated the stream.
+    ///
+    func registerByteStreamHandler(for topic: String, onNewStream: @escaping ByteStreamHandler) async throws {
+        try await incomingStreamManager.registerByteStreamHandler(for: topic, onNewStream)
+    }
+
+    /// Registers a handler for incoming text streams matching the given topic.
+    ///
+    /// - Parameters:
+    ///   - topic: Topic identifier that filters which streams will be handled.
+    ///     Only streams with a matching topic will trigger the handler.
+    ///   - onNewStream: Handler that is invoked whenever a remote participant
+    ///     opens a new stream with the matching topic. The handler receives a
+    ///     ``TextStreamReader`` for consuming the stream data and the identity of
+    ///     the remote participant who initiated the stream.
+    ///
+    func registerTextStreamHandler(for topic: String, onNewStream: @escaping TextStreamHandler) async throws {
+        try await incomingStreamManager.registerTextStreamHandler(for: topic, onNewStream)
+    }
+
+    /// Unregisters a byte stream handler that was previously registered for the given topic.
+    @objc
+    func unregisterByteStreamHandler(for topic: String) async {
+        await incomingStreamManager.unregisterByteStreamHandler(for: topic)
+    }
+
+    /// Unregisters a text stream handler that was previously registered for the given topic.
+    @objc
+    func unregisterTextStreamHandler(for topic: String) async {
+        await incomingStreamManager.unregisterTextStreamHandler(for: topic)
+    }
+}
+
+// MARK: - Objective-C Compatibility
+
+public extension Room {
+    @objc
+    @available(*, deprecated, message: "Use async registerByteStreamHandler(for:onNewStream:) method instead.")
+    func registerByteStreamHandler(
+        for topic: String,
+        onNewStream: @escaping (ByteStreamReader, Participant.Identity) -> Void,
+        onError: ((Error) -> Void)?
+    ) {
+        Task {
+            do { try await registerByteStreamHandler(for: topic, onNewStream: onNewStream) }
+            catch { onError?(error) }
+        }
+    }
+
+    @objc
+    @available(*, deprecated, message: "Use async registerTextStreamHandler(for:onNewStream:) method instead.")
+    func registerTextStreamHandler(
+        for topic: String,
+        onNewStream: @escaping (TextStreamReader, Participant.Identity) -> Void,
+        onError: ((Error) -> Void)?
+    ) {
+        Task {
+            do { try await registerTextStreamHandler(for: topic, onNewStream: onNewStream) }
+            catch { onError?(error) }
+        }
+    }
+}

Sources/LiveKit/Core/Room+Engine.swift

@@ -107,8 +107,7 @@ extension Room {
             packet.participantIdentity = identity
         }
 
-        // Should return true if successful
-        try publisherDataChannel.send(dataPacket: packet)
+        try await publisherDataChannel.send(dataPacket: packet)
     }
 }