Fix bugs in Reconnect logic (#157)

* Improve logging and diagnostics capabilities

* Expose new properties on IWebsocketClient (TextSenderRunning, BinarySenderRunning, IsInsideLock)
* Use Debug log level when receiving a close message and creating new send threads
* Write a Debug log message when the sending threads are shutting down

* Shorten test run times and make tests more debuggable

* Fix bugs in Reconnect logic which resulted in duplicate disconnection events

* Use ReconnectionType.ByServer when reconnecting in response to receiving a close message from the server
* Actually start the client a 2nd time in the test for that

* Implement a timeout for creating a new websocket connection

* Fixes an issue where the connection factory occasionally just hangs forever
* Expose a new property for configuring how long to wait before timing out (defaulted to 2 seconds)
* Ignore duplicate calls to Dispose
* Use try methods for closing channel writers in Dispose
* Update IsRunning flag immediately within exception handler in StartClient
* Skip reconnecting on error after waiting if the client is already reconnecting by user request

Author: shaynevanasperen

src/Websocket.Client/IWebsocketClient.cs

@@ -116,6 +116,12 @@ public interface IWebsocketClient : IDisposable
         /// </summary>
         IObservable<DisconnectionInfo> DisconnectionHappened { get; }
 
+        /// <summary>
+        /// Time range for how long to wait while connecting a new client.
+        /// Default: 2 seconds
+        /// </summary>
+        TimeSpan ConnectTimeout { get; set; }
+
         /// <summary>
         /// Time range for how long to wait before reconnecting if no message comes from server.
         /// Set null to disable this feature. 
@@ -158,6 +164,21 @@ public interface IWebsocketClient : IDisposable
         /// </summary>
         bool IsRunning { get; }
 
+        /// <summary>
+        /// Returns whether text message sender is running.
+        /// </summary>
+        bool TextSenderRunning { get; }
+
+        /// <summary>
+        /// Returns whether the binary message sender is running.
+        /// </summary>
+        bool BinarySenderRunning { get; }
+
+        /// <summary>
+        /// Indicates whether any thread has entered the lock.
+        /// </summary>
+        bool IsInsideLock { get; }
+
         /// <summary>
         /// Enable or disable text message conversion from binary to string (via 'MessageEncoding' property).
         /// Default: true

src/Websocket.Client/Threading/WebsocketAsyncLock.cs

@@ -31,6 +31,11 @@ public WebsocketAsyncLock()
             _releaserTask = Task.FromResult(_releaser);
         }
 
+        /// <summary>
+        /// True if the lock is currently taken
+        /// </summary>
+        public bool IsLocked => _semaphore.CurrentCount == 0;
+
         /// <summary>
         /// Use inside 'using' block
         /// </summary>

