From 48f45fc2826f22065204bfe3aa5767b49f2a35ac Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20Sierpi=C5=84ski?= <33436839+sierpinskid@users.noreply.github.com> Date: Tue, 18 Aug 2026 15:35:56 +0200 Subject: [PATCH 1/3] Fix setting ConnectionState Disconnected only from the main thread. Until now it could have been called from the background thread which would break things if the end user would react to this event and call Unity API --- .../StreamChatLowLevelClient.cs | 97 ++++++++++++++++++- .../Libs/Websockets/WebsocketClient.cs | 39 +++++--- .../StreamChatLowLevelClientTests.cs | 78 +++++++++++++++ 3 files changed, 198 insertions(+), 16 deletions(-) diff --git a/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs b/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs index 43f37098..b00e0df7 100644 --- a/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs +++ b/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs @@ -212,12 +212,12 @@ private set _logs.Warning($"Connection state changed from: {previous} to: {value}"); #endif - ConnectionStateChanged?.Invoke(previous, _connectionState); + RaiseConnectionStateChanged(previous, _connectionState); if (value == ConnectionState.Disconnected) { _disconnectionLastEventReceivedAt = _lastEventReceivedAt; - Disconnected?.Invoke(); + RaiseDisconnected(); } } } @@ -297,6 +297,8 @@ public StreamChatLowLevelClient(AuthCredentials authCredentials, IWebsocketClien IHttpClient httpClient, ISerializer serializer, ITimeService timeService, INetworkMonitor networkMonitor, IApplicationInfo applicationInfo, ILogs logs, IStreamClientConfig config) { + _mainThreadId = Thread.CurrentThread.ManagedThreadId; + _authCredentials = authCredentials; _websocketClient = websocketClient ?? throw new ArgumentNullException(nameof(websocketClient)); _httpClient = httpClient ?? throw new ArgumentNullException(nameof(httpClient)); @@ -410,6 +412,7 @@ public void Update(float deltaTime) #endif TryHandleWebsocketsConnectionFailed(); + TryHandleWebsocketDisconnected(); TryToReconnect(); UpdateHealthCheck(); @@ -577,6 +580,12 @@ internal async Task ConnectUserAsync(string apiKey, string u new Dictionary>(); private readonly object _websocketConnectionFailedFlagLock = new object(); + private readonly object _websocketDisconnectedFlagLock = new object(); + + /// + /// Thread that constructed this client. Every write must happen on it + /// + private readonly int _mainThreadId; private TaskCompletionSource _connectUserTaskSource; private CancellationToken _connectUserCancellationToken; @@ -592,6 +601,7 @@ internal async Task ConnectUserAsync(string apiKey, string u private bool _updateCallReceived; private bool _websocketConnectionFailed; + private bool _websocketDisconnected; private ITokenProvider _tokenProvider; /// @@ -646,14 +656,97 @@ private void TryCancelWaitingForUserConnection() } } + /// + /// This event can be called by a background thread and we must propagate it on the main thread + /// Otherwise any call to Unity API would result in Exception. Unity API can only be called from the main thread + /// private void OnWebsocketDisconnected() { #if STREAM_DEBUG_ENABLED _logs.Warning("Websocket Disconnected"); #endif + + // Applied inline when we're already on the main thread so that awaiting DisconnectAsync keeps + // guaranteeing that ConnectionState reads Disconnected once the task completes + if (Thread.CurrentThread.ManagedThreadId == _mainThreadId) + { + ConnectionState = ConnectionState.Disconnected; + return; + } + + lock (_websocketDisconnectedFlagLock) + { + _websocketDisconnected = true; + } + } + + private void TryHandleWebsocketDisconnected() + { + lock (_websocketDisconnectedFlagLock) + { + if (!_websocketDisconnected) + { + return; + } + + _websocketDisconnected = false; + } + + if (ConnectionState == ConnectionState.Closing) + { + return; + } + ConnectionState = ConnectionState.Disconnected; } + /// + /// Subscribers are invoked one by one so that a throwing handler cannot stop the remaining ones - + /// including the internal reconnect scheduling - from observing the transition + /// + private void RaiseConnectionStateChanged(ConnectionState previous, ConnectionState current) + { + var handler = ConnectionStateChanged; + if (handler == null) + { + return; + } + + foreach (var subscriber in handler.GetInvocationList()) + { + try + { + ((ConnectionStateChangeHandler)subscriber)(previous, current); + } + catch (Exception e) + { + _logs.Exception(e); + } + } + } + + /// + private void RaiseDisconnected() + { + var handler = Disconnected; + if (handler == null) + { + return; + } + + foreach (var subscriber in handler.GetInvocationList()) + { + try + { + ((Action)subscriber)(); + } + catch (Exception e) + { + _logs.Exception(e); + } + } + } + /// /// This event can be called by a background thread and we must propagate it on the main thread /// Otherwise any call to Unity API would result in Exception. Unity API can only be called from the main thread diff --git a/Assets/Plugins/StreamChat/Libs/Websockets/WebsocketClient.cs b/Assets/Plugins/StreamChat/Libs/Websockets/WebsocketClient.cs index 0b30ab8d..b424931e 100644 --- a/Assets/Plugins/StreamChat/Libs/Websockets/WebsocketClient.cs +++ b/Assets/Plugins/StreamChat/Libs/Websockets/WebsocketClient.cs @@ -124,7 +124,9 @@ public void Update() } #endif - var disconnect = false; + var serverClosedConnection = Interlocked.Exchange(ref _serverClosedConnectionFlag, 0) == 1; + + var disconnect = serverClosedConnection; while (_threadWebsocketExceptionsLog.TryDequeue(out var webSocketException)) { LogExceptionIfDebugMode(webSocketException); @@ -134,7 +136,15 @@ public void Update() if (disconnect) { - DisconnectAsync(WebSocketCloseStatus.ProtocolError, "WebSocket thrown an exception") + var closeStatus = serverClosedConnection + ? WebSocketCloseStatus.InternalServerError + : WebSocketCloseStatus.ProtocolError; + + var closeMessage = serverClosedConnection + ? "Server closed the connection" + : "WebSocket thrown an exception"; + + DisconnectAsync(closeStatus, closeMessage) .ContinueWith(_ => LogExceptionIfDebugMode(_.Exception), TaskContinuationOptions.OnlyOnFaulted); return; } @@ -200,6 +210,10 @@ public void Dispose() private WebSocketState _lastState; + // Set from the receive thread, consumed by Update. Int rather than bool so it can be read and + // reset in a single interlocked operation. + private int _serverClosedConnectionFlag; + private async void SendMessagesCallback(object state) { if (!IsConnected || _connectionCts == null || _connectionCts.IsCancellationRequested) @@ -283,6 +297,9 @@ private async void ReceiveMessagesCallback(object state) private async Task TryCloseAndDisposeAsync(WebSocketCloseStatus closeStatus, string closeMessage) { + // A close frame that arrived while we were tearing down must not disconnect the next connection + Interlocked.Exchange(ref _serverClosedConnectionFlag, 0); + try { #if UNITY_2021_2_OR_NEWER @@ -419,10 +436,12 @@ await TryCloseAndDisposeAsync(WebSocketCloseStatus.ProtocolError, ConnectionFailed?.Invoke(); } - // Called from a background thread - private void OnReceivedCloseMessage() - => DisconnectAsync(WebSocketCloseStatus.InternalServerError, "Server closed the connection") - .ContinueWith(t => LogThreadExceptionIfDebugMode(t.Exception), TaskContinuationOptions.OnlyOnFaulted); + /// + /// Called from a background thread. The disconnect is deferred to so that the + /// event is always raised from the main thread, like every other path that + /// raises it. Subscribers react to it with Unity API calls, which throw off the main thread. + /// + private void OnReceivedCloseMessage() => Interlocked.Exchange(ref _serverClosedConnectionFlag, 1); private async Task TryReceiveSingleMessageAsync() { @@ -478,14 +497,6 @@ private void LogExceptionIfDebugMode(Exception exception) } } - private void LogThreadExceptionIfDebugMode(Exception exception) - { - if (_isDebugMode) - { - _threadExceptionsLog.Enqueue(exception); - } - } - private void LogInfoIfDebugMode(string info) { if (_isDebugMode) diff --git a/Assets/Plugins/StreamChat/Tests/LowLevelClient/StreamChatLowLevelClientTests.cs b/Assets/Plugins/StreamChat/Tests/LowLevelClient/StreamChatLowLevelClientTests.cs index ef6f2420..75c56ec4 100644 --- a/Assets/Plugins/StreamChat/Tests/LowLevelClient/StreamChatLowLevelClientTests.cs +++ b/Assets/Plugins/StreamChat/Tests/LowLevelClient/StreamChatLowLevelClientTests.cs @@ -3,6 +3,7 @@ using System.Collections.Generic; using System.Linq; using System.Net.WebSockets; +using System.Threading; using System.Threading.Tasks; using NSubstitute; using NUnit.Framework; @@ -292,6 +293,60 @@ public void when_stream_client_health_check_timeout_detected_expect_client_disco Assert.IsFalse(client.ConnectionState == ConnectionState.Connected); } + [Test] + public void when_websocket_disconnected_raised_from_background_thread_expect_it_handled_on_main_thread_in_update() + { + var client = CreateConnectedClient(); + + var disconnectedCount = 0; + var disconnectedThreadId = 0; + client.Disconnected += () => + { + disconnectedCount++; + disconnectedThreadId = Thread.CurrentThread.ManagedThreadId; + }; + + Task.Run(() => _mockWebsocketClient.Disconnected += Raise.Event()).Wait(); + + Assert.AreEqual(0, disconnectedCount, + "Disconnected must not be raised from the thread that closed the websocket"); + Assert.IsTrue(client.ConnectionState == ConnectionState.Connected); + + client.Update(deltaTime: 0.2f); + + Assert.AreEqual(1, disconnectedCount); + Assert.AreEqual(Thread.CurrentThread.ManagedThreadId, disconnectedThreadId); + } + + [Test] + public void when_websocket_disconnected_raised_from_main_thread_expect_it_handled_immediately() + { + var client = CreateConnectedClient(); + + var disconnectedCount = 0; + client.Disconnected += () => disconnectedCount++; + + _mockWebsocketClient.Disconnected += Raise.Event(); + + // Awaiting DisconnectAsync must keep guaranteeing that the client is no longer connected + Assert.AreEqual(1, disconnectedCount); + Assert.IsFalse(client.ConnectionState == ConnectionState.Connected); + } + + [Test] + public void when_connection_state_changed_subscriber_throws_expect_remaining_subscribers_notified() + { + var client = CreateConnectedClient(Substitute.For()); + + var lastStateSeenByLateSubscriber = ConnectionState.Connected; + client.ConnectionStateChanged += (previous, current) => throw new Exception("Subscriber failed"); + client.ConnectionStateChanged += (previous, current) => lastStateSeenByLateSubscriber = current; + + _mockWebsocketClient.Disconnected += Raise.Event(); + + Assert.AreNotEqual(ConnectionState.Connected, lastStateSeenByLateSubscriber); + } + private readonly List _resourcesToDispose = new List(); private IStreamChatLowLevelClient _lowLevelClient; @@ -305,6 +360,29 @@ public void when_stream_client_health_check_timeout_detected_expect_client_disco private INetworkMonitor _mockNetworkMonitor; private IHttpClient _mockHttpClient; private IStreamClientConfig _mockStreamClientConfig; + + private StreamChatLowLevelClient CreateConnectedClient(ILogs logs = null) + { + var client = new StreamChatLowLevelClient(_authCredentials, _mockWebsocketClient, _mockHttpClient, + new NewtonsoftJsonSerializer(), _mockTimeService, _mockNetworkMonitor, _mockApplicationInfo, + logs ?? _mockLogs, _mockStreamClientConfig); + _resourcesToDispose.Add(client); + + _mockWebsocketClient.ConnectAsync(Arg.Any()).Returns(Task.CompletedTask); + + _mockWebsocketClient.TryDequeueMessage(out Arg.Any()).Returns(arg => + { + arg[0] = "{\"connection_id\":\"fakeId\", \"type\":\"health.check\"}"; + return true; + }, arg => false); + + client.Connect(); + client.Update(deltaTime: 0.2f); + + Assert.IsTrue(client.ConnectionState == ConnectionState.Connected); + + return client; + } } } #endif \ No newline at end of file From db050877ef91123f5213879df6180ddaf94b027d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20Sierpi=C5=84ski?= <33436839+sierpinskid@users.noreply.github.com> Date: Tue, 18 Aug 2026 15:38:26 +0200 Subject: [PATCH 2/3] Fix compiler errors --- .../Core/LowLevelClient/StreamChatLowLevelClient.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs b/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs index b00e0df7..49577dd0 100644 --- a/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs +++ b/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs @@ -297,7 +297,7 @@ public StreamChatLowLevelClient(AuthCredentials authCredentials, IWebsocketClien IHttpClient httpClient, ISerializer serializer, ITimeService timeService, INetworkMonitor networkMonitor, IApplicationInfo applicationInfo, ILogs logs, IStreamClientConfig config) { - _mainThreadId = Thread.CurrentThread.ManagedThreadId; + _mainThreadId = System.Threading.Thread.CurrentThread.ManagedThreadId; _authCredentials = authCredentials; _websocketClient = websocketClient ?? throw new ArgumentNullException(nameof(websocketClient)); @@ -668,7 +668,7 @@ private void OnWebsocketDisconnected() // Applied inline when we're already on the main thread so that awaiting DisconnectAsync keeps // guaranteeing that ConnectionState reads Disconnected once the task completes - if (Thread.CurrentThread.ManagedThreadId == _mainThreadId) + if (System.Threading.Thread.CurrentThread.ManagedThreadId == _mainThreadId) { ConnectionState = ConnectionState.Disconnected; return; From 60c4cfc296d7755d20f488a30df4ae99f3f9ae01 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20Sierpi=C5=84ski?= <33436839+sierpinskid@users.noreply.github.com> Date: Tue, 18 Aug 2026 15:48:32 +0200 Subject: [PATCH 3/3] cleanup --- .../Core/LowLevelClient/StreamChatLowLevelClient.cs | 9 ++++----- .../StreamChat/Libs/Websockets/WebsocketClient.cs | 3 +-- 2 files changed, 5 insertions(+), 7 deletions(-) diff --git a/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs b/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs index 49577dd0..2a491b7a 100644 --- a/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs +++ b/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs @@ -28,6 +28,7 @@ using StreamChat.Core.LowLevelClient.Requests; using System.Linq; using StreamChat.Core.Helpers; +using Thread = System.Threading.Thread; #if STREAM_TESTS_ENABLED || STREAM_RUNTIME_TESTS_ENABLED using System.Runtime.CompilerServices; @@ -297,7 +298,7 @@ public StreamChatLowLevelClient(AuthCredentials authCredentials, IWebsocketClien IHttpClient httpClient, ISerializer serializer, ITimeService timeService, INetworkMonitor networkMonitor, IApplicationInfo applicationInfo, ILogs logs, IStreamClientConfig config) { - _mainThreadId = System.Threading.Thread.CurrentThread.ManagedThreadId; + _mainThreadId = Thread.CurrentThread.ManagedThreadId; _authCredentials = authCredentials; _websocketClient = websocketClient ?? throw new ArgumentNullException(nameof(websocketClient)); @@ -583,7 +584,7 @@ internal async Task ConnectUserAsync(string apiKey, string u private readonly object _websocketDisconnectedFlagLock = new object(); /// - /// Thread that constructed this client. Every write must happen on it + /// Every write must happen on this thread /// private readonly int _mainThreadId; @@ -666,9 +667,7 @@ private void OnWebsocketDisconnected() _logs.Warning("Websocket Disconnected"); #endif - // Applied inline when we're already on the main thread so that awaiting DisconnectAsync keeps - // guaranteeing that ConnectionState reads Disconnected once the task completes - if (System.Threading.Thread.CurrentThread.ManagedThreadId == _mainThreadId) + if (Thread.CurrentThread.ManagedThreadId == _mainThreadId) { ConnectionState = ConnectionState.Disconnected; return; diff --git a/Assets/Plugins/StreamChat/Libs/Websockets/WebsocketClient.cs b/Assets/Plugins/StreamChat/Libs/Websockets/WebsocketClient.cs index b424931e..50ef8536 100644 --- a/Assets/Plugins/StreamChat/Libs/Websockets/WebsocketClient.cs +++ b/Assets/Plugins/StreamChat/Libs/Websockets/WebsocketClient.cs @@ -210,8 +210,7 @@ public void Dispose() private WebSocketState _lastState; - // Set from the receive thread, consumed by Update. Int rather than bool so it can be read and - // reset in a single interlocked operation. + // Int rather than bool so that Update can read and reset it in a single interlocked operation private int _serverClosedConnectionFlag; private async void SendMessagesCallback(object state)