diff --git a/.changes/keep-processor-across-restart b/.changes/keep-processor-across-restart new file mode 100644 index 000000000..22d050784 --- /dev/null +++ b/.changes/keep-processor-across-restart @@ -0,0 +1 @@ +patch type="fixed" "A track processor now stays attached across mute and unmute with the default stop-on-mute capture options, and across restartTrack such as a camera switch, instead of being silently dropped. It is restarted on the new media track, and the sender carries its processed track when it has one" \ No newline at end of file diff --git a/lib/src/track/local/local.dart b/lib/src/track/local/local.dart index 7252cb045..4b35ef4ce 100644 --- a/lib/src/track/local/local.dart +++ b/lib/src/track/local/local.dart @@ -196,7 +196,7 @@ abstract class LocalTrack extends Track { if (muted) return false; // already muted await disable(); if (!skipStopForTrackMute() && stopOnMute) { - await stop(); + await _stopKeepingProcessor(); } updateMuted(true, shouldSendSignal: true); return true; @@ -218,7 +218,24 @@ abstract class LocalTrack extends Track { @override Future stop() async { - final didStop = await super.stop() || !_stopped; + final didStop = await _disposeMediaTrack(await super.stop()); + // Also runs when a stop-on-mute already stopped the media track and kept + // the processor for the restart. + try { + await stopProcessor(); + } catch (error) { + logger.severe('LocalTrack.stopProcessor did throw: $error'); + } + return didStop; + } + + // Keeps the processor, which restartTrack restarts on the new media track. + Future _stopKeepingProcessor() async => _disposeMediaTrack(await super.stop()); + + // Takes the result of super.stop() because stop() has to call it directly + // (Track.stop is @mustCallSuper). + Future _disposeMediaTrack(bool trackStopped) async { + final didStop = trackStopped || !_stopped; if (didStop) { logger.fine('Stopping mediaStreamTrack...'); try { @@ -232,13 +249,6 @@ abstract class LocalTrack extends Track { logger.severe('MediaStreamTrack.dispose() did throw $error'); } _stopped = true; - try { - if (_processor != null) { - await stopProcessor(); - } - } catch (error) { - logger.severe('LocalTrack.stopProcessor did throw: $error'); - } } return didStop; } @@ -314,34 +324,43 @@ abstract class LocalTrack extends Track { currentOptions = options ?? currentOptions; // stop if not already stopped... - await stop(); + await _stopKeepingProcessor(); // create new track with options final newStream = await LocalTrack.createStream(currentOptions); final newTrack = newStream.getTracks().first; - final processor = _processor; + // set new stream & track to this object + updateMediaStreamAndTrack(newStream, newTrack); - await stopProcessor(); + // Restart the processor on the new track before the sender gets it, so + // the sender carries the processed track when the processor has one. + final processor = _processor; + if (processor != null) { + try { + await processor.restart( + kind == TrackType.VIDEO ? VideoProcessorOptions(track: newTrack) : AudioProcessorOptions(track: newTrack), + ); + final processedTrack = processor.processedTrack; + if (processedTrack != null) { + setProcessedTrack(processedTrack); + } + } catch (error) { + logger.severe('TrackProcessor.restart() did throw $error'); + } + } // replace track on sender try { - await sender?.replaceTrack(newTrack); + await sender?.replaceTrack(mediaStreamTrack); if (this is LocalVideoTrack) { final videoTrack = this as LocalVideoTrack; - await videoTrack.replaceTrackForMultiCodecSimulcast(newTrack); + await videoTrack.replaceTrackForMultiCodecSimulcast(mediaStreamTrack); } } catch (error) { logger.severe('RTCRtpSender.replaceTrack() did throw $error'); } - // set new stream & track to this object - updateMediaStreamAndTrack(newStream, newTrack); - - if (processor != null) { - await setProcessor(processor); - } - // mark as started await start(); diff --git a/test/track/local_track_processor_test.dart b/test/track/local_track_processor_test.dart new file mode 100644 index 000000000..4b1d8ae9a --- /dev/null +++ b/test/track/local_track_processor_test.dart @@ -0,0 +1,206 @@ +// 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/services.dart'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; + +import 'package:livekit_client/livekit_client.dart'; +import 'package:livekit_client/src/support/native.dart'; + +class _FakeProcessedTrack implements rtc.MediaStreamTrack { + _FakeProcessedTrack(this.id); + + @override + final String id; + + @override + String get kind => 'audio'; + + @override + bool enabled = true; + + @override + Future stop() async {} + + @override + Future dispose() async {} + + @override + dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation); +} + +class _FakeProcessor implements TrackProcessor { + _FakeProcessor({this.processes = false, this.throwOnRestart = false}); + + /// Hands out a processed track, as Krisp does on web. + final bool processes; + final bool throwOnRestart; + + final List initTrackIds = []; + final List restartTrackIds = []; + int destroyCount = 0; + rtc.MediaStreamTrack? _processedTrack; + + @override + String get name => 'fake'; + + @override + Future init(AudioProcessorOptions options) async { + initTrackIds.add(options.track.id!); + if (processes) _processedTrack = _FakeProcessedTrack('processed-${options.track.id}'); + } + + @override + Future restart(AudioProcessorOptions options) async { + restartTrackIds.add(options.track.id!); + if (throwOnRestart) throw Exception('restart failed'); + if (processes) _processedTrack = _FakeProcessedTrack('processed-${options.track.id}'); + } + + @override + Future destroy() async => destroyCount++; + + @override + Future onPublish(Room room) async {} + + @override + Future onUnpublish() async {} + + @override + rtc.MediaStreamTrack? get processedTrack => _processedTrack; +} + +class _FakeSender implements rtc.RTCRtpSender { + final List replacedTrackIds = []; + + @override + Future replaceTrack(rtc.MediaStreamTrack? track) async => replacedTrackIds.add(track?.id); + + @override + dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation); +} + +class _FakeTransceiver implements rtc.RTCRtpTransceiver { + _FakeTransceiver(this.sender); + + @override + final rtc.RTCRtpSender sender; + + @override + dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation); +} + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + final messenger = TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger; + const webrtcChannel = MethodChannel('FlutterWebRTC.Method'); + + var trackCounter = 0; + + setUp(() { + trackCounter = 0; + messenger.setMockMethodCallHandler(webrtcChannel, (call) async { + if (call.method == 'getUserMedia') { + trackCounter++; + return { + 'streamId': 'stream-$trackCounter', + 'audioTracks': [ + {'id': 'mic-$trackCounter', 'label': 'mic', 'kind': 'audio', 'enabled': true}, + ], + 'videoTracks': [], + }; + } + return null; + }); + messenger.setMockMethodCallHandler(Native.channel, (call) async => true); + }); + + tearDown(() { + messenger.setMockMethodCallHandler(webrtcChannel, null); + messenger.setMockMethodCallHandler(Native.channel, null); + }); + + Future<(LocalAudioTrack, _FakeProcessor, _FakeSender)> createPublishedTrack([_FakeProcessor? fake]) async { + final processor = fake ?? _FakeProcessor(); + final track = await LocalAudioTrack.create(AudioCaptureOptions(processor: processor)); + await track.start(); + final sender = _FakeSender(); + track.transceiver = _FakeTransceiver(sender); + return (track, processor, sender); + } + + test('mute and unmute restart the processor on the new track', () async { + final (track, processor, sender) = await createPublishedTrack(); + expect(processor.initTrackIds, ['mic-1']); + + await track.mute(); + expect(track.processor, same(processor)); + await track.unmute(); + + expect(sender.replacedTrackIds, ['mic-2']); + expect(track.processor, same(processor)); + expect(processor.initTrackIds, ['mic-1']); + expect(processor.restartTrackIds, ['mic-2']); + expect(processor.destroyCount, 0); + }); + + test('restartTrack restarts the processor on the new track', () async { + final (track, processor, sender) = await createPublishedTrack(); + + await track.restartTrack(); + + expect(sender.replacedTrackIds, ['mic-2']); + expect(track.processor, same(processor)); + expect(processor.initTrackIds, ['mic-1']); + expect(processor.restartTrackIds, ['mic-2']); + expect(processor.destroyCount, 0); + }); + + test('the sender gets the processed track after unmute and restartTrack', () async { + final (track, processor, sender) = await createPublishedTrack(_FakeProcessor(processes: true)); + expect(track.mediaStreamTrack.id, 'processed-mic-1'); + + await track.mute(); + await track.unmute(); + expect(track.mediaStreamTrack.id, 'processed-mic-2'); + + await track.restartTrack(); + expect(track.mediaStreamTrack.id, 'processed-mic-3'); + + expect(sender.replacedTrackIds, ['processed-mic-2', 'processed-mic-3']); + expect(processor.restartTrackIds, ['mic-2', 'mic-3']); + }); + + test('a processor that fails to restart still leaves the new track on the sender', () async { + final (track, processor, sender) = await createPublishedTrack(_FakeProcessor(throwOnRestart: true)); + + await track.restartTrack(); + + expect(sender.replacedTrackIds, ['mic-2']); + expect(track.mediaStreamTrack.id, 'mic-2'); + expect(track.processor, same(processor)); + }); + + test('stop releases the processor, also after a mute already stopped capture', () async { + final (track, processor, _) = await createPublishedTrack(); + + await track.mute(); + await track.stop(); + + expect(track.processor, isNull); + expect(processor.destroyCount, 1); + }); +}