async xzblock

src/SharpCompress/Compressors/Xz/BinaryUtils.cs

@@ -1,6 +1,8 @@
 using System;
 using System.Buffers.Binary;
 using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
 
 namespace SharpCompress.Compressors.Xz;
 
@@ -30,6 +32,28 @@ public static class BinaryUtils
     internal static uint ReadLittleEndianUInt32(this Stream stream) =>
         unchecked((uint)ReadLittleEndianInt32(stream));
 
+    public static async Task<int> ReadLittleEndianInt32Async(
+        this Stream stream,
+        CancellationToken cancellationToken = default
+    )
+    {
+        var bytes = new byte[4];
+        var read = await stream.ReadFullyAsync(bytes, cancellationToken).ConfigureAwait(false);
+        if (!read)
+        {
+            throw new EndOfStreamException();
+        }
+        return BinaryPrimitives.ReadInt32LittleEndian(bytes);
+    }
+
+    internal static async Task<uint> ReadLittleEndianUInt32Async(
+        this Stream stream,
+        CancellationToken cancellationToken = default
+    ) =>
+        unchecked(
+            (uint)await ReadLittleEndianInt32Async(stream, cancellationToken).ConfigureAwait(false)
+        );
+
     internal static byte[] ToBigEndianBytes(this uint uint32)
     {
         var result = BitConverter.GetBytes(uint32);

src/SharpCompress/Compressors/Xz/XZBlock.cs

@@ -4,6 +4,8 @@ using System;
 using System.Collections.Generic;
 using System.IO;
 using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
 using SharpCompress.Common;
 using SharpCompress.Compressors.Xz.Filters;
 
@@ -72,6 +74,49 @@ public sealed class XZBlock : XZReadOnlyStream
         return bytesRead;
     }
 
+    public override async Task<int> ReadAsync(
+        byte[] buffer,
+        int offset,
+        int count,
+        CancellationToken cancellationToken = default
+    )
+    {
+        var bytesRead = 0;
+        if (!HeaderIsLoaded)
+        {
+            await LoadHeaderAsync(cancellationToken).ConfigureAwait(false);
+        }
+
+        if (!_streamConnected)
+        {
+            ConnectStream();
+        }
+
+        if (!_endOfStream)
+        {
+            bytesRead = await _decomStream
+                .ReadAsync(buffer, offset, count, cancellationToken)
+                .ConfigureAwait(false);
+        }
+
+        if (bytesRead != count)
+        {
+            _endOfStream = true;
+        }
+
+        if (_endOfStream && !_paddingSkipped)
+        {
+            await SkipPaddingAsync(cancellationToken).ConfigureAwait(false);
+        }
+
+        if (_endOfStream && !_crcChecked)
+        {
+            await CheckCrcAsync(cancellationToken).ConfigureAwait(false);
+        }
+
+        return bytesRead;
+    }
+
     private void SkipPadding()
     {
         var bytes = (BaseStream.Position - _startPosition) % 4;
@@ -87,6 +132,23 @@ public sealed class XZBlock : XZReadOnlyStream
         _paddingSkipped = true;
     }
 
+    private async Task SkipPaddingAsync(CancellationToken cancellationToken = default)
+    {
+        var bytes = (BaseStream.Position - _startPosition) % 4;
+        if (bytes > 0)
+        {
+            var paddingBytes = new byte[4 - bytes];
+            await BaseStream
+                .ReadAsync(paddingBytes, 0, paddingBytes.Length, cancellationToken)
+                .ConfigureAwait(false);
+            if (paddingBytes.Any(b => b != 0))
+            {
+                throw new InvalidFormatException("Padding bytes were non-null");
+            }
+        }
+        _paddingSkipped = true;
+    }
+
     private void CheckCrc()
     {
         var crc = new byte[_checkSize];
@@ -96,6 +158,15 @@ public sealed class XZBlock : XZReadOnlyStream
         _crcChecked = true;
     }
 
