Merge pull request #991 from adamhathcock/async-reader-methods

Add more Async tests and complete Zip tests

Author: adamhathcock

src/SharpCompress/Common/Zip/StreamingZipHeaderFactory.cs

@@ -36,11 +36,10 @@ internal IEnumerable<ZipHeader> ReadStreamHeader(Stream stream)
                 throw new ArgumentException("Stream must be a SharpCompressStream", nameof(stream));
             }
         }
-        SharpCompressStream rewindableStream = (SharpCompressStream)stream;
+        var rewindableStream = (SharpCompressStream)stream;
 
         while (true)
         {
-            ZipHeader? header;
             var reader = new BinaryReader(rewindableStream);
             uint headerBytes = 0;
             if (
@@ -155,7 +154,7 @@ internal IEnumerable<ZipHeader> ReadStreamHeader(Stream stream)
             }
 
             _lastEntryHeader = null;
-            header = ReadHeader(headerBytes, reader);
+            var header = ReadHeader(headerBytes, reader);
             if (header is null)
             {
                 yield break;

src/SharpCompress/Compressors/Deflate64/Deflate64Stream.cs

@@ -2,12 +2,12 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
-#nullable disable
-
 using System;
 using System.Diagnostics;
 using System.IO;
 using System.Runtime.CompilerServices;
+using System.Threading;
+using System.Threading.Tasks;
 using SharpCompress.Common;
 using SharpCompress.Common.Zip;
 using SharpCompress.IO;
@@ -39,7 +39,6 @@ void IStreamStack.SetPosition(long position) { }
     private const int DEFAULT_BUFFER_SIZE = 8192;
 
     private Stream _stream;
-    private CompressionMode _mode;
     private InflaterManaged _inflater;
     private byte[] _buffer;
 
@@ -62,61 +61,23 @@ public Deflate64Stream(Stream stream, CompressionMode mode)
             throw new ArgumentException("Deflate64: input stream is not readable", nameof(stream));
         }
 
-        InitializeInflater(stream, ZipCompressionMethod.Deflate64);
-#if DEBUG_STREAMS
-        this.DebugConstruct(typeof(Deflate64Stream));
-#endif
-    }
-
-    /// <summary>
-    /// Sets up this DeflateManagedStream to be used for Inflation/Decompression
-    /// </summary>
-    private void InitializeInflater(
-        Stream stream,
-        ZipCompressionMethod method = ZipCompressionMethod.Deflate
-    )
-    {
-        Debug.Assert(stream != null);
-        Debug.Assert(
-            method == ZipCompressionMethod.Deflate || method == ZipCompressionMethod.Deflate64
-        );
         if (!stream.CanRead)
         {
             throw new ArgumentException("Deflate64: input stream is not readable", nameof(stream));
         }
 
-        _inflater = new InflaterManaged(method == ZipCompressionMethod.Deflate64);
+        _inflater = new InflaterManaged(true);
 
         _stream = stream;
-        _mode = CompressionMode.Decompress;
         _buffer = new byte[DEFAULT_BUFFER_SIZE];
+#if DEBUG_STREAMS
+        this.DebugConstruct(typeof(Deflate64Stream));
+#endif
     }
 
-    public override bool CanRead
-    {
-        get
-        {
-            if (_stream is null)
-            {
-                return false;
-            }
-
-            return (_mode == CompressionMode.Decompress && _stream.CanRead);
-        }
-    }
-
-    public override bool CanWrite
-    {
-        get
-        {
-            if (_stream is null)
-            {
-                return false;
-            }
+    public override bool CanRead => _stream.CanRead;
 
-            return (_mode == CompressionMode.Compress && _stream.CanWrite);
-        }
-    }
+    public override bool CanWrite => false;
 
     public override bool CanSeek => false;
 
@@ -138,7 +99,6 @@ public override void SetLength(long value) =>
 
     public override int Read(byte[] array, int offset, int count)
     {
-        EnsureDecompressionMode();
         ValidateParameters(array, offset, count);
         EnsureNotDisposed();
 
@@ -185,6 +145,106 @@ public override int Read(byte[] array, int offset, int count)
         return count - remainingCount;
     }
 
