Fixed #255

Author: sakno

src/DotNext.IO/IO/SparseStream.cs

@@ -1,39 +1,40 @@
 using System.Runtime.CompilerServices;
+using System.Runtime.InteropServices;
 
 namespace DotNext.IO;
 
+using Buffers;
+
 /// <summary>
 /// Represents multiple streams as a single stream.
 /// </summary>
 /// <remarks>
 /// The stream is available for read-only operations.
 /// </remarks>
-internal sealed class SparseStream : Stream, IFlushable
+internal abstract class SparseStream : Stream, IFlushable
 {
-    private readonly IEnumerator<Stream> enumerator;
-    private bool streamAvailable;
-
-    /// <summary>
-    /// Initializes a new sparse stream.
-    /// </summary>
-    /// <param name="streams">A collection of readable streams.</param>
-    public SparseStream(IEnumerable<Stream> streams)
+    private int runningIndex;
+    
+    protected abstract ReadOnlySpan<Stream> Streams { get; }
+
+    private Stream? Current
     {
-        enumerator = streams.GetEnumerator();
-        streamAvailable = enumerator.MoveNext();
-    }
+        get
+        {
+            var streams = Streams;
 
-    [MethodImpl(MethodImplOptions.AggressiveInlining)]
-    private void MoveToNextStream() => streamAvailable = enumerator.MoveNext();
+            return (uint)runningIndex < (uint)streams.Length ? streams[runningIndex] : null;
+        }
+    }
 
     /// <inheritdoc />
