Dynacast (#62)

* signal

* engine

* dynacast option

* implement

* update for non simulcast

* use low when rid is null

* firefox comment

example/pubspec.lock

@@ -56,7 +56,7 @@ packages:
       name: dart_webrtc
       url: "https://pub.dartlang.org"
     source: hosted
-    version: "1.0.2"
+    version: "1.0.3"
   eva_icons_flutter:
     dependency: "direct main"
     description:
@@ -134,14 +134,14 @@ packages:
       name: flutter_webrtc
       url: "https://pub.dartlang.org"
     source: hosted
-    version: "0.8.0"
+    version: "0.8.1"
   google_fonts:
     dependency: "direct main"
     description:
       name: google_fonts
       url: "https://pub.dartlang.org"
     source: hosted
-    version: "2.1.0"
+    version: "2.2.0"
   http:
     dependency: transitive
     description:
@@ -239,7 +239,7 @@ packages:
       name: path_provider_android
       url: "https://pub.dartlang.org"
     source: hosted
-    version: "2.0.9"
+    version: "2.0.11"
   path_provider_ios:
     dependency: transitive
     description:
@@ -316,7 +316,7 @@ packages:
       name: provider
       url: "https://pub.dartlang.org"
     source: hosted
-    version: "6.0.1"
+    version: "6.0.2"
   shared_preferences:
     dependency: "direct main"
     description:
@@ -461,7 +461,7 @@ packages:
       name: win32
       url: "https://pub.dartlang.org"
     source: hosted
-    version: "2.3.1"
+    version: "2.3.3"
   xdg_directories:
     dependency: transitive
     description:

lib/src/core/engine.dart

@@ -600,6 +600,8 @@ class Engine extends Disposable with EventsEmittable<EngineEvent> {
     ..on<SignalConnectionQualityUpdateEvent>((event) => events.emit(event))
     // relay
     ..on<SignalStreamStateUpdatedEvent>((event) => events.emit(event))
+    // relay to Room
+    ..on<SignalSubscribedQualityUpdatedEvent>((event) => events.emit(event))
     ..on<SignalLeaveEvent>((event) async {
       if (connectionState == ConnectionState.reconnecting) {
         logger.warning('Received leave signal while engine is reconnecting.');

lib/src/core/room.dart

@@ -56,6 +56,10 @@ class Room extends DisposableChangeNotifier with EventsEmittable<RoomEvent> {
   /// sid of the room
   String? sid;
 
+  /// Server version
+  String? get serverVersion => _serverVersion;
+  String? _serverVersion;
+
   List<Participant> _activeSpeakers = [];
 
   /// a list of participants that are actively speaking, including local participant.
@@ -109,6 +113,7 @@ class Room extends DisposableChangeNotifier with EventsEmittable<RoomEvent> {
 
     sid = joinResponse.room.sid;
     name = joinResponse.room.name;
+    _serverVersion = joinResponse.serverVersion;
 
     logger.fine(
         'Connected to LiveKit server, version: ${joinResponse.serverVersion}');
@@ -168,6 +173,20 @@ class Room extends DisposableChangeNotifier with EventsEmittable<RoomEvent> {
         await publication?.unmute();
       }
     })
+    ..on<SignalSubscribedQualityUpdatedEvent>((event) {
+      // Signal for Dynacast
+      final options = roomOptions ?? const RoomOptions();
+      // Dynacast is off or is unsupported
+      if (!options.dynacast || _serverVersion == '0.15.1') return;
+      // Find the publication
+      final publication = localParticipant?.trackPublications[event.trackSid];
+      if (publication == null) {
+        logger.warning(
+            'Received subscribed quality update for unknown track (${event.trackSid})');
+        return;
+      }
+      publication.updatePublishingLayers(event.updates);
+    })
     ..on<EngineTrackAddedEvent>((event) async {
       logger.fine('EngineTrackAddedEvent trackSid:${event.track.id}');
 

lib/src/core/signal_client.dart

@@ -180,6 +180,12 @@ class SignalClient extends Disposable with EventsEmittable<SignalEvent> {
           updates: msg.streamStateUpdate.streamStates,
         ));
         break;
+      case lk_rtc.SignalResponse_Message.subscribedQualityUpdate:
+        events.emit(SignalSubscribedQualityUpdatedEvent(
+          trackSid: msg.subscribedQualityUpdate.trackSid,
+          updates: msg.subscribedQualityUpdate.subscribedQualities,
+        ));
+        break;
       default:
         logger.warning('skipping unsupported signal message');
     }

lib/src/extensions.dart

@@ -132,3 +132,11 @@ extension PBStreamStateExt on lk_rtc.StreamState {
       }[this] ??
       StreamState.paused;
 }
+
+extension VideoQualityExt on lk_models.VideoQuality {
+  String toRid() => {
+        lk_models.VideoQuality.HIGH: 'f',
+        lk_models.VideoQuality.MEDIUM: 'h',
+        lk_models.VideoQuality.LOW: 'q',
+      }[this]!;
+}

lib/src/internal/events.dart

@@ -196,6 +196,17 @@ class SignalStreamStateUpdatedEvent
   });
 }
 
+@internal
+class SignalSubscribedQualityUpdatedEvent
+    with SignalEvent, EngineEvent, InternalEvent {
+  final String trackSid;
+  final List<lk_rtc.SubscribedQuality> updates;
+  const SignalSubscribedQualityUpdatedEvent({
+    required this.trackSid,
+    required this.updates,
+  });
+}
+
 // ----------------------------------------------------------------------
 // Engine events
 // ----------------------------------------------------------------------

lib/src/options.dart

@@ -52,6 +52,11 @@ class RoomOptions {
   /// Defaults to false.
   final bool adaptiveStream;
 
+  /// enable Dynacast, off by default. With Dynacast dynamically pauses
+  /// video layers that are not being consumed by any subscribers, significantly
+  /// reducing publishing CPU and bandwidth usage.
+  final bool dynacast;
+
   /// Set this to false in case you would like to stop the track yourself.
   /// If you set this to false, make sure you call [Track.stop].
   /// Defaults to true.
@@ -64,6 +69,7 @@ class RoomOptions {
     this.defaultVideoPublishOptions = const VideoPublishOptions(),
     this.defaultAudioPublishOptions = const AudioPublishOptions(),
     this.adaptiveStream = false,
+    this.dynacast = false,
     this.stopLocalTrackOnUnpublish = true,
   });
 }

