feat(llc): video pause feature (#1063)

* Participant source added

* changelog

* changelog tweak

* tweaks

* source exported

* update protocol model version

* added missing participant source when reconnecting

* video pause feature

* fixes

* added docs

Author: Brazol

packages/stream_video/CHANGELOG.md

@@ -3,6 +3,10 @@
 ✅ Added
 * Extended `CallParticipantState` with `participantSource`. This indicates the participant's source channel (e.g., WebRTC, RTMP, WHIP) and can be used in filtering and sorting criteria.
 * Livestream sorting preset now prioritises RTMP sources in layout sorting.
+* Automatic SFU-driven pausing of inbound video to save bandwidth and prevent visual artifacts:
+  * New `SfuClientCapability.subscriberVideoPause` (on by default). Control via `Call.enableClientCapabilities()` / `Call.disableClientCapabilities()`.
+  * New `SfuInboundStateNotificationEvent` notifies when inbound tracks are paused or resumed.
+  * `CallParticipantState.pausedTracks` and `CallParticipantState.isTrackPaused()` let you check which tracks are currently paused.
 * Added capability to remove a participant from a call via `call.kickUser()`. Requires the `kick-user` permission.
 
 🐞 Fixed

packages/stream_video/lib/src/call/call.dart

@@ -29,6 +29,7 @@ import '../models/models.dart';
 import '../platform_detector/platform_detector.dart';
 import '../retry/retry_policy.dart';
 import '../sfu/data/events/sfu_events.dart';
+import '../sfu/data/models/sfu_client_capability.dart';
 import '../sfu/data/models/sfu_error.dart';
 import '../sfu/data/models/sfu_track_type.dart';
 import '../shared_emitter.dart';
@@ -292,6 +293,9 @@ class Call {
   final Map<String, Timer> _reactionTimers = {};
   final Map<String, Timer> _captionsTimers = {};
   final List<CancelableOperation<void>> _sfuStatsTimers = [];
+  final Set<SfuClientCapability> _sfuClientCapabilities = {
+    SfuClientCapability.subscriberVideoPause, // on by default
+  };
 
   String get id => state.value.callId;
   StreamCallCid get callCid => state.value.callCid;
@@ -550,6 +554,28 @@ class Call {
     _stateManager.updateCallPreferences(preferences);
   }
 
+  /// Enables the given SFU client capabilities for this call.
+  ///
+  /// Should be configured before `call.join()` is called. Changes made after
+  /// joining will not affect the current session until the next join/reconnect.
+  void enableClientCapabilities(
+    List<SfuClientCapability> capabilities,
+  ) {
+    _logger.i(() => '[enableClientCapabilities] $capabilities');
+    capabilities.forEach(_sfuClientCapabilities.add);
+  }
+
+  /// Disables the given SFU client capabilities for this call.
+  ///
+  /// Should be configured before `call.join()` is called. Changes made after
+  /// joining will not affect the current session until the next join/reconnect.
+  void disableClientCapabilities(
+    List<SfuClientCapability> capabilities,
+  ) {
+    _logger.i(() => '[disableClientCapabilities] $capabilities');
+    capabilities.forEach(_sfuClientCapabilities.remove);
+  }
+
   /// Accepts the incoming call.
   Future<Result<None>> accept() async {
     final state = this.state.value;
@@ -1182,6 +1208,7 @@ class Call {
 
     final result = await session.start(
       reconnectDetails: reconnectDetails,
+      capabilities: _sfuClientCapabilities,
       onRtcManagerCreatedCallback: (_) async {
         _logger.v(() => '[startSession] applying connect options');
         await _applyConnectOptions();

packages/stream_video/lib/src/call/session/call_session.dart

@@ -20,6 +20,7 @@ import '../../errors/video_error.dart';
 import '../../errors/video_error_composer.dart';
 import '../../sfu/data/events/sfu_events.dart';
 import '../../sfu/data/models/sfu_call_state.dart';
+import '../../sfu/data/models/sfu_client_capability.dart';
 import '../../sfu/data/models/sfu_model_mapper_extensions.dart';
 import '../../sfu/data/models/sfu_subscription_details.dart';
 import '../../sfu/sfu_client.dart';
@@ -255,6 +256,7 @@ class CallSession extends Disposable {
             Duration fastReconnectDeadline,
           })>> start({
     sfu_events.ReconnectDetails? reconnectDetails,
+    Set<SfuClientCapability> capabilities = const {},
     FutureOr<void> Function(RtcManager)? onRtcManagerCreatedCallback,
     bool isAnonymousUser = false,
   }) async {
@@ -331,6 +333,7 @@ class CallSession extends Disposable {
         reconnectDetails: reconnectDetails,
         preferredPublishOptions: preferredPublishOptions,
         preferredSubscribeOptions: preferredSubscribeOptions,
+        capabilities: capabilities.map((c) => c.toDTO()).toList(),
         source:
             sfu_models.ParticipantSource.PARTICIPANT_SOURCE_WEBRTC_UNSPECIFIED,
       );
@@ -634,6 +637,8 @@ class CallSession extends Disposable {
         stateManager.sfuDominantSpeakerChanged(event);
       } else if (event is SfuPinsUpdatedEvent) {
         stateManager.sfuPinsUpdated(event.pins);
+      } else if (event is SfuInboundStateNotificationEvent) {
+        stateManager.sfuInboundStateNotification(event);
       }
     });
   }

packages/stream_video/lib/src/call/state/mixins/state_sfu_mixin.dart

@@ -69,6 +69,10 @@ mixin StateSfuMixin on StateNotifier<CallState> {
               ...participant.publishedTracks,
               if (trackState != null) event.trackType: trackState,
             },
+            pausedTracks: participant.pausedTracks
+                .toList()
+                .where((track) => track != event.trackType)
+                .toSet(),
           );
         }
 
@@ -267,4 +271,36 @@ mixin StateSfuMixin on StateNotifier<CallState> {
       ],
     );
   }
