Merge pull request #1002 from adamhathcock/adam/async-bzip2

async bzip2 and add

src/SharpCompress/Compressors/ADC/ADCBase.cs

@@ -24,10 +24,29 @@
 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 // THE SOFTWARE.
 using System;
+using System.Buffers;
 using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
 
 namespace SharpCompress.Compressors.ADC;
 
+/// <summary>
+/// Result of an ADC decompression operation
+/// </summary>
+public class AdcDecompressResult
+{
+    /// <summary>
+    /// Number of bytes read from input
+    /// </summary>
+    public int BytesRead { get; set; }
+
+    /// <summary>
+    /// Decompressed output buffer
+    /// </summary>
+    public byte[]? Output { get; set; }
+}
+
 /// <summary>
 /// Provides static methods for decompressing Apple Data Compression data
 /// </summary>
@@ -78,6 +97,173 @@ public static class ADCBase
     public static int Decompress(byte[] input, out byte[]? output, int bufferSize = 262144) =>
         Decompress(new MemoryStream(input), out output, bufferSize);
 
+    /// <summary>
+    /// Decompresses a byte buffer asynchronously that's compressed with ADC
+    /// </summary>
+    /// <param name="input">Compressed buffer</param>
+    /// <param name="bufferSize">Max size for decompressed data</param>
+    /// <param name="cancellationToken">Cancellation token</param>
+    /// <returns>Result containing bytes read and decompressed data</returns>
+    public static async Task<AdcDecompressResult> DecompressAsync(
+        byte[] input,
+        int bufferSize = 262144,
+        CancellationToken cancellationToken = default
+    ) => await DecompressAsync(new MemoryStream(input), bufferSize, cancellationToken);
+
+    /// <summary>
+    /// Decompresses a stream asynchronously that's compressed with ADC
+    /// </summary>
+    /// <param name="input">Stream containing compressed data</param>
+    /// <param name="bufferSize">Max size for decompressed data</param>
+    /// <param name="cancellationToken">Cancellation token</param>
+    /// <returns>Result containing bytes read and decompressed data</returns>
+    public static async Task<AdcDecompressResult> DecompressAsync(
+        Stream input,
+        int bufferSize = 262144,
+        CancellationToken cancellationToken = default
+    )
+    {
+        var result = new AdcDecompressResult();
+
+        if (input is null || input.Length == 0)
+        {
+            result.BytesRead = 0;
+            result.Output = null;
+            return result;
+        }
+
+        var start = (int)input.Position;
+        var position = (int)input.Position;
+        int chunkSize;
+        int offset;
+        int chunkType;
+        var buffer = ArrayPool<byte>.Shared.Rent(bufferSize);
+        var outPosition = 0;
+        var full = false;
+        byte[] temp = ArrayPool<byte>.Shared.Rent(3);
+
+        try
+        {
+            while (position < input.Length)
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+                var readByte = input.ReadByte();
+                if (readByte == -1)
+                {
+                    break;
+                }
+
+                chunkType = GetChunkType((byte)readByte);
+
+                switch (chunkType)
+                {
+                    case PLAIN:
+                        chunkSize = GetChunkSize((byte)readByte);
+                        if (outPosition + chunkSize > bufferSize)
+                        {
+                            full = true;
+                            break;
+                        }
+
+                        var readCount = await input.ReadAsync(
+                            buffer,
+                            outPosition,
+                            chunkSize,
+                            cancellationToken
+                        );
+                        outPosition += readCount;
+                        position += readCount + 1;
+                        break;
+                    case TWO_BYTE:
+                        chunkSize = GetChunkSize((byte)readByte);
+                        temp[0] = (byte)readByte;
+                        temp[1] = (byte)input.ReadByte();
+                        offset = GetOffset(temp.AsSpan(0, 2));
+                        if (outPosition + chunkSize > bufferSize)
+                        {
+                            full = true;
+                            break;
+                        }
+
+                        if (offset == 0)
+                        {
+                            var lastByte = buffer[outPosition - 1];
+                            for (var i = 0; i < chunkSize; i++)
+                            {
+                                buffer[outPosition] = lastByte;
+                                outPosition++;
+                            }
+
+                            position += 2;
+                        }
+                        else
+                        {
+                            for (var i = 0; i < chunkSize; i++)
+                            {
+                                buffer[outPosition] = buffer[outPosition - offset - 1];
+                                outPosition++;
+                            }
+
+                            position += 2;
+                        }
+
+                        break;
+                    case THREE_BYTE:
+                        chunkSize = GetChunkSize((byte)readByte);
+                        temp[0] = (byte)readByte;
+                        temp[1] = (byte)input.ReadByte();
+                        temp[2] = (byte)input.ReadByte();
+                        offset = GetOffset(temp.AsSpan(0, 3));
+                        if (outPosition + chunkSize > bufferSize)
+                        {
+                            full = true;
+                            break;
+                        }
+
+                        if (offset == 0)
+                        {
+                            var lastByte = buffer[outPosition - 1];
+                            for (var i = 0; i < chunkSize; i++)
+                            {
+                                buffer[outPosition] = lastByte;
+                                outPosition++;
+                            }
+
+                            position += 3;
+                        }
+                        else
+                        {
+                            for (var i = 0; i < chunkSize; i++)
+                            {
+                                buffer[outPosition] = buffer[outPosition - offset - 1];
+                                outPosition++;
+                            }
+
+                            position += 3;
+                        }
+
+                        break;
+                }
+
+                if (full)
+                {
+                    break;
+                }
+            }
+
+            var output = new byte[outPosition];
+            Array.Copy(buffer, output, outPosition);
+            result.BytesRead = position - start;
+            result.Output = output;
+            return result;
+        }
+        finally
+        {
+            ArrayPool<byte>.Shared.Return(buffer);
+            ArrayPool<byte>.Shared.Return(temp);
+        }
+    }
+
     /// <summary>
     /// Decompresses a stream that's compressed with ADC
     /// </summary>

