feat: 实现发送超时逻辑

Author: ArgoZhang

src/BootstrapBlazor/Services/TcpSocket/DataPackage/DataPackageHandlerBase.cs

@@ -16,33 +16,28 @@ public abstract class DataPackageHandlerBase : IDataPackageHandler
     private Memory<byte> _lastReceiveBuffer = Memory<byte>.Empty;
 
     /// <summary>
-    /// Gets or sets the callback function to handle received data.
+    /// <inheritdoc/>
     /// </summary>
-    /// <remarks>The callback function should be designed to handle the received data efficiently and
-    /// asynchronously.  Ensure that the implementation does not block or perform long-running operations, as this may
-    /// impact performance.</remarks>
     public Func<ReadOnlyMemory<byte>, ValueTask>? ReceivedCallBack { get; set; }
 
     /// <summary>
-    /// Sends the specified data asynchronously to the target destination.
+    /// <inheritdoc/>
     /// </summary>
-    /// <remarks>The method performs an asynchronous operation to send the provided data. The caller must
-    /// ensure  that the data is valid and non-empty. The returned memory block may contain a response or acknowledgment
-    /// depending on the implementation of the target destination.</remarks>
-    /// <param name="data">The data to be sent, represented as a block of memory.</param>
-    /// <returns>A task that represents the asynchronous operation. The task result contains a  <see cref="Memory{T}"/> of <see
-    /// cref="byte"/> representing the response or acknowledgment  received from the target destination.</returns>
-    public virtual ValueTask<ReadOnlyMemory<byte>> SendAsync(ReadOnlyMemory<byte> data)
+    /// <param name="data"></param>
+    /// <param name="token"></param>
+    /// <returns></returns>
+    public virtual ValueTask<ReadOnlyMemory<byte>> SendAsync(ReadOnlyMemory<byte> data, CancellationToken token = default)
     {
         return ValueTask.FromResult(data);
     }
 
     /// <summary>
-    /// Processes the received data asynchronously.
+    /// <inheritdoc/>
     /// </summary>
-    /// <param name="data">A memory buffer containing the data to be processed. The buffer must not be empty.</param>
-    /// <returns>A task that represents the asynchronous operation.</returns>
-    public virtual ValueTask ReceiveAsync(ReadOnlyMemory<byte> data)
+    /// <param name="data"></param>
+    /// <param name="token"></param>
+    /// <returns></returns>
+    public virtual ValueTask ReceiveAsync(ReadOnlyMemory<byte> data, CancellationToken token = default)
     {
         return ValueTask.CompletedTask;
     }

src/BootstrapBlazor/Services/TcpSocket/DataPackage/DelimiterDataPackageHandler.cs

@@ -50,8 +50,9 @@ public DelimiterDataPackageHandler(byte[] delimiter)
     /// <inheritdoc/>
     /// </summary>
     /// <param name="data"></param>
+    /// <param name="token"></param>
     /// <returns></returns>
