发送压缩功能重构完成。

Author: aaasoft

Quick.Protocol/QpChannel_Send.cs

@@ -10,6 +10,7 @@
 using System.Threading;
 using System.Threading.Tasks;
 using Nerdbank.Streams;
+using Quick.Protocol.Streams;
 
 namespace Quick.Protocol
 {
@@ -39,6 +40,8 @@ protected virtual void OnWriteError(Exception exception)
         private async Task writePackageBuffer(PipeReader currentReader, QpPackageType packageType, int packageBodyLength, bool ignoreCompressAndEncrypt = false)
         {
             var stream = QpPackageHandler_Stream;
+            if (stream == null)
+                throw new IOException("Not connected.");
 
             var readRet = await currentReader.ReadAtLeastAsync(packageBodyLength);
             var packageBodyBuffer = readRet.Buffer;
@@ -55,16 +58,20 @@ private async Task writePackageBuffer(PipeReader currentReader, QpPackageType pa
                     if (writeCompressPipe == null)
                         writeCompressPipe = new Pipe();
                     using (var inStream = packageBodyBuffer.AsStream())
-                    using (var outStream = writeCompressPipe.Writer.AsStream(true))
-                    using (var gzStream = new GZipStream(outStream, CompressionMode.Compress, true))
+                    using (var outStream = new PipeWriterStream(writeCompressPipe.Writer, true))
                     {
-                        await inStream.CopyToAsync(gzStream).ConfigureAwait(false);
+                        using (var gzStream = new GZipStream(outStream, CompressionMode.Compress, true))
+                        {
+                            await inStream.CopyToAsync(gzStream).ConfigureAwait(false);
+                        }
+                        packageTotalLength = Convert.ToInt32(outStream.Length);
+                        _ = writeCompressPipe.Writer.FlushAsync();
                     }
                     currentReader.AdvanceTo(packageBodyBuffer.End);
+
+                    readRet = await writeCompressPipe.Reader.ReadAtLeastAsync(packageTotalLength).ConfigureAwait(false);
 
-                    readRet = await writeCompressPipe.Reader.ReadAsync().ConfigureAwait(false);
                     packageBodyBuffer = readRet.Buffer;
-                    packageTotalLength += Convert.ToInt32(packageBodyBuffer.Length);
 
                     //包总长度
                     packageTotalLength += PACKAGE_TOTAL_LENGTH_LENGTH;
@@ -194,7 +201,7 @@ private async Task UseSendPipe(Func<Pipe, Task> handler)
         {
             if (options.EnableNetstat)
                 Interlocked.Increment(ref PackageSendQueueCount);
-            await sendLock.WaitAsync(5000).ConfigureAwait(false);
+            await sendLock.WaitAsync().ConfigureAwait(false);
             try
             {
                 await handler(sendPipe);

Quick.Protocol/Streams/PipeWriterStream.cs

@@ -92,7 +92,6 @@ public sealed override void EndWrite(IAsyncResult asyncResult)
         public override void Write(byte[] buffer, int offset, int count)
         {
             WriteAsync(buffer, offset, count).GetAwaiter().GetResult();
-            _Length += count;
         }
 
         public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)