src/SharpCompress/Compressors/ADC/ADCStream.cs

@@ -28,6 +28,8 @@
 
 using System;
 using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
 using SharpCompress.IO;
 
 namespace SharpCompress.Compressors.ADC;
@@ -187,6 +189,76 @@ public sealed class ADCStream : Stream, IStreamStack
         return copied;
     }
 
+    public override async Task<int> ReadAsync(
+        byte[] buffer,
+        int offset,
+        int count,
+        CancellationToken cancellationToken = default
+    )
+    {
+        if (count == 0)
+        {
+            return 0;
+        }
+        if (buffer is null)
+        {
+            throw new ArgumentNullException(nameof(buffer));
+        }
+        if (count < 0)
+        {
+            throw new ArgumentOutOfRangeException(nameof(count));
+        }
+        if (offset < buffer.GetLowerBound(0))
+        {
+            throw new ArgumentOutOfRangeException(nameof(offset));
+        }
+        if ((offset + count) > buffer.GetLength(0))
+        {
+            throw new ArgumentOutOfRangeException(nameof(count));
+        }
+
+        if (_outBuffer is null)
+        {
+            var result = await ADCBase.DecompressAsync(
+                _stream,
+                cancellationToken: cancellationToken
+            );
+            _outBuffer = result.Output;
+            _outPosition = 0;
+        }
+
+        var inPosition = offset;
+        var toCopy = count;
+        var copied = 0;
+
+        while (_outPosition + toCopy >= _outBuffer.Length)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+            var piece = _outBuffer.Length - _outPosition;
+            Array.Copy(_outBuffer, _outPosition, buffer, inPosition, piece);
+            inPosition += piece;
+            copied += piece;
+            _position += piece;
+            toCopy -= piece;
+            var result = await ADCBase.DecompressAsync(
+                _stream,
+                cancellationToken: cancellationToken
+            );
+            _outBuffer = result.Output;
+            _outPosition = 0;
+            if (result.BytesRead == 0 || _outBuffer is null || _outBuffer.Length == 0)
+            {
+                return copied;
+            }
+        }
+
+        Array.Copy(_outBuffer, _outPosition, buffer, inPosition, toCopy);
+        _outPosition += toCopy;
+        _position += toCopy;
+        copied += toCopy;
+        return copied;
+    }
+
     public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
 
     public override void SetLength(long value) => throw new NotSupportedException();

src/SharpCompress/Compressors/BZip2/BZip2Stream.cs

@@ -1,5 +1,7 @@
 using System;
 using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
 using SharpCompress.IO;
 
 namespace SharpCompress.Compressors.BZip2;
@@ -96,13 +98,37 @@ public sealed class BZip2Stream : Stream, IStreamStack
 
     public override void SetLength(long value) => stream.SetLength(value);
 
