Data streams v2 #1075
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1 @@ | ||
| minor type="changed" "Data streams: reimplemented on the Rust livekit-data-stream core over UniFFI, adding DEFLATE compression and single-packet inlining for one-shot sends when every recipient supports them, a configurable incoming payload cap via RoomOptions.dataStreamOptions, and Participant.capabilities" |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,132 @@ | ||
| /* | ||
| * Copyright 2026 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 | ||
|
|
||
| internal import LiveKitUniFFI | ||
|
|
||
| // The objects the Rust core calls into. Split from `DataStreams` itself so the coordinator file | ||
| // stays about coordination; each type here is a thin adapter that forwards across the boundary. | ||
| extension DataStreams { | ||
| // MARK: - Incoming delegate | ||
|
|
||
| /// Receives the incoming manager's stream-open callbacks and forwards them to the coordinator. | ||
| /// A separate object because the FFI manager retains its delegate strongly; holding the | ||
| /// coordinator weakly keeps this the weak link so teardown doesn't leak. | ||
| final class IncomingDelegate: LiveKitUniFFI.IncomingDataStreamManagerDelegate, @unchecked Sendable { | ||
|
Check failure on line 29 in Sources/LiveKit/DataStream/DataStreams+FFIDelegates.swift
|
||
| weak var coordinator: DataStreams? | ||
|
|
||
| func onByteStreamOpened(reader: LiveKitUniFFI.ByteStreamReader, identity: String) { | ||
|
Check failure on line 32 in Sources/LiveKit/DataStream/DataStreams+FFIDelegates.swift
|
||
| coordinator?.handleByteStreamOpened(reader, identity: identity) | ||
| } | ||
|
|
||
| func onTextStreamOpened(reader: LiveKitUniFFI.TextStreamReader, identity: String) { | ||
|
Check failure on line 36 in Sources/LiveKit/DataStream/DataStreams+FFIDelegates.swift
|
||
| coordinator?.handleTextStreamOpened(reader, identity: identity) | ||
| } | ||
|
|
||
| func onStreamClosed(streamId: String, identity: String) { | ||
| coordinator?.handleStreamClosed(streamID: streamId, identity: identity) | ||
| } | ||
| } | ||
|
|
||
| // MARK: - Outgoing delegate | ||
|
|
||
| /// Receives the outgoing manager's encoded `DataPacket`s and sends them over the reliable data | ||
| /// channel via `Room.send(dataPacket:)` — preserving E2EE, reliable sequencing, and identity | ||
| /// stamping. The room is held weakly to avoid retaining it through the FFI manager. | ||
| /// | ||
| /// The callback is `async`, which is what lets this honor the FFI's contract: it returns only | ||
| /// once the packets have actually reached the transport. Three properties follow, none of which | ||
| /// needs machinery on this side. | ||
| /// | ||
| /// - **Order.** The core awaits this call before pumping the next packet, so calls arrive — and | ||
| /// complete — strictly in emission order. That matters: the receiver drops a chunk that | ||
| /// arrives before its header and fails the stream on a non-consecutive chunk index. | ||
| /// - **Back-pressure.** The originating `write`/`send_*` stays pending until this returns, so a | ||
| /// producer can't outrun the transport and queue unboundedly. | ||
| /// - **Failures.** Throwing `PacketDeliveryError` fails that originating call with | ||
| /// `.sendFailed` and closes the stream, so `isOpen` reflects reality. | ||
| /// | ||
| /// `@unchecked Sendable` for the weak back-reference alone: it is assigned in `init` and never | ||
| /// mutated afterwards. | ||
| final class OutgoingDelegate: LiveKitUniFFI.OutgoingDataStreamManagerDelegate, @unchecked Sendable { | ||
|
Check failure on line 65 in Sources/LiveKit/DataStream/DataStreams+FFIDelegates.swift
|
||
| private weak var room: Room? | ||
|
|
||
| init(room: Room) { | ||
| self.room = room | ||
| } | ||
|
|
||
| func onPacketsAvailable(packets: [Data]) async throws { | ||
| // The room is gone, so there is no transport left to fail against; the manager is being | ||
| // torn down with it. | ||
| guard let room else { return } | ||
|
|
||
| for data in packets { | ||
| guard let packet = try? Livekit_DataPacket(serializedBytes: data) else { | ||
| room.log("Failed to decode outgoing data stream packet", .warning) | ||
| continue | ||
| } | ||
| do { | ||
| try await room.send(dataPacket: packet) | ||
| } catch { | ||
| throw LiveKitUniFFI.PacketDeliveryError.Failed(reason: String(describing: error)) | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // MARK: - Remote participant registry | ||
|
|
||
| /// Read access to the room's remote participants, used by the outgoing manager to resolve | ||
| /// broadcast recipients and decide compression eligibility. Client protocol/capabilities aren't | ||
| /// currently exposed on `RemoteParticipant`, so they default to none — compression stays off | ||
| /// until they're wired, which is a safe default (a non-compressed send always works). | ||
| final class Registry: LiveKitUniFFI.RemoteParticipantRegistryDelegate, @unchecked Sendable { | ||
|
Check failure on line 97 in Sources/LiveKit/DataStream/DataStreams+FFIDelegates.swift
|
||
| private weak var room: Room? | ||
|
|
||
| init(room: Room) { | ||
| self.room = room | ||
| } | ||
|
|
||
| private func participant(for identity: String) -> RemoteParticipant? { | ||
| room?.remoteParticipants.first { $0.key.stringValue == identity }?.value | ||
| } | ||
|
|
||
| func remoteClientProtocol(identity: String) -> Int32 { | ||
| Int32(participant(for: identity)?.clientProtocol.rawValue ?? 0) | ||
| } | ||
|
|
||
| func remoteCapabilities(identity: String) -> [LiveKitUniFFI.ClientCapability] { | ||
|
Check failure on line 112 in Sources/LiveKit/DataStream/DataStreams+FFIDelegates.swift
|
||
| participant(for: identity)?.capabilities.map(\.ffiValue) ?? [] | ||
| } | ||
|
|
||
| func remoteIdentities() -> [String] { | ||
| guard let room else { return [] } | ||
| return room.remoteParticipants.keys.map(\.stringValue) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private extension ClientCapability { | ||
| /// Bridges to the FFI enum. Kept here so `ClientCapability` itself stays free of any | ||
| /// `LiveKitUniFFI` import, matching how the rest of the public API is layered. | ||
| var ffiValue: LiveKitUniFFI.ClientCapability { | ||
|
Check failure on line 126 in Sources/LiveKit/DataStream/DataStreams+FFIDelegates.swift
|
||
| switch self { | ||
| case .packetTrailer: .packetTrailer | ||
| case .compressionDeflateRaw: .compressionDeflateRaw | ||
| } | ||
| } | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🟡 Encrypted data messages are reported to apps as unencrypted
The encryption kind handed to the data-message callback is read from the packet after it has already been unscrambled (
dataPacket.encryptedPacket.encryptionTypeatSources/LiveKit/Core/Room.swift:805), so apps are always told an encrypted message arrived in the clear even though the correct value is now available as a parameter.Impact: Applications that inspect whether received data was end-to-end encrypted always see "none", so they cannot distinguish encrypted from plaintext senders.
Mechanism: the oneof field carrying the encryption type is overwritten by decryption
The new delegate signature (
Sources/LiveKit/Core/DataChannelPair.swift:26-29) exists precisely becauseEncryptedPacketshares a protobufoneofwith the decrypted payload:decryptedPayload.applyTo(&$0)sets e.g.builder.user, clearingencryptedPacket(Sources/LiveKit/E2EE/Protos+E2EE.swift:68-88). The stream cases were migrated to the newencryptionTypeparameter, but the.usercase still reads the now-cleared field, soengine(_:didReceiveUserPacket:encryptionType:)— and ultimatelyroom(_:participant:didReceiveData:forTopic:encryptionType:)— always receives.nonefor decrypted packets.Prompt for agents
Was this helpful? React with 👍 or 👎 to provide feedback.