+
+  void sfuInboundStateNotification(SfuInboundStateNotificationEvent event) {
+    _logger.d(
+      () => '[sfuInboundStateNotification] ${state.sessionId}; event: $event',
+    );
+
+    state = state.copyWith(
+      callParticipants: state.callParticipants.map((participant) {
+        final inboundStates = event.inboundVideoStates.where((it) {
+          return it.userId == participant.userId &&
+              it.sessionId == participant.sessionId;
+        }).toList();
+
+        if (inboundStates.isEmpty) {
+          return participant;
+        }
+
+        final pausedTracks = {...participant.pausedTracks};
+        for (final inboundState in inboundStates) {
+          if (inboundState.paused) {
+            pausedTracks.add(inboundState.trackType);
+          } else {
+            pausedTracks.remove(inboundState.trackType);
+          }
+        }
+
+        return participant.copyWith(
+          pausedTracks: pausedTracks,
+        );
+      }).toList(),
+    );
+  }
 }

packages/stream_video/lib/src/models/call_participant_state.dart

@@ -25,6 +25,7 @@ class CallParticipantState
     required this.sessionId,
     required this.trackIdPrefix,
     this.publishedTracks = const {},
+    this.pausedTracks = const {},
     this.isLocal = false,
     this.connectionQuality = SfuConnectionQuality.unspecified,
     this.isOnline = false,
@@ -49,6 +50,7 @@ class CallParticipantState
     required this.sessionId,
     required this.trackIdPrefix,
     required this.publishedTracks,
+    required this.pausedTracks,
     required this.isLocal,
     required this.connectionQuality,
     required this.isOnline,
@@ -82,6 +84,14 @@ class CallParticipantState
   /// List of the last 10 audio levels.
   final List<double> audioLevels;
 
+  /// A list of tracks that are currently paused by our servers.
+  /// Typically, a server-side pause happens when the local participant doesn't
+  /// have enough bandwidth to receive all tracks. In this case, the server
+  /// will pause some tracks to optimize the bandwidth usage.
+  /// Once the bandwidth is restored, the server will resume the paused tracks.
+  /// This is useful to avoid any unwanted video and audio artifacts.
+  final Set<SfuTrackType> pausedTracks;
+
   final bool isSpeaking;
   final bool isDominantSpeaker;
   final CallParticipantPin? pin;
@@ -105,6 +115,7 @@ class CallParticipantState
     String? sessionId,
     String? trackIdPrefix,
     Map<SfuTrackType, TrackState>? publishedTracks,
