diff --git a/.changes/connect-room-options-ignored b/.changes/connect-room-options-ignored new file mode 100644 index 000000000..7827fc1da --- /dev/null +++ b/.changes/connect-room-options-ignored @@ -0,0 +1 @@ +patch type="fixed" "Room.connect no longer ignores the roomOptions argument passed to it" diff --git a/.changes/data-stream-options b/.changes/data-stream-options new file mode 100644 index 000000000..08695e2c9 --- /dev/null +++ b/.changes/data-stream-options @@ -0,0 +1 @@ +minor type="added" "ConnectOptions.dataStream with maxPayloadByteLength, bounding the payload a single incoming data stream may deliver" diff --git a/.changes/data-streams-v2 b/.changes/data-streams-v2 new file mode 100644 index 000000000..5504bc35b --- /dev/null +++ b/.changes/data-streams-v2 @@ -0,0 +1 @@ +minor type="changed" "Data streams are now backed by the Rust core (livekit-uniffi) on native platforms, adding data streams v2: single-packet inline sends, deflate-raw compression and MTU-bounded headers. Adds LocalParticipant.sendBytes, a compress option, Participant.capabilities and ClientProtocolVersion.v2. Web keeps the existing Dart implementation and interoperates as a pre-v2 peer." diff --git a/AGENTS.md b/AGENTS.md index 5e7661b9e..65fd4207b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -36,7 +36,16 @@ Web/native divergence is handled with conditional imports (e.g. `track/processor `lib/src/uniffi/` wraps `livekit_uniffi`, a Dart package generated from the `livekit-uniffi` crate in the sibling `rust-sdks` repo. It reaches Rust through Dart's Native Assets: the package's `hook/build.dart` bundles a `cdylib` into the host app and the generated bindings call into it with `@Native`. This is why the SDK requires Flutter >= 3.38 / Dart >= 3.10. -There is no dynamic library to load on the web, so `uniffi.dart` splits native/web the same way the rest of the SDK does. **`uniffi_io.dart` is the only file allowed to import `package:livekit_uniffi/...`** — importing it from anywhere reachable on web pulls `dart:ffi` into a web compile and breaks `flutter build web`/`--wasm`. Guard calls with `LiveKitUniffi.isAvailable`. +There is no dynamic library to load on the web, so `uniffi.dart` splits native/web the same way the rest of the SDK does. **Only `uniffi_io.dart` and files under `lib/src/data_stream/` whose names end in `_native.dart` (plus `ffi_bridged.dart`) may import `package:livekit_uniffi/...`** — importing it from anywhere reachable on web pulls `dart:ffi` into a web compile and breaks `flutter build web`/`--wasm`. No generated uniffi type may appear in a public API signature; convert at the boundary (`data_stream/ffi_bridged.dart`). Guard calls with `LiveKitUniffi.isAvailable`. + +### Data streams + +`lib/src/data_stream/` has two implementations behind one interface (`data_streams.dart`, conditional import): `data_streams_native.dart` delegates to the Rust core, which implements **data streams v2** (inline single-packet sends, deflate-raw compression, UTF-8-aware chunking, MTU-bounded headers); `data_streams_web.dart` is the original Dart v1 code, kept because the cdylib can't run in a browser. Web advertises `ClientProtocolVersion.v1` and no capabilities, so v2 senders fall back to uncompressed multi-packet for it. + +Two things to know when touching the native path: + +- **The core's push delegates cannot be used from Dart.** uniffi compiles a callback interface to `Pointer.fromFunction`, which is only valid on the isolate's thread, and the core invokes those delegates from its tokio runtime — the VM aborts with `Cannot invoke native callback outside an isolate`, which is not catchable. The managers are therefore built through the crate's `polled*` adapters (`livekit-uniffi/src/data_stream/polled.rs`), which implement the delegates *in Rust*, buffer into a channel, and expose an `async fn next_*` we await. `RemoteParticipantRegistryDelegate` is the one callback we implement directly, and it is safe: it is only called synchronously inside a `send*` future, which uniffi polls from the calling (Dart) thread. +- **Whoever awaits a uniffi object is the only thing that may dispose it.** Freeing the Rust handle while a `next()`/`nextPackets()` is in flight is a use-after-free that surfaces as a SIGBUS with no Dart stack. Hence readers are disposed by their pump rather than from a subscription's `onCancel`, and `dispose()` calls `close()` on the queues to wake their pumps instead of releasing them directly. ### Local development loop diff --git a/lib/livekit_client.dart b/lib/livekit_client.dart index f8527e14b..d2e5e3245 100644 --- a/lib/livekit_client.dart +++ b/lib/livekit_client.dart @@ -22,6 +22,7 @@ export 'src/connection_check/events.dart'; export 'src/constants.dart'; export 'src/core/room.dart'; export 'src/core/room_preconnect.dart'; +export 'src/data_stream/errors.dart'; export 'src/data_stream/stream_reader.dart'; export 'src/data_stream/stream_writer.dart'; export 'src/e2ee/e2ee_manager.dart'; @@ -69,6 +70,7 @@ export 'src/track/remote/remote.dart'; export 'src/track/remote/video.dart'; export 'src/track/track.dart'; export 'src/json/agent_attributes.dart'; +export 'src/types/client_capability.dart'; export 'src/types/data_stream.dart'; export 'src/types/audio_encoding.dart'; export 'src/types/other.dart'; diff --git a/lib/src/core/engine.dart b/lib/src/core/engine.dart index 89f990c4b..23ff72f63 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -1024,29 +1024,13 @@ class Engine extends Disposable with EventsEmittable { identity: dp.participantIdentity, ), ); - } else if (dp.whichValue() == lk_models.DataPacket_Value.streamHeader) { - // Data Stream Header + } else if (dp.whichValue() == lk_models.DataPacket_Value.streamHeader || + dp.whichValue() == lk_models.DataPacket_Value.streamChunk || + dp.whichValue() == lk_models.DataPacket_Value.streamTrailer) { + // Data stream header / chunk / trailer, forwarded whole — see EngineDataStreamPacketEvent. events.emit( - EngineDataStreamHeaderEvent( - header: dp.streamHeader, - identity: dp.participantIdentity, - encryptionType: encryptionType, - ), - ); - } else if (dp.whichValue() == lk_models.DataPacket_Value.streamChunk) { - // Data Stream Chunk - events.emit( - EngineDataStreamChunkEvent( - chunk: dp.streamChunk, - identity: dp.participantIdentity, - encryptionType: encryptionType, - ), - ); - } else if (dp.whichValue() == lk_models.DataPacket_Value.streamTrailer) { - // Data Stream trailer - events.emit( - EngineDataStreamTrailerEvent( - trailer: dp.streamTrailer, + EngineDataStreamPacketEvent( + packet: dp, identity: dp.participantIdentity, encryptionType: encryptionType, ), diff --git a/lib/src/core/room.dart b/lib/src/core/room.dart index ae2d8bb36..1b318c77a 100644 --- a/lib/src/core/room.dart +++ b/lib/src/core/room.dart @@ -20,10 +20,9 @@ import 'package:meta/meta.dart'; import '../audio/audio_manager.dart'; import '../core/signal_client.dart'; +import '../data_stream/data_streams.dart'; import '../data_stream/errors.dart'; -import '../data_stream/stream_reader.dart'; import '../e2ee/e2ee_manager.dart'; -import '../e2ee/options.dart'; import '../events.dart'; import '../exceptions.dart'; import '../extensions.dart'; @@ -138,13 +137,11 @@ class Room extends DisposableChangeNotifier with EventsEmittable { late final RpcClientManager _rpcClientManager; late final RpcServerManager _rpcServerManager; - final Map> _byteStreamControllers = {}; - - final Map> _textStreamControllers = {}; - - final Map _byteStreamHandlers = {}; - - final Map _textStreamHandlers = {}; + /// Owns the data-stream subsystem: the topic registry, the send path and inbound routing. On + /// native this is backed by the Rust core (data streams v2); on web by the original Dart + /// implementation. Created eagerly below so there is exactly one for the room's lifetime, and it + /// survives disconnect so handler registrations outlive a reconnect. + late final DataStreams dataStreams; @internal late final PreConnectAudioBuffer preConnectAudioBuffer; @@ -161,13 +158,13 @@ class Room extends DisposableChangeNotifier with EventsEmittable { // getter would surprise SDK consumers — filter them out here. @internal Map get textStreamHandlers => Map.fromEntries( - _textStreamHandlers.entries.where( + dataStreams.textStreamHandlers.entries.where( (e) => e.key != kRpcRequestTopic && e.key != kRpcResponseTopic, ), ); @internal - Map get byteStreamHandlers => _byteStreamHandlers; + Map get byteStreamHandlers => dataStreams.byteStreamHandlers; Room({ @Deprecated('deprecated, please use connectOptions in room.connect()') @@ -181,6 +178,8 @@ class Room extends DisposableChangeNotifier with EventsEmittable { roomOptions: roomOptions, ) { // + dataStreams = createDataStreams(this); + _engineListener = this.engine.createListener(); _setUpEngineListeners(); @@ -221,6 +220,7 @@ class Room extends DisposableChangeNotifier with EventsEmittable { await _cleanUp(); // reject any in-flight RPC calls _rpcClientManager.dispose(); + await dataStreams.dispose(); // dispose preConnectAudioBuffer await preConnectAudioBuffer.dispose(); // dispose events @@ -275,8 +275,12 @@ class Room extends DisposableChangeNotifier with EventsEmittable { @Deprecated('deprecated, please use roomOptions in Room constructor') RoomOptions? roomOptions, FastConnectOptions? fastConnectOptions, }) async { - var roomOptions = this.roomOptions; - if (lkPlatformIs(PlatformType.web) && (roomOptions.networkOptions.certificatePinning?.isEnabled ?? false)) { + // The deprecated `roomOptions` parameter still has to take effect when supplied. It was + // previously shadowed by a local of the same name declared right here — which Dart allows + // silently, with the local winning — so anything callers passed was discarded. + var effectiveRoomOptions = roomOptions ?? this.roomOptions; + if (lkPlatformIs(PlatformType.web) && + (effectiveRoomOptions.networkOptions.certificatePinning?.isEnabled ?? false)) { throw UnsupportedError( 'Certificate pinning is not supported on Flutter web, ' 'remove certificatePinning from NetworkOptions when targeting web', @@ -285,13 +289,17 @@ class Room extends DisposableChangeNotifier with EventsEmittable { connectOptions ??= ConnectOptions(); _pendingTrackQueue.updateTtl(connectOptions.timeouts.subscribe); // ignore: deprecated_member_use_from_same_package - if ((roomOptions.encryption != null || roomOptions.e2eeOptions != null) && engine.e2eeManager == null) { + if ((effectiveRoomOptions.encryption != null || effectiveRoomOptions.e2eeOptions != null) && + engine.e2eeManager == null) { if (!lkPlatformSupportsE2EE()) { throw LiveKitE2EEException('E2EE is not supported on this platform'); } // ignore: deprecated_member_use_from_same_package - final e2eeOptions = roomOptions.encryption ?? roomOptions.e2eeOptions; - _e2eeManager = E2EEManager(e2eeOptions!.keyProvider, dcEncryptionEnabled: roomOptions.encryption != null); + final e2eeOptions = effectiveRoomOptions.encryption ?? effectiveRoomOptions.e2eeOptions; + _e2eeManager = E2EEManager( + e2eeOptions!.keyProvider, + dcEncryptionEnabled: effectiveRoomOptions.encryption != null, + ); await _e2eeManager!.setup(this); engine.setE2eeManager(_e2eeManager); } else { @@ -300,8 +308,8 @@ class Room extends DisposableChangeNotifier with EventsEmittable { if (_e2eeManager != null) { // Disable backup codec when e2ee is enabled - roomOptions = roomOptions.copyWith( - defaultVideoPublishOptions: roomOptions.defaultVideoPublishOptions.copyWith( + effectiveRoomOptions = effectiveRoomOptions.copyWith( + defaultVideoPublishOptions: effectiveRoomOptions.defaultVideoPublishOptions.copyWith( backupVideoCodec: const BackupVideoCodec(enabled: false), ), ); @@ -313,7 +321,11 @@ class Room extends DisposableChangeNotifier with EventsEmittable { } if (isCloudUrl(Uri.parse(url))) { if (_regionUrlProvider == null) { - _regionUrlProvider = RegionUrlProvider(url: url, token: token, networkOptions: roomOptions.networkOptions); + _regionUrlProvider = RegionUrlProvider( + url: url, + token: token, + networkOptions: effectiveRoomOptions.networkOptions, + ); } else { _regionUrlProvider?.updateToken(token); } @@ -336,7 +348,7 @@ class Room extends DisposableChangeNotifier with EventsEmittable { // AudioManager once, on the first connect. Skipping it on a later manual // connect of the same Room keeps a runtime speaker change from being // reverted. New code should call setSpeakerOutputPreferred directly. - final legacySpeakerOn = roomOptions.defaultAudioOutputOptions.speakerOn; + final legacySpeakerOn = effectiveRoomOptions.defaultAudioOutputOptions.speakerOn; if (legacySpeakerOn != null && !_legacySpeakerBridged && lkPlatformIsMobile()) { _legacySpeakerBridged = true; await AudioManager.instance.setSpeakerOutputPreferred(legacySpeakerOn); @@ -351,7 +363,7 @@ class Room extends DisposableChangeNotifier with EventsEmittable { _regionUrl ?? url, token, connectOptions: connectOptions, - roomOptions: roomOptions, + roomOptions: effectiveRoomOptions, fastConnectOptions: fastConnectOptions, regionUrlProvider: _regionUrlProvider, ); @@ -374,7 +386,7 @@ class Room extends DisposableChangeNotifier with EventsEmittable { nextUrl, token, connectOptions: connectOptions, - roomOptions: roomOptions, + roomOptions: effectiveRoomOptions, fastConnectOptions: fastConnectOptions, regionUrlProvider: _regionUrlProvider, ); @@ -1081,6 +1093,11 @@ extension RoomPrivateMethods on Room { Future _cleanUp({bool disposeLocalParticipant = true}) async { logger.fine('[${objectId}] cleanUp()'); + // Fail any open data streams so their handlers return rather than awaiting a reader that will + // never finish. Handler registrations deliberately survive, so streams arriving after a + // reconnect are still routed. + await dataStreams.reset(); + // clean up RemoteParticipants final participants = _remoteParticipants.toList(); _remoteParticipants.clear(); @@ -1386,8 +1403,8 @@ extension RoomRPCMethods on Room { // Register v2 data-stream-based request/response handlers. These topics // are reserved by the SDK, so bypass the public registration guard. - _textStreamHandlers[kRpcRequestTopic] = _rpcServerManager.handleIncomingV2RequestStream; - _textStreamHandlers[kRpcResponseTopic] = _rpcClientManager.handleIncomingV2ResponseStream; + dataStreams.registerTextStreamHandler(kRpcRequestTopic, _rpcServerManager.handleIncomingV2RequestStream); + dataStreams.registerTextStreamHandler(kRpcResponseTopic, _rpcClientManager.handleIncomingV2ResponseStream); } /// Register a handler for incoming RPC requests. @@ -1420,48 +1437,41 @@ const _reservedRpcTopicPrefix = 'lk.rpc'; extension DataStreamRoomMethods on Room { void _setupDataStreamListeners() { - _engineListener - ..on((event) async { - await handleStreamHeader(event.header, event.identity, event.encryptionType); - }) - ..on((event) async { - handleStreamChunk(event.chunk, event.encryptionType); - }) - ..on((event) async { - await handleStreamTrailer(event.trailer, event.encryptionType); - }); + _engineListener.on((event) async { + dataStreams.handleIncomingPacket(event.packet, event.encryptionType); + }); } void registerTextStreamHandler(String topic, TextStreamHandler callback) { _ensureNotReservedRpcTopic(topic); - if (_textStreamHandlers.containsKey(topic)) { + if (dataStreams.textStreamHandlers.containsKey(topic)) { throw DataStreamError( message: 'A text stream handler for topic "${topic}" has already been set.', reason: DataStreamErrorReason.HandlerAlreadyRegistered, ); } - _textStreamHandlers[topic] = callback; + dataStreams.registerTextStreamHandler(topic, callback); } void unregisterTextStreamHandler(String topic) { if (_isReservedRpcTopic(topic)) return; - _textStreamHandlers.remove(topic); + dataStreams.unregisterTextStreamHandler(topic); } void registerByteStreamHandler(String topic, ByteStreamHandler callback) { _ensureNotReservedRpcTopic(topic); - if (_byteStreamHandlers.containsKey(topic)) { + if (dataStreams.byteStreamHandlers.containsKey(topic)) { throw DataStreamError( message: 'A byte stream handler for topic "${topic}" has already been set.', reason: DataStreamErrorReason.HandlerAlreadyRegistered, ); } - _byteStreamHandlers[topic] = callback; + dataStreams.registerByteStreamHandler(topic, callback); } void unregisterByteStreamHandler(String topic) { if (_isReservedRpcTopic(topic)) return; - _byteStreamHandlers.remove(topic); + dataStreams.unregisterByteStreamHandler(topic); } void _ensureNotReservedRpcTopic(String topic) { @@ -1476,208 +1486,6 @@ extension DataStreamRoomMethods on Room { bool _isReservedRpcTopic(String topic) => topic.startsWith(_reservedRpcTopicPrefix); @internal - Future handleStreamHeader( - lk_models.DataStream_Header streamHeader, - String participantIdentity, - EncryptionType encryptionType, - ) async { - if (streamHeader.hasByteHeader()) { - final streamHandlerCallback = _byteStreamHandlers[streamHeader.topic]; - - if (streamHandlerCallback == null) { - logger.info('ignoring incoming byte stream due to no handler for topic ${streamHeader.topic}'); - return; - } - - final info = ByteStreamInfo( - id: streamHeader.streamId, - name: streamHeader.byteHeader.name, - mimeType: streamHeader.mimeType, - size: streamHeader.hasTotalLength() ? streamHeader.totalLength.toInt() : 0, - topic: streamHeader.topic, - timestamp: streamHeader.timestamp.toInt(), - attributes: streamHeader.attributes, - encryptionType: encryptionType, - sendingParticipantIdentity: participantIdentity, - ); - - final streamController = DataStreamController( - info: info, - streamController: StreamController(), - startTime: DateTime.timestamp().millisecondsSinceEpoch, - ); - - if (_byteStreamControllers.containsKey(streamHeader.streamId)) { - throw DataStreamError( - message: 'A data stream read is already in progress for a stream with id ${streamHeader.streamId}.', - reason: DataStreamErrorReason.AlreadyOpened, - ); - } - - _byteStreamControllers[streamHeader.streamId] = streamController; - - streamHandlerCallback( - ByteStreamReader(info, streamController, streamHeader.totalLength.toInt()), - participantIdentity, - ); - } else if (streamHeader.hasTextHeader()) { - final streamHandlerCallback = _textStreamHandlers[streamHeader.topic]; - - if (streamHandlerCallback == null) { - logger.warning('ignoring incoming text stream due to no handler for topic ${streamHeader.topic}'); - return; - } - - final info = TextStreamInfo( - id: streamHeader.streamId, - mimeType: streamHeader.mimeType, - size: streamHeader.hasTotalLength() ? streamHeader.totalLength.toInt() : 0, - topic: streamHeader.topic, - timestamp: streamHeader.timestamp.toInt(), - attributes: streamHeader.attributes, - replyToStreamId: streamHeader.textHeader.hasReplyToStreamId() ? streamHeader.textHeader.replyToStreamId : null, - attachedStreamIds: streamHeader.textHeader.attachedStreamIds.toList(), - version: streamHeader.textHeader.hasVersion() ? streamHeader.textHeader.version : null, - generated: streamHeader.textHeader.hasGenerated() ? streamHeader.textHeader.generated : false, - operationType: streamHeader.textHeader.hasOperationType() - ? TextStreamOperationType.fromPBType(streamHeader.textHeader.operationType) - : null, - encryptionType: encryptionType, - sendingParticipantIdentity: participantIdentity, - ); - - final streamController = DataStreamController( - info: info, - streamController: StreamController(), - startTime: DateTime.timestamp().millisecondsSinceEpoch, - ); - - if (_textStreamControllers.containsKey(streamHeader.streamId)) { - throw DataStreamError( - message: 'A data stream read is already in progress for a stream with id ${streamHeader.streamId}.', - reason: DataStreamErrorReason.AlreadyOpened, - ); - } - - _textStreamControllers[streamHeader.streamId] = streamController; - - streamHandlerCallback( - TextStreamReader(info, streamController, streamHeader.totalLength.toInt()), - participantIdentity, - ); - } - } - - @internal - void handleStreamChunk(lk_models.DataStream_Chunk chunk, EncryptionType encryptionType) { - final fileBuffer = _byteStreamControllers[chunk.streamId]; - - if (fileBuffer != null) { - if (fileBuffer.info.encryptionType != encryptionType) { - fileBuffer.error( - DataStreamError( - message: - 'Encryption type mismatch for stream ${chunk.streamId}. Expected ${encryptionType}, got ${fileBuffer.info.encryptionType}', - reason: DataStreamErrorReason.EncryptionTypeMismatch, - ), - ); - - _byteStreamControllers.remove(chunk.streamId); - } else if (chunk.content.isNotEmpty) { - fileBuffer.write(chunk); - } - } - final textBuffer = _textStreamControllers[chunk.streamId]; - if (textBuffer != null) { - if (textBuffer.info.encryptionType != encryptionType) { - textBuffer.error( - DataStreamError( - message: - 'Encryption type mismatch for stream ${chunk.streamId}. Expected ${encryptionType}, got ${textBuffer.info.encryptionType}', - reason: DataStreamErrorReason.EncryptionTypeMismatch, - ), - ); - - logger.warning('encryption type mismatch for text stream ${chunk.streamId}'); - _textStreamControllers.remove(chunk.streamId); - } else if (chunk.content.isNotEmpty) { - textBuffer.write(chunk); - } - } - } - - @internal - Future handleStreamTrailer(lk_models.DataStream_Trailer trailer, EncryptionType encryptionType) async { - final textBuffer = _textStreamControllers[trailer.streamId]; - if (textBuffer != null) { - if (textBuffer.info.encryptionType != encryptionType) { - textBuffer.error( - DataStreamError( - message: - 'Encryption type mismatch for stream ${trailer.streamId}. Expected ${encryptionType}, got ${textBuffer.info.encryptionType}', - reason: DataStreamErrorReason.EncryptionTypeMismatch, - ), - ); - - _textStreamControllers.remove(trailer.streamId); - return; - } else { - textBuffer.info.attributes = { - ...textBuffer.info.attributes, - ...trailer.attributes, - }; - await textBuffer.close(); - _textStreamControllers.remove(trailer.streamId); - } - } - - final fileBuffer = _byteStreamControllers[trailer.streamId]; - if (fileBuffer != null) { - if (fileBuffer.info.encryptionType != encryptionType) { - fileBuffer.error( - DataStreamError( - message: - 'Encryption type mismatch for stream ${trailer.streamId}. Expected ${encryptionType}, got ${fileBuffer.info.encryptionType}', - reason: DataStreamErrorReason.EncryptionTypeMismatch, - ), - ); - - _byteStreamControllers.remove(trailer.streamId); - return; - } else { - fileBuffer.info.attributes = {...fileBuffer.info.attributes, ...trailer.attributes}; - await fileBuffer.close(); - _byteStreamControllers.remove(trailer.streamId); - } - } - } - - @internal - Future validateParticipantHasNoActiveDataStreams(String participantIdentity) async { - // Terminate any in flight data stream receives from the given participant - final textStreamsBeingSentByDisconnectingParticipant = _textStreamControllers.values - .where((controller) => controller.info.sendingParticipantIdentity == participantIdentity) - .toList(); - - final byteStreamsBeingSentByDisconnectingParticipant = _byteStreamControllers.values - .where((controller) => controller.info.sendingParticipantIdentity == participantIdentity) - .toList(); - if (textStreamsBeingSentByDisconnectingParticipant.isNotEmpty || - byteStreamsBeingSentByDisconnectingParticipant.isNotEmpty) { - final abnormalEndError = DataStreamError( - message: 'Participant ${participantIdentity} unexpectedly disconnected in the middle of sending data', - reason: DataStreamErrorReason.AbnormalEnd, - ); - for (var controller in byteStreamsBeingSentByDisconnectingParticipant) { - controller.error(abnormalEndError); - await controller.close(); - _byteStreamControllers.remove(controller.info.id); - } - for (var controller in textStreamsBeingSentByDisconnectingParticipant) { - controller.error(abnormalEndError); - await controller.close(); - _textStreamControllers.remove(controller.info.id); - } - } - } + Future validateParticipantHasNoActiveDataStreams(String participantIdentity) => + dataStreams.closeStreamsFrom(participantIdentity); } diff --git a/lib/src/data_stream/data_streams.dart b/lib/src/data_stream/data_streams.dart new file mode 100644 index 000000000..8db118c9a --- /dev/null +++ b/lib/src/data_stream/data_streams.dart @@ -0,0 +1,82 @@ +// Copyright 2026 LiveKit, Inc. +// +// 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 'dart:io'; + +import '../core/room.dart'; +import '../e2ee/options.dart'; +import '../proto/livekit_models.pb.dart' as lk_models; +import '../types/data_stream.dart'; +import 'data_streams_native.dart' if (dart.library.js_interop) 'data_streams_web.dart' as impl; +import 'stream_writer.dart'; + +/// Owns the data-stream subsystem for one [Room]: the topic→handler registry, the send path, and +/// the routing of inbound packets to open readers. +/// +/// Two implementations sit behind this interface, chosen by conditional import: +/// +/// - **native** ([createDataStreams] in `data_streams_native.dart`) delegates to the Rust core in +/// `package:livekit_uniffi`, which implements data streams v2 — single-packet inline sends, +/// deflate-raw compression, and MTU-bounded headers. +/// - **web** (`data_streams_web.dart`) keeps the original Dart implementation. There is no way to +/// load a cdylib in a browser, so web stays on the v1 wire format. That interoperates: a v2 +/// sender sees web's pre-v2 `clientProtocol` and falls back to uncompressed multi-packet. +/// +/// A [Room] owns exactly one of these for its whole lifetime, created eagerly in the constructor. +/// It outlives connect/disconnect because handler registrations must survive a reconnect and be +/// registrable before the first connect. +abstract class DataStreams { + /// Handlers registered for incoming text streams, keyed by topic. + Map get textStreamHandlers; + + /// Handlers registered for incoming byte streams, keyed by topic. + Map get byteStreamHandlers; + + void registerTextStreamHandler(String topic, TextStreamHandler callback); + + void unregisterTextStreamHandler(String topic); + + void registerByteStreamHandler(String topic, ByteStreamHandler callback); + + void unregisterByteStreamHandler(String topic); + + Future sendText(String text, SendTextOptions? options); + + /// Sends an in-memory byte payload. Returns info about the stream created for it. + Future sendBytes(List bytes, SendBytesOptions? options); + + Future> sendFile(File file, SendFileOptions options); + + Future streamText(StreamTextOptions? options); + + Future streamBytes(StreamBytesOptions? options); + + /// Routes one already-decrypted inbound [lk_models.DataPacket] carrying a stream header, chunk + /// or trailer. + void handleIncomingPacket(lk_models.DataPacket packet, EncryptionType encryptionType); + + /// Fails every open stream sent by [identity] — they disconnected mid-send, so their readers + /// error rather than hanging. + Future closeStreamsFrom(String identity); + + /// Fails every open stream, e.g. on disconnect. Handler registrations survive, so streams + /// arriving after a reconnect are still handled. + Future reset(); + + /// Releases the underlying resources. The owning [Room] is being disposed. + Future dispose(); +} + +/// Builds the implementation for the current platform. +DataStreams createDataStreams(Room room) => impl.createDataStreams(room); diff --git a/lib/src/data_stream/data_streams_native.dart b/lib/src/data_stream/data_streams_native.dart new file mode 100644 index 000000000..5f28e40b1 --- /dev/null +++ b/lib/src/data_stream/data_streams_native.dart @@ -0,0 +1,553 @@ +// Copyright 2026 LiveKit, Inc. +// +// 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 'dart:async'; +import 'dart:convert'; +import 'dart:io'; +import 'dart:typed_data'; + +import 'package:fixnum/fixnum.dart'; +import 'package:livekit_uniffi/livekit_uniffi.dart' as ffi; +import 'package:path/path.dart' show basename; +import 'package:uuid/uuid.dart'; + +import '../core/room.dart'; +import '../e2ee/options.dart'; +import '../extensions.dart'; +import '../logger.dart'; +import '../participant/participant.dart'; +import '../proto/livekit_models.pb.dart' as lk_models; +import '../types/data_stream.dart'; +import '../types/other.dart'; +import 'data_streams.dart'; +import 'errors.dart'; +import 'ffi_bridged.dart'; +import 'stream_reader.dart'; +import 'stream_writer.dart'; + +DataStreams createDataStreams(Room room) => NativeDataStreams(room); + +/// Data streams backed by the Rust core in `package:livekit_uniffi`, which implements v2: +/// single-packet inline sends, deflate-raw compression, UTF-8-aware chunking and MTU-bounded +/// headers. This layer owns topic routing, the public type conversions, and the transport hop — +/// the wire format itself is entirely Rust's. +/// +/// The FFI boundary is serialized `DataPacket` bytes in both directions. Inbound, [Room] hands over +/// already-decrypted packets; outbound, packets come back encoded and are re-sent through +/// [Engine.sendDataPacket] so E2EE wrapping, reliable sequencing and resume-resend all still apply. +/// +/// Both managers are built through the core's `polled*` adapters rather than constructed directly. +/// The core normally pushes its output to a foreign delegate from its tokio runtime, which Dart +/// cannot accept: uniffi compiles a callback interface to `Pointer.fromFunction`, valid only on the +/// thread owning the isolate, so such a call aborts the VM outright with "Cannot invoke native +/// callback outside an isolate". The adapters keep the delegate on the Rust side and buffer into a +/// channel we await, so nothing crosses the FFI until we pull. +/// +/// [ffi.RemoteParticipantRegistryDelegate] is the one callback we do implement, and it is safe: it +/// is only ever called synchronously inside a `send*` future, and uniffi polls those from whichever +/// thread called `rust_future_poll` — us. +class NativeDataStreams implements DataStreams { + NativeDataStreams(Room room) : _room = WeakReference(room) { + final outgoing = ffi.polledOutgoingDataStreamManager(registry: _Registry(room)); + _outgoing = outgoing.manager; + _outgoingPackets = outgoing.packets; + unawaited(_pumpOutgoing()); + } + + /// Weak so the Rust-side strong reference to the registry delegate can't keep the [Room] alive. + final WeakReference _room; + + late final ffi.OutgoingDataStreamManager _outgoing; + late final ffi.OutgoingPacketQueue _outgoingPackets; + + /// Created on the first inbound packet rather than here, so a + /// [DataStreamOptions.maxPayloadByteLength] supplied at connect time is picked up. + ffi.IncomingDataStreamManager? _incoming; + ffi.IncomingStreamQueue? _incomingStreams; + + final Map _textStreamHandlers = {}; + final Map _byteStreamHandlers = {}; + + /// Serializes outbound sends so packet order survives the hop from the pump into the engine's + /// async send. + Future _sendChain = Future.value(); + + bool _disposed = false; + + @override + Map get textStreamHandlers => _textStreamHandlers; + + @override + Map get byteStreamHandlers => _byteStreamHandlers; + + @override + void registerTextStreamHandler(String topic, TextStreamHandler callback) => _textStreamHandlers[topic] = callback; + + @override + void unregisterTextStreamHandler(String topic) => _textStreamHandlers.remove(topic); + + @override + void registerByteStreamHandler(String topic, ByteStreamHandler callback) => _byteStreamHandlers[topic] = callback; + + @override + void unregisterByteStreamHandler(String topic) => _byteStreamHandlers.remove(topic); + + // MARK: - Send + + @override + Future sendText(String text, SendTextOptions? options) async { + // Attachments are still composed here: the core sends one stream, and each attachment is its + // own byte stream referenced by `attachedStreamIds` in the text header. + final attachments = options?.attachments ?? const []; + final attachmentIds = [for (var i = 0; i < attachments.length; i++) const Uuid().v4()]; + + final info = await mappingFfiErrors( + () => _outgoing.sendText( + text: text, + options: ffi.StreamTextOptions( + topic: options?.topic ?? '', + attributes: options?.attributes ?? const {}, + destinationIdentities: options?.destinationIdentities ?? const [], + attachedStreamIds: attachmentIds, + compress: options?.compress, + ), + ), + ); + + // The core does its own chunking, so there is no per-chunk progress to report; the text part + // is simply done. Attachments still report individually. + options?.onProgress?.call(attachments.isEmpty ? 1 : 1 / (attachments.length + 1)); + + for (var i = 0; i < attachments.length; i++) { + await _sendFileWithId( + attachmentIds[i], + attachments[i], + SendFileOptions(topic: options?.topic, destinationIdentities: options?.destinationIdentities ?? const []), + ); + options?.onProgress?.call((i + 2) / (attachments.length + 1)); + } + + return info.toLK( + sendingParticipantIdentity: _localIdentity, + encryptionType: _currentEncryptionType, + ); + } + + @override + Future sendBytes(List bytes, SendBytesOptions? options) async { + final info = await mappingFfiErrors( + () => _outgoing.sendBytes( + data: Uint8List.fromList(bytes), + options: ffi.StreamByteOptions( + topic: options?.topic ?? '', + attributes: options?.attributes ?? const {}, + destinationIdentities: options?.destinationIdentities ?? const [], + name: options?.name, + mimeType: options?.mimeType, + compress: options?.compress, + ), + ), + ); + return info.toLK( + sendingParticipantIdentity: _localIdentity, + encryptionType: _currentEncryptionType, + ); + } + + @override + Future> sendFile(File file, SendFileOptions options) async { + final id = const Uuid().v4(); + await _sendFileWithId(id, file, options); + return {'id': id}; + } + + Future _sendFileWithId(String id, File file, SendFileOptions options) async { + await mappingFfiErrors( + () => _outgoing.sendFile( + // The core streams the file from disk rather than buffering it. + path: file.path, + options: ffi.StreamByteOptions( + topic: options.topic ?? '', + attributes: const {}, + destinationIdentities: options.destinationIdentities, + id: id, + mimeType: options.mimeType, + name: basename(file.path), + ), + ), + ); + options.onProgress?.call(1); + } + + @override + Future streamText(StreamTextOptions? options) async { + final writer = await mappingFfiErrors( + () => _outgoing.streamText( + options: ffi.StreamTextOptions( + topic: options?.topic ?? '', + attributes: options?.attributes ?? const {}, + destinationIdentities: options?.destinationIdentities ?? const [], + id: options?.streamId, + operationType: options?.type?.toFfi(), + version: options?.version, + replyToStreamId: options?.replyToStreamId, + attachedStreamIds: options?.attachedStreamIds ?? const [], + generated: options?.generated, + ), + ), + ); + return TextStreamWriter( + writableStream: _FfiTextStreamWriter(writer), + info: writer.info().toLK( + sendingParticipantIdentity: _localIdentity, + encryptionType: _currentEncryptionType, + ), + onClose: () async => writer.dispose(), + ); + } + + @override + Future streamBytes(StreamBytesOptions? options) async { + final writer = await mappingFfiErrors( + () => _outgoing.streamBytes( + options: ffi.StreamByteOptions( + topic: options?.topic ?? '', + attributes: options?.attributes ?? const {}, + destinationIdentities: options?.destinationIdentities ?? const [], + id: options?.streamId, + mimeType: options?.mimeType, + name: options?.name, + totalLength: options?.totalSize, + ), + ), + ); + return ByteStreamWriter( + writableStream: _FfiByteStreamWriter(writer), + info: writer.info().toLK( + sendingParticipantIdentity: _localIdentity, + encryptionType: _currentEncryptionType, + ), + onClose: () async => writer.dispose(), + ); + } + + /// Drains outbound packets from the core and puts them on the wire, in order. + /// + /// The pump owns the queue's lifetime and is the only thing that may dispose it — freeing it + /// while a `nextPackets` is in flight is a use-after-free. [dispose] wakes us by closing the + /// queue rather than releasing it. + Future _pumpOutgoing() async { + try { + while (true) { + final batch = await _outgoingPackets.nextPackets(); + if (batch == null) break; // closed or shutting down + for (final encoded in batch) { + _enqueueSend(encoded); + } + } + } catch (e) { + logger.warning('[DataStreams] outgoing pump failed: $e'); + } finally { + _outgoingPackets.dispose(); + } + } + + void _enqueueSend(Uint8List encoded) { + _sendChain = _sendChain.then((_) async { + final room = _room.target; + if (room == null || _disposed) return; + try { + // Back through the engine rather than the data channel directly, so E2EE wrapping, + // reliable sequencing and resume-resend all still apply. + await room.engine.sendDataPacket( + lk_models.DataPacket.fromBuffer(encoded), + reliability: Reliability.reliable, + ); + } catch (e) { + // The core acknowledges sends unconditionally, so there is nobody to propagate this to. + logger.warning('[DataStreams] failed to send outbound packet: $e'); + } + }); + } + + // MARK: - Receive + + ffi.IncomingDataStreamManager _incomingManager() { + final existing = _incoming; + if (existing != null) return existing; + // Read now rather than at construction: this runs on the first inbound packet, i.e. after + // connect, so a cap supplied via `connect(connectOptions:)` is in effect by this point. + final incoming = ffi.polledIncomingDataStreamManager( + maxPayloadByteLength: _room.target?.connectOptions.dataStream.maxPayloadByteLength, + ); + _incoming = incoming.manager; + _incomingStreams = incoming.streams; + unawaited(_pumpIncoming(incoming.streams)); + return incoming.manager; + } + + @override + void handleIncomingPacket(lk_models.DataPacket packet, EncryptionType encryptionType) { + if (_disposed) return; + // The core decodes the header/chunk/trailer itself, so hand it the whole packet. + _incomingManager().handlePacketReceived(packet: packet.writeToBuffer()); + } + + /// Drains opened streams from the core and dispatches them to the registered topic handler. + /// + /// Owns the queue's lifetime, for the same reason as [_pumpOutgoing]. + Future _pumpIncoming(ffi.IncomingStreamQueue streams) async { + try { + while (true) { + final opened = await streams.nextOpenedStream(); + if (opened == null) break; // closed or shutting down + try { + _dispatchOpenedStream(opened); + } catch (e) { + logger.warning('[DataStreams] failed to dispatch opened stream: $e'); + } + } + } catch (e) { + logger.warning('[DataStreams] incoming pump failed: $e'); + } finally { + streams.dispose(); + } + } + + void _dispatchOpenedStream(ffi.OpenedStream opened) { + final identity = opened.identity; + final encryptionType = _currentEncryptionType; + + final textReader = opened.textReader; + if (textReader != null) { + final info = textReader.info().toLK( + sendingParticipantIdentity: identity, + encryptionType: encryptionType, + ); + final handler = _textStreamHandlers[info.topic]; + if (handler == null) { + logger.info('[DataStreams] ignoring text stream on unhandled topic "${info.topic}"'); + textReader.dispose(); + return; + } + // The core yields decoded pieces; re-frame them as protobuf chunks so the public reader — + // which is a Stream — behaves exactly as it did before. + final controller = _controllerFor( + info: info, + next: textReader.next, + toBytes: (piece) => Uint8List.fromList(utf8.encode(piece)), + streamId: info.id, + dispose: textReader.dispose, + ); + handler(TextStreamReader(info, controller, info.size), identity); + return; + } + + final byteReader = opened.byteReader; + if (byteReader != null) { + final info = byteReader.info().toLK( + sendingParticipantIdentity: identity, + encryptionType: encryptionType, + ); + final handler = _byteStreamHandlers[info.topic]; + if (handler == null) { + logger.info('[DataStreams] ignoring byte stream on unhandled topic "${info.topic}"'); + byteReader.dispose(); + return; + } + final controller = _controllerFor( + info: info, + next: byteReader.next, + toBytes: (piece) => piece, + streamId: info.id, + dispose: byteReader.dispose, + ); + handler(ByteStreamReader(info, controller, info.size), identity); + } + } + + /// Adapts the core's pull-based reader onto the [DataStreamController] the public readers wrap. + /// + /// Pulling is driven by the subscription: nothing is read until someone listens, and the loop + /// stops while the subscription is paused, so the core's backpressure is preserved rather than + /// buffering the whole stream into Dart. + /// + /// The pump owns the reader's lifetime and is the only thing that may dispose it. Disposing from + /// `onCancel` instead would free the Rust handle while a `next()` is still in flight — a + /// use-after-free that shows up as a SIGBUS, not a Dart exception. + DataStreamController _controllerFor({ + required BaseStreamInfo info, + required Future Function() next, + required Uint8List Function(T piece) toBytes, + required String streamId, + required void Function() dispose, + }) { + late final StreamController controller; + late final DataStreamController wrapper; + var chunkIndex = 0; + var running = false; + var cancelled = false; + var disposed = false; + + void disposeOnce() { + if (disposed) return; + disposed = true; + dispose(); + } + + Future pump() async { + if (running) return; + running = true; + try { + while (!cancelled && !controller.isClosed && !controller.isPaused) { + final piece = await next(); + if (piece == null) break; + if (cancelled || controller.isClosed) break; + wrapper.write( + lk_models.DataStream_Chunk( + streamId: streamId, + chunkIndex: Int64(chunkIndex++), + content: toBytes(piece), + ), + ); + } + // Paused means the consumer will resume us later, so leave the reader open. + if (cancelled || (!controller.isPaused && !controller.isClosed)) { + await wrapper.close(); + disposeOnce(); + } + } on ffi.DataStreamException catch (e) { + wrapper.error(toLKError(e)); + await wrapper.close(); + disposeOnce(); + } catch (e) { + wrapper.error( + DataStreamError( + reason: DataStreamErrorReason.AbnormalEnd, + message: 'Data stream failed: $e', + ), + ); + await wrapper.close(); + disposeOnce(); + } finally { + running = false; + } + } + + controller = StreamController( + onListen: () => unawaited(pump()), + onResume: () => unawaited(pump()), + // Only flag it: the pump disposes once it has stopped touching the reader. If it is blocked + // in `next()` the reader stays alive until that resolves, which is the safe order. + onCancel: () { + cancelled = true; + if (!running) disposeOnce(); + }, + ); + wrapper = DataStreamController( + info: info, + streamController: controller, + startTime: DateTime.timestamp().millisecondsSinceEpoch, + ); + return wrapper; + } + + // MARK: - Lifecycle + + @override + Future closeStreamsFrom(String identity) async { + _incoming?.abortStreamsFrom(identity: identity); + } + + @override + Future reset() async { + _incoming?.abortAllStreams(); + } + + @override + Future dispose() async { + if (_disposed) return; + _disposed = true; + // Error out open readers first so their handlers unwind, then close the queues so each pump + // wakes, exits and disposes the queue it owns. The managers have nothing awaiting them, so + // they can be released here. + _incoming?.abortAllStreams(); + _incomingStreams?.close(); + _incomingStreams = null; + _outgoingPackets.close(); + _incoming?.dispose(); + _incoming = null; + _outgoing.dispose(); + } + + // MARK: - Helpers + + String get _localIdentity => _room.target?.localParticipant?.identity ?? ''; + + /// The FFI normalizes every stream's encryption type to none — payload crypto happens in the + /// engine — so surface the room's data-channel setting to preserve the previous behavior. + EncryptionType get _currentEncryptionType { + final room = _room.target; + final enabled = room?.e2eeManager?.isDataChannelEncryptionEnabled ?? false; + return enabled ? EncryptionType.kGcm : EncryptionType.kNone; + } +} + +/// Answers the core's per-send eligibility questions from the room's current participant list. +/// +/// A separate object rather than [NativeDataStreams] itself because the Rust manager retains its +/// registry strongly; holding the room weakly here keeps that from pinning the room alive. +class _Registry implements ffi.RemoteParticipantRegistryDelegate { + _Registry(Room room) : _room = WeakReference(room); + + final WeakReference _room; + + @override + int remoteClientProtocol(String identity) => + _participant(identity)?.clientProtocol.toIntValue() ?? ClientProtocolVersion.v0.wireValue; + + @override + List remoteCapabilities(String identity) => + _participant(identity)?.capabilities.map((c) => c.toFfi()).toList() ?? const []; + + @override + List remoteIdentities() => _room.target?.remoteParticipants.keys.toList() ?? const []; + + Participant? _participant(String identity) => _room.target?.remoteParticipants[identity]; +} + +/// Bridges the core's text writer onto the [StreamWriter] the public writer wraps. +class _FfiTextStreamWriter implements StreamWriter { + _FfiTextStreamWriter(this._writer); + + final ffi.TextStreamWriter _writer; + + @override + Future write(String chunk) => mappingFfiErrors(() => _writer.write(text: chunk)); + + @override + Future close() => mappingFfiErrors(() => _writer.close()); +} + +class _FfiByteStreamWriter implements StreamWriter { + _FfiByteStreamWriter(this._writer); + + final ffi.ByteStreamWriter _writer; + + @override + Future write(Uint8List chunk) => mappingFfiErrors(() => _writer.write(data: chunk)); + + @override + Future close() => mappingFfiErrors(() => _writer.close()); +} diff --git a/lib/src/data_stream/data_streams_web.dart b/lib/src/data_stream/data_streams_web.dart new file mode 100644 index 000000000..fa759a3e9 --- /dev/null +++ b/lib/src/data_stream/data_streams_web.dart @@ -0,0 +1,549 @@ +// Copyright 2026 LiveKit, Inc. +// +// 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 'dart:async'; +import 'dart:io'; +import 'dart:typed_data'; + +import 'package:async/async.dart'; +import 'package:fixnum/fixnum.dart'; +import 'package:mime_type/mime_type.dart'; +import 'package:path/path.dart' show basename; +import 'package:uuid/uuid.dart'; + +import '../core/room.dart'; +import '../e2ee/options.dart'; +import '../internal/events.dart'; +import '../logger.dart'; +import '../options.dart'; +import '../proto/livekit_models.pb.dart' as lk_models; +import '../types/data_stream.dart'; +import '../types/other.dart'; +import 'data_streams.dart'; +import 'errors.dart'; +import 'stream_reader.dart'; +import 'stream_writer.dart'; + +DataStreams createDataStreams(Room room) => WebDataStreams(room); + +/// The original pure-Dart data-stream implementation, retained for web. +/// +/// `package:livekit_uniffi` ships a cdylib and cannot run in a browser, so web stays on the v1 +/// wire format: no single-packet inline sends, no compression. That interoperates cleanly — web +/// advertises [ClientProtocolVersion.v1], and a v2 sender seeing a pre-v2 recipient falls back to +/// uncompressed multi-packet framing, which this code understands. +/// +/// This is a move of the logic that previously lived in `Room` and `LocalParticipant`, unchanged +/// in behavior. +class WebDataStreams implements DataStreams { + WebDataStreams(this._room); + + final Room _room; + + @override + final Map textStreamHandlers = {}; + + @override + final Map byteStreamHandlers = {}; + + final Map> _byteStreamControllers = {}; + final Map> _textStreamControllers = {}; + + /// Content bytes delivered so far per stream id, checked against + /// [DataStreamOptions.maxPayloadByteLength]. + final Map _receivedBytes = {}; + + int get _maxPayloadByteLength => _room.connectOptions.dataStream.maxPayloadByteLength ?? kDefaultMaxPayloadByteLength; + + /// Whether a stream declaring [totalLength] is over the payload cap. Streams of unknown length + /// pass here and are capped as their chunks arrive instead. + bool _declaresOverCap(int? totalLength) => totalLength != null && totalLength > _maxPayloadByteLength; + + /// Fails an oversized stream's reader, after its handler has been given it. + /// + /// Matches the Rust core, which emits the stream-opened event before applying the cap: the + /// consumer is told the stream failed rather than never hearing about it. + Future _failOverCap( + DataStreamController controller, + String streamId, + ) async { + logger.warning( + 'incoming stream $streamId exceeds the maxPayloadByteLength of $_maxPayloadByteLength', + ); + controller.error(_payloadTooLarge()); + await controller.close(); + _forgetStream(streamId); + } + + @override + void registerTextStreamHandler(String topic, TextStreamHandler callback) => textStreamHandlers[topic] = callback; + + @override + void unregisterTextStreamHandler(String topic) => textStreamHandlers.remove(topic); + + @override + void registerByteStreamHandler(String topic, ByteStreamHandler callback) => byteStreamHandlers[topic] = callback; + + @override + void unregisterByteStreamHandler(String topic) => byteStreamHandlers.remove(topic); + + // MARK: - Receive + + @override + void handleIncomingPacket(lk_models.DataPacket packet, EncryptionType encryptionType) { + if (packet.hasStreamHeader()) { + unawaited(_handleStreamHeader(packet.streamHeader, packet.participantIdentity, encryptionType)); + } else if (packet.hasStreamChunk()) { + _handleStreamChunk(packet.streamChunk, encryptionType); + } else if (packet.hasStreamTrailer()) { + unawaited(_handleStreamTrailer(packet.streamTrailer, encryptionType)); + } + } + + Future _handleStreamHeader( + lk_models.DataStream_Header streamHeader, + String participantIdentity, + EncryptionType encryptionType, + ) async { + if (streamHeader.hasByteHeader()) { + final streamHandlerCallback = byteStreamHandlers[streamHeader.topic]; + + if (streamHandlerCallback == null) { + logger.info('ignoring incoming byte stream due to no handler for topic ${streamHeader.topic}'); + return; + } + + final info = ByteStreamInfo( + id: streamHeader.streamId, + name: streamHeader.byteHeader.name, + mimeType: streamHeader.mimeType, + size: streamHeader.hasTotalLength() ? streamHeader.totalLength.toInt() : 0, + topic: streamHeader.topic, + timestamp: streamHeader.timestamp.toInt(), + attributes: streamHeader.attributes, + sendingParticipantIdentity: participantIdentity, + encryptionType: encryptionType, + ); + + if (_byteStreamControllers.containsKey(streamHeader.streamId)) { + throw DataStreamError( + message: 'A byte stream with id "${streamHeader.streamId}" is already open.', + reason: DataStreamErrorReason.AlreadyOpened, + ); + } + + final controller = DataStreamController( + info: info, + streamController: StreamController(), + startTime: DateTime.timestamp().millisecondsSinceEpoch, + ); + _byteStreamControllers[streamHeader.streamId] = controller; + + streamHandlerCallback(ByteStreamReader(info, controller, info.size), participantIdentity); + if (_declaresOverCap(streamHeader.hasTotalLength() ? info.size : null)) { + await _failOverCap(controller, streamHeader.streamId); + } + return; + } + + if (streamHeader.hasTextHeader()) { + final streamHandlerCallback = textStreamHandlers[streamHeader.topic]; + + if (streamHandlerCallback == null) { + logger.warning('ignoring incoming text stream due to no handler for topic ${streamHeader.topic}'); + return; + } + + final info = TextStreamInfo( + id: streamHeader.streamId, + mimeType: streamHeader.mimeType, + size: streamHeader.hasTotalLength() ? streamHeader.totalLength.toInt() : 0, + topic: streamHeader.topic, + timestamp: streamHeader.timestamp.toInt(), + attributes: streamHeader.attributes, + replyToStreamId: streamHeader.textHeader.replyToStreamId, + attachedStreamIds: streamHeader.textHeader.attachedStreamIds, + version: streamHeader.textHeader.version, + generated: streamHeader.textHeader.generated, + operationType: TextStreamOperationType.fromPBType(streamHeader.textHeader.operationType), + sendingParticipantIdentity: participantIdentity, + encryptionType: encryptionType, + ); + + if (_textStreamControllers.containsKey(streamHeader.streamId)) { + throw DataStreamError( + message: 'A text stream with id "${streamHeader.streamId}" is already open.', + reason: DataStreamErrorReason.AlreadyOpened, + ); + } + + final controller = DataStreamController( + info: info, + streamController: StreamController(), + startTime: DateTime.timestamp().millisecondsSinceEpoch, + ); + _textStreamControllers[streamHeader.streamId] = controller; + + streamHandlerCallback(TextStreamReader(info, controller, info.size), participantIdentity); + if (_declaresOverCap(streamHeader.hasTotalLength() ? info.size : null)) { + await _failOverCap(controller, streamHeader.streamId); + } + } + } + + void _handleStreamChunk(lk_models.DataStream_Chunk chunk, EncryptionType encryptionType) { + final textController = _textStreamControllers[chunk.streamId]; + if (textController != null) { + if (textController.info.encryptionType != encryptionType) { + textController.error(_encryptionMismatch()); + _forgetStream(chunk.streamId); + } else if (chunk.content.isNotEmpty) { + if (_exceedsPayloadCap(chunk)) { + textController.error(_payloadTooLarge()); + unawaited(textController.close()); + _forgetStream(chunk.streamId); + return; + } + textController.write(chunk); + } + } + + final byteController = _byteStreamControllers[chunk.streamId]; + if (byteController != null) { + if (byteController.info.encryptionType != encryptionType) { + byteController.error(_encryptionMismatch()); + _forgetStream(chunk.streamId); + } else if (chunk.content.isNotEmpty) { + if (_exceedsPayloadCap(chunk)) { + byteController.error(_payloadTooLarge()); + unawaited(byteController.close()); + _forgetStream(chunk.streamId); + return; + } + byteController.write(chunk); + } + } + } + + /// Accumulates this chunk against the stream's running total, returning true once the payload + /// cap is passed. + bool _exceedsPayloadCap(lk_models.DataStream_Chunk chunk) { + final total = (_receivedBytes[chunk.streamId] ?? 0) + chunk.content.length; + _receivedBytes[chunk.streamId] = total; + return total > _maxPayloadByteLength; + } + + DataStreamError _payloadTooLarge() => DataStreamError( + message: 'Stream payload exceeds the maxPayloadByteLength of $_maxPayloadByteLength', + reason: DataStreamErrorReason.LengthExceeded, + ); + + void _forgetStream(String streamId) { + _textStreamControllers.remove(streamId); + _byteStreamControllers.remove(streamId); + _receivedBytes.remove(streamId); + } + + Future _handleStreamTrailer(lk_models.DataStream_Trailer trailer, EncryptionType encryptionType) async { + final textController = _textStreamControllers[trailer.streamId]; + if (textController != null) { + if (textController.info.encryptionType != encryptionType) { + textController.error(_encryptionMismatch()); + _forgetStream(trailer.streamId); + return; + } + textController.info.attributes = {...textController.info.attributes, ...trailer.attributes}; + await textController.close(); + _forgetStream(trailer.streamId); + } + + final byteController = _byteStreamControllers[trailer.streamId]; + if (byteController != null) { + if (byteController.info.encryptionType != encryptionType) { + byteController.error(_encryptionMismatch()); + _forgetStream(trailer.streamId); + return; + } + byteController.info.attributes = {...byteController.info.attributes, ...trailer.attributes}; + await byteController.close(); + _forgetStream(trailer.streamId); + } + } + + DataStreamError _encryptionMismatch() => DataStreamError( + message: 'Encryption type mismatch', + reason: DataStreamErrorReason.EncryptionTypeMismatch, + ); + + // MARK: - Send + + @override + Future sendText(String text, SendTextOptions? options) async { + final streamId = const Uuid().v4(); + final totalTextLength = text.codeUnits.length; + + final fileIds = options?.attachments.map((f) => const Uuid().v4()).toList(); + final len = (fileIds != null && fileIds.isNotEmpty) ? fileIds.length + 1 : 1; + final progresses = List.filled(len, 0); + + void handleProgress(num progress, int idx) { + progresses[idx] = progress; + final totalProgress = progresses.reduce((acc, val) => acc + val); + options?.onProgress?.call(totalProgress.toDouble() / len); + } + + final writer = await streamText( + StreamTextOptions( + streamId: streamId, + totalSize: totalTextLength, + destinationIdentities: options?.destinationIdentities ?? [], + topic: options?.topic, + attachedStreamIds: fileIds ?? [], + attributes: options?.attributes ?? {}, + ), + ); + + await writer.write(text); + handleProgress(1, 0); + await writer.close(); + + if (options?.attachments != null) { + var idx = 0; + await Future.wait( + options?.attachments.map((file) { + final curIdx = idx++; + return _sendFile( + fileIds![curIdx], + file, + SendFileOptions( + topic: options.topic, + mimeType: mime(basename(file.path)), + onProgress: (progress) => handleProgress(progress, curIdx + 1), + ), + ); + }).toList() ?? + [], + ); + } + return writer.info; + } + + @override + Future sendBytes(List bytes, SendBytesOptions? options) async { + final writer = await streamBytes( + StreamBytesOptions( + name: options?.name ?? 'unknown', + mimeType: options?.mimeType ?? 'application/octet-stream', + topic: options?.topic, + destinationIdentities: options?.destinationIdentities ?? [], + attributes: options?.attributes ?? {}, + totalSize: bytes.length, + ), + ); + await writer.write(Uint8List.fromList(bytes)); + await writer.close(); + return writer.info; + } + + @override + Future> sendFile(File file, SendFileOptions options) async { + final streamId = const Uuid().v4(); + await _sendFile(streamId, file, options); + return {'id': streamId}; + } + + Future _sendFile(String streamId, File file, SendFileOptions options) async { + final totalLength = await file.length(); + final writer = await streamBytes( + StreamBytesOptions( + streamId: streamId, + totalSize: totalLength, + name: basename(file.path), + mimeType: options.mimeType, + topic: options.topic, + destinationIdentities: options.destinationIdentities, + encryptionType: options.encryptionType, + ), + ); + + final totalChunks = (totalLength / kStreamChunkSize).ceil(); + final reader = ChunkedStreamReader(file.openRead()); + try { + for (var i = 0; i < totalChunks; i++) { + final chunk = await reader.readBytes(kStreamChunkSize); + if (chunk.isEmpty) break; + await writer.write(Uint8List.fromList(chunk)); + options.onProgress?.call((i + 1) / totalChunks); + } + } finally { + await reader.cancel(); + await writer.close(); + } + } + + @override + Future streamText(StreamTextOptions? options) async { + final streamId = options?.streamId ?? const Uuid().v4(); + final timestamp = DateTime.timestamp().millisecondsSinceEpoch; + + final info = TextStreamInfo( + id: streamId, + mimeType: 'text/plain', + timestamp: timestamp, + topic: options?.topic ?? '', + size: options?.totalSize ?? 0, + replyToStreamId: options?.replyToStreamId, + attachedStreamIds: options?.attachedStreamIds ?? [], + version: options?.version, + generated: options?.generated ?? false, + operationType: options?.type, + sendingParticipantIdentity: _room.localParticipant?.identity ?? '', + attributes: options?.attributes ?? {}, + ); + + final header = lk_models.DataStream_Header( + streamId: streamId, + mimeType: info.mimeType, + topic: info.topic, + timestamp: Int64(timestamp), + totalLength: options?.totalSize != null ? Int64(options!.totalSize!) : null, + attributes: options?.attributes.entries, + textHeader: lk_models.DataStream_TextHeader( + version: options?.version, + attachedStreamIds: options?.attachedStreamIds, + replyToStreamId: options?.replyToStreamId, + generated: options?.generated ?? false, + operationType: options?.type?.toPBType(), + ), + ); + + final destinationIdentities = options?.destinationIdentities ?? const []; + final packet = lk_models.DataPacket( + kind: lk_models.DataPacket_Kind.RELIABLE, + destinationIdentities: destinationIdentities, + streamHeader: header, + ); + await _room.engine.sendDataPacket(packet, reliability: Reliability.reliable); + + final writableStream = WritableStream( + destinationIdentities: destinationIdentities, + engine: _room.engine, + streamId: streamId, + ); + + return TextStreamWriter( + writableStream: writableStream, + info: info, + onClose: _closeOnEngineClose(writableStream), + ); + } + + @override + Future streamBytes(StreamBytesOptions? options) async { + final streamId = options?.streamId ?? const Uuid().v4(); + final timestamp = DateTime.timestamp().millisecondsSinceEpoch; + + final info = ByteStreamInfo( + id: streamId, + name: options?.name ?? 'unknown', + mimeType: options?.mimeType ?? 'application/octet-stream', + timestamp: timestamp, + topic: options?.topic ?? '', + size: options?.totalSize ?? 0, + attributes: options?.attributes ?? {}, + sendingParticipantIdentity: _room.localParticipant?.identity ?? '', + ); + + final header = lk_models.DataStream_Header( + streamId: streamId, + mimeType: info.mimeType, + topic: info.topic, + timestamp: Int64(timestamp), + totalLength: options?.totalSize != null ? Int64(options!.totalSize!) : null, + attributes: options?.attributes.entries, + encryptionType: options?.encryptionType, + byteHeader: lk_models.DataStream_ByteHeader(name: info.name), + ); + + final destinationIdentities = options?.destinationIdentities ?? const []; + final packet = lk_models.DataPacket( + kind: lk_models.DataPacket_Kind.RELIABLE, + destinationIdentities: destinationIdentities, + streamHeader: header, + ); + await _room.engine.sendDataPacket(packet, reliability: Reliability.reliable); + + final writableStream = WritableStream( + destinationIdentities: destinationIdentities, + engine: _room.engine, + streamId: streamId, + ); + + return ByteStreamWriter( + writableStream: writableStream, + info: info, + onClose: _closeOnEngineClose(writableStream), + ); + } + + /// Closes the stream if the engine shuts down first, and unsubscribes that listener once the + /// writer closes normally. + Future Function() _closeOnEngineClose(WritableStream writableStream) { + final cancel = _room.engine.events.once((_) { + unawaited(writableStream.close()); + }); + return () async => cancel?.call(); + } + + // MARK: - Lifecycle + + @override + Future closeStreamsFrom(String participantIdentity) async { + final texts = _textStreamControllers.values + .where((c) => c.info.sendingParticipantIdentity == participantIdentity) + .toList(); + final bytes = _byteStreamControllers.values + .where((c) => c.info.sendingParticipantIdentity == participantIdentity) + .toList(); + if (texts.isEmpty && bytes.isEmpty) return; + + final abnormalEndError = DataStreamError( + message: 'Participant $participantIdentity unexpectedly disconnected in the middle of sending data', + reason: DataStreamErrorReason.AbnormalEnd, + ); + for (final controller in bytes) { + controller.error(abnormalEndError); + await controller.close(); + _forgetStream(controller.info.id); + } + for (final controller in texts) { + controller.error(abnormalEndError); + await controller.close(); + _forgetStream(controller.info.id); + } + } + + @override + Future reset() async { + for (final controller in [..._textStreamControllers.values, ..._byteStreamControllers.values]) { + await controller.close(); + } + _textStreamControllers.clear(); + _byteStreamControllers.clear(); + _receivedBytes.clear(); + } + + @override + Future dispose() => reset(); +} diff --git a/lib/src/data_stream/ffi_bridged.dart b/lib/src/data_stream/ffi_bridged.dart new file mode 100644 index 000000000..78d6c3b62 --- /dev/null +++ b/lib/src/data_stream/ffi_bridged.dart @@ -0,0 +1,121 @@ +// Copyright 2026 LiveKit, Inc. +// +// 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 'package:livekit_uniffi/livekit_uniffi.dart' as ffi; + +import '../e2ee/options.dart'; +import '../types/client_capability.dart'; +import '../types/data_stream.dart'; +import 'errors.dart'; + +/// Conversions between this SDK's public data-stream types and the generated `livekit_uniffi` +/// ones. +/// +/// Mirrors the `FFIBridged` marker the Swift SDK uses: bridging lives here rather than on the +/// public types, so those stay free of any `livekit_uniffi` import. Dart has no `internal import` +/// to enforce that, so the rule is by convention — see AGENTS.md. +/// +/// The FFI's stream info carries no encryption type (the Rust core normalizes it to `none` and +/// expects already-decrypted packets), so callers inject the room's current one. +extension FfiTextStreamInfo on ffi.TextStreamInfo { + TextStreamInfo toLK({ + required String sendingParticipantIdentity, + required EncryptionType encryptionType, + }) => TextStreamInfo( + id: id, + mimeType: mimeType, + topic: topic, + timestamp: timestampMs, + size: totalLength ?? 0, + attributes: attributes, + replyToStreamId: replyToStreamId, + attachedStreamIds: attachedStreamIds, + version: version, + generated: generated, + operationType: operationType.toLK(), + sendingParticipantIdentity: sendingParticipantIdentity, + encryptionType: encryptionType, + ); +} + +extension FfiByteStreamInfo on ffi.ByteStreamInfo { + ByteStreamInfo toLK({ + required String sendingParticipantIdentity, + required EncryptionType encryptionType, + }) => ByteStreamInfo( + id: id, + mimeType: mimeType, + topic: topic, + timestamp: timestampMs, + size: totalLength ?? 0, + attributes: attributes, + name: name, + sendingParticipantIdentity: sendingParticipantIdentity, + encryptionType: encryptionType, + ); +} + +extension FfiOperationType on ffi.OperationType { + TextStreamOperationType toLK() => switch (this) { + ffi.OperationType.create => TextStreamOperationType.create, + ffi.OperationType.update => TextStreamOperationType.update, + ffi.OperationType.delete => TextStreamOperationType.delete, + ffi.OperationType.reaction => TextStreamOperationType.reaction, + }; +} + +extension LKTextStreamOperationType on TextStreamOperationType { + ffi.OperationType toFfi() => switch (this) { + TextStreamOperationType.create => ffi.OperationType.create, + TextStreamOperationType.update => ffi.OperationType.update, + TextStreamOperationType.delete => ffi.OperationType.delete, + TextStreamOperationType.reaction => ffi.OperationType.reaction, + }; +} + +extension LKClientCapability on ClientCapability { + ffi.ClientCapability toFfi() => switch (this) { + ClientCapability.packetTrailer => ffi.ClientCapability.packetTrailer, + ClientCapability.compressionDeflateRaw => ffi.ClientCapability.compressionDeflateRaw, + }; +} + +/// Maps an FFI error onto the public [DataStreamError] set. +/// +/// Lossy: the public reasons predate the Rust core and don't cover every case, so several collapse +/// onto the closest existing one. The FFI's own message is kept so nothing is lost for debugging. +DataStreamError toLKError(ffi.DataStreamException e) { + final reason = switch (e) { + ffi.AbnormalEndDataStreamException() => DataStreamErrorReason.AbnormalEnd, + ffi.IoDataStreamException() => DataStreamErrorReason.AbnormalEnd, + ffi.Utf8DataStreamException() => DataStreamErrorReason.DecodeFailed, + ffi.DecompressionDataStreamException() => DataStreamErrorReason.DecodeFailed, + ffi.LengthExceededDataStreamException() => DataStreamErrorReason.LengthExceeded, + ffi.HeaderTooLargeDataStreamException() => DataStreamErrorReason.LengthExceeded, + ffi.PayloadTooLargeDataStreamException() => DataStreamErrorReason.LengthExceeded, + ffi.IncompleteDataStreamException() => DataStreamErrorReason.Incomplete, + ffi.EncryptionTypeMismatchDataStreamException() => DataStreamErrorReason.EncryptionTypeMismatch, + _ => DataStreamErrorReason.AbnormalEnd, + }; + return DataStreamError(reason: reason, message: e.toString()); +} + +/// Runs [body], translating any FFI error into the public [DataStreamError]. +Future mappingFfiErrors(Future Function() body) async { + try { + return await body(); + } on ffi.DataStreamException catch (e) { + throw toLKError(e); + } +} diff --git a/lib/src/internal/events.dart b/lib/src/internal/events.dart index 63c54a831..4b6cc6089 100644 --- a/lib/src/internal/events.dart +++ b/lib/src/internal/events.dart @@ -697,46 +697,16 @@ class EngineRPCAckReceivedEvent with EngineEvent, InternalEvent { } @internal -class EngineDataStreamHeaderEvent with EngineEvent, InternalEvent { - final lk_models.DataStream_Header header; +class EngineDataStreamPacketEvent with EngineEvent, InternalEvent { + /// The whole already-decrypted packet, carrying a stream header, chunk or trailer. + /// + /// Kept intact rather than split per part: the native data-stream path hands the serialized + /// packet straight to the Rust core, which decodes it itself. + final lk_models.DataPacket packet; final String identity; final EncryptionType encryptionType; - const EngineDataStreamHeaderEvent({ - required this.header, - required this.identity, - required this.encryptionType, - }); - - @override - String toString() => - '${runtimeType}' - '(header: ${header}, identity: ${identity}, encryptionType: ${encryptionType})'; -} - -@internal -class EngineDataStreamChunkEvent with EngineEvent, InternalEvent { - final lk_models.DataStream_Chunk chunk; - final EncryptionType encryptionType; - final String identity; - const EngineDataStreamChunkEvent({ - required this.chunk, - required this.identity, - required this.encryptionType, - }); - - @override - String toString() => - '${runtimeType}' - '(chunk: ${chunk}, identity: ${identity}, encryptionType: ${encryptionType})'; -} - -@internal -class EngineDataStreamTrailerEvent with EngineEvent, InternalEvent { - final lk_models.DataStream_Trailer trailer; - final String identity; - final EncryptionType encryptionType; - const EngineDataStreamTrailerEvent({ - required this.trailer, + const EngineDataStreamPacketEvent({ + required this.packet, required this.identity, required this.encryptionType, }); @@ -744,7 +714,7 @@ class EngineDataStreamTrailerEvent with EngineEvent, InternalEvent { @override String toString() => '${runtimeType}' - '(trailer: ${trailer}, identity: ${identity}, encryptionType: ${encryptionType})'; + '(packet: ${packet.whichValue()}, identity: ${identity}, encryptionType: ${encryptionType})'; } @internal diff --git a/lib/src/options.dart b/lib/src/options.dart index 34d193f6a..51608e802 100644 --- a/lib/src/options.dart +++ b/lib/src/options.dart @@ -219,15 +219,32 @@ class ConnectOptions { final Timeouts timeouts; + /// Tuning for incoming data streams. + final DataStreamOptions dataStream; + const ConnectOptions({ this.autoSubscribe = true, this.rtcConfiguration = const RTCConfiguration(), this.protocolVersion = ProtocolVersion.v16, this.clientProtocolVersion = ClientProtocolVersion.current, this.timeouts = Timeouts.defaultTimeouts, + this.dataStream = const DataStreamOptions(), }); } +/// Options for receiving data streams. +/// {@category Room} +class DataStreamOptions { + /// Largest payload, in bytes, that a single incoming stream may deliver. If unset, defaults to + /// 5gb. + final int? maxPayloadByteLength; + + const DataStreamOptions({this.maxPayloadByteLength}); +} + +/// Default for [DataStreamOptions.maxPayloadByteLength]; matches the Rust core's own default. +const int kDefaultMaxPayloadByteLength = 5000000000; + /// Options used to modify the behavior of the [Room]. /// {@category Room} class RoomOptions { diff --git a/lib/src/participant/local.dart b/lib/src/participant/local.dart index 89a7e71c4..4956e1b5c 100644 --- a/lib/src/participant/local.dart +++ b/lib/src/participant/local.dart @@ -16,18 +16,11 @@ import 'dart:async'; import 'dart:io'; -import 'dart:math'; -import 'dart:typed_data' show Uint8List; import 'package:flutter/foundation.dart' show kIsWeb; -import 'package:async/async.dart'; -import 'package:fixnum/fixnum.dart'; import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; import 'package:meta/meta.dart'; -import 'package:mime_type/mime_type.dart'; -import 'package:path/path.dart'; -import 'package:uuid/uuid.dart'; import '../core/engine.dart'; import '../core/room.dart'; @@ -1010,221 +1003,20 @@ class LocalParticipant extends Participant { } extension DataStreamParticipantMethods on LocalParticipant { - Future sendText(String text, {SendTextOptions? options}) async { - final streamId = Uuid().v4(); - final textInBytes = text.codeUnits; - final totalTextLength = textInBytes.length; - - final fileIds = options?.attachments.map((f) => Uuid().v4()).toList(); - var len = 0; - if (fileIds != null && fileIds.isNotEmpty) { - len = fileIds.length + 1; - } else { - len = 1; - } - final progresses = List.filled(len, 0); - - handleProgress(num progress, int idx) { - progresses[idx] = progress; - final totalProgress = progresses.reduce((acc, val) => acc + val); - options?.onProgress?.call(totalProgress.toDouble() / len); - } - - final writer = await streamText( - StreamTextOptions( - streamId: streamId, - totalSize: totalTextLength, - destinationIdentities: options?.destinationIdentities ?? [], - topic: options?.topic, - attachedStreamIds: fileIds ?? [], - attributes: options?.attributes ?? {}, - ), - ); - - await writer.write(text); - // set text part of progress to 1 - handleProgress(1, 0); - - await writer.close(); - - if (options?.attachments != null) { - var idx = 0; - await Future.wait( - options?.attachments.map( - (file) { - final curIdx = idx++; - return _sendFile( - fileIds![curIdx], - file, - SendFileOptions( - topic: options.topic, - mimeType: mime(basename(file.path)), - onProgress: (progress) { - handleProgress(progress, curIdx + 1); - }, - ), - ); - }, - ).toList() ?? - [], - ); - } - return writer.info; - } - - Future streamText(StreamTextOptions? options) async { - final streamId = options?.streamId ?? Uuid().v4(); - final timestamp = DateTime.timestamp().millisecondsSinceEpoch; - - final info = TextStreamInfo( - id: streamId, - mimeType: 'text/plain', - timestamp: timestamp, - topic: options?.topic ?? '', - size: options?.totalSize ?? 0, - replyToStreamId: options?.replyToStreamId, - attachedStreamIds: options?.attachedStreamIds ?? [], - version: options?.version, - generated: options?.generated ?? false, - operationType: options?.type, - sendingParticipantIdentity: identity, - ); - - final header = lk_models.DataStream_Header( - streamId: streamId, - mimeType: info.mimeType, - topic: info.topic, - timestamp: Int64(timestamp), - totalLength: options?.totalSize != null ? Int64(options!.totalSize!) : null, - attributes: options?.attributes.entries, - textHeader: lk_models.DataStream_TextHeader( - version: options?.version, - attachedStreamIds: options?.attachedStreamIds, - replyToStreamId: options?.replyToStreamId, - generated: options?.generated ?? false, - operationType: options?.type?.toPBType(), - ), - ); - - final destinationIdentities = options?.destinationIdentities; - final packet = lk_models.DataPacket( - kind: lk_models.DataPacket_Kind.RELIABLE, - destinationIdentities: destinationIdentities, - streamHeader: header, - ); - await room.engine.sendDataPacket(packet, reliability: Reliability.reliable); - - final writableStream = WritableStream( - destinationIdentities: destinationIdentities!, - engine: room.engine, - streamId: streamId, - ); - - onEngineClose() async { - await writableStream.close(); - } - - final cancelFun = room.engine.events.once((_) => onEngineClose); - - final writer = TextStreamWriter(writableStream: writableStream, info: info, onClose: cancelFun); - - return writer; - } - - Future> sendFile( - File file, { - required SendFileOptions options, - }) async { - final streamId = Uuid().v4(); - await _sendFile(streamId, file, options); - return {'id': streamId}; - } - - Future _sendFile( - String streamId, - File file, - SendFileOptions options, - ) async { - final totalLength = await file.length(); - - final streamBytesOptions = StreamBytesOptions( - streamId: streamId, - totalSize: totalLength, - topic: options.topic, - mimeType: options.mimeType ?? mime(basename(file.path)), - name: basename(file.path), - destinationIdentities: options.destinationIdentities, - encryptionType: options.encryptionType, - ); - - final writer = await streamBytes(streamBytesOptions); - - final reader = ChunkedStreamReader(file.openRead()); - - final totalChunks = (totalLength / kStreamChunkSize).ceil(); - for (var i = 0; i < totalChunks; i++) { - final chunkData = await reader.readBytes(min((i + 1) * kStreamChunkSize, kStreamChunkSize)); - await writer.write(chunkData); - options.onProgress?.call((i + 1) / totalChunks); - } - await writer.close(); - } - - Future streamBytes(StreamBytesOptions? options) async { - final streamId = options?.streamId ?? Uuid().v4(); - final timestamp = DateTime.timestamp().millisecondsSinceEpoch; - - final info = ByteStreamInfo( - name: options?.name ?? 'unknown', - id: streamId, - mimeType: options?.mimeType ?? 'application/octet-stream', - timestamp: timestamp, - topic: options?.topic ?? '', - size: options?.totalSize ?? 0, - attributes: options?.attributes ?? {}, - sendingParticipantIdentity: identity, - ); + /// Sends a complete text payload as a data stream. + Future sendText(String text, {SendTextOptions? options}) => room.dataStreams.sendText(text, options); - final header = lk_models.DataStream_Header( - totalLength: options?.totalSize != null ? Int64(options!.totalSize!) : null, - mimeType: info.mimeType, - streamId: streamId, - topic: options?.topic, - encryptionType: options?.encryptionType, - timestamp: Int64(timestamp), - byteHeader: lk_models.DataStream_ByteHeader( - name: info.name, - ), - attributes: options?.attributes.entries, - ); + /// Sends a complete in-memory byte payload as a data stream. + Future sendBytes(List bytes, {SendBytesOptions? options}) => + room.dataStreams.sendBytes(bytes, options); - final destinationIdentities = options?.destinationIdentities; - final packet = lk_models.DataPacket( - kind: lk_models.DataPacket_Kind.RELIABLE, - destinationIdentities: destinationIdentities, - streamHeader: header, - ); - - await room.engine.sendDataPacket(packet, reliability: Reliability.reliable); + /// Sends a file as a byte data stream, returning `{'id': streamId}`. + Future> sendFile(File file, {required SendFileOptions options}) => + room.dataStreams.sendFile(file, options); - final writableStream = WritableStream( - destinationIdentities: destinationIdentities, - streamId: streamId, - engine: room.engine, - ); + /// Opens an incremental text stream. Incremental writers are never compressed or inlined. + Future streamText(StreamTextOptions? options) => room.dataStreams.streamText(options); - onEngineClose() async { - await writableStream.close(); - } - - final cancelFun = room.engine.events.once((_) => onEngineClose); - - final byteWriter = ByteStreamWriter( - writableStream: writableStream, - info: info, - onClose: cancelFun, - ); - - return byteWriter; - } + /// Opens an incremental byte stream. Incremental writers are never compressed or inlined. + Future streamBytes(StreamBytesOptions? options) => room.dataStreams.streamBytes(options); } diff --git a/lib/src/participant/participant.dart b/lib/src/participant/participant.dart index 90a33e654..75e30a454 100644 --- a/lib/src/participant/participant.dart +++ b/lib/src/participant/participant.dart @@ -24,6 +24,7 @@ import '../managers/event.dart'; import '../proto/livekit_models.pb.dart' as lk_models; import '../publication/track_publication.dart'; import '../support/disposable.dart'; +import '../types/client_capability.dart'; import '../types/other.dart'; import '../types/participant_permissions.dart'; import '../types/participant_state.dart'; @@ -104,6 +105,16 @@ abstract class Participant extends DisposableChangeN /// supported version. ClientProtocolVersion get clientProtocol => ClientProtocolVersion.fromIntValue(_participantInfo?.clientProtocol); + /// Optional feature capabilities this participant advertises, mirrored by the server from its + /// `ClientInfo`. Consulted by the data-stream send path to decide per-recipient eligibility for + /// v2 features such as deflate-raw compression. + /// + /// The protocol's `CAP_UNUSED` placeholder and any value newer than this SDK are omitted. + List get capabilities => [ + for (final value in _participantInfo?.capabilities ?? const []) + ?ClientCapability.fromProto(value), + ]; + /// if [Participant] is currently speaking. bool get isSpeaking => _isSpeaking; diff --git a/lib/src/proto/livekit_models.pb.dart b/lib/src/proto/livekit_models.pb.dart index 6ffa0300e..b7a824a97 100644 --- a/lib/src/proto/livekit_models.pb.dart +++ b/lib/src/proto/livekit_models.pb.dart @@ -728,6 +728,7 @@ class ParticipantInfo extends $pb.GeneratedMessage { $core.Iterable? kindDetails, $core.Iterable? dataTracks, $core.int? clientProtocol, + $core.Iterable? capabilities, }) { final result = create(); if (sid != null) result.sid = sid; @@ -748,6 +749,7 @@ class ParticipantInfo extends $pb.GeneratedMessage { if (kindDetails != null) result.kindDetails.addAll(kindDetails); if (dataTracks != null) result.dataTracks.addAll(dataTracks); if (clientProtocol != null) result.clientProtocol = clientProtocol; + if (capabilities != null) result.capabilities.addAll(capabilities); return result; } @@ -786,6 +788,10 @@ class ParticipantInfo extends $pb.GeneratedMessage { defaultEnumValue: ParticipantInfo_KindDetail.CLOUD_AGENT) ..pPM(19, _omitFieldNames ? '' : 'dataTracks', subBuilder: DataTrackInfo.create) ..aI(20, _omitFieldNames ? '' : 'clientProtocol') + ..pc(21, _omitFieldNames ? '' : 'capabilities', $pb.PbFieldType.KE, + valueOf: ClientInfo_Capability.valueOf, + enumValues: ClientInfo_Capability.values, + defaultEnumValue: ClientInfo_Capability.CAP_UNUSED) ..hasRequiredFields = false; @$core.Deprecated('See https://github.com/google/protobuf.dart/issues/998.') @@ -950,6 +956,11 @@ class ParticipantInfo extends $pb.GeneratedMessage { $core.bool hasClientProtocol() => $_has(17); @$pb.TagNumber(20) void clearClientProtocol() => $_clearField(20); + + /// capabilities the participant's client advertises, mirrored from ClientInfo. + /// Lets other participants perform client-side feature detection. + @$pb.TagNumber(21) + $pb.PbList get capabilities => $_getList(18); } class Encryption extends $pb.GeneratedMessage { @@ -5397,6 +5408,8 @@ class DataStream_Header extends $pb.GeneratedMessage { $core.Iterable<$core.MapEntry<$core.String, $core.String>>? attributes, DataStream_TextHeader? textHeader, DataStream_ByteHeader? byteHeader, + $core.List<$core.int>? inlineContent, + DataStream_CompressionType? compression, }) { final result = create(); if (streamId != null) result.streamId = streamId; @@ -5408,6 +5421,8 @@ class DataStream_Header extends $pb.GeneratedMessage { if (attributes != null) result.attributes.addEntries(attributes); if (textHeader != null) result.textHeader = textHeader; if (byteHeader != null) result.byteHeader = byteHeader; + if (inlineContent != null) result.inlineContent = inlineContent; + if (compression != null) result.compression = compression; return result; } @@ -5441,6 +5456,9 @@ class DataStream_Header extends $pb.GeneratedMessage { packageName: const $pb.PackageName('livekit')) ..aOM(9, _omitFieldNames ? '' : 'textHeader', subBuilder: DataStream_TextHeader.create) ..aOM(10, _omitFieldNames ? '' : 'byteHeader', subBuilder: DataStream_ByteHeader.create) + ..a<$core.List<$core.int>>(11, _omitFieldNames ? '' : 'inlineContent', $pb.PbFieldType.OY) + ..aE(12, _omitFieldNames ? '' : 'compression', + enumValues: DataStream_CompressionType.values) ..hasRequiredFields = false; @$core.Deprecated('See https://github.com/google/protobuf.dart/issues/998.') @@ -5550,6 +5568,25 @@ class DataStream_Header extends $pb.GeneratedMessage { void clearByteHeader() => $_clearField(10); @$pb.TagNumber(10) DataStream_ByteHeader ensureByteHeader() => $_ensure(8); + + /// Optional inline content so that a data stream can be sent as a single packet for short payloads. + @$pb.TagNumber(11) + $core.List<$core.int> get inlineContent => $_getN(9); + @$pb.TagNumber(11) + set inlineContent($core.List<$core.int> value) => $_setBytes(9, value); + @$pb.TagNumber(11) + $core.bool hasInlineContent() => $_has(9); + @$pb.TagNumber(11) + void clearInlineContent() => $_clearField(11); + + @$pb.TagNumber(12) + DataStream_CompressionType get compression => $_getN(10); + @$pb.TagNumber(12) + set compression(DataStream_CompressionType value) => $_setField(12, value); + @$pb.TagNumber(12) + $core.bool hasCompression() => $_has(10); + @$pb.TagNumber(12) + void clearCompression() => $_clearField(12); } class DataStream_Chunk extends $pb.GeneratedMessage { diff --git a/lib/src/proto/livekit_models.pbenum.dart b/lib/src/proto/livekit_models.pbenum.dart index 35b6fad6a..402797ccf 100644 --- a/lib/src/proto/livekit_models.pbenum.dart +++ b/lib/src/proto/livekit_models.pbenum.dart @@ -581,13 +581,16 @@ class ClientInfo_Capability extends $pb.ProtobufEnum { static const ClientInfo_Capability CAP_UNUSED = ClientInfo_Capability._(0, _omitEnumNames ? '' : 'CAP_UNUSED'); static const ClientInfo_Capability CAP_PACKET_TRAILER = ClientInfo_Capability._(1, _omitEnumNames ? '' : 'CAP_PACKET_TRAILER'); + static const ClientInfo_Capability CAP_COMPRESSION_DEFLATE_RAW = + ClientInfo_Capability._(2, _omitEnumNames ? '' : 'CAP_COMPRESSION_DEFLATE_RAW'); static const $core.List values = [ CAP_UNUSED, CAP_PACKET_TRAILER, + CAP_COMPRESSION_DEFLATE_RAW, ]; - static final $core.List _byValue = $pb.ProtobufEnum.$_initByValueList(values, 1); + static final $core.List _byValue = $pb.ProtobufEnum.$_initByValueList(values, 2); static ClientInfo_Capability? valueOf($core.int value) => value < 0 || value >= _byValue.length ? null : _byValue[value]; @@ -615,4 +618,25 @@ class DataStream_OperationType extends $pb.ProtobufEnum { const DataStream_OperationType._(super.value, super.name); } +/// The compression type of the whole data stream +/// +/// This will only get populated when send to participants with a +/// client protocol >= 2 which advertise a client capability of CAP_COMPRESSION_DEFLATE_RAW +class DataStream_CompressionType extends $pb.ProtobufEnum { + static const DataStream_CompressionType NONE = DataStream_CompressionType._(0, _omitEnumNames ? '' : 'NONE'); + static const DataStream_CompressionType DEFLATE_RAW = + DataStream_CompressionType._(1, _omitEnumNames ? '' : 'DEFLATE_RAW'); + + static const $core.List values = [ + NONE, + DEFLATE_RAW, + ]; + + static final $core.List _byValue = $pb.ProtobufEnum.$_initByValueList(values, 1); + static DataStream_CompressionType? valueOf($core.int value) => + value < 0 || value >= _byValue.length ? null : _byValue[value]; + + const DataStream_CompressionType._(super.value, super.name); +} + const $core.bool _omitEnumNames = $core.bool.fromEnvironment('protobuf.omit_enum_names'); diff --git a/lib/src/proto/livekit_models.pbjson.dart b/lib/src/proto/livekit_models.pbjson.dart index 4f66de6bd..ffe893442 100644 --- a/lib/src/proto/livekit_models.pbjson.dart +++ b/lib/src/proto/livekit_models.pbjson.dart @@ -340,8 +340,8 @@ final $typed_data.Uint8List roomDescriptor = '50cxIjCg1jcmVhdGlvbl90aW1lGAUgASgDUgxjcmVhdGlvblRpbWUSKAoQY3JlYXRpb25fdGlt' 'ZV9tcxgPIAEoA1IOY3JlYXRpb25UaW1lTXMSIwoNdHVybl9wYXNzd29yZBgGIAEoCVIMdHVybl' 'Bhc3N3b3JkEjUKDmVuYWJsZWRfY29kZWNzGAcgAygLMg4ubGl2ZWtpdC5Db2RlY1INZW5hYmxl' - 'ZENvZGVjcxJACghtZXRhZGF0YRgIIAEoCUIkqFABslAePHJlZGFjdGVkICh7eyAuU2l6ZSB9fS' - 'BieXRlcyk+UghtZXRhZGF0YRIpChBudW1fcGFydGljaXBhbnRzGAkgASgNUg9udW1QYXJ0aWNp' + 'ZENvZGVjcxJACghtZXRhZGF0YRgIIAEoCUIkslAePHJlZGFjdGVkICh7eyAuU2l6ZSB9fSBieX' + 'Rlcyk+wFABUghtZXRhZGF0YRIpChBudW1fcGFydGljaXBhbnRzGAkgASgNUg9udW1QYXJ0aWNp' 'cGFudHMSJQoObnVtX3B1Ymxpc2hlcnMYCyABKA1SDW51bVB1Ymxpc2hlcnMSKQoQYWN0aXZlX3' 'JlY29yZGluZxgKIAEoCFIPYWN0aXZlUmVjb3JkaW5nEi8KB3ZlcnNpb24YDSABKAsyFS5saXZl' 'a2l0LlRpbWVkVmVyc2lvblIHdmVyc2lvbg=='); @@ -447,6 +447,7 @@ const ParticipantInfo$json = { {'1': 'kind_details', '3': 18, '4': 3, '5': 14, '6': '.livekit.ParticipantInfo.KindDetail', '10': 'kindDetails'}, {'1': 'data_tracks', '3': 19, '4': 3, '5': 11, '6': '.livekit.DataTrackInfo', '10': 'dataTracks'}, {'1': 'client_protocol', '3': 20, '4': 1, '5': 5, '10': 'clientProtocol'}, + {'1': 'capabilities', '3': 21, '4': 3, '5': 14, '6': '.livekit.ClientInfo.Capability', '10': 'capabilities'}, ], '3': [ParticipantInfo_AttributesEntry$json], '4': [ParticipantInfo_State$json, ParticipantInfo_Kind$json, ParticipantInfo_KindDetail$json], @@ -504,25 +505,26 @@ final $typed_data.Uint8List participantInfoDescriptor = $convert.base64Decode('Cg9QYXJ0aWNpcGFudEluZm8SEAoDc2lkGAEgASgJUgNzaWQSGgoIaWRlbnRpdHkYAiABKAlSCG' 'lkZW50aXR5EjQKBXN0YXRlGAMgASgOMh4ubGl2ZWtpdC5QYXJ0aWNpcGFudEluZm8uU3RhdGVS' 'BXN0YXRlEioKBnRyYWNrcxgEIAMoCzISLmxpdmVraXQuVHJhY2tJbmZvUgZ0cmFja3MSQAoIbW' - 'V0YWRhdGEYBSABKAlCJKhQAbJQHjxyZWRhY3RlZCAoe3sgLlNpemUgfX0gYnl0ZXMpPlIIbWV0' + 'V0YWRhdGEYBSABKAlCJLJQHjxyZWRhY3RlZCAoe3sgLlNpemUgfX0gYnl0ZXMpPsBQAVIIbWV0' 'YWRhdGESGwoJam9pbmVkX2F0GAYgASgDUghqb2luZWRBdBIgCgxqb2luZWRfYXRfbXMYESABKA' - 'NSCmpvaW5lZEF0TXMSFwoEbmFtZRgJIAEoCUIDqFABUgRuYW1lEhgKB3ZlcnNpb24YCiABKA1S' + 'NSCmpvaW5lZEF0TXMSFwoEbmFtZRgJIAEoCUIDwFABUgRuYW1lEhgKB3ZlcnNpb24YCiABKA1S' 'B3ZlcnNpb24SPgoKcGVybWlzc2lvbhgLIAEoCzIeLmxpdmVraXQuUGFydGljaXBhbnRQZXJtaX' 'NzaW9uUgpwZXJtaXNzaW9uEhYKBnJlZ2lvbhgMIAEoCVIGcmVnaW9uEiEKDGlzX3B1Ymxpc2hl' 'chgNIAEoCFILaXNQdWJsaXNoZXISMQoEa2luZBgOIAEoDjIdLmxpdmVraXQuUGFydGljaXBhbn' 'RJbmZvLktpbmRSBGtpbmQSbgoKYXR0cmlidXRlcxgPIAMoCzIoLmxpdmVraXQuUGFydGljaXBh' - 'bnRJbmZvLkF0dHJpYnV0ZXNFbnRyeUIkqFABslAePHJlZGFjdGVkICh7eyAuU2l6ZSB9fSBieX' - 'Rlcyk+UgphdHRyaWJ1dGVzEkYKEWRpc2Nvbm5lY3RfcmVhc29uGBAgASgOMhkubGl2ZWtpdC5E' + 'bnRJbmZvLkF0dHJpYnV0ZXNFbnRyeUIkslAePHJlZGFjdGVkICh7eyAuU2l6ZSB9fSBieXRlcy' + 'k+wFABUgphdHRyaWJ1dGVzEkYKEWRpc2Nvbm5lY3RfcmVhc29uGBAgASgOMhkubGl2ZWtpdC5E' 'aXNjb25uZWN0UmVhc29uUhBkaXNjb25uZWN0UmVhc29uEkYKDGtpbmRfZGV0YWlscxgSIAMoDj' 'IjLmxpdmVraXQuUGFydGljaXBhbnRJbmZvLktpbmREZXRhaWxSC2tpbmREZXRhaWxzEjcKC2Rh' 'dGFfdHJhY2tzGBMgAygLMhYubGl2ZWtpdC5EYXRhVHJhY2tJbmZvUgpkYXRhVHJhY2tzEicKD2' - 'NsaWVudF9wcm90b2NvbBgUIAEoBVIOY2xpZW50UHJvdG9jb2waPQoPQXR0cmlidXRlc0VudHJ5' - 'EhAKA2tleRgBIAEoCVIDa2V5EhQKBXZhbHVlGAIgASgJUgV2YWx1ZToCOAEiPgoFU3RhdGUSCw' - 'oHSk9JTklORxAAEgoKBkpPSU5FRBABEgoKBkFDVElWRRACEhAKDERJU0NPTk5FQ1RFRBADIlwK' - 'BEtpbmQSDAoIU1RBTkRBUkQQABILCgdJTkdSRVNTEAESCgoGRUdSRVNTEAISBwoDU0lQEAMSCQ' - 'oFQUdFTlQQBBINCglDT05ORUNUT1IQBxIKCgZCUklER0UQCCJrCgpLaW5kRGV0YWlsEg8KC0NM' - 'T1VEX0FHRU5UEAASDQoJRk9SV0FSREVEEAESFgoSQ09OTkVDVE9SX1dIQVRTQVBQEAISFAoQQ0' - '9OTkVDVE9SX1RXSUxJTxADEg8KC0JSSURHRV9SVFNQEAQ='); + 'NsaWVudF9wcm90b2NvbBgUIAEoBVIOY2xpZW50UHJvdG9jb2wSQgoMY2FwYWJpbGl0aWVzGBUg' + 'AygOMh4ubGl2ZWtpdC5DbGllbnRJbmZvLkNhcGFiaWxpdHlSDGNhcGFiaWxpdGllcxo9Cg9BdH' + 'RyaWJ1dGVzRW50cnkSEAoDa2V5GAEgASgJUgNrZXkSFAoFdmFsdWUYAiABKAlSBXZhbHVlOgI4' + 'ASI+CgVTdGF0ZRILCgdKT0lOSU5HEAASCgoGSk9JTkVEEAESCgoGQUNUSVZFEAISEAoMRElTQ0' + '9OTkVDVEVEEAMiXAoES2luZBIMCghTVEFOREFSRBAAEgsKB0lOR1JFU1MQARIKCgZFR1JFU1MQ' + 'AhIHCgNTSVAQAxIJCgVBR0VOVBAEEg0KCUNPTk5FQ1RPUhAHEgoKBkJSSURHRRAIImsKCktpbm' + 'REZXRhaWwSDwoLQ0xPVURfQUdFTlQQABINCglGT1JXQVJERUQQARIWChJDT05ORUNUT1JfV0hB' + 'VFNBUFAQAhIUChBDT05ORUNUT1JfVFdJTElPEAMSDwoLQlJJREdFX1JUU1AQBA=='); @$core.Deprecated('Use encryptionDescriptor instead') const Encryption$json = { @@ -639,7 +641,7 @@ const TrackInfo$json = { /// Descriptor for `TrackInfo`. Decode as a `google.protobuf.DescriptorProto`. final $typed_data.Uint8List trackInfoDescriptor = $convert.base64Decode('CglUcmFja0luZm8SEAoDc2lkGAEgASgJUgNzaWQSJgoEdHlwZRgCIAEoDjISLmxpdmVraXQuVH' - 'JhY2tUeXBlUgR0eXBlEhcKBG5hbWUYAyABKAlCA6hQAVIEbmFtZRIUCgVtdXRlZBgEIAEoCFIF' + 'JhY2tUeXBlUgR0eXBlEhcKBG5hbWUYAyABKAlCA8BQAVIEbmFtZRIUCgVtdXRlZBgEIAEoCFIF' 'bXV0ZWQSFAoFd2lkdGgYBSABKA1SBXdpZHRoEhYKBmhlaWdodBgGIAEoDVIGaGVpZ2h0EiAKCX' 'NpbXVsY2FzdBgHIAEoCEICGAFSCXNpbXVsY2FzdBIjCgtkaXNhYmxlX2R0eBgIIAEoCEICGAFS' 'CmRpc2FibGVEdHgSLAoGc291cmNlGAkgASgOMhQubGl2ZWtpdC5UcmFja1NvdXJjZVIGc291cm' @@ -1210,6 +1212,7 @@ const ClientInfo_Capability$json = { '2': [ {'1': 'CAP_UNUSED', '2': 0}, {'1': 'CAP_PACKET_TRAILER', '2': 1}, + {'1': 'CAP_COMPRESSION_DEFLATE_RAW', '2': 2}, ], }; @@ -1226,8 +1229,8 @@ final $typed_data.Uint8List clientInfoDescriptor = 'bGl0aWVzIrMBCgNTREsSCwoHVU5LTk9XThAAEgYKAkpTEAESCQoFU1dJRlQQAhILCgdBTkRST0' 'lEEAMSCwoHRkxVVFRFUhAEEgYKAkdPEAUSCQoFVU5JVFkQBhIQCgxSRUFDVF9OQVRJVkUQBxII' 'CgRSVVNUEAgSCgoGUFlUSE9OEAkSBwoDQ1BQEAoSDQoJVU5JVFlfV0VCEAsSCAoETk9ERRAMEg' - 'oKBlVOUkVBTBANEgkKBUVTUDMyEA4iNAoKQ2FwYWJpbGl0eRIOCgpDQVBfVU5VU0VEEAASFgoS' - 'Q0FQX1BBQ0tFVF9UUkFJTEVSEAE='); + 'oKBlVOUkVBTBANEgkKBUVTUDMyEA4iVQoKQ2FwYWJpbGl0eRIOCgpDQVBfVU5VU0VEEAASFgoS' + 'Q0FQX1BBQ0tFVF9UUkFJTEVSEAESHwobQ0FQX0NPTVBSRVNTSU9OX0RFRkxBVEVfUkFXEAI='); @$core.Deprecated('Use clientConfigurationDescriptor instead') const ClientConfiguration$json = { @@ -1533,7 +1536,7 @@ const DataStream$json = { DataStream_Chunk$json, DataStream_Trailer$json ], - '4': [DataStream_OperationType$json], + '4': [DataStream_OperationType$json, DataStream_CompressionType$json], }; @$core.Deprecated('Use dataStreamDescriptor instead') @@ -1577,11 +1580,14 @@ const DataStream_Header$json = { {'1': 'attributes', '3': 8, '4': 3, '5': 11, '6': '.livekit.DataStream.Header.AttributesEntry', '10': 'attributes'}, {'1': 'text_header', '3': 9, '4': 1, '5': 11, '6': '.livekit.DataStream.TextHeader', '9': 0, '10': 'textHeader'}, {'1': 'byte_header', '3': 10, '4': 1, '5': 11, '6': '.livekit.DataStream.ByteHeader', '9': 0, '10': 'byteHeader'}, + {'1': 'inline_content', '3': 11, '4': 1, '5': 12, '9': 2, '10': 'inlineContent', '17': true}, + {'1': 'compression', '3': 12, '4': 1, '5': 14, '6': '.livekit.DataStream.CompressionType', '10': 'compression'}, ], '3': [DataStream_Header_AttributesEntry$json], '8': [ {'1': 'content_header'}, {'1': '_total_length'}, + {'1': '_inline_content'}, ], }; @@ -1658,6 +1664,15 @@ const DataStream_OperationType$json = { ], }; +@$core.Deprecated('Use dataStreamDescriptor instead') +const DataStream_CompressionType$json = { + '1': 'CompressionType', + '2': [ + {'1': 'NONE', '2': 0}, + {'1': 'DEFLATE_RAW', '2': 1}, + ], +}; + /// Descriptor for `DataStream`. Decode as a `google.protobuf.DescriptorProto`. final $typed_data.Uint8List dataStreamDescriptor = $convert.base64Decode('CgpEYXRhU3RyZWFtGv8BCgpUZXh0SGVhZGVyEkgKDm9wZXJhdGlvbl90eXBlGAEgASgOMiEubG' @@ -1665,7 +1680,7 @@ final $typed_data.Uint8List dataStreamDescriptor = 'bhgCIAEoBVIHdmVyc2lvbhI/ChJyZXBseV90b19zdHJlYW1faWQYAyABKAlCErpQD3JlcGx5VG' '9TdHJlYW1JRFIPcmVwbHlUb1N0cmVhbUlkEi4KE2F0dGFjaGVkX3N0cmVhbV9pZHMYBCADKAlS' 'EWF0dGFjaGVkU3RyZWFtSWRzEhwKCWdlbmVyYXRlZBgFIAEoCFIJZ2VuZXJhdGVkGiAKCkJ5dG' - 'VIZWFkZXISEgoEbmFtZRgBIAEoCVIEbmFtZRqmBAoGSGVhZGVyEigKCXN0cmVhbV9pZBgBIAEo' + 'VIZWFkZXISEgoEbmFtZRgBIAEoCVIEbmFtZRqsBQoGSGVhZGVyEigKCXN0cmVhbV9pZBgBIAEo' 'CUILulAIc3RyZWFtSURSCHN0cmVhbUlkEhwKCXRpbWVzdGFtcBgCIAEoA1IJdGltZXN0YW1wEh' 'QKBXRvcGljGAMgASgJUgV0b3BpYxIbCgltaW1lX3R5cGUYBCABKAlSCG1pbWVUeXBlEiYKDHRv' 'dGFsX2xlbmd0aBgFIAEoBEgBUgt0b3RhbExlbmd0aIgBARJFCg9lbmNyeXB0aW9uX3R5cGUYBy' @@ -1673,17 +1688,20 @@ final $typed_data.Uint8List dataStreamDescriptor = 'dHJpYnV0ZXMYCCADKAsyKi5saXZla2l0LkRhdGFTdHJlYW0uSGVhZGVyLkF0dHJpYnV0ZXNFbn' 'RyeVIKYXR0cmlidXRlcxJBCgt0ZXh0X2hlYWRlchgJIAEoCzIeLmxpdmVraXQuRGF0YVN0cmVh' 'bS5UZXh0SGVhZGVySABSCnRleHRIZWFkZXISQQoLYnl0ZV9oZWFkZXIYCiABKAsyHi5saXZla2' - 'l0LkRhdGFTdHJlYW0uQnl0ZUhlYWRlckgAUgpieXRlSGVhZGVyGj0KD0F0dHJpYnV0ZXNFbnRy' - 'eRIQCgNrZXkYASABKAlSA2tleRIUCgV2YWx1ZRgCIAEoCVIFdmFsdWU6AjgBQhAKDmNvbnRlbn' - 'RfaGVhZGVyQg8KDV90b3RhbF9sZW5ndGgapgEKBUNodW5rEigKCXN0cmVhbV9pZBgBIAEoCUIL' - 'ulAIc3RyZWFtSURSCHN0cmVhbUlkEh8KC2NodW5rX2luZGV4GAIgASgEUgpjaHVua0luZGV4Eh' - 'gKB2NvbnRlbnQYAyABKAxSB2NvbnRlbnQSGAoHdmVyc2lvbhgEIAEoBVIHdmVyc2lvbhIXCgJp' - 'dhgFIAEoDEICGAFIAFICaXaIAQFCBQoDX2l2GtcBCgdUcmFpbGVyEigKCXN0cmVhbV9pZBgBIA' - 'EoCUILulAIc3RyZWFtSURSCHN0cmVhbUlkEhYKBnJlYXNvbhgCIAEoCVIGcmVhc29uEksKCmF0' - 'dHJpYnV0ZXMYAyADKAsyKy5saXZla2l0LkRhdGFTdHJlYW0uVHJhaWxlci5BdHRyaWJ1dGVzRW' - '50cnlSCmF0dHJpYnV0ZXMaPQoPQXR0cmlidXRlc0VudHJ5EhAKA2tleRgBIAEoCVIDa2V5EhQK' - 'BXZhbHVlGAIgASgJUgV2YWx1ZToCOAEiQQoNT3BlcmF0aW9uVHlwZRIKCgZDUkVBVEUQABIKCg' - 'ZVUERBVEUQARIKCgZERUxFVEUQAhIMCghSRUFDVElPThAD'); + 'l0LkRhdGFTdHJlYW0uQnl0ZUhlYWRlckgAUgpieXRlSGVhZGVyEioKDmlubGluZV9jb250ZW50' + 'GAsgASgMSAJSDWlubGluZUNvbnRlbnSIAQESRQoLY29tcHJlc3Npb24YDCABKA4yIy5saXZla2' + 'l0LkRhdGFTdHJlYW0uQ29tcHJlc3Npb25UeXBlUgtjb21wcmVzc2lvbho9Cg9BdHRyaWJ1dGVz' + 'RW50cnkSEAoDa2V5GAEgASgJUgNrZXkSFAoFdmFsdWUYAiABKAlSBXZhbHVlOgI4AUIQCg5jb2' + '50ZW50X2hlYWRlckIPCg1fdG90YWxfbGVuZ3RoQhEKD19pbmxpbmVfY29udGVudBqmAQoFQ2h1' + 'bmsSKAoJc3RyZWFtX2lkGAEgASgJQgu6UAhzdHJlYW1JRFIIc3RyZWFtSWQSHwoLY2h1bmtfaW' + '5kZXgYAiABKARSCmNodW5rSW5kZXgSGAoHY29udGVudBgDIAEoDFIHY29udGVudBIYCgd2ZXJz' + 'aW9uGAQgASgFUgd2ZXJzaW9uEhcKAml2GAUgASgMQgIYAUgAUgJpdogBAUIFCgNfaXYa1wEKB1' + 'RyYWlsZXISKAoJc3RyZWFtX2lkGAEgASgJQgu6UAhzdHJlYW1JRFIIc3RyZWFtSWQSFgoGcmVh' + 'c29uGAIgASgJUgZyZWFzb24SSwoKYXR0cmlidXRlcxgDIAMoCzIrLmxpdmVraXQuRGF0YVN0cm' + 'VhbS5UcmFpbGVyLkF0dHJpYnV0ZXNFbnRyeVIKYXR0cmlidXRlcxo9Cg9BdHRyaWJ1dGVzRW50' + 'cnkSEAoDa2V5GAEgASgJUgNrZXkSFAoFdmFsdWUYAiABKAlSBXZhbHVlOgI4ASJBCg1PcGVyYX' + 'Rpb25UeXBlEgoKBkNSRUFURRAAEgoKBlVQREFURRABEgoKBkRFTEVURRACEgwKCFJFQUNUSU9O' + 'EAMiLAoPQ29tcHJlc3Npb25UeXBlEggKBE5PTkUQABIPCgtERUZMQVRFX1JBVxAB'); @$core.Deprecated('Use filterParamsDescriptor instead') const FilterParams$json = { diff --git a/lib/src/proto/livekit_rtc.pbjson.dart b/lib/src/proto/livekit_rtc.pbjson.dart index f9eff6e6c..2f541d1d8 100644 --- a/lib/src/proto/livekit_rtc.pbjson.dart +++ b/lib/src/proto/livekit_rtc.pbjson.dart @@ -922,11 +922,11 @@ const UpdateParticipantMetadata_AttributesEntry$json = { /// Descriptor for `UpdateParticipantMetadata`. Decode as a `google.protobuf.DescriptorProto`. final $typed_data.Uint8List updateParticipantMetadataDescriptor = - $convert.base64Decode('ChlVcGRhdGVQYXJ0aWNpcGFudE1ldGFkYXRhEkAKCG1ldGFkYXRhGAEgASgJQiSoUAGyUB48cm' - 'VkYWN0ZWQgKHt7IC5TaXplIH19IGJ5dGVzKT5SCG1ldGFkYXRhEjgKBG5hbWUYAiABKAlCJKhQ' - 'AbJQHjxyZWRhY3RlZCAoe3sgLlNpemUgfX0gYnl0ZXMpPlIEbmFtZRJ4CgphdHRyaWJ1dGVzGA' + $convert.base64Decode('ChlVcGRhdGVQYXJ0aWNpcGFudE1ldGFkYXRhEkAKCG1ldGFkYXRhGAEgASgJQiSyUB48cmVkYW' + 'N0ZWQgKHt7IC5TaXplIH19IGJ5dGVzKT7AUAFSCG1ldGFkYXRhEjgKBG5hbWUYAiABKAlCJLJQ' + 'HjxyZWRhY3RlZCAoe3sgLlNpemUgfX0gYnl0ZXMpPsBQAVIEbmFtZRJ4CgphdHRyaWJ1dGVzGA' 'MgAygLMjIubGl2ZWtpdC5VcGRhdGVQYXJ0aWNpcGFudE1ldGFkYXRhLkF0dHJpYnV0ZXNFbnRy' - 'eUIkqFABslAePHJlZGFjdGVkICh7eyAuU2l6ZSB9fSBieXRlcyk+UgphdHRyaWJ1dGVzEisKCn' + 'eUIkslAePHJlZGFjdGVkICh7eyAuU2l6ZSB9fSBieXRlcyk+wFABUgphdHRyaWJ1dGVzEisKCn' 'JlcXVlc3RfaWQYBCABKA1CDLpQCXJlcXVlc3RJRFIJcmVxdWVzdElkGj0KD0F0dHJpYnV0ZXNF' 'bnRyeRIQCgNrZXkYASABKAlSA2tleRIUCgV2YWx1ZRgCIAEoCVIFdmFsdWU6AjgB'); @@ -942,8 +942,8 @@ const ICEServer$json = { /// Descriptor for `ICEServer`. Decode as a `google.protobuf.DescriptorProto`. final $typed_data.Uint8List iCEServerDescriptor = - $convert.base64Decode('CglJQ0VTZXJ2ZXISEgoEdXJscxgBIAMoCVIEdXJscxIfCgh1c2VybmFtZRgCIAEoCUIDqFABUg' - 'h1c2VybmFtZRIjCgpjcmVkZW50aWFsGAMgASgJQgOoUAFSCmNyZWRlbnRpYWw='); + $convert.base64Decode('CglJQ0VTZXJ2ZXISEgoEdXJscxgBIAMoCVIEdXJscxIfCgh1c2VybmFtZRgCIAEoCUIDwFABUg' + 'h1c2VybmFtZRIjCgpjcmVkZW50aWFsGAMgASgJQgPAUAJSCmNyZWRlbnRpYWw='); @$core.Deprecated('Use speakersChangedDescriptor instead') const SpeakersChanged$json = { @@ -1541,10 +1541,10 @@ const JoinRequest_ParticipantAttributesEntry$json = { final $typed_data.Uint8List joinRequestDescriptor = $convert.base64Decode('CgtKb2luUmVxdWVzdBI0CgtjbGllbnRfaW5mbxgBIAEoCzITLmxpdmVraXQuQ2xpZW50SW5mb1' 'IKY2xpZW50SW5mbxJMChNjb25uZWN0aW9uX3NldHRpbmdzGAIgASgLMhsubGl2ZWtpdC5Db25u' - 'ZWN0aW9uU2V0dGluZ3NSEmNvbm5lY3Rpb25TZXR0aW5ncxJACghtZXRhZGF0YRgDIAEoCUIkqF' - 'ABslAePHJlZGFjdGVkICh7eyAuU2l6ZSB9fSBieXRlcyk+UghtZXRhZGF0YRKMAQoWcGFydGlj' + 'ZWN0aW9uU2V0dGluZ3NSEmNvbm5lY3Rpb25TZXR0aW5ncxJACghtZXRhZGF0YRgDIAEoCUIksl' + 'AePHJlZGFjdGVkICh7eyAuU2l6ZSB9fSBieXRlcyk+wFABUghtZXRhZGF0YRKMAQoWcGFydGlj' 'aXBhbnRfYXR0cmlidXRlcxgEIAMoCzIvLmxpdmVraXQuSm9pblJlcXVlc3QuUGFydGljaXBhbn' - 'RBdHRyaWJ1dGVzRW50cnlCJKhQAbJQHjxyZWRhY3RlZCAoe3sgLlNpemUgfX0gYnl0ZXMpPlIV' + 'RBdHRyaWJ1dGVzRW50cnlCJLJQHjxyZWRhY3RlZCAoe3sgLlNpemUgfX0gYnl0ZXMpPsBQAVIV' 'cGFydGljaXBhbnRBdHRyaWJ1dGVzEkYKEmFkZF90cmFja19yZXF1ZXN0cxgFIAMoCzIYLmxpdm' 'VraXQuQWRkVHJhY2tSZXF1ZXN0UhBhZGRUcmFja1JlcXVlc3RzEkQKD3B1Ymxpc2hlcl9vZmZl' 'chgGIAEoCzIbLmxpdmVraXQuU2Vzc2lvbkRlc2NyaXB0aW9uUg5wdWJsaXNoZXJPZmZlchIcCg' diff --git a/lib/src/types/client_capability.dart b/lib/src/types/client_capability.dart new file mode 100644 index 000000000..86996f371 --- /dev/null +++ b/lib/src/types/client_capability.dart @@ -0,0 +1,58 @@ +// Copyright 2026 LiveKit, Inc. +// +// 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 'package:flutter/foundation.dart' show kIsWeb; + +import '../proto/livekit_models.pb.dart' as lk_models; + +/// An optional feature a client advertises to its peers. +/// +/// Distinct from `clientProtocol`, which is a monotonic baseline version: capabilities cover +/// features a client may or may not support depending on its platform, and each is advertised +/// independently. Wire values match `livekit.ClientInfo.Capability`. +enum ClientCapability { + /// The client understands packet trailers. + packetTrailer(1), + + /// The client can decompress a deflate-raw compressed data stream. + compressionDeflateRaw(2) + ; + + const ClientCapability(this.wireValue); + + final int wireValue; + + /// The capabilities this SDK advertises. + /// + /// Compression is advertised unconditionally on native, where the Rust core always provides + /// deflate-raw. Web has no Rust core and so advertises nothing — a v2 sender then falls back to + /// uncompressed multi-packet for it, which every client understands. + static const List advertised = kIsWeb + ? [] + : [ClientCapability.compressionDeflateRaw]; + + static ClientCapability? fromProto(lk_models.ClientInfo_Capability value) { + for (final capability in ClientCapability.values) { + if (capability.wireValue == value.value) return capability; + } + // CAP_UNUSED, and anything newer than this SDK knows about. + return null; + } + + /// The name the server expects for this capability in the signal URL's `capabilities` param. + String toWireName() => switch (this) { + ClientCapability.packetTrailer => 'CAP_PACKET_TRAILER', + ClientCapability.compressionDeflateRaw => 'CAP_COMPRESSION_DEFLATE_RAW', + }; +} diff --git a/lib/src/types/data_stream.dart b/lib/src/types/data_stream.dart index 3cfa23756..9e65d3daa 100644 --- a/lib/src/types/data_stream.dart +++ b/lib/src/types/data_stream.dart @@ -20,12 +20,41 @@ class SendTextOptions { /// user defined attributes map that can carry additional info Map attributes; + /// Whether to deflate-raw compress the payload when every recipient supports it. Defaults to + /// true; set false to opt out. Only honored on native platforms — web always sends uncompressed. + bool compress; + SendTextOptions({ this.topic, this.destinationIdentities = const [], this.attachments = const [], this.onProgress, this.attributes = const {}, + this.compress = true, + }); +} + +/// Options for sending an in-memory byte payload with `sendBytes`. +/// +/// Unlike a file send, nothing is inferred from the input: [name] defaults to `unknown` and the +/// mime type to `application/octet-stream`. +class SendBytesOptions { + String? topic; + String? name; + String? mimeType; + List destinationIdentities = []; + Map attributes; + + /// See [SendTextOptions.compress]. + bool compress; + + SendBytesOptions({ + this.topic, + this.name, + this.mimeType, + this.destinationIdentities = const [], + this.attributes = const {}, + this.compress = true, }); } diff --git a/lib/src/types/other.dart b/lib/src/types/other.dart index 85afa368f..8b0a4db6b 100644 --- a/lib/src/types/other.dart +++ b/lib/src/types/other.dart @@ -14,6 +14,7 @@ // ignore_for_file: constant_identifier_names +import 'package:flutter/foundation.dart' show kIsWeb; import 'package:flutter/material.dart'; import '../extensions.dart'; @@ -51,7 +52,12 @@ enum ClientProtocolVersion implements Comparable { v0(0), /// Spec: `CLIENT_PROTOCOL_DATA_STREAM_RPC`. Supports RPC v2 (data-stream payloads). - v1(1) + v1(1), + + /// Spec: `CLIENT_PROTOCOL_DATA_STREAM_V2`. Understands data streams v2 — in particular + /// single-packet inline sends. Crossing this threshold is a baseline commitment with no opt-out; + /// optional v2 features such as compression are negotiated separately via [ClientCapability]. + v2(2) ; const ClientProtocolVersion(this.wireValue); @@ -62,11 +68,16 @@ enum ClientProtocolVersion implements Comparable { /// The highest version this SDK build supports. Used as the default for /// [ConnectOptions.clientProtocolVersion] and in tests that need to advertise /// "the current SDK". - static const ClientProtocolVersion current = v1; + /// + /// Web stays at [v1]: data streams v2 is implemented by the Rust core, which cannot run in a + /// browser. Advertising a lower protocol there is what makes a v2 sender fall back to + /// uncompressed multi-packet framing, which the Dart implementation understands. + static const ClientProtocolVersion current = kIsWeb ? v1 : v2; /// Maps wire values to the highest protocol version this SDK can use. static ClientProtocolVersion fromIntValue(int? value) { if (value == null) return v0; + if (value >= v2.wireValue) return v2; if (value >= v1.wireValue) return v1; return v0; } diff --git a/lib/src/utils.dart b/lib/src/utils.dart index c29d3ca23..6c3b5463d 100644 --- a/lib/src/utils.dart +++ b/lib/src/utils.dart @@ -33,6 +33,7 @@ import 'logger.dart'; import 'options.dart'; import 'support/platform.dart'; import 'track/local/video.dart'; +import 'types/client_capability.dart'; import 'types/other.dart'; import 'types/priority.dart'; import 'types/video_dimensions.dart'; @@ -199,6 +200,10 @@ class Utils { if (reconnect && sid != null) 'sid': sid, 'protocol': connectOptions.protocolVersion.toStringValue(), 'client_protocol': connectOptions.clientProtocolVersion.toStringValue(), + // Optional feature flags, negotiated per-peer independently of `client_protocol`. Omitted + // entirely when empty (web), which peers read as "no optional features". + if (ClientCapability.advertised.isNotEmpty) + 'capabilities': ClientCapability.advertised.map((c) => c.toWireName()).join(','), 'sdk': 'flutter', 'version': LiveKitClient.version, 'network': networkType, diff --git a/test/core/connect_options_test.dart b/test/core/connect_options_test.dart new file mode 100644 index 000000000..1587e0e48 --- /dev/null +++ b/test/core/connect_options_test.dart @@ -0,0 +1,58 @@ +// Copyright 2026 LiveKit, Inc. +// +// 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. + +@Timeout(Duration(seconds: 10)) +library; + +import 'package:flutter_test/flutter_test.dart'; + +import 'package:livekit_client/livekit_client.dart'; +import '../mock/e2e_container.dart'; +import '../mock/peerconnection_mock.dart'; + +void main() { + setUp(resetMockDataChannels); + + group('Room.connect options', () { + // Regression: the deprecated `roomOptions` parameter was shadowed by a local of the same name + // in the first line of `connect`, which Dart permits silently. Everything passed here was + // discarded, so callers saw the Room's own options with no indication anything was wrong. + test('honors the roomOptions passed to connect', () async { + final container = E2EContainer( + roomOptions: const RoomOptions(dynacast: false, adaptiveStream: false), + ); + addTearDown(container.dispose); + + await container.connectRoom( + // ignore: deprecated_member_use_from_same_package + roomOptions: const RoomOptions(dynacast: true, adaptiveStream: true), + ); + + expect(container.room.roomOptions.dynacast, isTrue); + expect(container.room.roomOptions.adaptiveStream, isTrue); + }); + + test('falls back to the Room\'s options when connect is given none', () async { + final container = E2EContainer( + roomOptions: const RoomOptions(dynacast: true, adaptiveStream: true), + ); + addTearDown(container.dispose); + + await container.connectRoom(); + + expect(container.room.roomOptions.dynacast, isTrue); + expect(container.room.roomOptions.adaptiveStream, isTrue); + }); + }); +} diff --git a/test/core/data_stream_v2_test.dart b/test/core/data_stream_v2_test.dart new file mode 100644 index 000000000..49fc27134 --- /dev/null +++ b/test/core/data_stream_v2_test.dart @@ -0,0 +1,371 @@ +// Copyright 2026 LiveKit, Inc. +// +// 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. + +// Data streams v2 wire behavior, asserted on the packets that actually reach the data channel. +// +// These only apply to the native path, where the Rust core does the framing. Web keeps the v1 +// Dart implementation and advertises a pre-v2 clientProtocol, so a v2 sender falls back for it — +// there is nothing v2-shaped to assert there. +@TestOn('vm') +@Timeout(Duration(seconds: 20)) +library; + +import 'dart:async'; +import 'dart:convert'; +import 'dart:typed_data'; + +import 'package:fixnum/fixnum.dart'; +import 'package:flutter_test/flutter_test.dart'; + +import 'package:livekit_client/livekit_client.dart'; +import 'package:livekit_client/src/proto/livekit_models.pb.dart' as lk_models; +import '../mock/e2e_container.dart'; +import '../mock/peerconnection_mock.dart'; + +/// A recipient that understands v2 and can decompress. +const _v2WithCompression = [lk_models.ClientInfo_Capability.CAP_COMPRESSION_DEFLATE_RAW]; + +void main() { + late E2EContainer container; + late Room room; + + setUp(() async { + resetMockDataChannels(); + container = E2EContainer(); + await container.connectRoom(captureOutbound: true); + room = container.room; + }); + + tearDown(() async { + await container.dispose(); + }); + + /// The stream packets emitted since the last clear, in order. + List streamPackets() => container.capturedDataPackets + .where((p) => p.hasStreamHeader() || p.hasStreamChunk() || p.hasStreamTrailer()) + .toList(); + + group('send side', () { + test('a v2 recipient that can decompress gets one compressed inline packet', () async { + await container.simulateRemoteParticipantJoin( + 'alice', + clientProtocol: 2, + capabilities: _v2WithCompression, + ); + container.capturedDataPackets.clear(); + + const text = 'hello hello compressible world'; + await room.localParticipant!.sendText( + text, + options: SendTextOptions(topic: 'chat', destinationIdentities: ['alice']), + ); + + final packets = streamPackets(); + expect(packets, hasLength(1), reason: 'inline send is a single packet'); + final header = packets.single.streamHeader; + expect(header.hasTextHeader(), isTrue); + expect(header.compression, lk_models.DataStream_CompressionType.DEFLATE_RAW); + expect(header.hasInlineContent(), isTrue); + expect( + header.inlineContent, + isNot(equals(utf8.encode(text))), + reason: 'inline content should be the compressed bytes, not the raw UTF-8', + ); + }); + + test('a v2 recipient without the compression capability gets inline but raw', () async { + await container.simulateRemoteParticipantJoin('noCompression', clientProtocol: 2); + container.capturedDataPackets.clear(); + + const text = 'hello hello compressible world'; + await room.localParticipant!.sendText( + text, + options: SendTextOptions(topic: 'chat', destinationIdentities: ['noCompression']), + ); + + final packets = streamPackets(); + expect(packets, hasLength(1), reason: 'inline is gated on clientProtocol alone'); + final header = packets.single.streamHeader; + expect(header.compression, lk_models.DataStream_CompressionType.NONE); + expect(header.inlineContent, equals(utf8.encode(text))); + }); + + test('a pre-v2 recipient gets legacy header + chunk + trailer', () async { + await container.simulateRemoteParticipantJoin('legacy', clientProtocol: 0); + container.capturedDataPackets.clear(); + + const text = 'hello world'; + await room.localParticipant!.sendText( + text, + options: SendTextOptions(topic: 'chat', destinationIdentities: ['legacy']), + ); + + final packets = streamPackets(); + expect(packets, hasLength(3)); + expect(packets[0].hasStreamHeader(), isTrue); + expect(packets[0].streamHeader.compression, lk_models.DataStream_CompressionType.NONE); + expect(packets[0].streamHeader.hasInlineContent(), isFalse); + expect(packets[1].hasStreamChunk(), isTrue); + expect(packets[1].streamChunk.content, equals(utf8.encode(text))); + expect(packets[2].hasStreamTrailer(), isTrue); + expect(packets[2].streamTrailer.streamId, equals(packets[0].streamHeader.streamId)); + }); + + test('a broadcast to a mixed room falls back to legacy framing', () async { + await container.simulateRemoteParticipantJoin('alice', clientProtocol: 2, capabilities: _v2WithCompression); + await container.simulateRemoteParticipantJoin('legacy', clientProtocol: 0); + container.capturedDataPackets.clear(); + + // No destinationIdentities => every remote participant is a recipient, and one is pre-v2. + await room.localParticipant!.sendText('hello world', options: SendTextOptions(topic: 'chat')); + + final packets = streamPackets(); + expect(packets, hasLength(3), reason: 'one pre-v2 recipient disables inline for everyone'); + expect(packets[0].streamHeader.hasInlineContent(), isFalse); + }); + + test('compress: false keeps inline but sends raw bytes', () async { + await container.simulateRemoteParticipantJoin('alice', clientProtocol: 2, capabilities: _v2WithCompression); + container.capturedDataPackets.clear(); + + const text = 'hello hello compressible world'; + await room.localParticipant!.sendText( + text, + options: SendTextOptions(topic: 'chat', destinationIdentities: ['alice'], compress: false), + ); + + final header = streamPackets().single.streamHeader; + expect(header.compression, lk_models.DataStream_CompressionType.NONE); + expect(header.inlineContent, equals(utf8.encode(text))); + }); + + test('streamText never inlines or compresses', () async { + await container.simulateRemoteParticipantJoin('alice', clientProtocol: 2, capabilities: _v2WithCompression); + container.capturedDataPackets.clear(); + + final writer = await room.localParticipant!.streamText( + StreamTextOptions(topic: 'chat', destinationIdentities: ['alice']), + ); + expect(streamPackets(), hasLength(1), reason: 'the header goes out when the stream opens'); + expect(streamPackets().single.streamHeader.compression, lk_models.DataStream_CompressionType.NONE); + + await writer.write('hello world'); + expect(streamPackets(), hasLength(2)); + expect(streamPackets()[1].streamChunk.content, equals(utf8.encode('hello world'))); + + await writer.close(); + expect(streamPackets(), hasLength(3)); + expect(streamPackets()[2].hasStreamTrailer(), isTrue); + }); + + test('sendBytes produces a byte header and defaults name/mimeType', () async { + await container.simulateRemoteParticipantJoin('alice', clientProtocol: 2, capabilities: _v2WithCompression); + container.capturedDataPackets.clear(); + + final info = await room.localParticipant!.sendBytes( + utf8.encode('hello hello compressible world'), + options: SendBytesOptions(topic: 'files', destinationIdentities: ['alice']), + ); + + final header = streamPackets().single.streamHeader; + expect(header.hasByteHeader(), isTrue); + expect(header.compression, lk_models.DataStream_CompressionType.DEFLATE_RAW); + expect(info.name, equals('unknown')); + expect(info.mimeType, equals('application/octet-stream')); + }); + }); + + group('receive side', () { + /// Feeds a single inline text header, as a v2 sender would emit it. + void feedInlineText({ + required String streamId, + required String topic, + required List inlineContent, + required int totalLength, + lk_models.DataStream_CompressionType compression = lk_models.DataStream_CompressionType.NONE, + Map attributes = const {}, + }) { + container.deliverInboundDataPacket( + lk_models.DataPacket( + kind: lk_models.DataPacket_Kind.RELIABLE, + participantIdentity: 'alice', + streamHeader: lk_models.DataStream_Header( + streamId: streamId, + topic: topic, + mimeType: 'text/plain', + timestamp: Int64(DateTime.timestamp().millisecondsSinceEpoch), + totalLength: Int64(totalLength), + attributes: attributes.entries, + inlineContent: Uint8List.fromList(inlineContent), + compression: compression, + textHeader: lk_models.DataStream_TextHeader(), + ), + ), + ); + } + + test('an inline uncompressed text stream is delivered whole', () async { + const text = 'hello inline world'; + final received = Completer(); + final gotInfo = Completer(); + + room.registerTextStreamHandler('inline', (reader, identity) async { + gotInfo.complete(reader.info!); + received.complete(await reader.readAll()); + }); + + feedInlineText( + streamId: 'inline-1', + topic: 'inline', + inlineContent: utf8.encode(text), + totalLength: utf8.encode(text).length, + attributes: {'foo': 'bar'}, + ); + + expect(await received.future, equals(text)); + final info = await gotInfo.future; + expect(info.attributes['foo'], equals('bar')); + expect(info.sendingParticipantIdentity, equals('alice')); + }); + + test('an inline compressed text stream round-trips through the core', () async { + // Genuinely compressed bytes are hard to hand-write, so let the send path produce them and + // read them back over the harness's data-channel loopback — a real compress/decompress pass + // through the Rust core in both directions. + await container.simulateRemoteParticipantJoin('alice', clientProtocol: 2, capabilities: _v2WithCompression); + container.capturedDataPackets.clear(); + + final received = Completer(); + room.registerTextStreamHandler('compressed', (reader, identity) async { + received.complete(await reader.readAll()); + }); + + const text = 'hello hello compressible world'; + await room.localParticipant!.sendText( + text, + options: SendTextOptions(topic: 'compressed', destinationIdentities: ['alice']), + ); + + expect( + streamPackets().single.streamHeader.compression, + lk_models.DataStream_CompressionType.DEFLATE_RAW, + reason: 'the payload really was compressed on the way out', + ); + expect(await received.future, equals(text)); + }); + + test('a stream on an unregistered topic is ignored', () async { + var fired = false; + room.registerTextStreamHandler('registered', (reader, identity) async { + fired = true; + }); + + feedInlineText( + streamId: 'inline-2', + topic: 'not-registered', + inlineContent: utf8.encode('nobody wants this'), + totalLength: 17, + ); + + await Future.delayed(const Duration(milliseconds: 100)); + expect(fired, isFalse); + }); + }); + + group('maxPayloadByteLength', () { + test('a stream declaring more than the cap fails its reader', () async { + // A fresh container so the cap is set at connect time, which is when the native manager + // reads it. + resetMockDataChannels(); + final capped = E2EContainer(); + addTearDown(capped.dispose); + await capped.connectRoom( + connectOptions: const ConnectOptions( + dataStream: DataStreamOptions(maxPayloadByteLength: 16), + ), + ); + + // The handler is still invoked — the core reports the stream opened before applying the + // cap — and it is the read that fails. + final outcome = Completer(); + capped.room.registerTextStreamHandler('capped', (reader, identity) async { + try { + await reader.readAll(); + outcome.complete(null); + } catch (e) { + outcome.complete(e); + } + }); + + capped.deliverInboundDataPacket( + lk_models.DataPacket( + kind: lk_models.DataPacket_Kind.RELIABLE, + participantIdentity: 'alice', + streamHeader: lk_models.DataStream_Header( + streamId: 'too-big', + topic: 'capped', + mimeType: 'text/plain', + timestamp: Int64(DateTime.timestamp().millisecondsSinceEpoch), + totalLength: Int64(1000), + inlineContent: Uint8List.fromList(utf8.encode('x' * 1000)), + textHeader: lk_models.DataStream_TextHeader(), + ), + ), + ); + + final error = await outcome.future.timeout(const Duration(seconds: 5)); + expect(error, isA()); + expect( + (error as DataStreamError).reason, + DataStreamErrorReason.LengthExceeded, + reason: 'the payload exceeds maxPayloadByteLength', + ); + }); + + test('a stream within the cap is delivered', () async { + resetMockDataChannels(); + final capped = E2EContainer(); + addTearDown(capped.dispose); + await capped.connectRoom( + connectOptions: const ConnectOptions( + dataStream: DataStreamOptions(maxPayloadByteLength: 1000), + ), + ); + + const text = 'small enough'; + final received = Completer(); + capped.room.registerTextStreamHandler('capped', (reader, identity) async { + received.complete(await reader.readAll()); + }); + + capped.deliverInboundDataPacket( + lk_models.DataPacket( + kind: lk_models.DataPacket_Kind.RELIABLE, + participantIdentity: 'alice', + streamHeader: lk_models.DataStream_Header( + streamId: 'small', + topic: 'capped', + mimeType: 'text/plain', + timestamp: Int64(DateTime.timestamp().millisecondsSinceEpoch), + totalLength: Int64(utf8.encode(text).length), + inlineContent: Uint8List.fromList(utf8.encode(text)), + textHeader: lk_models.DataStream_TextHeader(), + ), + ), + ); + + expect(await received.future, equals(text)); + }); + }); +} diff --git a/test/core/rpc_test.dart b/test/core/rpc_test.dart index fe30eda0b..036f1e80d 100644 --- a/test/core/rpc_test.dart +++ b/test/core/rpc_test.dart @@ -18,7 +18,6 @@ library; import 'package:flutter_test/flutter_test.dart'; import 'package:livekit_client/livekit_client.dart'; -import 'package:livekit_client/src/data_stream/errors.dart'; import 'package:livekit_client/src/proto/livekit_models.pb.dart' as lk_models; import '../mock/e2e_container.dart'; import '../mock/peerconnection_mock.dart'; diff --git a/test/mock/e2e_container.dart b/test/mock/e2e_container.dart index 25de66597..c991075cb 100644 --- a/test/mock/e2e_container.dart +++ b/test/mock/e2e_container.dart @@ -37,12 +37,12 @@ class E2EContainer { /// since [connectRoom] returned. Populated only when [captureOutbound] is true. final List capturedDataPackets = []; - E2EContainer() { + E2EContainer({RoomOptions roomOptions = const RoomOptions()}) { wsConnector = MockWebSocketConnector(); client = SignalClient(wsConnector.connect); engine = Engine( connectOptions: const ConnectOptions(), - roomOptions: const RoomOptions(), + roomOptions: roomOptions, signalClient: client, peerConnectionCreate: MockPeerConnection.create, ); @@ -58,8 +58,19 @@ class E2EContainer { /// that value (used to exercise v1 vs v2 caller paths in self-loop tests). /// When [captureOutbound] is true, all DataPackets sent over the reliable /// data channel are recorded in [capturedDataPackets]. - Future connectRoom({int? localClientProtocol, bool captureOutbound = false}) async { - final connectFuture = room.connect(exampleUri, token); + Future connectRoom({ + int? localClientProtocol, + bool captureOutbound = false, + ConnectOptions? connectOptions, + @Deprecated('mirrors the deprecated Room.connect parameter') RoomOptions? roomOptions, + }) async { + final connectFuture = room.connect( + exampleUri, + token, + connectOptions: connectOptions, + // ignore: deprecated_member_use_from_same_package + roomOptions: roomOptions, + ); Future.delayed(const Duration(milliseconds: 1), () { final resp = _buildJoinResponse(localClientProtocol); wsConnector.onData(resp.writeToBuffer()); @@ -140,6 +151,7 @@ class E2EContainer { String identity, { int? clientProtocol, String? sid, + List capabilities = const [], }) async { clientProtocol ??= ClientProtocolVersion.current.toIntValue(); final info = lk_models.ParticipantInfo( @@ -147,6 +159,7 @@ class E2EContainer { identity: identity, state: lk_models.ParticipantInfo_State.ACTIVE, clientProtocol: clientProtocol, + capabilities: capabilities, ); final resp = lk_rtc.SignalResponse( update: lk_rtc.ParticipantUpdate(participants: [info]),