async lzma

src/SharpCompress/Compressors/LZMA/AesDecoderStream.cs

@@ -2,6 +2,8 @@
 using System.IO;
 using System.Security.Cryptography;
 using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
 using SharpCompress.Compressors.LZMA.Utilites;
 using SharpCompress.IO;
 
@@ -283,5 +285,70 @@ private int HandleUnderflow(byte[] buffer, int offset, int count)
         return count;
     }
 
+    public override async Task<int> ReadAsync(
+        byte[] buffer,
+        int offset,
+        int count,
+        CancellationToken cancellationToken = default
+    )
+    {
+        if (count == 0 || mWritten == mLimit)
+        {
+            return 0;
+        }
+
+        if (mUnderflow > 0)
+        {
+            return HandleUnderflow(buffer, offset, count);
+        }
+
+        // Need at least 16 bytes to proceed.
+        if (mEnding - mOffset < 16)
+        {
+            Buffer.BlockCopy(mBuffer, mOffset, mBuffer, 0, mEnding - mOffset);
+            mEnding -= mOffset;
+            mOffset = 0;
+
+            do
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+                var read = await mStream
+                    .ReadAsync(mBuffer, mEnding, mBuffer.Length - mEnding, cancellationToken)
+                    .ConfigureAwait(false);
+                if (read == 0)
+                {
+                    // We are not done decoding and have less than 16 bytes.
+                    throw new EndOfStreamException();
+                }
+
+                mEnding += read;
+            } while (mEnding - mOffset < 16);
+        }
+
+        // We shouldn't return more data than we are limited to.
+        if (count > mLimit - mWritten)
+        {
+            count = (int)(mLimit - mWritten);
+        }
+
+        // We cannot transform less than 16 bytes into the target buffer,
+        // but we also cannot return zero, so we need to handle this.
+        if (count < 16)
+        {
+            return HandleUnderflow(buffer, offset, count);
+        }
+
+        if (count > mEnding - mOffset)
+        {
+            count = mEnding - mOffset;
+        }
+
+        // Otherwise we transform directly into the target buffer.
+        var processed = mDecoder.TransformBlock(mBuffer, mOffset, count & ~15, buffer, offset);
+        mOffset += processed;
+        mWritten += processed;
+        return processed;
+    }
+
     #endregion
 }

src/SharpCompress/Compressors/LZMA/Bcj2DecoderStream.cs

@@ -1,6 +1,8 @@
 using System;
 using System.Collections.Generic;
 using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
 using SharpCompress.IO;
 
 namespace SharpCompress.Compressors.LZMA;
@@ -191,6 +193,18 @@ public override int Read(byte[] buffer, int offset, int count)
         return count;
     }
 
+    public override Task<int> ReadAsync(
+        byte[] buffer,
+        int offset,
+        int count,
+        CancellationToken cancellationToken = default
+    )
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+        // Bcj2DecoderStream uses complex state machine with multiple streams
+        return Task.FromResult(Read(buffer, offset, count));
+    }
+
     public override int ReadByte()
     {
         if (_mFinished)

src/SharpCompress/Compressors/LZMA/LZ/LzOutWindow.cs

@@ -3,6 +3,8 @@
 using System;
 using System.Buffers;
 using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
 
 namespace SharpCompress.Compressors.LZMA.LZ;
 
@@ -104,6 +106,27 @@ private void Flush()
         _streamPos = _pos;
     }
 
+    private async Task FlushAsync(CancellationToken cancellationToken = default)
+    {
+        if (_stream is null)
+        {
+            return;
+        }
+        var size = _pos - _streamPos;
+        if (size == 0)
+        {
+            return;
+        }
+        await _stream
+            .WriteAsync(_buffer, _streamPos, size, cancellationToken)
+            .ConfigureAwait(false);
+        if (_pos >= _windowSize)
+        {
+            _pos = 0;
+        }
+        _streamPos = _pos;
+    }
+
     public void CopyPending()
     {
         if (_pendingLen < 1)
@@ -207,6 +230,44 @@ public int CopyStream(Stream stream, int len)
         return len - size;
     }
 
+    public async Task<int> CopyStreamAsync(
+        Stream stream,
+        int len,
+        CancellationToken cancellationToken = default
+    )
+    {
+        var size = len;
+        while (size > 0 && _pos < _windowSize && _total < _limit)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+
+            var curSize = _windowSize - _pos;
+            if (curSize > _limit - _total)
+            {
+                curSize = (int)(_limit - _total);
+            }
+            if (curSize > size)
+            {
+                curSize = size;
+            }
+            var numReadBytes = await stream
+                .ReadAsync(_buffer, _pos, curSize, cancellationToken)
+                .ConfigureAwait(false);
+            if (numReadBytes == 0)
+            {
+                throw new DataErrorException();
+            }
+            size -= numReadBytes;
+            _pos += numReadBytes;
+            _total += numReadBytes;
+            if (_pos >= _windowSize)
+            {
+                await FlushAsync(cancellationToken).ConfigureAwait(false);
+            }
+        }
+        return len - size;
+    }
+
     public void SetLimit(long size) => _limit = _total + size;
 
     public bool HasSpace => _pos < _windowSize && _total < _limit;

src/SharpCompress/Compressors/LZMA/LZipStream.cs

@@ -1,6 +1,8 @@
 using System;
 using System.Buffers.Binary;
 using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
 using SharpCompress.Common;
 using SharpCompress.Crypto;
 using SharpCompress.IO;
@@ -157,6 +159,11 @@ public override int Read(byte[] buffer, int offset, int count) =>
 
 #if !NETFRAMEWORK && !NETSTANDARD2_0
 
+    public override ValueTask<int> ReadAsync(
+        Memory<byte> buffer,
+        CancellationToken cancellationToken = default
+    ) => _stream.ReadAsync(buffer, cancellationToken);
+
     public override int Read(Span<byte> buffer) => _stream.Read(buffer);
 
     public override void Write(ReadOnlySpan<byte> buffer)
@@ -179,6 +186,25 @@ public override void WriteByte(byte value)
         ++_writeCount;
     }
 
+    public override Task<int> ReadAsync(
+        byte[] buffer,
+        int offset,
+        int count,
+        CancellationToken cancellationToken = default
+    ) => _stream.ReadAsync(buffer, offset, count, cancellationToken);
+
+    public override async Task WriteAsync(
+        byte[] buffer,
+        int offset,
+        int count,
+        CancellationToken cancellationToken = default
+    )
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+        await _stream.WriteAsync(buffer, offset, count, cancellationToken);
+        _writeCount += count;
+    }
+
     #endregion
 
     /// <summary>