+    Set<SfuTrackType>? pausedTracks,
     bool? isLocal,
     SfuConnectionQuality? connectionQuality,
     bool? isOnline,
@@ -127,6 +138,7 @@ class CallParticipantState
       sessionId: sessionId ?? this.sessionId,
       trackIdPrefix: trackIdPrefix ?? this.trackIdPrefix,
       publishedTracks: publishedTracks ?? this.publishedTracks,
+      pausedTracks: pausedTracks ?? this.pausedTracks,
       isLocal: isLocal ?? this.isLocal,
       connectionQuality: connectionQuality ?? this.connectionQuality,
       isOnline: isOnline ?? this.isOnline,
@@ -173,6 +185,7 @@ class CallParticipantState
       sessionId: sessionId,
       trackIdPrefix: trackIdPrefix,
       publishedTracks: publishedTracks,
+      pausedTracks: pausedTracks,
       isLocal: isLocal,
       connectionQuality: connectionQuality,
       isOnline: isOnline,
@@ -204,6 +217,7 @@ class CallParticipantState
       sessionId: sessionId,
       trackIdPrefix: trackIdPrefix,
       publishedTracks: publishedTracks,
+      pausedTracks: pausedTracks,
       isLocal: isLocal,
       connectionQuality: connectionQuality,
       isOnline: isOnline,
@@ -242,6 +256,7 @@ class CallParticipantState
         'sessionId: $sessionId, '
         'trackId: $trackIdPrefix, image: $image, '
         'publishedTracks: $publishedTracks, '
+        'pausedTracks: $pausedTracks, '
         'isLocal: $isLocal, '
         'connectionQuality: $connectionQuality, isOnline: $isOnline, '
         'audioLevel: $audioLevel, audioLevels: $audioLevels, isSpeaking: $isSpeaking, '
@@ -261,6 +276,7 @@ class CallParticipantState
         sessionId,
         trackIdPrefix,
         publishedTracks,
+        pausedTracks,
         isLocal,
         connectionQuality,
         isOnline,
@@ -299,6 +315,10 @@ class CallParticipantState
     return !(screenShareTrack?.muted ?? true);
   }
 
