Merge remote-tracking branch 'origin/main' into feature/better-suppor…

…t-for-disposable-assets-in-tests

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

@@ -0,0 +1,20 @@
+using System;
+
+namespace StreamVideo.Core.LowLevelClient
+{
+    internal class DisposedDuringOperationException : Exception
+    {
+
+    }
+
+    internal static class AppDisposedDuringOperationExceptionExt
+    {
+        public static void ThrowDisposedDuringOperationIfNull(this StreamPeerConnection streamPeerConnection)
+        {
+            if (streamPeerConnection == null)
+            {
+                throw new DisposedDuringOperationException();
+            }
+        }
+    }
+}

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

@@ -527,12 +527,15 @@ private async void OnSfuSubscriberOffer(SubscriberOffer subscriberOffer)
                 };
 
                 await _subscriber.SetRemoteDescriptionAsync(rtcSessionDescription);
+                _subscriber.ThrowDisposedDuringOperationIfNull();
 
                 var answer = await _subscriber.CreateAnswerAsync();
+                _subscriber.ThrowDisposedDuringOperationIfNull();
 
                 //StreamTodo: mangle SDP
 
                 await _subscriber.SetLocalDescriptionAsync(ref answer);
+                _subscriber.ThrowDisposedDuringOperationIfNull();
 
                 var sendAnswerRequest = new SendAnswerRequest
                 {
@@ -543,6 +546,10 @@ private async void OnSfuSubscriberOffer(SubscriberOffer subscriberOffer)
 
                 await RpcCallAsync(sendAnswerRequest, GeneratedAPI.SendAnswer, nameof(GeneratedAPI.SendAnswer),
                     preLog: true);
+            }
+            catch (DisposedDuringOperationException)
+            {
+
             }
             catch (Exception e)
             {
@@ -765,7 +772,8 @@ private async void OnPublisherNegotiationNeeded()
                 }
 
                 var offer = await _publisher.CreateOfferAsync();
-
+                _publisher.ThrowDisposedDuringOperationIfNull();
+
                 //StreamTodo: ignored the _config.Audio.EnableRed because with current webRTC version this modification causes a crash
                 //We're also forcing the red codec in the StreamPeerConnection but atm this results in "InvalidModification"
                 //This is most likely issue with the webRTC lib
@@ -782,6 +790,7 @@ private async void OnPublisherNegotiationNeeded()
                 }
 
                 await _publisher.SetLocalDescriptionAsync(ref offer);
+                _publisher.ThrowDisposedDuringOperationIfNull();
 
                 // //StreamTodo: timeout + break if we're disconnecting/reconnecting
                 // while (_sfuWebSocket.ConnectionState != ConnectionState.Connected)
@@ -809,6 +818,7 @@ private async void OnPublisherNegotiationNeeded()
 #endif
 
                 var result = await RpcCallAsync(request, GeneratedAPI.SetPublisher, nameof(GeneratedAPI.SetPublisher));
+                _publisher.ThrowDisposedDuringOperationIfNull();
 
 #if STREAM_DEBUG_ENABLED
                 _logs.Warning($"[Publisher] RemoteDesc (SDP Answer):\n{result.Sdp}");
@@ -819,6 +829,10 @@ await _publisher.SetRemoteDescriptionAsync(new RTCSessionDescription()
                     type = RTCSdpType.Answer,
                     sdp = result.Sdp
                 });
+            }
+            catch (DisposedDuringOperationException)
+            {
+
             }
             catch (Exception e)
             {

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

@@ -163,7 +163,7 @@ public void Dispose()
             WebsocketClient.ConnectionFailed -= OnConnectionFailed;
             WebsocketClient.Disconnected -= OnDisconnected;
 
-            //StreamTodo: we're disposing the WS but we're not the owner, would be better to accept a factory method so we own the obj
+            //StreamTodo: we're disposing the WS but we're not the owner, would be better to receive a factory method so we own the obj
             WebsocketClient.Dispose();
         }
 
@@ -294,7 +294,7 @@ private void OnConnectionFailed()
         private void OnDisconnected()
         {
 #if STREAM_DEBUG_ENABLED
-            Logs.Warning($"{LogsPrefix} Websocket Disconnected");
+            Logs.Warning($"{LogsPrefix} Websocket Disconnected. Messages left: {WebsocketClient.QueuedMessagesCount}");
 #endif
             ConnectionState = ConnectionState.Disconnected;
         }

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