+    public override async Task<int> ReadAsync(
+        byte[] array,
+        int offset,
+        int count,
+        CancellationToken cancellationToken
+    )
+    {
+        ValidateParameters(array, offset, count);
+        EnsureNotDisposed();
+
+        int bytesRead;
+        var currentOffset = offset;
+        var remainingCount = count;
+
+        while (true)
+        {
+            bytesRead = _inflater.Inflate(array, currentOffset, remainingCount);
+            currentOffset += bytesRead;
+            remainingCount -= bytesRead;
+
+            if (remainingCount == 0)
+            {
+                break;
+            }
+
+            if (_inflater.Finished())
+            {
+                // if we finished decompressing, we can't have anything left in the outputwindow.
+                Debug.Assert(
+                    _inflater.AvailableOutput == 0,
+                    "We should have copied all stuff out!"
+                );
+                break;
+            }
+
+            var bytes = await _stream
+                .ReadAsync(_buffer, 0, _buffer.Length, cancellationToken)
+                .ConfigureAwait(false);
+            if (bytes <= 0)
+            {
+                break;
+            }
+            else if (bytes > _buffer.Length)
+            {
+                // The stream is either malicious or poorly implemented and returned a number of
+                // bytes larger than the buffer supplied to it.
+                throw new InvalidFormatException("Deflate64: invalid data");
+            }
+
+            _inflater.SetInput(_buffer, 0, bytes);
+        }
+
+        return count - remainingCount;
+    }
+
+#if !NETFRAMEWORK && !NETSTANDARD2_0
+    public override async ValueTask<int> ReadAsync(
+        Memory<byte> buffer,
+        CancellationToken cancellationToken = default
+    )
+    {
+        EnsureNotDisposed();
+
+        // InflaterManaged doesn't have a Span-based Inflate method, so we need to work with arrays
+        // For large buffers, we could rent from ArrayPool, but for simplicity we'll use the buffer's array if available
+        if (
+            System.Runtime.InteropServices.MemoryMarshal.TryGetArray<byte>(
+                buffer,
+                out var arraySegment
+            )
+        )
+        {
+            // Fast path: the Memory<byte> is backed by an array
+            return await ReadAsync(
+                    arraySegment.Array!,
+                    arraySegment.Offset,
+                    arraySegment.Count,
+                    cancellationToken
+                )
+                .ConfigureAwait(false);
+        }
+        else
+        {
+            // Slow path: rent a temporary array
+            var tempBuffer = System.Buffers.ArrayPool<byte>.Shared.Rent(buffer.Length);
+            try
+            {
+                var bytesRead = await ReadAsync(tempBuffer, 0, buffer.Length, cancellationToken)
+                    .ConfigureAwait(false);
+                tempBuffer.AsMemory(0, bytesRead).CopyTo(buffer);
+                return bytesRead;
+            }
+            finally
+            {
+                System.Buffers.ArrayPool<byte>.Shared.Return(tempBuffer);
+            }
+        }
+    }
+#endif
+
     private void ValidateParameters(byte[] array, int offset, int count)
     {
         if (array is null)
@@ -220,26 +280,6 @@ private void EnsureNotDisposed()
     private static void ThrowStreamClosedException() =>
         throw new ObjectDisposedException(null, "Deflate64: stream has been disposed");
 
-    private void EnsureDecompressionMode()
-    {
-        if (_mode != CompressionMode.Decompress)
-        {
-            ThrowCannotReadFromDeflateManagedStreamException();
-        }
-    }
-
-    [MethodImpl(MethodImplOptions.NoInlining)]
-    private static void ThrowCannotReadFromDeflateManagedStreamException() =>
-        throw new InvalidOperationException("Deflate64: cannot read from this stream");
-
-    private void EnsureCompressionMode()
-    {
-        if (_mode != CompressionMode.Compress)
-        {
-            ThrowCannotWriteToDeflateManagedStreamException();
-        }
-    }
-
     [MethodImpl(MethodImplOptions.NoInlining)]
     private static void ThrowCannotWriteToDeflateManagedStreamException() =>
         throw new InvalidOperationException("Deflate64: cannot write to this stream");
@@ -281,20 +321,17 @@ protected override void Dispose(bool disposing)
 #endif
                 if (disposing)
                 {
-                    _stream?.Dispose();
+                    _stream.Dispose();
                 }
             }
             finally
             {
-                _stream = null;
-
                 try
                 {
-                    _inflater?.Dispose();
+                    _inflater.Dispose();
                 }
                 finally
                 {
-                    _inflater = null;
                     base.Dispose(disposing);
                 }
             }

src/SharpCompress/IO/ReadOnlySubStream.cs

@@ -1,6 +1,8 @@
 using System;
 using System.Diagnostics;
 using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
 
 namespace SharpCompress.IO;
 
@@ -93,6 +95,47 @@ public override int Read(Span<byte> buffer)
     }
 #endif
 
+    public override async Task<int> ReadAsync(
+        byte[] buffer,
+        int offset,
+        int count,
+        CancellationToken cancellationToken
+    )
+    {
+        if (BytesLeftToRead < count)
+        {
+            count = (int)BytesLeftToRead;
+        }
+        var read = await Stream
+            .ReadAsync(buffer, offset, count, cancellationToken)
+            .ConfigureAwait(false);
+        if (read > 0)
+        {
+            BytesLeftToRead -= read;
+            _position += read;
+        }
+        return read;
+    }
+
+#if !NETFRAMEWORK && !NETSTANDARD2_0
+    public override async ValueTask<int> ReadAsync(
+        Memory<byte> buffer,
+        CancellationToken cancellationToken = default
+    )
+    {
+        var sliceLen = BytesLeftToRead < buffer.Length ? BytesLeftToRead : buffer.Length;
+        var read = await Stream
+            .ReadAsync(buffer.Slice(0, (int)sliceLen), cancellationToken)
+            .ConfigureAwait(false);
+        if (read > 0)
+        {
+            BytesLeftToRead -= read;
+            _position += read;
+        }
+        return read;
+    }
+#endif
+
     public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
 
     public override void SetLength(long value) => throw new NotSupportedException();