src/Websocket.Client/WebsocketClient.Reconnecting.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Net.WebSockets;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
@@ -67,7 +68,7 @@ private async Task Reconnect(ReconnectionType type, bool failFast, Exception? ca
 
             var disType = TranslateTypeToDisconnection(type);
             var disInfo = DisconnectionInfo.Create(disType, _client, causedException);
-            if (type != ReconnectionType.Error)
+            if (type != ReconnectionType.Error && _client?.State != WebSocketState.CloseReceived && _client?.State != WebSocketState.Closed)
             {
                 _disconnectedSubject.OnNext(disInfo);
                 if (disInfo.CancelReconnection)
@@ -88,7 +89,7 @@ private async Task Reconnect(ReconnectionType type, bool failFast, Exception? ca
             }
             _client?.Dispose();
 
-            if (!IsReconnectionEnabled || disInfo.CancelReconnection)
+            if (type != ReconnectionType.Error && (!IsReconnectionEnabled || disInfo.CancelReconnection))
             {
                 // reconnection disabled, do nothing
                 IsStarted = false;

src/Websocket.Client/WebsocketClient.Sending.cs

@@ -24,6 +24,12 @@ public partial class WebsocketClient
 
         private static readonly byte[] _emptyArray = { };
 
+        /// <inheritdoc />
+        public bool TextSenderRunning { get; private set; }
+
+        /// <inheritdoc />
+        public bool BinarySenderRunning { get; private set; }
+
         /// <summary>
         /// Send text message to the websocket channel. 
         /// It inserts the message to the queue and actual sending is done on another thread
@@ -159,6 +165,7 @@ public void StreamFakeMessage(ResponseMessage message)
 
         private async Task SendTextFromQueue()
         {
+            TextSenderRunning = true;
             try
             {
                 while (await _messagesTextToSendQueue.Reader.WaitToReadAsync())
@@ -176,30 +183,35 @@ private async Task SendTextFromQueue()
                     }
                 }
             }
-            catch (TaskCanceledException)
+            catch (TaskCanceledException e)
             {
                 // task was canceled, ignore
+                _logger.LogDebug(e, L("Sending text thread failed, error: {error}. Shutting down."), Name, e.Message);
             }
-            catch (OperationCanceledException)
+            catch (OperationCanceledException e)
             {
                 // operation was canceled, ignore
+                _logger.LogDebug(e, L("Sending text thread failed, error: {error}. Shutting down."), Name, e.Message);
             }
             catch (Exception e)
             {
                 if (_cancellationTotal?.IsCancellationRequested == true || _disposing)
                 {
                     // disposing/canceling, do nothing and exit
+                    _logger.LogDebug(e, L("Sending text thread failed, error: {error}. Shutting down."), Name, e.Message);
+                    TextSenderRunning = false;
                     return;
                 }
 
-                _logger.LogTrace(L("Sending text thread failed, error: {error}. Creating a new sending thread."), Name, e.Message);
+                _logger.LogDebug(e, L("Sending text thread failed, error: {error}. Creating a new sending thread."), Name, e.Message);
                 StartBackgroundThreadForSendingText();
             }
-
+            TextSenderRunning = false;
         }
 
         private async Task SendBinaryFromQueue()
         {
+            BinarySenderRunning = true;
             try
             {
                 while (await _messagesBinaryToSendQueue.Reader.WaitToReadAsync())
@@ -217,26 +229,30 @@ private async Task SendBinaryFromQueue()
                     }
                 }
             }
-            catch (TaskCanceledException)
+            catch (TaskCanceledException e)
             {
                 // task was canceled, ignore
+                _logger.LogDebug(e, L("Sending binary thread failed, error: {error}. Shutting down."), Name, e.Message);
             }
-            catch (OperationCanceledException)
+            catch (OperationCanceledException e)
             {
                 // operation was canceled, ignore
+                _logger.LogDebug(e, L("Sending binary thread failed, error: {error}. Shutting down."), Name, e.Message);
             }
             catch (Exception e)
             {
                 if (_cancellationTotal?.IsCancellationRequested == true || _disposing)
                 {
                     // disposing/canceling, do nothing and exit
+                    _logger.LogDebug(e, L("Sending binary thread failed, error: {error}. Shutting down."), Name, e.Message);
+                    BinarySenderRunning = false;
                     return;
                 }
 
-                _logger.LogTrace(L("Sending binary thread failed, error: {error}. Creating a new sending thread."), Name, e.Message);
+                _logger.LogDebug(e, L("Sending binary thread failed, error: {error}. Creating a new sending thread."), Name, e.Message);
                 StartBackgroundThreadForSendingBinary();
             }
-
+            BinarySenderRunning = false;
         }
 
         private void StartBackgroundThreadForSendingText()

src/Websocket.Client/WebsocketClient.cs

@@ -40,6 +40,7 @@ public partial class WebsocketClient : IWebsocketClient
         private bool _isReconnectionEnabled = true;
         private WebSocket? _client;
         private CancellationTokenSource? _cancellation;
+        private CancellationTokenSource? _cancellationConnection;
         private CancellationTokenSource? _cancellationTotal;
 
         private readonly Subject<ResponseMessage> _messageReceivedSubject = new Subject<ResponseMessage>();
@@ -123,6 +124,12 @@ public Uri Url
         /// </summary>
         public IObservable<DisconnectionInfo> DisconnectionHappened => _disconnectedSubject.AsObservable();
 
