diff --git a/.changes/refresh-subscribed-codecs-after-publisher-answer b/.changes/refresh-subscribed-codecs-after-publisher-answer new file mode 100644 index 000000000..59f8ba65d --- /dev/null +++ b/.changes/refresh-subscribed-codecs-after-publisher-answer @@ -0,0 +1 @@ +patch type="fixed" "The last subscribed codecs are now re-applied to the local video senders after each publisher renegotiation, so simulcast layers that dynacast paused are no longer re-enabled by the SFU answer" diff --git a/lib/src/core/engine.dart b/lib/src/core/engine.dart index 48d29c8e7..7ac40969f 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -1564,6 +1564,7 @@ class Engine extends Disposable with EventsEmittable { logger.fine('received answer (type: ${event.sd.type})'); logger.finer('sdp: ${event.sd.sdp}'); await publisher!.setRemoteDescription(event.sd); + events.emit(const EngineSubscribedCodecRefreshRequestedEvent()); }) ..on((event) async { if (publisher == null || subscriber == null) { diff --git a/lib/src/core/room.dart b/lib/src/core/room.dart index 4ef28bbf0..f10199abb 100644 --- a/lib/src/core/room.dart +++ b/lib/src/core/room.dart @@ -667,6 +667,15 @@ class Room extends DisposableChangeNotifier with EventsEmittable { ), ), ) + ..on((event) async { + if (!roomOptions.dynacast || _serverVersion == '0.15.1') { + return; + } + final publications = localParticipant?.videoTrackPublications.toList() ?? []; + for (final publication in publications) { + await publication.track?.refreshSubscribedCodecs(); + } + }) ..on((event) => _onEngineActiveSpeakersUpdateEvent(event.speakers)) ..on(_onDataMessageEvent) ..on(_onTranscriptionEvent) diff --git a/lib/src/internal/events.dart b/lib/src/internal/events.dart index 63c54a831..85624644b 100644 --- a/lib/src/internal/events.dart +++ b/lib/src/internal/events.dart @@ -227,6 +227,17 @@ class EngineLocalTrackSubscribedEvent with InternalEvent, EngineEvent { String toString() => '${runtimeType}(trackSid: ${trackSid})'; } +/// Emitted after a publisher answer is applied. The answer re-enables every +/// simulcast layer, so the last subscribed codecs have to be re-applied or the +/// layers dynacast paused stay up until the subscribed quality changes again. +@internal +class EngineSubscribedCodecRefreshRequestedEvent with InternalEvent, EngineEvent { + const EngineSubscribedCodecRefreshRequestedEvent(); + + @override + String toString() => '${runtimeType}()'; +} + @internal class EngineFullRestartingEvent with InternalEvent, EngineEvent { const EngineFullRestartingEvent(); diff --git a/lib/src/track/local/video.dart b/lib/src/track/local/video.dart index d1e4773dc..613adc391 100644 --- a/lib/src/track/local/video.dart +++ b/lib/src/track/local/video.dart @@ -50,6 +50,8 @@ class SimulcastTrackInfo { }); } +bool _hasPausedLayer(lk_rtc.SubscribedCodec codec) => codec.qualities.any((q) => !q.enabled); + /// A video track from the local device. Use static methods in this class to create /// video tracks. class LocalVideoTrack extends LocalTrack with VideoTrack { @@ -323,15 +325,48 @@ extension LocalVideoTrackExt on LocalVideoTrack { }); } + /// Re-applies the last subscribed codecs to the sender. A publisher answer + /// re-enables every layer, and dynacast only signals changes, so without this + /// the layers it paused keep uploading. + /// + /// Re-applying a pause needs a forced write: the answer re-enabled the layers + /// without going through `setParameters`, so where `sender.parameters` is a + /// local cache it still reports the layers as paused, and a conditional write + /// would find no change and send nothing. Only the codecs that actually have + /// a paused layer are forced; see [setPublishingCodecs]. + @internal + Future refreshSubscribedCodecs() async { + if (subscribedCodecs.isEmpty) { + return; + } + try { + await setPublishingCodecs(subscribedCodecs, this, force: true); + } catch (e) { + logger.warning('Failed to refresh subscribed codecs: $e'); + } + } + + /// Re-applies [codecs] to the senders publishing them. + /// + /// [force] asks for a write even where the sender already reports the state + /// being applied, for a caller re-asserting a pause the sender lost without + /// being told. It is narrowed per codec: one with every layer subscribed has + /// nothing to re-assert, so its sender is left alone. Future> setPublishingCodecs( List codecs, - LocalTrack track, - ) async { + LocalTrack track, { + bool force = false, + }) async { logger.fine('setPublishingCodecs $codecs'); // only enable simulcast codec for preference codec setted if (codec == null && codecs.isNotEmpty) { - await setPublishingLayers(track, codecs[0].qualities, isSVC: isSVCCodec(codecs[0].codec)); + await setPublishingLayers( + track, + codecs[0].qualities, + isSVC: isSVCCodec(codecs[0].codec), + force: force && _hasPausedLayer(codecs[0]), + ); return []; } @@ -340,8 +375,9 @@ extension LocalVideoTrackExt on LocalVideoTrack { final List newCodecs = []; for (var codec in codecs) { + final forceCodec = force && _hasPausedLayer(codec); if (this.codec?.toLowerCase() == codec.codec.toLowerCase()) { - await setPublishingLayers(track, codec.qualities, isSVC: isSVCCodec(codec.codec)); + await setPublishingLayers(track, codec.qualities, isSVC: isSVCCodec(codec.codec), force: forceCodec); } else { final simulcastCodecInfo = simulcastCodecs[codec.codec]; logger.fine('setPublishingCodecs $codecs'); @@ -359,6 +395,7 @@ extension LocalVideoTrackExt on LocalVideoTrack { simulcastCodecInfo.encodings!, codec.qualities, isSVC: isSVCCodec(codec.codec), + force: forceCodec, ); } } @@ -371,6 +408,7 @@ extension LocalVideoTrackExt on LocalVideoTrack { LocalTrack? track, List layers, { bool isSVC = false, + bool force = false, }) async { logger.fine('Update publishing layers: $layers'); @@ -391,7 +429,7 @@ extension LocalVideoTrackExt on LocalVideoTrack { return; } - return setPublishingLayersForSender(track!.sender!, encodings, layers, isSVC: isSVC); + return setPublishingLayersForSender(track!.sender!, encodings, layers, isSVC: isSVC, force: force); } lk_models.VideoQuality _videoQualityForRid(String rid) { @@ -412,12 +450,13 @@ extension LocalVideoTrackExt on LocalVideoTrack { List encodings, List layers, { bool isSVC = false, + bool force = false, }) async { logger.fine('Update publishing layers: $layers'); final params = sender.parameters; - var hasChanged = false; + var hasChanged = force; // NOTE: closable spatial layer is disabled due to video blur / frozen issues // with Chrome 113+ and LiveKit SFU PLI handling. See JS SDK LocalVideoTrack.ts:529-568. diff --git a/test/mock/mediastream_mock.dart b/test/mock/mediastream_mock.dart new file mode 100644 index 000000000..2cd128a18 --- /dev/null +++ b/test/mock/mediastream_mock.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 'dart:typed_data'; + +import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; + +class FakeMediaStream extends rtc.MediaStream { + final List _tracks = []; + + FakeMediaStream(String id) : super(id, 'fake-owner'); + + @override + bool? get active => true; + + @override + Future addTrack(rtc.MediaStreamTrack track, {bool addToNative = true}) async { + _tracks.add(track); + } + + @override + Future clone() async => FakeMediaStream('${id}_clone'); + + @override + List getAudioTracks() => _tracks.where((t) => t.kind == 'audio').toList(); + + @override + Future getMediaTracks() async {} + + @override + List getTracks() => List.from(_tracks); + + @override + List getVideoTracks() => _tracks.where((t) => t.kind == 'video').toList(); + + @override + Future removeTrack(rtc.MediaStreamTrack track, {bool removeFromNative = true}) async { + _tracks.remove(track); + } +} + +class FakeMediaStreamTrack implements rtc.MediaStreamTrack { + @override + rtc.StreamTrackCallback? onEnded; + + @override + rtc.StreamTrackCallback? onMute; + + @override + rtc.StreamTrackCallback? onUnMute; + + @override + bool enabled; + + @override + final String id; + + @override + final String kind; + + @override + String? get label => '$kind-track'; + + @override + bool? get muted => false; + + FakeMediaStreamTrack({ + required this.id, + required this.kind, + this.enabled = true, + }); + + @override + Future adaptRes(int width, int height) async {} + + @override + Future applyConstraints([Map? constraints]) async {} + + @override + Future captureFrame() { + throw UnimplementedError(); + } + + @override + Future clone() async => FakeMediaStreamTrack(id: id, kind: kind, enabled: enabled); + + @override + Future dispose() async {} + + @override + Map getConstraints() => const {}; + + @override + Map getSettings() => const {}; + + @override + Future hasTorch() async => false; + + @override + void enableSpeakerphone(bool enable) {} + + @override + Future setTorch(bool torch) async {} + + @override + Future stop() async {} + + @override + Future switchCamera() async => false; +} diff --git a/test/track/simulcast_state_test.dart b/test/track/simulcast_state_test.dart index 26992194c..a8b764f08 100644 --- a/test/track/simulcast_state_test.dart +++ b/test/track/simulcast_state_test.dart @@ -12,21 +12,20 @@ // See the License for the specific language governing permissions and // limitations under the License. -import 'dart:typed_data'; - import 'package:flutter_test/flutter_test.dart'; import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; import 'package:livekit_client/src/track/local/video.dart'; import 'package:livekit_client/src/track/options.dart'; import 'package:livekit_client/src/types/other.dart'; +import '../mock/mediastream_mock.dart'; void main() { TestWidgetsFlutterBinding.ensureInitialized(); LocalVideoTrack createTrack() { - final mediaTrack = _FakeMediaStreamTrack(id: 'video-1', kind: 'video'); - final stream = _FakeMediaStream('stream-1'); + final mediaTrack = FakeMediaStreamTrack(id: 'video-1', kind: 'video'); + final stream = FakeMediaStream('stream-1'); return LocalVideoTrack( TrackSource.camera, stream, @@ -59,107 +58,3 @@ void main() { }); }); } - -class _FakeMediaStream extends rtc.MediaStream { - final List _tracks = []; - - _FakeMediaStream(String id) : super(id, 'fake-owner'); - - @override - bool? get active => true; - - @override - Future addTrack(rtc.MediaStreamTrack track, {bool addToNative = true}) async { - _tracks.add(track); - } - - @override - Future clone() async => _FakeMediaStream('${id}_clone'); - - @override - List getAudioTracks() => _tracks.where((t) => t.kind == 'audio').toList(); - - @override - Future getMediaTracks() async {} - - @override - List getTracks() => List.from(_tracks); - - @override - List getVideoTracks() => _tracks.where((t) => t.kind == 'video').toList(); - - @override - Future removeTrack(rtc.MediaStreamTrack track, {bool removeFromNative = true}) async { - _tracks.remove(track); - } -} - -class _FakeMediaStreamTrack implements rtc.MediaStreamTrack { - @override - rtc.StreamTrackCallback? onEnded; - - @override - rtc.StreamTrackCallback? onMute; - - @override - rtc.StreamTrackCallback? onUnMute; - - @override - bool enabled; - - @override - final String id; - - @override - final String kind; - - @override - String? get label => '$kind-track'; - - @override - bool? get muted => false; - - _FakeMediaStreamTrack({ - required this.id, - required this.kind, - this.enabled = true, - }); - - @override - Future adaptRes(int width, int height) async {} - - @override - Future applyConstraints([Map? constraints]) async {} - - @override - Future captureFrame() { - throw UnimplementedError(); - } - - @override - Future clone() async => _FakeMediaStreamTrack(id: id, kind: kind, enabled: enabled); - - @override - Future dispose() async {} - - @override - Map getConstraints() => const {}; - - @override - Map getSettings() => const {}; - - @override - Future hasTorch() async => false; - - @override - void enableSpeakerphone(bool enable) {} - - @override - Future setTorch(bool torch) async {} - - @override - Future stop() async {} - - @override - Future switchCamera() async => false; -} diff --git a/test/track/subscribed_codec_refresh_test.dart b/test/track/subscribed_codec_refresh_test.dart new file mode 100644 index 000000000..951565922 --- /dev/null +++ b/test/track/subscribed_codec_refresh_test.dart @@ -0,0 +1,235 @@ +// 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:flutter_webrtc/flutter_webrtc.dart' as rtc; +import 'package:mockito/mockito.dart'; + +import 'package:livekit_client/livekit_client.dart'; +import 'package:livekit_client/src/proto/livekit_models.pb.dart' as lk_models; +import 'package:livekit_client/src/proto/livekit_rtc.pb.dart' as lk_rtc; +import '../mock/e2e_container.dart'; +import '../mock/mediastream_mock.dart'; + +lk_rtc.SubscribedCodec _codec(String name, {required bool medium, required bool low}) => lk_rtc.SubscribedCodec( + codec: name, + qualities: [ + lk_rtc.SubscribedQuality(quality: lk_models.VideoQuality.HIGH, enabled: true), + lk_rtc.SubscribedQuality(quality: lk_models.VideoQuality.MEDIUM, enabled: medium), + lk_rtc.SubscribedQuality(quality: lk_models.VideoQuality.LOW, enabled: low), + ], +); + +/// The dynacast state the SFU last asked for: only the high layer is subscribed. +List _onlyHighSubscribed() => [_codec('vp8', medium: false, low: false)]; + +/// Every layer subscribed, so dynacast has paused nothing. +List _allSubscribed() => [_codec('vp8', medium: true, low: true)]; + +LocalVideoTrack _createTrack(_FakeSender sender) { + final track = LocalVideoTrack( + TrackSource.camera, + FakeMediaStream('stream-1'), + FakeMediaStreamTrack(id: 'video-1', kind: 'video'), + const CameraCaptureOptions(), + ); + track.codec = 'vp8'; + track.transceiver = _FakeTransceiver(sender); + return track; +} + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + + group('LocalVideoTrack.refreshSubscribedCodecs', () { + test('writes to the sender even when its parameters already read as paused', () async { + // The state a native sender is in after the answer: the layers are live + // again, but `parameters` is a local cache that still reports the pause. + // Nothing here differs from the target, so only the forced write reaches + // the sender at all. + final sender = _FakeSender(cachedActive: [true, false, false]); + final track = _createTrack(sender); + track.subscribedCodecs = _onlyHighSubscribed(); + + await track.refreshSubscribedCodecs(); + + expect(sender.setParametersCalls, 1); + expect(sender.writtenActiveFlags, [true, false, false]); + }); + + test('re-pauses the layers when the sender reports them re-enabled', () async { + final sender = _FakeSender(cachedActive: [true, true, true]); + final track = _createTrack(sender); + track.subscribedCodecs = _onlyHighSubscribed(); + + await track.refreshSubscribedCodecs(); + + expect(sender.setParametersCalls, 1); + expect(sender.writtenActiveFlags, [true, false, false]); + }); + + test('does not touch the sender when no layer is paused', () async { + // Nothing was paused, so the answer undid nothing and the renegotiation + // owes no native call. + final sender = _FakeSender(cachedActive: [true, true, true]); + final track = _createTrack(sender); + track.subscribedCodecs = _allSubscribed(); + + await track.refreshSubscribedCodecs(); + + expect(sender.setParametersCalls, 0); + }); + + test('does nothing when no subscribed codecs have been recorded', () async { + final sender = _FakeSender(cachedActive: [true, true, true]); + final track = _createTrack(sender); + + await track.refreshSubscribedCodecs(); + + expect(sender.setParametersCalls, 0); + }); + + test('leaves a fully-subscribed backup codec alone while re-pausing the primary', () async { + // The backup path writes an SDK-owned encoding list wholesale over the + // sender, so forcing a codec that has nothing paused would clobber + // whatever the sender had adapted to. + final primarySender = _FakeSender(cachedActive: [true, false, false]); + final backupSender = _FakeSender(cachedActive: [true, true, true]); + final track = _createTrack(primarySender); + track.addSimulcastTrack('h264', backupSender.parameters.encodings!).sender = backupSender; + track.subscribedCodecs = [ + _codec('vp8', medium: false, low: false), + _codec('h264', medium: true, low: true), + ]; + + await track.refreshSubscribedCodecs(); + + expect(primarySender.setParametersCalls, 1); + expect(backupSender.setParametersCalls, 0); + }); + + test('swallows a sender failure rather than throwing into its caller', () async { + final sender = _FakeSender(throwOnParameters: true); + final track = _createTrack(sender); + track.subscribedCodecs = _onlyHighSubscribed(); + + // It runs from an engine event handler, where a throw has nowhere to go. + await expectLater(track.refreshSubscribedCodecs(), completes); + }); + }); + + group('a dynacast update still writes only on a real change', () { + test('leaves the sender alone when the subscribed qualities already hold', () async { + final sender = _FakeSender(cachedActive: [true, false, false]); + final track = _createTrack(sender); + + await track.setPublishingCodecs(_onlyHighSubscribed(), track); + + expect(sender.setParametersCalls, 0); + }); + }); + + group('publisher answer', () { + Future<_FakeSender> publishedTrackAfterAnswer({bool dynacast = true}) async { + final container = E2EContainer(roomOptions: RoomOptions(dynacast: dynacast)); + addTearDown(container.dispose); + await container.connectRoom(); + + final sender = _FakeSender(cachedActive: [true, false, false]); + final track = _createTrack(sender); + track.subscribedCodecs = _onlyHighSubscribed(); + container.room.localParticipant!.addTrackPublication( + LocalTrackPublication( + participant: container.room.localParticipant!, + info: lk_models.TrackInfo(sid: 'video_sid', type: lk_models.TrackType.VIDEO), + track: track, + ), + ); + + container.wsConnector.onData( + lk_rtc.SignalResponse( + answer: lk_rtc.SessionDescription(type: 'answer', sdp: 'v=0\r\n'), + ).writeToBuffer(), + ); + await Future.delayed(const Duration(milliseconds: 50)); + + return sender; + } + + test('re-applies the pause the answer undid', () async { + final sender = await publishedTrackAfterAnswer(); + + expect(sender.setParametersCalls, 1); + expect(sender.writtenActiveFlags, [true, false, false]); + }); + + test('leaves the sender alone when dynacast is off', () async { + expect((await publishedTrackAfterAnswer(dynacast: false)).setParametersCalls, 0); + }); + }); +} + +class _FakeTransceiver extends Mock implements rtc.RTCRtpTransceiver { + @override + final rtc.RTCRtpSender sender; + + _FakeTransceiver(this.sender); +} + +/// Models a native sender, whose `parameters` is a local cache that only +/// [setParameters] updates — never a live read of what the layers are doing. +class _FakeSender extends Mock implements rtc.RTCRtpSender { + final bool throwOnParameters; + int setParametersCalls = 0; + + rtc.RTCRtpParameters _parameters; + + _FakeSender({ + List cachedActive = const [true, false, false], + this.throwOnParameters = false, + }) : _parameters = rtc.RTCRtpParameters( + encodings: [ + rtc.RTCRtpEncoding(rid: 'f', active: cachedActive[0]), + rtc.RTCRtpEncoding(rid: 'h', active: cachedActive[1]), + rtc.RTCRtpEncoding(rid: 'q', active: cachedActive[2]), + ], + ); + + /// The layer states as of the last [setParameters]. Snapshotted on the way + /// in, because the caller mutates the very encodings this fake handed out — + /// reading them back would pass even if the write never happened. + List? writtenActiveFlags; + + @override + rtc.RTCRtpParameters get parameters { + if (throwOnParameters) { + throw Exception('sender is gone'); + } + return _parameters; + } + + @override + String get senderId => 'sender-1'; + + @override + Future setParameters(rtc.RTCRtpParameters parameters) async { + setParametersCalls++; + writtenActiveFlags = parameters.encodings!.map((e) => e.active).toList(); + _parameters = parameters; + return true; + } +}