-    public override int ReadByte()
+    public sealed override int ReadByte()
     {
         var result = -1;
 
-        for (; streamAvailable; MoveToNextStream())
+        for (; Current is { } current; runningIndex++)
         {
-            result = enumerator.Current.ReadByte();
+            result = current.ReadByte();
 
             if (result >= 0)
                 break;
@@ -43,12 +44,12 @@ public override int ReadByte()
     }
 
     /// <inheritdoc />
-    public override int Read(Span<byte> buffer)
+    public sealed override int Read(Span<byte> buffer)
     {
         int count;
-        for (count = 0; streamAvailable; MoveToNextStream())
+        for (count = 0; Current is { } current; runningIndex++)
         {
-            count = enumerator.Current.Read(buffer);
+            count = current.Read(buffer);
 
             if (count > 0)
                 break;
@@ -58,7 +59,7 @@ public override int Read(Span<byte> buffer)
     }
 
     /// <inheritdoc />
-    public override int Read(byte[] buffer, int offset, int count)
+    public sealed override int Read(byte[] buffer, int offset, int count)
     {
         ValidateBufferArguments(buffer, offset, count);
 
@@ -67,12 +68,12 @@ public override int Read(byte[] buffer, int offset, int count)
 
     /// <inheritdoc />
     [AsyncMethodBuilder(typeof(PoolingAsyncValueTaskMethodBuilder<>))]
-    public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken token = default)
+    public sealed override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken token = default)
     {
         int count;
-        for (count = 0; streamAvailable; MoveToNextStream())
+        for (count = 0; Current is { } current; runningIndex++)
         {
-            count = await enumerator.Current.ReadAsync(buffer, token).ConfigureAwait(false);
+            count = await current.ReadAsync(buffer, token).ConfigureAwait(false);
 
             if (count > 0)
                 break;
@@ -82,95 +83,118 @@ public override async ValueTask<int> ReadAsync(Memory<byte> buffer, Cancellation
     }
 
     /// <inheritdoc />
-    public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken token)
+    public sealed override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken token)
         => ReadAsync(buffer.AsMemory(offset, count), token).AsTask();
 
+    public sealed override IAsyncResult BeginRead(byte[] buffer, int offset, int count, AsyncCallback? callback, object? state)
+        => TaskToAsyncResult.Begin(ReadAsync(buffer, offset, count), callback, state);
+
+    public sealed override int EndRead(IAsyncResult asyncResult)
+        => TaskToAsyncResult.End<int>(asyncResult);
+
     /// <inheritdoc />
-    public override void CopyTo(Stream destination, int bufferSize)
+    public sealed override void CopyTo(Stream destination, int bufferSize)
     {
         ValidateCopyToArguments(destination, bufferSize);
 
-        for (; streamAvailable; MoveToNextStream())
-            enumerator.Current.CopyTo(destination, bufferSize);
+        for (; Current is { } current; runningIndex++)
+            current.CopyTo(destination, bufferSize);
     }
 
     /// <inheritdoc />
-    public override async Task CopyToAsync(Stream destination, int bufferSize, CancellationToken token)
+    public sealed override async Task CopyToAsync(Stream destination, int bufferSize, CancellationToken token)
     {
         ValidateCopyToArguments(destination, bufferSize);
 
-        for (; streamAvailable; MoveToNextStream())
-            await enumerator.Current.CopyToAsync(destination, bufferSize, token).ConfigureAwait(false);
+        for (; Current is { } current; runningIndex++)
+            await current.CopyToAsync(destination, bufferSize, token).ConfigureAwait(false);
     }
 
     /// <inheritdoc />
-    public override bool CanRead => true;
+    public sealed override bool CanRead => true;
 
     /// <inheritdoc />
-    public override bool CanWrite => false;
+    public sealed override bool CanWrite => false;
 
     /// <inheritdoc />
-    public override bool CanSeek => false;
+    public sealed override bool CanSeek => false;
 
     /// <inheritdoc />
-    public override long Position
+    public sealed override long Position
     {
         get => throw new NotSupportedException();
         set => throw new NotSupportedException();
     }
 
     /// <inheritdoc cref="Stream.Flush"/>
-    public override void Flush()
-    {
-        if (streamAvailable)
-            enumerator.Current.Flush();
-    }
+    public sealed override void Flush() => Current?.Flush();
 
     /// <inheritdoc cref="Stream.FlushAsync(CancellationToken)"/>
-    public override Task FlushAsync(CancellationToken token)
-        => streamAvailable ? enumerator.Current.FlushAsync(token) : Task.CompletedTask;
+    public sealed override Task FlushAsync(CancellationToken token)
+        => Current?.FlushAsync(token) ?? Task.CompletedTask;
 
     /// <inheritdoc />
-    public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
+    public sealed override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
 
     /// <inheritdoc />
-    public override long Length => throw new NotSupportedException();
+    public sealed override long Length
+    {
+        get
+        {
+            var length = 0L;
+
+            foreach (var stream in Streams)
+            {
+                length += stream.Length;
+            }
+
+            return length;
+        }
+    }
 
     /// <inheritdoc />
-    public override void SetLength(long value) => throw new NotSupportedException();
+    public sealed override void SetLength(long value) => throw new NotSupportedException();
 
     /// <inheritdoc/>
-    public override void WriteByte(byte value) => throw new NotSupportedException();
+    public sealed override void WriteByte(byte value) => throw new NotSupportedException();
 
     /// <inheritdoc/>
-    public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();
+    public sealed override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();
 
     /// <inheritdoc/>
-    public override void Write(ReadOnlySpan<byte> buffer) => throw new NotSupportedException();
+    public sealed override void Write(ReadOnlySpan<byte> buffer) => throw new NotSupportedException();
 
     /// <inheritdoc/>
-    public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken token) => Task.FromException(new NotSupportedException());
+    public sealed override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken token) => Task.FromException(new NotSupportedException());
 
     /// <inheritdoc/>
-    public override ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
+    public sealed override ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
         => ValueTask.FromException(new NotSupportedException());
 
     /// <inheritdoc/>
-    public override IAsyncResult BeginWrite(byte[] buffer, int offset, int count, AsyncCallback? callback, object? state)
+    public sealed override IAsyncResult BeginWrite(byte[] buffer, int offset, int count, AsyncCallback? callback, object? state)
         => throw new NotSupportedException();
 
     /// <inheritdoc/>
-    public override void EndWrite(IAsyncResult asyncResult) => throw new InvalidOperationException();
+    public sealed override void EndWrite(IAsyncResult asyncResult) => throw new InvalidOperationException();
+}
+
+internal sealed class SparseStream<T>(T streams) : SparseStream
+    where T : struct, ITuple
+{
+    protected override ReadOnlySpan<Stream> Streams
+        => MemoryMarshal.CreateReadOnlySpan(in Unsafe.As<T, Stream>(ref Unsafe.AsRef(in streams)), streams.Length);
+}
+
+internal sealed class UnboundedSparseStream(ReadOnlySpan<Stream> streams) : SparseStream
+{
+    private MemoryOwner<Stream> streams = streams.Copy();
+
+    protected override ReadOnlySpan<Stream> Streams => streams.Span;
 
-    /// <inheritdoc />
     protected override void Dispose(bool disposing)
     {
-        if (disposing)
-        {
-            enumerator.Dispose();
-        }
-
-        streamAvailable = false;
+        streams.Dispose();
         base.Dispose(disposing);
     }
 }