lib/src/publication/local.dart

@@ -1,5 +1,11 @@
+import 'package:collection/collection.dart';
+import 'package:meta/meta.dart';
+
+import '../extensions.dart';
+import '../logger.dart';
 import '../participant/local.dart';
 import '../proto/livekit_models.pb.dart' as lk_models;
+import '../proto/livekit_rtc.pb.dart' as lk_rtc;
 import '../track/local/local.dart';
 import 'track_publication.dart';
 
@@ -27,4 +33,40 @@ class LocalTrackPublication<T extends LocalTrack> extends TrackPublication<T> {
 
   /// Unmute the track associated with this publication
   Future<void> unmute() async => await track?.unmute();
+
+  @internal
+  void updatePublishingLayers(List<lk_rtc.SubscribedQuality> layers) async {
+    //
+    final params = track?.sender?.parameters;
+    if (params == null) return;
+
+    final encodings = params.encodings;
+    if (encodings == null) return;
+
+    bool didChange = false;
+
+    for (final encoding in encodings) {
+      final layer = layers.firstWhereOrNull((e) =>
+          // If there is exact match, use it
+          (e.quality.toRid() == encoding.rid) ||
+          // Use low layer if rid is null (not simulcast)
+          (encoding.rid == null && e.quality == lk_models.VideoQuality.LOW));
+      if (layer != null && encoding.active != layer.enabled) {
+        encoding.active = layer.enabled;
+        logger.fine('Setting layer ${layer.quality} to ${layer.enabled}');
+        // FireFox does not support setting encoding.active to false, so we
+        // have a workaround of lowering its bitrate and resolution to the min.
+        // TODO: Workaround for firefox
+        didChange = true;
+      }
+    }
+
+    if (didChange) {
+      params.encodings = encodings;
+      final result = await track?.sender?.setParameters(params);
+      if (result == false) {
+        logger.warning('Failed to update sender parameters');
+      }
+    }
+  }
 }