优化传输性能。。。

Author: aaasoft

Quick.Protocol.WebSocket.Server.AspNetCore/WebSocketServerStream.cs

@@ -63,6 +63,7 @@ public override Task WriteAsync(byte[] buffer, int offset, int count, Cancellati
 
         protected override void Dispose(bool disposing)
         {
+            cts.Cancel();
             cts.Dispose();
             webSocket.Dispose();
             base.Dispose(disposing);

Quick.Protocol/QpChannel_Send.cs

@@ -176,14 +176,17 @@ private static void writePackageTotalLengthToBuffer(Memory<byte> memory, int pac
             writePackageTotalLengthToBuffer(memory.Span, packageTotalLength);
         }
 
-        private async Task UseSendPipe(Func<Pipe, Task> handler)
+        private async Task UseSendPipe(QpPackageType packageType, Func<Pipe, Task<int>> packageBodyHandler = null, bool ignoreCompressAndEncrypt = false)
         {
             if (options.EnableNetstat)
                 Interlocked.Increment(ref PackageSendQueueCount);
             await sendLock.WaitAsync().ConfigureAwait(false);
             try
             {
-                await handler(sendPipe);
+                var packageBodyLength = 0;
+                if (packageBodyHandler != null)
+                    packageBodyLength = await packageBodyHandler(sendPipe);
+                await writePackageBuffer(sendPipe.Reader, packageType, packageBodyLength, ignoreCompressAndEncrypt);
             }
             catch (Exception ex)
             {
@@ -203,18 +206,12 @@ private async Task UseSendPipe(Func<Pipe, Task> handler)
         /// </summary>
         public async Task SendHeartbeatPackage()
         {
-            await UseSendPipe(async pipe =>
-            {
-                var bodyLength = 0;
-                await writePackageBuffer(pipe.Reader,
-                    QpPackageType.Heartbeat,
-                    bodyLength).ConfigureAwait(false);
-            });
+            await UseSendPipe(QpPackageType.Heartbeat);
         }
 
         public async Task SendNoticePackage(string noticePackageTypeName, string noticePackageContent)
         {
-            await UseSendPipe(async pipe =>
+            await UseSendPipe(QpPackageType.Notice, async pipe =>
             {
                 var writer = pipe.Writer;
 
@@ -247,12 +244,10 @@ await UseSendPipe(async pipe =>
                         bodyLength += byteCount;
                     }
                 }
-                _ = writer.FlushAsync();
+                await writer.FlushAsync();
                 if (LogUtils.LogNotice)
                     LogUtils.Log("{0}: [Send-NoticePackage]Type:{1},Content:{2}", DateTime.Now, typeName, LogUtils.LogContent ? content : LogUtils.NOT_SHOW_CONTENT_MESSAGE);
-                await writePackageBuffer(pipe.Reader,
-                        QpPackageType.Notice,
-                        bodyLength).ConfigureAwait(false);
+                return bodyLength;
             });
         }
 
@@ -266,7 +261,7 @@ public Task SendCommandRequestPackage(string commandId, string typeName, string
         /// </summary>
         private async Task SendCommandRequestPackage(string commandId, string typeName, string content, bool ignoreCompressAndEncrypt)
         {
-            await UseSendPipe(async pipe =>
+            await UseSendPipe(QpPackageType.CommandRequest, async pipe =>
             {
                 var writer = pipe.Writer;
                 var bodyLength = 0;
@@ -295,24 +290,21 @@ await UseSendPipe(async pipe =>
                     writer.Advance(contentLength);
                     bodyLength += contentLength;
                 }
-                _ = writer.FlushAsync();
+                await writer.FlushAsync();
 
                 if (LogUtils.LogCommand)
                     LogUtils.Log("{0}: [Send-CommandRequestPackage]CommandId:{1},Type:{2},Content:{3}", DateTime.Now, commandId, typeName, LogUtils.LogContent ? content : LogUtils.NOT_SHOW_CONTENT_MESSAGE);
 
-                await writePackageBuffer(pipe.Reader,
-                        QpPackageType.CommandRequest,
-                        bodyLength,
-                        ignoreCompressAndEncrypt).ConfigureAwait(false);
-            });
+                return bodyLength;
+            }, ignoreCompressAndEncrypt);
         }
 
         /// <summary>
         /// 发送命令响应包
         /// </summary>
         public async Task SendCommandResponsePackage(string commandId, byte code, string message, string typeName, string content)
         {
-            await UseSendPipe(async pipe =>
+            await UseSendPipe(QpPackageType.CommandResponse, async pipe =>
             {
                 var writer = pipe.Writer;
                 var bodyLength = 0;
@@ -369,9 +361,7 @@ await UseSendPipe(async pipe =>
                         LogUtils.Log("{0}: [Send-CommandResponsePackage]CommandId:{1},Code:{2},Message:{3}", DateTime.Now, commandId, code, message);
                 }
                 _ = writer.FlushAsync();
-                await writePackageBuffer(pipe.Reader,
-                        QpPackageType.CommandResponse,
-                        bodyLength).ConfigureAwait(false);
+                return bodyLength;
             });
         }
     }