+    private async Task CheckCrcAsync(CancellationToken cancellationToken = default)
+    {
+        var crc = new byte[_checkSize];
+        await BaseStream.ReadAsync(crc, 0, _checkSize, cancellationToken).ConfigureAwait(false);
+        // Actually do a check (and read in the bytes
+        //   into the function throughout the stream read).
+        _crcChecked = true;
+    }
+
     private void ConnectStream()
     {
         _decomStream = BaseStream;
@@ -123,6 +194,21 @@ public sealed class XZBlock : XZReadOnlyStream
         HeaderIsLoaded = true;
     }
 
+    private async Task LoadHeaderAsync(CancellationToken cancellationToken = default)
+    {
+        await ReadHeaderSizeAsync(cancellationToken).ConfigureAwait(false);
+        var headerCache = await CacheHeaderAsync(cancellationToken).ConfigureAwait(false);
+
+        using (var cache = new MemoryStream(headerCache))
+        using (var cachedReader = new BinaryReader(cache))
+        {
+            cachedReader.BaseStream.Position = 1; // skip the header size byte
+            ReadBlockFlags(cachedReader);
+            ReadFilters(cachedReader);
+        }
+        HeaderIsLoaded = true;
+    }
+
     private void ReadHeaderSize()
     {
         _blockHeaderSizeByte = (byte)BaseStream.ReadByte();
@@ -132,6 +218,17 @@ public sealed class XZBlock : XZReadOnlyStream
         }
     }
 
+    private async Task ReadHeaderSizeAsync(CancellationToken cancellationToken = default)
+    {
+        var buffer = new byte[1];
+        await BaseStream.ReadAsync(buffer, 0, 1, cancellationToken).ConfigureAwait(false);
+        _blockHeaderSizeByte = buffer[0];
+        if (_blockHeaderSizeByte == 0)
+        {
+            throw new XZIndexMarkerReachedException();
+        }
+    }
+
     private byte[] CacheHeader()
     {
         var blockHeaderWithoutCrc = new byte[BlockHeaderSize - 4];
@@ -152,6 +249,30 @@ public sealed class XZBlock : XZReadOnlyStream
         return blockHeaderWithoutCrc;
     }
 
+    private async Task<byte[]> CacheHeaderAsync(CancellationToken cancellationToken = default)
+    {
+        var blockHeaderWithoutCrc = new byte[BlockHeaderSize - 4];
+        blockHeaderWithoutCrc[0] = _blockHeaderSizeByte;
+        var read = await BaseStream
+            .ReadAsync(blockHeaderWithoutCrc, 1, BlockHeaderSize - 5, cancellationToken)
+            .ConfigureAwait(false);
+        if (read != BlockHeaderSize - 5)
+        {
+            throw new EndOfStreamException("Reached end of stream unexectedly");
+        }
+
+        var crc = await BaseStream
+            .ReadLittleEndianUInt32Async(cancellationToken)
+            .ConfigureAwait(false);
+        var calcCrc = Crc32.Compute(blockHeaderWithoutCrc);
+        if (crc != calcCrc)
+        {
+            throw new InvalidFormatException("Block header corrupt");
+        }
+
+        return blockHeaderWithoutCrc;
+    }
+
     private void ReadBlockFlags(BinaryReader reader)
     {
         var blockFlags = reader.ReadByte();

src/SharpCompress/Utility.cs

@@ -447,6 +447,31 @@ internal static class Utility
     }
 #endif
 
+    public static async Task<bool> ReadFullyAsync(
+        this Stream stream,
+        byte[] buffer,
+        CancellationToken cancellationToken = default
+    )
+    {
+        var total = 0;
+        int read;
+        while (
+            (
+                read = await stream
+                    .ReadAsync(buffer, total, buffer.Length - total, cancellationToken)
+                    .ConfigureAwait(false)
+            ) > 0
+        )
+        {
+            total += read;
+            if (total >= buffer.Length)
+            {
+                return true;
+            }
+        }
+        return (total >= buffer.Length);
+    }
+
     public static string TrimNulls(this string source) => source.Replace('\0', ' ').Trim();
 
     /// <summary>