Fix interop test. Dispose GZipStream before accessing content. (#637)

grpc · Oct 31, 2019 · ce0dcd5 · ce0dcd5
1 parent 542066a
commit ce0dcd5
Show file tree

Hide file tree

Showing 6 changed files with 48 additions and 14 deletions.
diff --git a/src/Grpc.AspNetCore.Server/Internal/HttpContextSerializationContext.cs b/src/Grpc.AspNetCore.Server/Internal/HttpContextSerializationContext.cs
@@ -225,13 +225,15 @@ private ReadOnlySpan<byte> CompressMessage(ReadOnlySpan<byte> messageData)
             GrpcServerLog.CompressingMessage(_serverCallContext.Logger, _compressionProvider.EncodingName);
 
             var output = new MemoryStream();
+
+            // Compression stream must be disposed before its content is read.
+            // GZipStream writes final Adler32 at the end of the stream on dispose.
             using (var compressionStream = _compressionProvider.CreateCompressionStream(output, _serverCallContext.MethodContext.ResponseCompressionLevel))
             {
                 compressionStream.Write(messageData);
-                compressionStream.Flush();
-
-                return output.GetBuffer().AsSpan(0, (int)output.Length);
             }
+
+            return output.GetBuffer().AsSpan(0, (int)output.Length);
         }
     }
 }
diff --git a/src/Grpc.AspNetCore.Server/Internal/PipeExtensions.cs b/src/Grpc.AspNetCore.Server/Internal/PipeExtensions.cs
@@ -409,10 +409,10 @@ private static bool TryDecompressMessage(ILogger logger, string compressionEncod
                 using (var compressionStream = compressionProvider.CreateDecompressionStream(new ReadOnlySequenceStream(messageData)))
                 {
                     compressionStream.CopyTo(output);
-
-                    result = new ReadOnlySequence<byte>(output.GetBuffer(), 0, (int)output.Length);
-                    return true;
                 }
+
+                result = new ReadOnlySequence<byte>(output.GetBuffer(), 0, (int)output.Length);
+                return true;
             }
 
             result = null;

diff --git a/src/Grpc.Net.Client/Internal/StreamExtensions.cs b/src/Grpc.Net.Client/Internal/StreamExtensions.cs
@@ -191,8 +191,10 @@ private static bool TryDecompressMessage(ILogger logger, string compressionEncod
                 GrpcCallLog.DecompressingMessage(logger, compressionProvider.EncodingName);
 
                 var output = new MemoryStream();
-                var compressionStream = compressionProvider.CreateDecompressionStream(new MemoryStream(messageData));
-                compressionStream.CopyTo(output);
+                using (var compressionStream = compressionProvider.CreateDecompressionStream(new MemoryStream(messageData)))
+                {
+                    compressionStream.CopyTo(output);
+                }
 
                 result = new ReadOnlySequence<byte>(output.GetBuffer(), 0, (int)output.Length);
                 return true;
@@ -282,6 +284,9 @@ private static byte[] CompressMessage(ILogger logger, string compressionEncoding
                 GrpcCallLog.CompressingMessage(logger, compressionProvider.EncodingName);
 
                 var output = new MemoryStream();
+
+                // Compression stream must be disposed before its content is read.
+                // GZipStream writes final Adler32 at the end of the stream.
                 using (var compressionStream = compressionProvider.CreateCompressionStream(output, compressionLevel))
                 {
                     compressionStream.Write(messageData.Span);

diff --git a/src/Grpc.Net.Common/Compression/GzipCompressionProvider.cs b/src/Grpc.Net.Common/Compression/GzipCompressionProvider.cs
@@ -50,7 +50,7 @@ public GzipCompressionProvider(CompressionLevel defaultCompressionLevel)
         /// <returns>A stream used to compress data.</returns>
         public Stream CreateCompressionStream(Stream stream, CompressionLevel? compressionLevel)
         {
-            return new GZipStream(stream, compressionLevel ?? _defaultCompressionLevel);
+            return new GZipStream(stream, compressionLevel ?? _defaultCompressionLevel, leaveOpen: true);
         }
 
         /// <summary>
@@ -60,7 +60,7 @@ public Stream CreateCompressionStream(Stream stream, CompressionLevel? compressi
         /// <returns>A stream used to decompress data.</returns>
         public Stream CreateDecompressionStream(Stream stream)
         {
-            return new GZipStream(stream, CompressionMode.Decompress);
+            return new GZipStream(stream, CompressionMode.Decompress, leaveOpen: true);
         }
     }
 }
diff --git a/test/FunctionalTests/Server/CompressionTests.cs b/test/FunctionalTests/Server/CompressionTests.cs
@@ -345,12 +345,39 @@ private class DoesNotExistCompressionProvider : ICompressionProvider
 
             public Stream CreateCompressionStream(Stream stream, System.IO.Compression.CompressionLevel? compressionLevel)
             {
-                return stream;
+                return new WrapperStream(stream);
             }
 
             public Stream CreateDecompressionStream(Stream stream)
             {
-                return stream;
+                return new WrapperStream(stream);
+            }
+
+            // Returned stream is disposed. Wrapper leaves the inner stream open.
+            private class WrapperStream : Stream
+            {
+                private readonly Stream _innerStream;
+
+                public WrapperStream(Stream innerStream)
+                {
+                    _innerStream = innerStream;
+                }
+
+                public override bool CanRead => _innerStream.CanRead;
+                public override bool CanSeek => _innerStream.CanSeek;
+                public override bool CanWrite => _innerStream.CanWrite;
+                public override long Length => _innerStream.Length;
+                public override long Position
+                {
+                    get => _innerStream.Position;
+                    set => _innerStream.Position = value;
+                }
+
+                public override void Flush() => _innerStream.Flush();
+                public override int Read(byte[] buffer, int offset, int count) => _innerStream.Read(buffer, offset, count);
+                public override long Seek(long offset, SeekOrigin origin) => _innerStream.Seek(offset, origin);
+                public override void SetLength(long value) => _innerStream.SetLength(value);
+                public override void Write(byte[] buffer, int offset, int count) => _innerStream.Write(buffer, offset, count);
             }
         }
 

diff --git a/test/Grpc.AspNetCore.Server.Tests/PipeExtensionsTestsBase.cs b/test/Grpc.AspNetCore.Server.Tests/PipeExtensionsTestsBase.cs
@@ -709,7 +709,7 @@ public async Task WriteMessageAsync_GzipCompressed_WriteCompressedData()
             var messageData = ms.ToArray();
 
             Assert.AreEqual(1, messageData[0]); // compression
-            Assert.AreEqual(17, messageData[4]); // message length
+            Assert.AreEqual(21, messageData[4]); // message length
 
             byte[] result = Decompress(compressionProviders["gzip"], messageData);
             Assert.AreEqual(1, result.Length);
@@ -747,7 +747,7 @@ public async Task WriteMessageAsync_HasCustomCompressionLevel_WriteCompressedDat
             var messageData = ms.ToArray();
 
             Assert.AreEqual(1, messageData[0]); // compression
-            Assert.AreEqual(17, messageData[4]); // message length
+            Assert.AreEqual(21, messageData[4]); // message length
 
             byte[] result = Decompress(mockCompressionProvider, messageData);
             Assert.AreEqual(1, result.Length);