-#if !NETFRAMEWORK&& !NETSTANDARD2_0
+#if !NETFRAMEWORK && !NETSTANDARD2_0
 
     public override int Read(Span<byte> buffer) => stream.Read(buffer);
 
     public override void Write(ReadOnlySpan<byte> buffer) => stream.Write(buffer);
+
+    public override async ValueTask<int> ReadAsync(
+        Memory<byte> buffer,
+        CancellationToken cancellationToken = default
+    ) => await stream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
+
+    public override async ValueTask WriteAsync(
+        ReadOnlyMemory<byte> buffer,
+        CancellationToken cancellationToken = default
+    ) => await stream.WriteAsync(buffer, cancellationToken).ConfigureAwait(false);
 #endif
 
+    public override async Task<int> ReadAsync(
+        byte[] buffer,
+        int offset,
+        int count,
+        CancellationToken cancellationToken = default
+    ) => await stream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
+
+    public override async Task WriteAsync(
+        byte[] buffer,
+        int offset,
+        int count,
+        CancellationToken cancellationToken = default
+    ) => await stream.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
+
     public override void Write(byte[] buffer, int offset, int count) =>
         stream.Write(buffer, offset, count);
 

src/SharpCompress/Compressors/BZip2/CBZip2InputStream.cs

@@ -2,6 +2,8 @@
 
 using System;
 using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
 using SharpCompress.IO;
 
 /*
@@ -1127,6 +1129,28 @@ internal class CBZip2InputStream : Stream, IStreamStack
         return k;
     }
 
+    public override Task<int> ReadAsync(
+        byte[] buffer,
+        int offset,
+        int count,
+        CancellationToken cancellationToken = default
+    )
+    {
+        var c = -1;
+        int k;
+        for (k = 0; k < count; ++k)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+            c = ReadByte();
+            if (c == -1)
+            {
+                break;
+            }
+            buffer[k + offset] = (byte)c;
+        }
+        return Task.FromResult(k);
+    }
+
     public override long Seek(long offset, SeekOrigin origin) => 0;
 
     public override void SetLength(long value) { }

src/SharpCompress/Compressors/BZip2/CBZip2OutputStream.cs

@@ -1,5 +1,7 @@
 using System;
 using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
 using SharpCompress.IO;
 
 /*
@@ -2022,6 +2024,21 @@ internal sealed class CBZip2OutputStream : Stream, IStreamStack
         }
     }
 
+    public override Task WriteAsync(
+        byte[] buffer,
+        int offset,
+        int count,
+        CancellationToken cancellationToken = default
+    )
+    {
+        for (var k = 0; k < count; ++k)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+            WriteByte(buffer[k + offset]);
+        }
+        return Task.CompletedTask;
+    }
+
     public override bool CanRead => false;
 
     public override bool CanSeek => false;

src/SharpCompress/Writers/Zip/ZipWriter.cs

@@ -160,7 +160,7 @@ public class ZipWriter : AbstractWriter
         WriteDirectoryEntry(normalizedName, options);
     }
 
-    public override async Task WriteDirectoryAsync(
+    public override Task WriteDirectoryAsync(
         string directoryName,
         DateTime? modificationTime,
         CancellationToken cancellationToken = default
@@ -168,7 +168,7 @@ public class ZipWriter : AbstractWriter
     {
         // Synchronous implementation is sufficient for directory entries
         WriteDirectory(directoryName, modificationTime);
-        await Task.CompletedTask.ConfigureAwait(false);
+        return Task.CompletedTask;
     }
 
     private void WriteDirectoryEntry(string directoryPath, ZipWriterEntryOptions options)

tests/SharpCompress.Test/AdcAsyncTest.cs

@@ -0,0 +1,61 @@
+using System.IO;
+using System.Threading.Tasks;
+using SharpCompress.Compressors.ADC;
+using Xunit;
+
+namespace SharpCompress.Test;
+
+public class AdcAsyncTest : TestBase
+{
+    [Fact]
+    public async Task TestAdcStreamAsyncWholeChunk()
+    {
+        using var decFs = File.OpenRead(Path.Combine(TEST_ARCHIVES_PATH, "adc_decompressed.bin"));
+        var decompressed = new byte[decFs.Length];
+        decFs.Read(decompressed, 0, decompressed.Length);
+
+        using var cmpFs = File.OpenRead(Path.Combine(TEST_ARCHIVES_PATH, "adc_compressed.bin"));
+        using var decStream = new ADCStream(cmpFs);
+        var test = new byte[262144];
+
+        await decStream.ReadAsync(test, 0, test.Length);
+
+        Assert.Equal(decompressed, test);
+    }
+
+    [Fact]
+    public async Task TestAdcStreamAsync()
+    {
+        using var decFs = File.OpenRead(Path.Combine(TEST_ARCHIVES_PATH, "adc_decompressed.bin"));
+        var decompressed = new byte[decFs.Length];
+        decFs.Read(decompressed, 0, decompressed.Length);
+
+        using var cmpFs = File.OpenRead(Path.Combine(TEST_ARCHIVES_PATH, "adc_compressed.bin"));
+        using var decStream = new ADCStream(cmpFs);
+        using var decMs = new MemoryStream();
+        var test = new byte[512];
+        var count = 0;
+
+        do
+        {
+            count = await decStream.ReadAsync(test, 0, test.Length);
+            decMs.Write(test, 0, count);
+        } while (count > 0);
+
+        Assert.Equal(decompressed, decMs.ToArray());
+    }
+
+    [Fact]
+    public async Task TestAdcStreamAsyncWithCancellation()
+    {
+        using var cmpFs = File.OpenRead(Path.Combine(TEST_ARCHIVES_PATH, "adc_compressed.bin"));
+        using var decStream = new ADCStream(cmpFs);
+        var test = new byte[512];
+        using var cts = new System.Threading.CancellationTokenSource();
+
+        // Read should complete without cancellation
+        var bytesRead = await decStream.ReadAsync(test, 0, test.Length, cts.Token);
+
+        Assert.True(bytesRead > 0);
+    }
+}

tests/SharpCompress.Test/BZip2/BZip2StreamAsyncTests.cs

@@ -0,0 +1,231 @@
+using System;
+using System.Buffers;
+using System.IO;
+using System.Text;
+using System.Threading.Tasks;
+using SharpCompress.Compressors.BZip2;
+using Xunit;
+
+namespace SharpCompress.Test.BZip2;
+
+public class BZip2StreamAsyncTests
+{
+    private byte[] CreateTestData(int size)
+    {
+        var data = new byte[size];
+        // Create compressible data with repetitive pattern
+        for (int i = 0; i < size; i++)
+        {
+            data[i] = (byte)('A' + (i % 26));
+        }
+        return data;
+    }
+
+    [Fact]
+    public async Task BZip2CompressDecompressAsyncTest()
+    {
+        var testData = CreateTestData(10000);
+        byte[] compressed;
+
+        // Compress
+        using (var memoryStream = new MemoryStream())
+        {
+            using (
+                var bzip2Stream = new BZip2Stream(
+                    memoryStream,
+                    SharpCompress.Compressors.CompressionMode.Compress,
+                    false
+                )
+            )
+            {
+                await bzip2Stream.WriteAsync(testData, 0, testData.Length);
+                (bzip2Stream as BZip2Stream)?.Finish();
+            }
+            compressed = memoryStream.ToArray();
+        }
+
+        // Verify compression occurred
+        Assert.True(compressed.Length > 0);
+        Assert.True(compressed.Length < testData.Length);
+
+        // Decompress
+        byte[] decompressed;
+        using (var memoryStream = new MemoryStream(compressed))
+        {
+            using (
+                var bzip2Stream = new BZip2Stream(
+                    memoryStream,
+                    SharpCompress.Compressors.CompressionMode.Decompress,
+                    false
+                )
+            )
+            {
+                decompressed = new byte[testData.Length];
+                var totalRead = 0;
+                int bytesRead;
+                while (
+                    (
+                        bytesRead = await bzip2Stream.ReadAsync(
+                            decompressed,
+                            totalRead,
+                            testData.Length - totalRead
+                        )
+                    ) > 0
+                )
+                {
+                    totalRead += bytesRead;
+                }
+            }
+        }
+
+        // Verify decompression
+        Assert.Equal(testData, decompressed);
+    }
+
+    [Fact]
+    public async Task BZip2ReadAsyncWithCancellationTest()
+    {
+        var testData = Encoding.ASCII.GetBytes(new string('A', 5000)); // Repetitive data compresses well
+        byte[] compressed;
+
+        // Compress
+        using (var memoryStream = new MemoryStream())
+        {
+            using (
+                var bzip2Stream = new BZip2Stream(
+                    memoryStream,
+                    SharpCompress.Compressors.CompressionMode.Compress,
+                    false
+                )
+            )
+            {
+                await bzip2Stream.WriteAsync(testData, 0, testData.Length);
+                (bzip2Stream as BZip2Stream)?.Finish();
+            }
+            compressed = memoryStream.ToArray();
+        }
+
+        // Decompress with cancellation support
+        using (var memoryStream = new MemoryStream(compressed))
+        {
+            using (
+                var bzip2Stream = new BZip2Stream(
+                    memoryStream,
+                    SharpCompress.Compressors.CompressionMode.Decompress,
+                    false
+                )
+            )
+            {
+                var buffer = new byte[1024];
+                using var cts = new System.Threading.CancellationTokenSource();
+
+                // Read should complete without cancellation
+                var bytesRead = await bzip2Stream.ReadAsync(buffer, 0, buffer.Length, cts.Token);
+                Assert.True(bytesRead > 0);
+            }
+        }
+    }
+
+    [Fact]
+    public async Task BZip2MultipleAsyncWritesTest()
+    {
+        using (var memoryStream = new MemoryStream())
+        {
+            using (
+                var bzip2Stream = new BZip2Stream(
+                    memoryStream,
+                    SharpCompress.Compressors.CompressionMode.Compress,
+                    false
+                )
+            )
+            {
+                var data1 = Encoding.ASCII.GetBytes("Hello ");
+                var data2 = Encoding.ASCII.GetBytes("World");
+                var data3 = Encoding.ASCII.GetBytes("!");
+
+                await bzip2Stream.WriteAsync(data1, 0, data1.Length);
+                await bzip2Stream.WriteAsync(data2, 0, data2.Length);
+                await bzip2Stream.WriteAsync(data3, 0, data3.Length);
+
+                (bzip2Stream as BZip2Stream)?.Finish();
+            }
+
+            var compressed = memoryStream.ToArray();
+            Assert.True(compressed.Length > 0);
+
+            // Decompress and verify
+            using (var readStream = new MemoryStream(compressed))
+            {
+                using (
+                    var bzip2Stream = new BZip2Stream(
+                        readStream,
+                        SharpCompress.Compressors.CompressionMode.Decompress,
+                        false
+                    )
+                )
+                {
+                    var result = new StringBuilder();
+                    var buffer = new byte[256];
+                    int bytesRead;
+                    while ((bytesRead = await bzip2Stream.ReadAsync(buffer, 0, buffer.Length)) > 0)
+                    {
+                        result.Append(Encoding.ASCII.GetString(buffer, 0, bytesRead));
+                    }
+
+                    Assert.Equal("Hello World!", result.ToString());
+                }
+            }
+        }
+    }
+
+    [Fact]
+    public async Task BZip2LargeDataAsyncTest()
+    {
+        var largeData = CreateTestData(100000);
+
+        // Compress
+        byte[] compressed;
+        using (var memoryStream = new MemoryStream())
+        {
+            using (
+                var bzip2Stream = new BZip2Stream(
+                    memoryStream,
+                    SharpCompress.Compressors.CompressionMode.Compress,
+                    false
+                )
+            )
+            {
+                await bzip2Stream.WriteAsync(largeData, 0, largeData.Length);
+                (bzip2Stream as BZip2Stream)?.Finish();
+            }
+            compressed = memoryStream.ToArray();
+        }
+
+        // Decompress
+        byte[] decompressed;
+        using (var memoryStream = new MemoryStream(compressed))
+        {
+            using (
+                var bzip2Stream = new BZip2Stream(
+                    memoryStream,
+                    SharpCompress.Compressors.CompressionMode.Decompress,
+                    false
+                )
+            )
+            {
+                decompressed = new byte[largeData.Length];
+                var totalRead = 0;
+                int bytesRead;
+                var buffer = new byte[4096];
+                while ((bytesRead = await bzip2Stream.ReadAsync(buffer, 0, buffer.Length)) > 0)
+                {
+                    Array.Copy(buffer, 0, decompressed, totalRead, bytesRead);
+                    totalRead += bytesRead;
+                }
+            }
+        }
+
+        // Verify
+        Assert.Equal(largeData, decompressed);
+    }
+}