+        /// <summary>
+        /// Time range for how long to wait while connecting a new client.
+        /// Default: 2 seconds
+        /// </summary>
+        public TimeSpan ConnectTimeout { get; set; } = TimeSpan.FromSeconds(2);
+
         /// <summary>
         /// Time range for how long to wait before reconnecting if no message comes from server.
         /// Set null to disable this feature. 
@@ -189,6 +196,9 @@ public bool IsReconnectionEnabled
         /// Default: true
         /// </summary>
         public bool IsTextMessageConversionEnabled { get; set; } = true;
+        
+        /// <inheritdoc />
+        public bool IsInsideLock => _locker.IsLocked;
 
         /// <summary>
         /// Enable or disable automatic <see cref="MemoryStream.Dispose(bool)"/> of the <see cref="MemoryStream"/> 
@@ -210,19 +220,24 @@ public bool IsReconnectionEnabled
         /// </summary>
         public void Dispose()
         {
+            if (_disposing)
+                return;
+
             _disposing = true;
             _logger.LogDebug(L("Disposing.."), Name);
             try
             {
-                _messagesTextToSendQueue.Writer.Complete();
-                _messagesBinaryToSendQueue.Writer.Complete();
+                _messagesTextToSendQueue.Writer.TryComplete();
+                _messagesBinaryToSendQueue.Writer.TryComplete();
                 _lastChanceTimer?.Dispose();
                 _errorReconnectTimer?.Dispose();
                 _cancellation?.Cancel();
+                _cancellationConnection?.Cancel();
                 _cancellationTotal?.Cancel();
                 _client?.Abort();
                 _client?.Dispose();
                 _cancellation?.Dispose();
+                _cancellationConnection?.Dispose();
                 _cancellationTotal?.Dispose();
                 _messageReceivedSubject.OnCompleted();
                 _reconnectionSubject.OnCompleted();
@@ -402,7 +417,9 @@ private async Task StartClient(Uri uri, CancellationToken token, ReconnectionTyp
 
             try
             {
-                _client = await _connectionFactory(uri, token).ConfigureAwait(false);
+                _cancellationConnection = CancellationTokenSource.CreateLinkedTokenSource(token);
+                _cancellationConnection.CancelAfter(ConnectTimeout);
+                _client = await _connectionFactory(uri, _cancellationConnection.Token).ConfigureAwait(false);
                 _ = Listen(_client, token);
                 IsRunning = true;
                 IsStarted = true;
@@ -412,6 +429,7 @@ private async Task StartClient(Uri uri, CancellationToken token, ReconnectionTyp
             }
             catch (Exception e)
             {
+                IsRunning = _client?.State == WebSocketState.Open;
                 var info = DisconnectionInfo.Create(DisconnectionType.Error, _client, e);
                 _disconnectedSubject.OnNext(info);
 
@@ -447,7 +465,9 @@ private async Task StartClient(Uri uri, CancellationToken token, ReconnectionTyp
 
         private void ReconnectOnError(object? state)
         {
-            // await Task.Delay(timeout, token).ConfigureAwait(false);
+            if (_client != null && ShouldIgnoreReconnection(_client))
+                return;
+
             _ = Reconnect(ReconnectionType.Error, false, state as Exception).ConfigureAwait(false);
         }
 
@@ -491,7 +511,7 @@ private async Task Listen(WebSocket client, CancellationToken token)
                     }
                     else if (result.MessageType == WebSocketMessageType.Close)
                     {
-                        _logger.LogTrace(L("Received close message"), Name);
+                        _logger.LogDebug(L("Received close message"), Name);
 
                         if (!IsStarted || _stopping)
                         {
@@ -512,13 +532,12 @@ private async Task Listen(WebSocket client, CancellationToken token)
                             continue;
                         }
 
-                        await StopInternal(client, WebSocketCloseStatus.NormalClosure, "Closing",
-                            token, false, true);
+                        await StopInternal(client, WebSocketCloseStatus.NormalClosure, "Closing", token, false, true);
 
                         // reconnect if enabled
                         if (IsReconnectionEnabled && !ShouldIgnoreReconnection(client))
                         {
-                            _ = ReconnectSynchronized(ReconnectionType.Lost, false, null);
+                            _ = ReconnectSynchronized(ReconnectionType.ByServer, false, null);
                         }
 
                         return;