Merge pull request #213 from GetStream/feature/uni-162-implement-handling-sfu-inboundstatenotification

Implement handling InboundStateNotification - this allows SFU to pause tracks if needed

Author: sierpinskid

Packages/StreamVideo/Runtime/Core/LowLevelClient/RtcSession.cs

@@ -1580,6 +1580,11 @@ var updateLocalParticipantState
             UpdateParticipantTracksState(userId, sessionId, type, isEnabled: false, updateLocalParticipantState,
                 out var participant);
 
+            if (participant != null)
+            {
+                participant.ClearTrackPausedByServer(type);
+            }
+
             if (participantSfuDto != null && participant != null)
             {
                 participant.UpdateFromSfu(participantSfuDto);
@@ -1982,10 +1987,25 @@ private void OnSfuWebSocketOnChangePublishOptions(ChangePublishOptions obj)
             // StreamTODO: Implement OnSfuWebSocketOnChangePublishOptions
         }
 
-        private void OnSfuInboundStateNotification(InboundStateNotification obj)
+        private void OnSfuInboundStateNotification(InboundStateNotification inboundStateNotification)
         {
-            _sfuTracer?.Trace("inboundStateNotification", obj);
-            //StreamTODO: implement
+            _sfuTracer?.Trace("inboundStateNotification", inboundStateNotification);
+
+            foreach (var state in inboundStateNotification.InboundVideoStates)
+            {
+                var trackType = state.TrackType.ToPublicEnum();
+                var participant = (StreamVideoCallParticipant)ActiveCall?.Participants
+                    .FirstOrDefault(p => p.SessionId == state.SessionId);
+
+                if (participant == null)
+                {
+                    _logs.WarningIfDebug(
+                        $"[InboundState] Received pause notification for unknown session: {state.SessionId}");
+                    continue;
+                }
+
+                participant.SetTrackPausedByServer(trackType, state.Paused);
+            }
         }
 
         private void OnSfuWebSocketOnParticipantMigrationComplete()

Packages/StreamVideo/Runtime/Core/LowLevelClient/WebSockets/SfuWebSocket.cs

@@ -180,6 +180,8 @@ protected override async Task<JoinResponse> ExecuteConnectAsync(SfuConnectReques
                     Source = ParticipantSource.WebrtcUnspecified,
                 };
 
+                joinRequest.Capabilities.Add(ClientCapability.SubscriberVideoPause);
+
                 var sfuJoinRequest = new SfuRequest
                 {
                     JoinRequest = joinRequest,

Packages/StreamVideo/Runtime/Core/Models/Sfu/TrackType.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 using System.Collections.Generic;
 using System.Linq;
 using Unity.WebRTC;

Packages/StreamVideo/Runtime/Core/StatefulModels/StreamVideoCallParticipant.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 using System.Collections.Generic;
 using System.Threading.Tasks;
 using StreamVideo.Core.InternalDTO.Responses;
@@ -300,6 +300,11 @@ internal void SetTrack(TrackType type, MediaStreamTrack mediaStreamTrack, out IS
                     throw new ArgumentOutOfRangeException(nameof(type), type, null);
             }
 
+            if (_pausedTracks.Contains(type))
+            {
+                ((BaseStreamTrack)streamTrack).SetServerPaused(true);
+            }
+
             TrackAdded?.Invoke(this, streamTrack);
         }
 
@@ -323,12 +328,38 @@ internal void NotifyTrackEnabled(TrackType type, bool enabled)
             // Join call by host and watcher
             // Disable track on host -> this causes watcher to disable the track as well
             // Leave the call as host and re-join -> the track is re-enabled on host side but watcher still has it disabled
-            streamTrack.SetEnabled(enabled);
+            streamTrack.SetPublisherEnabled(enabled);
 
             //StreamTodo: we should trigger some event that track status changed
             TrackIsEnabledChanged?.Invoke(this, streamTrack);
         }
 
+        internal void SetTrackPausedByServer(TrackType trackType, bool paused)
+        {
+            bool changed;
+            if (paused)
+            {
+                changed = _pausedTracks.Add(trackType);
+            }
+            else
+            {
+                changed = _pausedTracks.Remove(trackType);
+            }
+
+            if (changed)
+            {
+                GetStreamTrack(trackType)?.SetServerPaused(paused);
+            }
+        }
+
+        internal void ClearTrackPausedByServer(TrackType trackType)
+        {
+            if (_pausedTracks.Remove(trackType))
+            {
+                GetStreamTrack(trackType)?.SetServerPaused(false);
+            }
+        }
+
         internal void SetIsPinned(bool isPinned) => IsPinned = isPinned;
 
         protected override string InternalUniqueId
