Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .changes/keep-processor-across-restart
Original file line number Diff line number Diff line change
@@ -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"
61 changes: 40 additions & 21 deletions lib/src/track/local/local.dart
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -218,7 +218,24 @@ abstract class LocalTrack extends Track {

@override
Future<bool> 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<bool> _stopKeepingProcessor() async => _disposeMediaTrack(await super.stop());

// Takes the result of super.stop() because stop() has to call it directly
// (Track.stop is @mustCallSuper).
Future<bool> _disposeMediaTrack(bool trackStopped) async {
final didStop = trackStopped || !_stopped;
if (didStop) {
logger.fine('Stopping mediaStreamTrack...');
try {
Expand All @@ -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;
}
Expand Down Expand Up @@ -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();

Expand Down
206 changes: 206 additions & 0 deletions test/track/local_track_processor_test.dart
Original file line number Diff line number Diff line change
@@ -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<void> stop() async {}

@override
Future<void> dispose() async {}

@override
dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation);
}

class _FakeProcessor implements TrackProcessor<AudioProcessorOptions> {
_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<String> initTrackIds = [];
final List<String> restartTrackIds = [];
int destroyCount = 0;
rtc.MediaStreamTrack? _processedTrack;

@override
String get name => 'fake';

@override
Future<void> init(AudioProcessorOptions options) async {
initTrackIds.add(options.track.id!);
if (processes) _processedTrack = _FakeProcessedTrack('processed-${options.track.id}');
}

@override
Future<void> 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<void> destroy() async => destroyCount++;

@override
Future<void> onPublish(Room room) async {}

@override
Future<void> onUnpublish() async {}

@override
rtc.MediaStreamTrack? get processedTrack => _processedTrack;
}

class _FakeSender implements rtc.RTCRtpSender {
final List<String?> replacedTrackIds = [];

@override
Future<void> 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 <String, dynamic>{
'streamId': 'stream-$trackCounter',
'audioTracks': [
<String, dynamic>{'id': 'mic-$trackCounter', 'label': 'mic', 'kind': 'audio', 'enabled': true},
],
'videoTracks': <dynamic>[],
};
}
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);
});
}
Loading