Have MergedLinesEnumerable implement IAsyncEnumerable<string>

MedallionShell.Tests/MedallionShell.Tests.csproj

@@ -6,9 +6,9 @@
     <IsPackable>false</IsPackable>
     <LangVersion>Latest</LangVersion>
     <Nullable>enable</Nullable>
-	  <GenerateDocumentationFile>True</GenerateDocumentationFile>
-	  <CodeAnalysisRuleSet>..\stylecop.analyzers.ruleset</CodeAnalysisRuleSet>
-	  <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
+    <GenerateDocumentationFile>True</GenerateDocumentationFile>
+    <CodeAnalysisRuleSet>..\stylecop.analyzers.ruleset</CodeAnalysisRuleSet>
+    <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
     <NoWarn>1591</NoWarn>
     <RootNamespace>Medallion.Shell.Tests</RootNamespace>
   </PropertyGroup>
@@ -17,15 +17,16 @@
     <PackageReference Include="nunit" Version="3.13.3" />
     <PackageReference Include="NUnit3TestAdapter" Version="4.4.0-beta.1" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.4.1" />
-	  <PackageReference Include="Moq" Version="4.7.63" />
-	  <PackageReference Include="StyleCop.Analyzers" Version="1.2.0-beta.435">
+    <PackageReference Include="Moq" Version="4.7.63" />
+    <PackageReference Include="Microsoft.Bcl.AsyncInterfaces" Version="8.0.0" />
+    <PackageReference Include="StyleCop.Analyzers" Version="1.2.0-beta.435">
       <PrivateAssets>All</PrivateAssets>
     </PackageReference>
   </ItemGroup>
 
   <ItemGroup>
     <ProjectReference Include="..\MedallionShell\MedallionShell.csproj" />
-	  <ProjectReference Include="..\SampleCommand\SampleCommand.csproj" />
+    <ProjectReference Include="..\SampleCommand\SampleCommand.csproj" />
   </ItemGroup>
 
   <ItemGroup Condition="'$(TargetFramework)' == 'net462'">

MedallionShell.Tests/Streams/AsyncEnumerableAdapter.cs

@@ -0,0 +1,39 @@
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Medallion.Shell.Tests.Streams;
+
+public class AsyncEnumerableAdapter : IAsyncEnumerable<string>
+{
+    private readonly IEnumerable<string> strings;
+
+    public AsyncEnumerableAdapter(IEnumerable<string> strings)
+    {
+        this.strings = strings;
+    }
+
+    public IAsyncEnumerator<string> GetAsyncEnumerator(CancellationToken cancellationToken = default) =>
+        // this does not allow consuming the same IEnumerable twice
+        new AsyncEnumeratorAdapter(this.strings.GetEnumerator());
+
+    private class AsyncEnumeratorAdapter : IAsyncEnumerator<string>
+    {
+        private readonly IEnumerator<string> enumerator;
+
+        public AsyncEnumeratorAdapter(IEnumerator<string> enumerator)
+        {
+            this.enumerator = enumerator;
+        }
+
+        public string Current => this.enumerator.Current;
+
+        public ValueTask DisposeAsync()
+        {
+            this.enumerator.Dispose();
+            return default;
+        }
+
+        public ValueTask<bool> MoveNextAsync() => new(this.enumerator.MoveNext());
+    }
+}

MedallionShell.Tests/Streams/MergedLinesEnumerableTestAsync.cs

@@ -0,0 +1,13 @@
+#if NETCOREAPP3_0_OR_GREATER || NETSTANDARD2_0_OR_GREATER
+using System.Collections.Generic;
+using System.IO;
+using Medallion.Shell.Streams;
+
+namespace Medallion.Shell.Tests.Streams;
+
+public class MergedLinesEnumerableTestAsync : MergedLinesEnumerableTestBase
+{
+    protected override IAsyncEnumerable<string> Create(TextReader reader1, TextReader reader2) =>
+        new MergedLinesEnumerable(reader1, reader2);
+}
+#endif

MedallionShell.Tests/Streams/MergedLinesEnumerableTestBase.cs