@@ -356,6 +387,7 @@ protected override Task UploadCustomDataAsync()
         #region Sfu State
 
         private readonly List<TrackType> _publishedTracks = new List<TrackType>();
+        private readonly HashSet<TrackType> _pausedTracks = new HashSet<TrackType>();
         private readonly List<string> _roles = new List<string>();
         private float _audioLevel;
         private bool _isSpeaking;

Packages/StreamVideo/Runtime/Core/StatefulModels/Tracks/BaseStreamTrack.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 using Unity.WebRTC;
 
 namespace StreamVideo.Core.StatefulModels.Tracks
@@ -12,36 +12,39 @@ public abstract class BaseStreamTrack : IStreamTrack
     {
         public event StreamTrackStateChangeHandler EnabledChanged;
 
-        public bool IsEnabled
-        {
-            get => _isEnabled;
-            private set
-            {
-                if(value == _isEnabled)
-                {
-                    return;
-                }
-                
-                _isEnabled = value;
-                EnabledChanged?.Invoke(_isEnabled);
-            }
-        }
+        /// <summary>
+        /// Effective enabled state: true only when the publisher has the track enabled
+        /// AND the Stream Server (SFU) has not paused it.
+        /// </summary>
+        public bool IsEnabled => _publisherEnabled && !_serverPaused;
+
+        /// <summary>
+        /// Whether the Stream Server (SFU) has paused this inbound track (e.g. due to insufficient bandwidth).
+        /// </summary>
+        public bool IsPausedByServer => _serverPaused;
 
         public void Dispose() => OnDisposing();
 
         internal BaseStreamTrack(MediaStreamTrack track)
         {
             InternalTrack = track ?? throw new ArgumentNullException(nameof(track));
-            _isEnabled = track.Enabled;
+            _publisherEnabled = track.Enabled;
         }
 
         internal virtual void Update()
         {
         }
 
-        internal void SetEnabled(bool enabled)
+        internal void SetPublisherEnabled(bool enabled)
         {
-            IsEnabled = enabled;
+            if (_publisherEnabled == enabled)
+            {
+                return;
+            }
+
+            var wasEnabled = IsEnabled;
+            _publisherEnabled = enabled;
+            NotifyIfEffectiveStateChanged(wasEnabled);
 
             //StreamTodo: investigate this. In theory we should disable track whenever the remote user disabled it.
             //But there's and edge case where:
@@ -53,14 +56,35 @@ internal void SetEnabled(bool enabled)
             // InternalTrack.Enabled = enabled;
         }
 
+        internal void SetServerPaused(bool paused)
+        {
+            if (_serverPaused == paused)
+            {
+                return;
+            }
+
+            var wasEnabled = IsEnabled;
+            _serverPaused = paused;
+            NotifyIfEffectiveStateChanged(wasEnabled);
+        }
+
         protected MediaStreamTrack InternalTrack { get; set; }
 
         protected virtual void OnDisposing()
         {
 
         }
-        
-        private bool _isEnabled;
+
+        private bool _publisherEnabled;
+        private bool _serverPaused;
+
+        private void NotifyIfEffectiveStateChanged(bool wasEnabled)
+        {
+            if (IsEnabled != wasEnabled)
+            {
+                EnabledChanged?.Invoke(IsEnabled);
+            }
+        }
     }
 
     public abstract class BaseStreamTrack<TTrack> : BaseStreamTrack

Packages/StreamVideo/Runtime/Core/StatefulModels/Tracks/IStreamTrack.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 
 namespace StreamVideo.Core.StatefulModels.Tracks
 {
@@ -10,8 +10,16 @@ public interface IStreamTrack : IDisposable
         event StreamTrackStateChangeHandler EnabledChanged;
 
         /// <summary>
-        /// Is this track active.
+        /// Is this track active. This is false when either the publisher has disabled
+        /// the track or the Stream Server (SFU) has paused it (e.g. due to insufficient bandwidth).
         /// </summary>
         bool IsEnabled { get; }
+
+        /// <summary>
+        /// Whether the Stream Server (SFU) has paused this inbound track due to bandwidth constraints.
+        /// Use this to distinguish "publisher turned off the camera" from
+        /// "video paused by the server due to poor network conditions".
+        /// </summary>
+        bool IsPausedByServer { get; }
     }
 }