src/DotNext.IO/IO/StreamExtensions.cs

@@ -3,6 +3,8 @@
 
 namespace DotNext.IO;
 
+using Collections.Generic;
+
 /// <summary>
 /// Represents high-level read/write methods for the stream.
 /// </summary>
@@ -24,8 +26,41 @@ internal static void ThrowIfEmpty<T>(in Memory<T> buffer, [CallerArgumentExpress
     /// <param name="stream">The stream to combine.</param>
     /// <param name="others">A collection of streams.</param>
     /// <returns>An object that represents multiple streams as one logical stream.</returns>
-    public static Stream Combine(this Stream stream, ReadOnlySpan<Stream> others)
-        => others is { Length: > 0 } ? new SparseStream([stream, .. others]) : stream;
+    public static Stream Combine(this Stream stream, ReadOnlySpan<Stream> others) // TODO: Use params in future
+        => others switch
+        {
+            [] => stream,
+            [var s] => new SparseStream<(Stream, Stream)>((stream, s)),
+            [var s1, var s2] => new SparseStream<(Stream, Stream, Stream)>((stream, s1, s2)),
+            [var s1, var s2, var s3] => new SparseStream<(Stream, Stream, Stream, Stream)>((stream, s1, s2, s3)),
+            [var s1, var s2, var s3, var s4] => new SparseStream<(Stream, Stream, Stream, Stream, Stream)>((stream, s1, s2, s3, s4)),
+            [var s1, var s2, var s3, var s4, var s5] => new SparseStream<(Stream, Stream, Stream, Stream, Stream, Stream)>((stream, s1, s2, s3, s4,
+                s5)),
+            _ => new UnboundedSparseStream(others.ToArray()),
+        };
+
+    /// <summary>
+    /// Combines multiple readable streams.
+    /// </summary>
+    /// <param name="streams">A collection of streams.</param>
+    /// <returns>An object that represents multiple streams as one logical stream.</returns>
+    /// <exception cref="ArgumentException"><paramref name="streams"/> is empty.</exception>
+    public static Stream Combine(this ReadOnlySpan<Stream> streams)
+        => streams is [var first, .. var rest]
+            ? Combine(first, rest)
+            : throw new ArgumentException(ExceptionMessages.BufferTooSmall, nameof(streams));
+
+    /// <summary>
+    /// Combines multiple readable streams.
+    /// </summary>
+    /// <param name="streams">A collection of streams.</param>
+    /// <returns>An object that represents multiple streams as one logical stream.</returns>
+    /// <exception cref="ArgumentException"><paramref name="streams"/> is empty.</exception>
+    public static Stream Combine(this IEnumerable<Stream> streams)
+    {
+        using var buffer = streams.Copy();
+        return Combine(buffer.Span);
+    }
 
     /// <summary>
     /// Creates a stream for the specified file handle.

src/DotNext.Tests/IO/StreamExtensionsTests.cs

@@ -1,14 +1,10 @@
 using System.Buffers;
-using System.Globalization;
 using System.Text;
-using static System.Globalization.CultureInfo;
-using DateTimeStyles = System.Globalization.DateTimeStyles;
 
 namespace DotNext.IO;
 
-using DotNext.Buffers.Binary;
-using Net.Cluster;
-using Text;
+using Buffers.Binary;
+using Collections.Generic;
 
 public sealed class StreamExtensionsTests : Test
 {
@@ -156,7 +152,7 @@ public static async Task CombineStreamsAsync()
     {
         await using var ms1 = new MemoryStream([1, 2, 3]);
         await using var ms2 = new MemoryStream([4, 5, 6]);
-        await using var combined = ms1.Combine([ms2]);
+        await using var combined = StreamExtensions.Combine([ms1, ms2]);
 
         var buffer = new byte[6];
         await combined.ReadExactlyAsync(buffer);
@@ -169,7 +165,7 @@ public static void CopyCombinedStreams()
     {
         using var ms1 = new MemoryStream([1, 2, 3]);
         using var ms2 = new MemoryStream([4, 5, 6]);
-        using var combined = ms1.Combine([ms2]);
+        using var combined = List.Singleton(ms1).Append(ms2).Combine();
         using var result = new MemoryStream();
 
         combined.CopyTo(result, 128);