+  bool isTrackPaused(SfuTrackType trackType) {
+    return pausedTracks.contains(trackType);
+  }
+
   UserInfo toUserInfo() => UserInfo(
         id: userId,
         role: roles.firstOrNull ?? '',

packages/stream_video/lib/src/sfu/data/events/sfu_event_mapper_extensions.dart

@@ -11,11 +11,13 @@ import '../models/sfu_connection_info.dart';
 import '../models/sfu_connection_quality.dart';
 import '../models/sfu_error.dart';
 import '../models/sfu_goaway_reason.dart';
+import '../models/sfu_inbound_video_state.dart';
 import '../models/sfu_model_mapper_extensions.dart';
 import '../models/sfu_participant.dart';
 import '../models/sfu_participant_source.dart';
 import '../models/sfu_pin.dart';
 import '../models/sfu_publish_options.dart';
+import '../models/sfu_client_capability.dart';
 import '../models/sfu_track_type.dart';
 import '../models/sfu_video_layer_setting.dart';
 import '../models/sfu_video_sender.dart';
@@ -178,6 +180,18 @@ extension SfuEventMapper on sfu_events.SfuEvent {
         return SfuCallEndedEvent(
           callEndedReason: payload.reason.toDomain(),
         );
+      case sfu_events.SfuEvent_EventPayload.inboundStateNotification:
+        final payload = inboundStateNotification;
+        return SfuInboundStateNotificationEvent(
+          inboundVideoStates: payload.inboundVideoStates
+              .map((s) => SfuInboundVideoState(
+                    userId: s.userId,
+                    sessionId: s.sessionId,
+                    trackType: s.trackType.toDomain(),
+                    paused: s.paused,
+                  ))
+              .toList(),
+        );
       case sfu_events.SfuEvent_EventPayload.participantUpdated:
         final payload = participantUpdated;
         return SfuParticipantUpdatedEvent(
@@ -192,6 +206,18 @@ extension SfuEventMapper on sfu_events.SfuEvent {
   }
 }
 
+extension SfuClientCapabilityExtension on sfu_models.ClientCapability {
+  SfuClientCapability toDomain() {
+    switch (this) {
+      case sfu_models.ClientCapability.CLIENT_CAPABILITY_SUBSCRIBER_VIDEO_PAUSE:
+        return SfuClientCapability.subscriberVideoPause;
+      case sfu_models.ClientCapability.CLIENT_CAPABILITY_UNSPECIFIED:
+      default:
+        return SfuClientCapability.unspecified;
+    }
+  }
+}
+
 /// TODO
 extension SfuCallStateExtension on sfu_models.CallState {
   SfuCallState toDomain() {

packages/stream_video/lib/src/sfu/data/events/sfu_events.dart

@@ -12,6 +12,7 @@ import '../models/sfu_call_state.dart';
 import '../models/sfu_connection_info.dart';
 import '../models/sfu_error.dart';
 import '../models/sfu_goaway_reason.dart';
+import '../models/sfu_inbound_video_state.dart';
 import '../models/sfu_participant.dart';
 import '../models/sfu_pin.dart';
 import '../models/sfu_publish_options.dart';
@@ -266,6 +267,22 @@ class SfuCallGrantsUpdated extends SfuEvent {
   List<Object> get props => [currentGrants, message];
 }
 
+class SfuInboundStateNotificationEvent extends SfuEvent {
+  const SfuInboundStateNotificationEvent({
+    required this.inboundVideoStates,
+  });
+
+  final List<SfuInboundVideoState> inboundVideoStates;
+
+  @override
+  List<Object> get props => [inboundVideoStates];
+
+  @override
+  String toString() {
+    return 'SfuInboundStateNotificationEvent{inboundVideoStates: $inboundVideoStates}';
+  }
+}
+
 class SfuErrorEvent extends SfuEvent {
   const SfuErrorEvent({required this.error});
 

packages/stream_video/lib/src/sfu/data/models/sfu_client_capability.dart

@@ -0,0 +1,9 @@
+enum SfuClientCapability {
+  unspecified,
+  subscriberVideoPause;
+
+  @override
+  String toString() {
+    return name;
+  }
+}

packages/stream_video/lib/src/sfu/data/models/sfu_inbound_video_state.dart

@@ -0,0 +1,25 @@
+import 'package:equatable/equatable.dart';
+
+import 'sfu_track_type.dart';
+
+class SfuInboundVideoState with EquatableMixin {
+  const SfuInboundVideoState({
+    required this.userId,
+    required this.sessionId,
+    required this.trackType,
+    required this.paused,
+  });
+
+  final String userId;
+  final String sessionId;
+  final SfuTrackType trackType;
+  final bool paused;
+
+  @override
+  List<Object?> get props => [userId, sessionId, trackType, paused];
+
+  @override
+  String toString() {
+    return 'SfuInboundVideoState{userId: $userId, sessionId: $sessionId, trackType: $trackType, paused: $paused}';
+  }
+}

packages/stream_video/lib/src/sfu/data/models/sfu_model_mapper_extensions.dart

@@ -2,6 +2,7 @@ import '../../../../protobuf/video/sfu/models/models.pb.dart' as sfu_models;
 import '../../../../protobuf/video/sfu/signal_rpc/signal.pb.dart' as sfu;
 import '../../../webrtc/model/rtc_video_encoding.dart';
 import '../../../webrtc/peer_type.dart';
+import 'sfu_client_capability.dart';
 import 'sfu_codec.dart';
 import 'sfu_publish_options.dart';
 import 'sfu_subscription_details.dart';
@@ -21,6 +22,18 @@ extension SfuPeerTypeMapper on sfu_models.PeerType {
   }
 }
 
+extension SfuClientCapabilityMapper on SfuClientCapability {
+  sfu_models.ClientCapability toDTO() {
+    switch (this) {
+      case SfuClientCapability.subscriberVideoPause:
+        return sfu_models
+            .ClientCapability.CLIENT_CAPABILITY_SUBSCRIBER_VIDEO_PAUSE;
+      case SfuClientCapability.unspecified:
+        return sfu_models.ClientCapability.CLIENT_CAPABILITY_UNSPECIFIED;
+    }
+  }
+}
+
 extension SfuTrackTypeMapper on SfuTrackType {
   sfu_models.TrackType toDTO() {
     if (this == SfuTrackType.audio) {