feat(ITcpSocketClient): add AutoConnect parameter (#6372)

* feat: 增加自动重连配置

* refactor: 使用取消令牌

* refactor: 更改参数为 IsAutoReconnect

* test: 更新单元测试

* feat: 断线时也需要触发 ReceivedCallBack 回调

* refactor: 提高性能避免线程切换

* feat: 连接方法增加保护

* refactor: 增加 Reconnect 方法

* test: 增加重连单元测试

* feat: 增加发送接收时断线重连机制

* test: 增加发送接收时断开重连单元测试

Author: ArgoZhang

src/BootstrapBlazor/Services/TcpSocket/DefaultSocketClientProvider.cs

@@ -33,7 +33,7 @@ class DefaultSocketClientProvider : ISocketClientProvider
     public async ValueTask<bool> ConnectAsync(IPEndPoint endPoint, CancellationToken token = default)
     {
         _client = new TcpClient(LocalEndPoint);
-        await _client.ConnectAsync(endPoint, token);
+        await _client.ConnectAsync(endPoint, token).ConfigureAwait(false);
         if (_client.Connected)
         {
             if (_client.Client.LocalEndPoint is IPEndPoint localEndPoint)
@@ -53,7 +53,7 @@ public async ValueTask<bool> SendAsync(ReadOnlyMemory<byte> data, CancellationTo
         if (_client != null)
         {
             var stream = _client.GetStream();
-            await stream.WriteAsync(data, token);
+            await stream.WriteAsync(data, token).ConfigureAwait(false);
             ret = true;
         }
         return ret;
@@ -68,7 +68,7 @@ public async ValueTask<int> ReceiveAsync(Memory<byte> buffer, CancellationToken
         if (_client is { Connected: true })
         {
             var stream = _client.GetStream();
-            len = await stream.ReadAsync(buffer, token);
+            len = await stream.ReadAsync(buffer, token).ConfigureAwait(false);
         }
         return len;
     }

src/BootstrapBlazor/Services/TcpSocket/SocketClientOptions.cs

@@ -55,4 +55,14 @@ public class SocketClientOptions
     /// Gets or sets a value indicating whether logging is enabled. Default value is false.
     /// </summary>
     public bool EnableLog { get; set; }
+    
+    /// <summary>    
+    /// Gets or sets a value indicating whether the system should automatically attempt to reconnect  after a connection is lost. Default value is false.
+    /// </summary>
+    public bool IsAutoReconnect { get; set; }
+
+    /// <summary>
+    /// Gets or sets the interval, in milliseconds, between reconnection attempts. Default value is 5000.
+    /// </summary>
+    public int ReconnectInterval { get; set; } = 5000;
 }

src/BootstrapBlazor/Services/TcpSocket/TcpSocketClientBase.cs

@@ -63,6 +63,7 @@ public abstract class TcpSocketClientBase(SocketClientOptions options) : ITcpSoc
     private IPEndPoint? _remoteEndPoint;
     private IPEndPoint? _localEndPoint;
     private CancellationTokenSource? _receiveCancellationTokenSource;
+    private CancellationTokenSource? _autoConnectTokenSource;
 
     /// <summary>
     /// <inheritdoc/>
@@ -72,39 +73,21 @@ public abstract class TcpSocketClientBase(SocketClientOptions options) : ITcpSoc
     /// <returns></returns>
     public async ValueTask<bool> ConnectAsync(IPEndPoint endPoint, CancellationToken token = default)
     {
+        if (IsConnected)
+        {
+            return true;
+        }
+
         var ret = false;
         SocketClientProvider = ServiceProvider?.GetRequiredService<ISocketClientProvider>()
             ?? throw new InvalidOperationException("SocketClientProvider is not registered in the service provider.");
 
         try
         {
-            // 释放资源
-            await CloseAsync();
-
-            // 创建新的 TcpClient 实例
-            SocketClientProvider.LocalEndPoint = Options.LocalEndPoint;
-
-            _localEndPoint = Options.LocalEndPoint;
-            _remoteEndPoint = null;
-
-            var connectionToken = token;
-            if (Options.ConnectTimeout > 0)
+            ret = await ConnectCoreAsync(SocketClientProvider, endPoint, token);
+            if (ret == false)
             {
-                // 设置连接超时时间
-                var connectTokenSource = new CancellationTokenSource(options.ConnectTimeout);
-                connectionToken = CancellationTokenSource.CreateLinkedTokenSource(token, connectTokenSource.Token).Token;
-            }
-            ret = await SocketClientProvider.ConnectAsync(endPoint, connectionToken);
-
-            if (ret)
-            {
-                _localEndPoint = SocketClientProvider.LocalEndPoint;
-                _remoteEndPoint = endPoint;
-
-                if (options.IsAutoReceive)
-                {
-                    _ = Task.Run(AutoReceiveAsync, token);
-                }
+                Reconnect();
             }
         }
         catch (OperationCanceledException ex)
@@ -116,15 +99,75 @@ public async ValueTask<bool> ConnectAsync(IPEndPoint endPoint, CancellationToken
             else
             {
                 Log(LogLevel.Warning, ex, $"TCP Socket connect operation timed out from {LocalEndPoint} to {endPoint}");
+                Reconnect();
             }
         }
         catch (Exception ex)
         {
             Log(LogLevel.Error, ex, $"TCP Socket connection failed from {LocalEndPoint} to {endPoint}");
+            Reconnect();
+        }
+        return ret;
+    }
+
+    private void Reconnect()
+    {
+        if (options.IsAutoReconnect)
+        {
+            Task.Run(async () =>
+            {
+                try
+                {
+                    _autoConnectTokenSource ??= new();
+                    await Task.Delay(options.ReconnectInterval, _autoConnectTokenSource.Token).ConfigureAwait(false);
+                    HandleDisconnection();
+                }
+                catch { }
+            }, CancellationToken.None).ConfigureAwait(false);
+        }
+    }
+
+    private async ValueTask<bool> ConnectCoreAsync(ISocketClientProvider provider, IPEndPoint endPoint, CancellationToken token)
+    {
+        // 释放资源
+        await CloseAsync();
+
+        // 创建新的 TcpClient 实例
+        provider.LocalEndPoint = Options.LocalEndPoint;
+
+        _localEndPoint = Options.LocalEndPoint;
+        _remoteEndPoint = endPoint;
+
+        var connectionToken = token;
+        if (Options.ConnectTimeout > 0)
+        {
+            // 设置连接超时时间
+            var connectTokenSource = new CancellationTokenSource(options.ConnectTimeout);
+            connectionToken = CancellationTokenSource.CreateLinkedTokenSource(token, connectTokenSource.Token).Token;
+        }
+        var ret = await provider.ConnectAsync(endPoint, connectionToken);
+
+        if (ret)
+        {
+            _localEndPoint = provider.LocalEndPoint;
+
+            if (options.IsAutoReceive)
+            {
+                _ = Task.Run(AutoReceiveAsync, CancellationToken.None).ConfigureAwait(false);
+            }
         }
         return ret;
     }
 
+    private void HandleDisconnection()
+    {
+        if (options.IsAutoReconnect && _remoteEndPoint != null)
+        {
+            _autoConnectTokenSource ??= new();
+            _ = Task.Run(() => ConnectAsync(_remoteEndPoint, _autoConnectTokenSource.Token)).ConfigureAwait(false);
+        }
+    }
+
     /// <summary>
     /// <inheritdoc/>
     /// </summary>
@@ -149,6 +192,10 @@ public virtual async ValueTask<bool> SendAsync(ReadOnlyMemory<byte> data, Cancel
                 sendToken = CancellationTokenSource.CreateLinkedTokenSource(token, sendTokenSource.Token).Token;
             }
             ret = await SocketClientProvider.SendAsync(data, sendToken);
+            if (ret == false)
+            {
+                HandleDisconnection();
+            }
         }
         catch (OperationCanceledException ex)
         {
@@ -164,6 +211,8 @@ public virtual async ValueTask<bool> SendAsync(ReadOnlyMemory<byte> data, Cancel
         catch (Exception ex)
         {
             Log(LogLevel.Error, ex, $"TCP Socket send failed from {_localEndPoint} to {_remoteEndPoint}");
+
+            HandleDisconnection();
         }
 
         if (options.EnableLog)
@@ -193,14 +242,18 @@ public virtual async ValueTask<Memory<byte>> ReceiveAsync(CancellationToken toke
         using var block = MemoryPool<byte>.Shared.Rent(options.ReceiveBufferSize);
         var buffer = block.Memory;
         var len = await ReceiveCoreAsync(SocketClientProvider, buffer, token);
+        if (len == 0)
+        {
+            HandleDisconnection();
+        }
         return buffer[..len];
     }
 
     private async ValueTask AutoReceiveAsync()
     {
         // 自动接收方法
         _receiveCancellationTokenSource ??= new();
-        while (true)
+        while (_receiveCancellationTokenSource is { IsCancellationRequested: false })
         {
             if (SocketClientProvider is not { IsConnected: true })
             {
@@ -216,6 +269,8 @@ private async ValueTask AutoReceiveAsync()
                 break;
             }
         }
+
+        HandleDisconnection();
     }
 
     private async ValueTask<int> ReceiveCoreAsync(ISocketClientProvider client, Memory<byte> buffer, CancellationToken token)
@@ -236,16 +291,19 @@ private async ValueTask<int> ReceiveCoreAsync(ISocketClientProvider client, Memo
             {
                 // 远端主机关闭链路
                 Log(LogLevel.Information, null, $"TCP Socket {_localEndPoint} received 0 data closed by {_remoteEndPoint}");
+                buffer = Memory<byte>.Empty;
+
+                HandleDisconnection();
             }
             else
             {
                 buffer = buffer[..len];
+            }
 
-                if (ReceivedCallBack != null)
-                {
-                    // 如果订阅回调则触发回调
-                    await ReceivedCallBack(buffer);
-                }
+            if (ReceivedCallBack != null)
+            {
+                // 如果订阅回调则触发回调
+                await ReceivedCallBack(buffer);
             }
         }
         catch (OperationCanceledException ex)
@@ -262,6 +320,8 @@ private async ValueTask<int> ReceiveCoreAsync(ISocketClientProvider client, Memo
         catch (Exception ex)
         {
             Log(LogLevel.Error, ex, $"TCP Socket receive failed from {_localEndPoint} to {_remoteEndPoint}");
+
+            HandleDisconnection();
         }
 
         if (options.EnableLog)
@@ -283,9 +343,19 @@ protected virtual void Log(LogLevel logLevel, Exception? ex, string? message)
     /// <summary>
     /// <inheritdoc/>
     /// </summary>
-    public virtual ValueTask CloseAsync()
+    public virtual async ValueTask CloseAsync()
     {
-        return DisposeAsync(true);
+        // 取消接收数据的任务
+        if (_receiveCancellationTokenSource != null)
+        {
+            _receiveCancellationTokenSource.Cancel();
+            _receiveCancellationTokenSource.Dispose();
+            _receiveCancellationTokenSource = null;
+        }
+        if (SocketClientProvider != null)
+        {
+            await SocketClientProvider.CloseAsync();
+        }
     }
 
     /// <summary>
@@ -300,18 +370,15 @@ protected virtual async ValueTask DisposeAsync(bool disposing)
     {
         if (disposing)
         {
-            // 取消接收数据的任务
-            if (_receiveCancellationTokenSource != null)
+            // 取消重连任务
+            if (_autoConnectTokenSource != null)
             {
-                _receiveCancellationTokenSource.Cancel();
-                _receiveCancellationTokenSource.Dispose();
-                _receiveCancellationTokenSource = null;
+                _autoConnectTokenSource.Cancel();
+                _autoConnectTokenSource.Dispose();
+                _autoConnectTokenSource = null;
             }
 
-            if (SocketClientProvider != null)
-            {
-                await SocketClientProvider.CloseAsync();
-            }
+            await CloseAsync();
         }
     }