-    public override async ValueTask ReceiveAsync(ReadOnlyMemory<byte> data)
+    public override async ValueTask ReceiveAsync(ReadOnlyMemory<byte> data, CancellationToken token = default)
     {
         data = ConcatBuffer(data);
 

src/BootstrapBlazor/Services/TcpSocket/DataPackage/FixLengthDataPackageHandler.cs

@@ -22,8 +22,9 @@ public class FixLengthDataPackageHandler(int length) : DataPackageHandlerBase
     /// <inheritdoc/>
     /// </summary>
     /// <param name="data"></param>
+    /// <param name="token"></param>
     /// <returns></returns>
-    public override async ValueTask ReceiveAsync(ReadOnlyMemory<byte> data)
+    public override async ValueTask ReceiveAsync(ReadOnlyMemory<byte> data, CancellationToken token = default)
     {
         while (data.Length > 0)
         {

src/BootstrapBlazor/Services/TcpSocket/DataPackage/IDataPackageHandler.cs

@@ -25,17 +25,19 @@ public interface IDataPackageHandler
     /// ensure  that the data is valid and non-empty. The returned memory block may contain a response or acknowledgment
     /// depending on the implementation of the target destination.</remarks>
     /// <param name="data">The data to be sent, represented as a block of memory.</param>
+    /// <param name="token">An optional <see cref="CancellationToken"/> to observe while waiting for the operation to complete.</param>
     /// <returns>A task that represents the asynchronous operation. The task result contains a  <see cref="Memory{T}"/> of <see
     /// cref="byte"/> representing the response or acknowledgment  received from the target destination.</returns>
-    ValueTask<ReadOnlyMemory<byte>> SendAsync(ReadOnlyMemory<byte> data);
+    ValueTask<ReadOnlyMemory<byte>> SendAsync(ReadOnlyMemory<byte> data, CancellationToken token = default);
 
     /// <summary>
     /// Asynchronously receives data from a source and writes it into the provided memory buffer.
     /// </summary>
     /// <remarks>This method does not guarantee that the entire buffer will be filled. The number of bytes
     /// written depends on the availability of data.</remarks>
     /// <param name="data">The memory buffer to store the received data. The buffer must be writable and have sufficient capacity.</param>
+    /// <param name="token">A cancellation token that can be used to cancel the operation. The default value is <see langword="default"/>.</param>
     /// <returns>A task that represents the asynchronous operation. The task result contains the number of bytes written to the
     /// buffer. Returns 0 if the end of the data stream is reached.</returns>
-    ValueTask ReceiveAsync(ReadOnlyMemory<byte> data);
+    ValueTask ReceiveAsync(ReadOnlyMemory<byte> data, CancellationToken token = default);
 }

src/BootstrapBlazor/Services/TcpSocket/DefaultTcpSocketClient.cs

@@ -102,17 +102,34 @@ public async ValueTask<bool> SendAsync(ReadOnlyMemory<byte> data, CancellationTo
         var ret = false;
         try
         {
+            var stream = _client.GetStream();
+
+            var sendToken = token;
+            if (SendTimeout > 0)
+            {
+                // 设置发送超时时间
+                var sendTokenSource = new CancellationTokenSource(SendTimeout);
+                sendToken = CancellationTokenSource.CreateLinkedTokenSource(token, sendTokenSource.Token).Token;
+            }
+
             if (_dataPackageHandler != null)
             {
-                data = await _dataPackageHandler.SendAsync(data);
+                data = await _dataPackageHandler.SendAsync(data, sendToken);
             }
-            var stream = _client.GetStream();
-            await stream.WriteAsync(data, token);
+
+            await stream.WriteAsync(data, sendToken);
             ret = true;
         }
         catch (OperationCanceledException ex)
         {
-            Logger.LogWarning(ex, "TCP Socket send operation was canceled from {LocalEndPoint} to {RemoteEndPoint}", LocalEndPoint, _remoteEndPoint);
+            if (token.IsCancellationRequested)
+            {
+                Logger.LogWarning(ex, "TCP Socket send operation was canceled from {LocalEndPoint} to {RemoteEndPoint}", LocalEndPoint, _remoteEndPoint);
+            }
+            else
+            {
+                Logger.LogWarning(ex, "TCP Socket send operation timed out from {LocalEndPoint} to {RemoteEndPoint}", LocalEndPoint, _remoteEndPoint);
+            }
         }
         catch (Exception ex)
         {

test/UnitTest/Services/TcpSocketFactoryTest.cs

@@ -76,6 +76,23 @@ public async Task ConnectAsync_Failed()
         Assert.False(connect);
     }
 
+    [Fact]
+    public async Task Send_Timeout()
+    {
+        var port = 8887;
+        var server = StartTcpServer(port, MockSplitPackageAsync);
+
+        var client = CreateClient();
+        client.SendTimeout = 100;
+        client.SetDataHandler(new MockSendTimeoutHandler());
+
+        await client.ConnectAsync("localhost", port);
+
+        var data = new ReadOnlyMemory<byte>([1, 2, 3, 4, 5]);
+        var result = await client.SendAsync(data);
+        Assert.False(result);
+    }
+
     [Fact]
     public async Task SendAsync_Error()
     {
@@ -499,27 +516,37 @@ class MockSendErrorHandler : DataPackageHandlerBase
     {
         public ITcpSocketClient? Socket { get; set; }
 
-        public override async ValueTask<ReadOnlyMemory<byte>> SendAsync(ReadOnlyMemory<byte> data)
+        public override async ValueTask<ReadOnlyMemory<byte>> SendAsync(ReadOnlyMemory<byte> data, CancellationToken token = default)
         {
             Socket?.Close();
-            await Task.Delay(10);
+            await Task.Delay(10, token);
             return data;
         }
     }
 
     class MockReceiveErrorHandler : DataPackageHandlerBase
     {
-        public override ValueTask<ReadOnlyMemory<byte>> SendAsync(ReadOnlyMemory<byte> data)
+        public override ValueTask<ReadOnlyMemory<byte>> SendAsync(ReadOnlyMemory<byte> data, CancellationToken token = default)
         {
             return ValueTask.FromResult(data);
         }
 
-        public override async ValueTask ReceiveAsync(ReadOnlyMemory<byte> data)
+        public override async ValueTask ReceiveAsync(ReadOnlyMemory<byte> data, CancellationToken token = default)
         {
-            await base.ReceiveAsync(data);
+            await base.ReceiveAsync(data, token);
 
             // 模拟接收数据时报错
             throw new InvalidOperationException("Test Error");
         }
     }
+
+    class MockSendTimeoutHandler : DataPackageHandlerBase
+    {
+        public override async ValueTask<ReadOnlyMemory<byte>> SendAsync(ReadOnlyMemory<byte> data, CancellationToken token = default)
+        {
+            // 模拟发送超时
+            await Task.Delay(200, token);
+            return data;
+        }
+    }
 }