Added support for copying readable streams to writable streams and ad…

…ded test coverage.

Schema Tests/TextSchemaTestUtil.cs

@@ -1,7 +1,5 @@
 using System.IO;
 
-using schema.text.reader;
-
 using TextReader = schema.text.reader.TextReader;
 
 namespace schema.text {

Schema Tests/util/SubstreamSharpTests.cs → Schema Tests/util/streams/RangedReadableSubstreamTests.cs

@@ -1,9 +1,7 @@
 using NUnit.Framework;
 
-using schema.util.streams;
-
-namespace schema.lib.SubstreamSharp {
-  public class RangedSubstreamTests {
+namespace schema.util.streams {
+  public class RangedReadableSubstreamTests {
     [Test]
     public void TestFullSubstream() {
       var s = new ReadableStream(new byte[] { 1, 2, 3, 4, 5, 6, 7 });

Schema Tests/util/streams/ReadableStreamTests.cs

@@ -0,0 +1,64 @@
+using System;
+using System.IO;
+
+using NUnit.Framework;
+
+namespace schema.util.streams {
+  public class ReadableStreamTests {
+    [Test]
+    public void TestPosition() {
+      var ms = new MemoryStream(new byte[] { 1, 2, 3 });
+      var rs = new ReadableStream(ms);
+
+      Assert.AreEqual(0, ms.Position);
+      Assert.AreEqual(0, rs.Position);
+
+      rs.Position = 2;
+      Assert.AreEqual(2, ms.Position);
+      Assert.AreEqual(2, rs.Position);
+    }
+
+    [Test]
+    public void TestLength() {
+      var ms = new MemoryStream(new byte[] { 1, 2, 3 });
+      var rs = new ReadableStream(ms);
+
+      Assert.AreEqual(3, ms.Length);
+      Assert.AreEqual(3, rs.Length);
+    }
+
+    [Test]
+    public void TestReadByte() {
+      var ms = new MemoryStream(new byte[] { 1, 2, 3 });
+      var rs = new ReadableStream(ms);
+
+      Assert.AreEqual(0, ms.Position);
+      Assert.AreEqual(0, rs.Position);
+
+      Assert.AreEqual(1, rs.ReadByte());
+      Assert.AreEqual(1, ms.Position);
+      Assert.AreEqual(1, rs.Position);
+
+      Assert.AreEqual(2, rs.ReadByte());
+      Assert.AreEqual(2, ms.Position);
+      Assert.AreEqual(2, rs.Position);
+    }
+
+    [Test]
+    public void TestReadSpan() {
+      var ms = new MemoryStream(new byte[] { 1, 2, 3 });
+      var rs = new ReadableStream(ms);
+
+      Assert.AreEqual(0, ms.Position);
+      Assert.AreEqual(0, rs.Position);
+
+      Span<byte> span = stackalloc byte[5];
+
+      Assert.AreEqual(3, rs.Read(span));
+      Assert.AreEqual(3, ms.Position);
+      Assert.AreEqual(3, rs.Position);
+
+      CollectionAssert.AreEqual(new[] { 1, 2, 3, 0, 0 }, span.ToArray());
+    }
+  }
+}

Schema Tests/util/streams/WritableStreamTests.cs

@@ -0,0 +1,107 @@
+using System;
+using System.IO;
+using System.Linq;
+
+using NUnit.Framework;
+
+namespace schema.util.streams {
+  public class WritableStreamTests {
+    [Test]
+    public void TestPosition() {
+      var ms = new MemoryStream(new byte[] { 1, 2, 3 });
+      var ws = new WritableStream(ms);
+
+      Assert.AreEqual(0, ms.Position);
+      Assert.AreEqual(0, ws.Position);
+
+      ws.Position = 2;
+      Assert.AreEqual(2, ms.Position);
+      Assert.AreEqual(2, ws.Position);
+    }
+
+    [Test]
+    public void TestLength() {
+      var ms = new MemoryStream(new byte[] { 1, 2, 3 });
+      var ws = new WritableStream(ms);
+
+      Assert.AreEqual(3, ms.Length);
+      Assert.AreEqual(3, ws.Length);
+    }
+
+    [Test]
+    public void TestWriteByte() {
+      var data = new byte[] { 1, 2, 3 };
+      var ms = new MemoryStream(data);
+      var ws = new WritableStream(ms);
+
+      Assert.AreEqual(0, ms.Position);
+      Assert.AreEqual(0, ws.Position);
+
+      ws.WriteByte(5);
+      Assert.AreEqual(5, data[0]);
+      Assert.AreEqual(1, ms.Position);
+      Assert.AreEqual(1, ws.Position);
+
+      ws.WriteByte(6);
+      Assert.AreEqual(6, data[1]);
+      Assert.AreEqual(2, ms.Position);
+      Assert.AreEqual(2, ws.Position);
+    }
+
+    [Test]
+    public void TestWriteSpan() {
+      var ms = new MemoryStream();
+      var ws = new WritableStream(ms);
+
+      Assert.AreEqual(0, ms.Position);
+      Assert.AreEqual(0, ws.Position);
+
+      ReadOnlySpan<byte> span = stackalloc byte[5] { 5, 6, 7, 8, 9 };
+
+      ws.Write(span);
+      Assert.AreEqual(5, ms.Position);
+      Assert.AreEqual(5, ws.Position);
+
+      CollectionAssert.AreEqual(new[] { 5, 6, 7, 8, 9 }, ms.ToArray());
+    }
+
+    [Test]
+    public void TestWriteReadableStream() {
+      var ms = new MemoryStream();
+      var ws = new WritableStream(ms);
+      var rs = new ReadableStream(new byte[] { 5, 6, 7, 8, 9 });
+
+      Assert.AreEqual(0, ms.Position);
+      Assert.AreEqual(0, ws.Position);
+      Assert.AreEqual(0, rs.Position);
+
+      ws.Write(rs);
+      Assert.AreEqual(rs.Length, ms.Position);
+      Assert.AreEqual(rs.Length, ws.Position);
+      Assert.AreEqual(rs.Length, rs.Position);
+
+      CollectionAssert.AreEqual(new[] { 5, 6, 7, 8, 9 }, ms.ToArray());
+    }
+
+    [Test]
+    public void TestWriteLongReadableStream() {
+      var readData =
+          Enumerable.Range(0, 300_000).Select(i => (byte) i).ToArray();
+
+      var ms = new MemoryStream();
+      var ws = new WritableStream(ms);
+      var rs = new ReadableStream(readData);
+
+      Assert.AreEqual(0, ms.Position);
+      Assert.AreEqual(0, ws.Position);
+      Assert.AreEqual(0, rs.Position);
+
+      ws.Write(rs);
+      Assert.AreEqual(readData.Length, ms.Position);
+      Assert.AreEqual(readData.Length, ws.Position);
+      Assert.AreEqual(readData.Length, rs.Position);
+
+      CollectionAssert.AreEqual(readData, ms.ToArray());
+    }
+  }
+}

Schema/src/util/streams/Interfaces.cs

@@ -27,6 +27,7 @@ public interface IReadableStream {
 
   public interface IWritableStream {
     void Write(ReadOnlySpan<byte> src);
+    void Write(IReadableStream readableStream);
   }
 
   public interface ISizedReadableStream : IReadableStream,

Schema/src/util/streams/ReadableStream.cs

@@ -5,27 +5,29 @@
 
 namespace schema.util.streams {
   public class ReadableStream : ISeekableReadableStream {
-    private readonly Stream impl_;
+    internal Stream Impl { get; }
+
+    public static implicit operator ReadableStream(Stream impl) => new(impl);
 
     public ReadableStream(Stream impl) {
       if (!impl.CanRead) {
         throw new ArgumentException(nameof(impl));
       }
 
-      this.impl_ = impl;
+      this.Impl = impl;
     }
 
     public ReadableStream(byte[] impl) : this(new MemoryStream(impl)) { }
 
-    public void Dispose() => this.impl_.Dispose();
+    public void Dispose() => this.Impl.Dispose();
 
     public long Position {
-      get => this.impl_.Position;
-      set => this.impl_.Position = value;
+      get => this.Impl.Position;
+      set => this.Impl.Position = value;
     }
 
-    public long Length => this.impl_.Length;
+    public long Length => this.Impl.Length;
 
-    public int Read(Span<byte> dst) => this.impl_.Read(dst);
+    public int Read(Span<byte> dst) => this.Impl.Read(dst);
   }
 }

Schema/src/util/streams/WritableStream.cs

@@ -5,6 +5,16 @@
 
 namespace schema.util.streams {
   public class WritableStream(Stream impl) : ISeekableWritableStream {
+    /// <summary>
+    ///   (Straight-up copied from the implementation of Stream.CopyTo())
+    ///   We pick a value that is the largest multiple of 4096 that is still smaller than the large object heap threshold (85K).
+    ///   The CopyTo/CopyToAsync buffer is short-lived and is likely to be collected at Gen0, and it offers a significant
+    ///   improvement in Copy performance.
+    /// </summary>
+    private const int DEFAULT_COPY_BUFFER_SIZE = 81920;
+
+    public static implicit operator WritableStream(Stream impl) => new(impl);
+
     public void Dispose() => impl.Dispose();
 
     public long Position {
@@ -15,5 +25,18 @@ public long Position {
     public long Length => impl.Length;
 
     public void Write(ReadOnlySpan<byte> src) => impl.Write(src);
+
+    public void Write(IReadableStream readableStream) {
+      if (readableStream is ReadableStream readableStreamImpl) {
+        readableStreamImpl.Impl.CopyTo(impl);
+        return;
+      }
+
+      Span<byte> buffer = stackalloc byte[DEFAULT_COPY_BUFFER_SIZE];
+      int bytesRead;
+      while ((bytesRead = readableStream.Read(buffer)) != 0) {
+        impl.Write(buffer.Slice(0, bytesRead));
+      }
+    }
   }
 }