@@ -138,7 +138,6 @@ protected override void ProcessMessages()
                 var sfuEvent = SfuEvent.Parser.ParseFrom(msg);
 
 #if STREAM_DEBUG_ENABLED
-
                 DebugLogEvent(sfuEvent);
 #endif
 

Packages/StreamVideo/Runtime/Libs/Websockets/IWebsocketClient.cs

@@ -12,6 +12,7 @@ public interface IWebsocketClient : IDisposable
         event Action Connected;
         event Action Disconnected;
         event Action ConnectionFailed;
+        int QueuedMessagesCount { get; }
 
         bool TryDequeueMessage(out byte[] message);
 

Packages/StreamVideo/Runtime/Libs/Websockets/NativeWebSocketWrapper.cs

@@ -19,6 +19,8 @@ public class NativeWebSocketWrapper : IWebsocketClient
         public event Action Connected;
         public event Action Disconnected;
         public event Action ConnectionFailed;
+
+        public int QueuedMessagesCount => _messages.Count;
 
         public NativeWebSocketWrapper(ILogs logs, bool isDebugMode)
         {
@@ -82,6 +84,7 @@ public async Task DisconnectAsync()
 
             UnsubscribeFromEvents();
             await _webSocket.Close();
+            _messages.Clear();
             Disconnected?.Invoke();
         }
 

Packages/StreamVideo/Runtime/Libs/Websockets/WebsocketClient.cs

@@ -25,6 +25,8 @@ public class WebsocketClient : IWebsocketClient
 
         public WebSocketState State => _internalClient?.State ?? WebSocketState.None;
 
+        public int QueuedMessagesCount => _receiveQueue.Count;
+
         /// <param name="isDebugMode">Additional logs will be printed</param>
         public WebsocketClient(ILogs logs, Encoding encoding = default, bool isDebugMode = false)
         {
@@ -90,7 +92,7 @@ public void Send(string message)
 
             _sendQueue.Add(messageSegment);
         }
-        
+
         public void Send(byte[] message)
         {
             var messageSegment = new ArraySegment<byte>(message);
@@ -125,6 +127,11 @@ public async Task DisconnectAsync(WebSocketCloseStatus closeStatus, string close
         {
             LogInfoIfDebugMode("Disconnect");
             await TryDisposeResourcesAsync(closeStatus, closeMessage);
+
+            _receiveQueue.Clear();
+            while (_sendQueue.TryTake(out _))
+            {
+            }
 
             Disconnected?.Invoke();
         }
@@ -135,7 +142,7 @@ public void Dispose()
             DisconnectAsync(WebSocketCloseStatus.NormalClosure, "WebSocket client is disposed")
                 .ContinueWith(_ => LogExceptionIfDebugMode(_.Exception), TaskContinuationOptions.OnlyOnFaulted);
         }
-        
+
         private const int UpdatesPerSecond = 20;
         private const int UpdatePeriod = 1000 / UpdatesPerSecond;
         private const int UpdatePeriodOffset = UpdatePeriod / 2;
@@ -255,10 +262,15 @@ private async Task TryDisposeResourcesAsync(WebSocketCloseStatus closeStatus, st
         {
             try
             {
-                _backgroundReceiveTimer?.Dispose();
-                _backgroundReceiveTimer = null;
-                _backgroundSendTimer?.Dispose();
-                _backgroundSendTimer = null;
+                if (_backgroundReceiveTimer != null)
+                {
+                    await _backgroundReceiveTimer.DisposeAsync();
+                }
+
+                if (_backgroundSendTimer != null)
+                {
+                    await _backgroundSendTimer.DisposeAsync();
+                }
             }
             catch (Exception e)
             {
@@ -297,8 +309,12 @@ private async Task TryDisposeResourcesAsync(WebSocketCloseStatus closeStatus, st
             }
             finally
             {
-                _internalClient.Dispose();
-                _internalClient = null;
+                //StreamTOdo: this fixes possible null ref if Dispose was called multiple times but perhaps this logic should not be called multiple times
+                if (_internalClient == null)
+                {
+                    _internalClient.Dispose();
+                    _internalClient = null;
+                }
             }
         }