@@ -0,0 +1,142 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Medallion.Shell.Streams;
+using Moq;
+using NUnit.Framework;
+
+namespace Medallion.Shell.Tests.Streams;
+
+public abstract class MergedLinesEnumerableTestBase
+{
+    protected abstract IAsyncEnumerable<string> Create(TextReader reader1, TextReader reader2);
+
+    [Test]
+    public async Task TestOneIsEmpty()
+    {
+        var empty1 = new StringReader(string.Empty);
+        var nonEmpty1 = new StringReader("abc\r\ndef\r\nghi\r\njkl");
+
+        var enumerable1 = this.Create(empty1, nonEmpty1);
+        var list1 = await enumerable1.ToListAsync();
+        list1.SequenceEqual(new[] { "abc", "def", "ghi", "jkl" })
+            .ShouldEqual(true, string.Join(", ", list1));
+
+        var empty2 = new StringReader(string.Empty);
+        var nonEmpty2 = new StringReader("a\nbb\nccc\n");
+        var enumerable2 = this.Create(nonEmpty2, empty2);
+        var list2 = await enumerable2.ToListAsync();
+        list2.SequenceEqual(new[] { "a", "bb", "ccc" })
+            .ShouldEqual(true, string.Join(", ", list2));
+    }
+
+    [Test]
+    public async Task TestBothAreEmpty()
+    {
+        var list = await this.Create(new StringReader(string.Empty), new StringReader(string.Empty)).ToListAsync();
+        list.Count.ShouldEqual(0, string.Join(", ", list));
+    }
+
+    [Test]
+    public async Task TestBothArePopulatedEqualSizes()
+    {
+        var list = await this.Create(
+                new StringReader("a\nbb\nccc"),
+                new StringReader("1\r\n22\r\n333")
+            )
+            .ToListAsync();
+        string.Join(", ", list).ShouldEqual("a, 1, bb, 22, ccc, 333");
+    }
+
+    [Test]
+    public async Task TestBothArePopulatedDifferenceSizes()
+    {
+        var lines1 = string.Join("\n", new[] { "x", "y", "z" });
+        var lines2 = string.Join("\n", new[] { "1", "2", "3", "4", "5" });
+
+        var list1 = await this.Create(new StringReader(lines1), new StringReader(lines2))
+            .ToListAsync();
+        string.Join(", ", list1).ShouldEqual("x, 1, y, 2, z, 3, 4, 5");
+
+        var list2 = await this.Create(new StringReader(lines2), new StringReader(lines1))
+            .ToListAsync();
+        string.Join(", ", list2).ShouldEqual("1, x, 2, y, 3, z, 4, 5");
+    }
+
+    [Test]
+    public void TestConsumeTwice()
+    {
+        var asyncEnumerable = this.Create(new StringReader("a"), new StringReader("b"));
+        asyncEnumerable.GetAsyncEnumerator();
+        Assert.Throws<InvalidOperationException>(() => asyncEnumerable.GetAsyncEnumerator());
+    }
+
+    [Test]
+    public void TestOneThrows()
+    {
+        void TestOneThrows(bool reverse)
+        {
+            var reader1 = new StringReader("a\nb\nc");
+            var count = 0;
+            var mockReader = new Mock<TextReader>(MockBehavior.Strict);
+            mockReader.Setup(r => r.ReadLineAsync())
+                .ReturnsAsync(() => ++count < 3 ? "LINE" : throw new TimeZoneNotFoundException());
+
+            Assert.ThrowsAsync<TimeZoneNotFoundException>(
+                async () => await this.Create(
+                        reverse ? mockReader.Object : reader1,
+                        reverse ? reader1 : mockReader.Object
+                    ).ToListAsync()
+            );
+        }
+
+        TestOneThrows(reverse: false);
+        TestOneThrows(reverse: true);
+    }
+
+    [Test, Timeout(5_000)] // something's wrong if it's taking more than 5 seconds
+    public async Task FuzzTest()
+    {
+        Pipe pipe1 = new(), pipe2 = new();
+
+        var asyncEnumerable = this.Create(new StreamReader(pipe1.OutputStream), new StreamReader(pipe2.OutputStream));
+
+        var strings1 = Enumerable.Range(0, 2000).Select(_ => Guid.NewGuid().ToString()).ToArray();
+        var strings2 = Enumerable.Range(0, 2300).Select(_ => Guid.NewGuid().ToString()).ToArray();
+
+        static void WriteStrings(IReadOnlyList<string> strings, Pipe pipe)
+        {
+            SpinWait spinWait = default;
+            Random random = new(Guid.NewGuid().GetHashCode());
+            using StreamWriter writer = new(pipe.InputStream);
+            foreach (var line in strings)
+            {
+                if (random.Next(10) == 1)
+                {
+                    spinWait.SpinOnce();
+                }
+
+                writer.WriteLine(line);
+            }
+        }
+
+        var task1 = Task.Run(() => WriteStrings(strings1, pipe1));
+        var task2 = Task.Run(() => WriteStrings(strings2, pipe2));
+        Task.WaitAll(task1, task2); // need to dispose the writer to end the stream
+
+        CollectionAssert.AreEquivalent(strings1.Concat(strings2), await asyncEnumerable.ToListAsync());
+    }
+}
+
+public static class AsyncEnumerableExtensions
+{
+    public static async Task<List<string>> ToListAsync(this IAsyncEnumerable<string> strings)
+    {
+        List<string> result = [];
+        await foreach (var item in strings) { result.Add(item); }
+        return result;
+    }
+}

MedallionShell.Tests/Streams/MergedLinesEnumerableTestSync.cs

@@ -0,0 +1,11 @@
+using System.Collections.Generic;
+using System.IO;
+using Medallion.Shell.Streams;
+
+namespace Medallion.Shell.Tests.Streams;
+
+public class MergedLinesEnumerableTestSync : MergedLinesEnumerableTestBase
+{
+    protected override IAsyncEnumerable<string> Create(TextReader reader1, TextReader reader2) =>
+        new AsyncEnumerableAdapter(new MergedLinesEnumerable(reader1, reader2));
+}

MedallionShell/MedallionShell.csproj

@@ -45,7 +45,7 @@
   </PropertyGroup>
 
   <ItemGroup Condition="'$(TargetFramework)' == 'netstandard1.3'">
-	  <PackageReference Include="System.Diagnostics.Process" version="4.3.0" />
+    <PackageReference Include="System.Diagnostics.Process" version="4.3.0" />
   </ItemGroup>
   <ItemGroup Condition="'$(TargetFramework)' == 'net46' or '$(TargetFramework)' == 'net45'">
     <PackageReference Include="System.Runtime.InteropServices.RuntimeInformation" Version="4.3.0" />
@@ -80,4 +80,7 @@
       <LogicalName>MedallionShell.ProcessSignaler.exe</LogicalName>
     </EmbeddedResource>
   </ItemGroup>
+  <ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.0'">
+    <PackageReference Include="Microsoft.Bcl.AsyncInterfaces" Version="8.0.0" />
+  </ItemGroup>
 </Project>