Have MergedLinesEnumerable implement IAsyncEnumerable<string>

MedallionShell.Tests/AttachingTests.cs

@@ -106,7 +106,7 @@ public void TestTimeout()
                 "Did not time out"
             );
 
-            Assert.IsInstanceOf<TimeoutException>(exception.GetBaseException());
+            Assert.IsInstanceOf<TimeoutException>(exception!.GetBaseException());
         }
     }
 }

MedallionShell.Tests/GeneralTest.cs

@@ -112,7 +112,7 @@ public void TestExitCode()
 
             var shell = MakeTestShell(o => o.ThrowOnError());
             var ex = Assert.Throws<AggregateException>(() => shell.Run(SampleCommand, "exit", -1).Task.Wait());
-            ex.InnerExceptions.Select(e => e.GetType()).SequenceEqual(new[] { typeof(ErrorExitCodeException) })
+            ex!.InnerExceptions.Select(e => e.GetType()).SequenceEqual(new[] { typeof(ErrorExitCodeException) })
                 .ShouldEqual(true);
 
             shell.Run(SampleCommand, "exit", 0).Task.Wait();
@@ -123,7 +123,7 @@ public void TestThrowOnErrorWithTimeout()
         {
             var command = TestShell.Run(SampleCommand, new object[] { "exit", 1 }, o => o.ThrowOnError().Timeout(TimeSpan.FromDays(1)));
             var ex = Assert.Throws<AggregateException>(() => command.Task.Wait());
-            ex.InnerExceptions.Select(e => e.GetType()).SequenceEqual(new[] { typeof(ErrorExitCodeException) })
+            ex!.InnerExceptions.Select(e => e.GetType()).SequenceEqual(new[] { typeof(ErrorExitCodeException) })
                 .ShouldEqual(true);
         }
 
@@ -132,15 +132,15 @@ public void TestTimeout()
         {
             var willTimeout = TestShell.Run(SampleCommand, new object[] { "sleep", 1000000 }, o => o.Timeout(TimeSpan.FromMilliseconds(200)));
             var ex = Assert.Throws<AggregateException>(() => willTimeout.Task.Wait());
-            Assert.IsInstanceOf<TimeoutException>(ex.InnerException);
+            Assert.IsInstanceOf<TimeoutException>(ex!.InnerException);
         }
 
         [Test]
         public void TestZeroTimeout()
         {
             var willTimeout = TestShell.Run(SampleCommand, new object[] { "sleep", 1000000 }, o => o.Timeout(TimeSpan.Zero));
             var ex = Assert.Throws<AggregateException>(() => willTimeout.Task.Wait());
-            Assert.IsInstanceOf<TimeoutException>(ex.InnerException);
+            Assert.IsInstanceOf<TimeoutException>(ex!.InnerException);
         }
 
         [Test]
@@ -177,7 +177,7 @@ public void TestCancellationCanceledPartway()
             results.Count.ShouldEqual(1);
             cancellationTokenSource.Cancel();
             var aggregateException = Assert.Throws<AggregateException>(() => command.Task.Wait(1000));
-            Assert.IsInstanceOf<TaskCanceledException>(aggregateException.GetBaseException());
+            Assert.IsInstanceOf<TaskCanceledException>(aggregateException!.GetBaseException());
             CollectionAssert.AreEqual(results, new[] { "hello" });
         }
 
@@ -350,7 +350,7 @@ public void TestVersioning()
             var version = typeof(Command).GetTypeInfo().Assembly.GetName().Version.ToString();
             var informationalVersion = (AssemblyInformationalVersionAttribute)typeof(Command).GetTypeInfo().Assembly.GetCustomAttribute(typeof(AssemblyInformationalVersionAttribute));
             Assert.IsNotNull(informationalVersion);
-            version.ShouldEqual(Regex.Replace(informationalVersion.InformationalVersion, "-.*$", string.Empty) + ".0");
+            version.ShouldEqual(Regex.Replace(informationalVersion.InformationalVersion, "[+-].*$", string.Empty) + ".0");
         }
 
         [Test]
@@ -469,11 +469,11 @@ void TestHelper(bool disposeOnExit)
                     if (disposeOnExit)
                     {
                         // invalid due to DisposeOnExit()
-                        Assert.Throws<InvalidOperationException>(() => command1.Process.ToString())
+                        Assert.Throws<InvalidOperationException>(() => command1.Process.ToString())!
                             .Message.ShouldContain("dispose on exit");
-                        Assert.Throws<InvalidOperationException>(() => command2.Processes.Count())
+                        Assert.Throws<InvalidOperationException>(() => command2.Processes.Count())!
                             .Message.ShouldContain("dispose on exit");
-                        Assert.Throws<InvalidOperationException>(() => pipeCommand.Processes.Count())
+                        Assert.Throws<InvalidOperationException>(() => pipeCommand.Processes.Count())!
                             .Message.ShouldContain("dispose on exit");
                     }
                     else

MedallionShell.Tests/IoCommandTest.cs

@@ -19,12 +19,12 @@ public void TestStandardOutCannotBeAccessedAfterRedirectingIt()
             var command = TestShell.Run(SampleCommand, "argecho", "a");
             var ioCommand = command.RedirectTo(output);
 
-            var errorMessage = Assert.Throws<InvalidOperationException>(() => ioCommand.StandardOutput.GetHashCode()).Message;
+            var errorMessage = Assert.Throws<InvalidOperationException>(() => ioCommand.StandardOutput.GetHashCode())!.Message;
             errorMessage.ShouldEqual("StandardOutput is unavailable because it is already being piped to System.Collections.Generic.List`1[System.String]");
 
             Assert.DoesNotThrow(() => command.StandardOutput.GetHashCode());
 
-            Assert.Throws<InvalidOperationException>(() => ioCommand.Result.StandardOutput.GetHashCode())
+            Assert.Throws<InvalidOperationException>(() => ioCommand.Result.StandardOutput.GetHashCode())!
                 .Message
                 .ShouldEqual(errorMessage);
             Assert.Throws<ObjectDisposedException>(() => command.Result.StandardOutput.GetHashCode());
@@ -40,12 +40,12 @@ public void TestStandardErrorCannotBeAccessedAfterRedirectingIt()
             var command = TestShell.Run(SampleCommand, "argecho", "a");
             var ioCommand = command.RedirectStandardErrorTo(output);
 
-            var errorMessage = Assert.Throws<InvalidOperationException>(() => ioCommand.StandardError.GetHashCode()).Message;
+            var errorMessage = Assert.Throws<InvalidOperationException>(() => ioCommand.StandardError.GetHashCode())!.Message;
             errorMessage.ShouldEqual("StandardError is unavailable because it is already being piped to System.Collections.Generic.List`1[System.String]");
 
             Assert.DoesNotThrow(() => command.StandardError.GetHashCode());
 
-            Assert.Throws<InvalidOperationException>(() => ioCommand.Result.StandardError.GetHashCode())
+            Assert.Throws<InvalidOperationException>(() => ioCommand.Result.StandardError.GetHashCode())!
                 .Message
                 .ShouldEqual(errorMessage);
             Assert.Throws<ObjectDisposedException>(() => command.Result.StandardError.GetHashCode());
@@ -60,7 +60,7 @@ public void TestStandardInputCannotBeAccessedAfterRedirectingIt()
             var command = TestShell.Run(SampleCommand, "echo");
             var ioCommand = command.RedirectFrom(new[] { "a" });
 
-            var errorMessage = Assert.Throws<InvalidOperationException>(() => ioCommand.StandardInput.GetHashCode()).Message;
+            var errorMessage = Assert.Throws<InvalidOperationException>(() => ioCommand.StandardInput.GetHashCode())!.Message;
             errorMessage.ShouldEqual("StandardInput is unavailable because it is already being piped from System.String[]");
 
             Assert.DoesNotThrow(() => command.StandardInput.GetHashCode());

MedallionShell.Tests/MedallionShell.Tests.csproj

@@ -8,15 +8,16 @@
 	  <GenerateDocumentationFile>True</GenerateDocumentationFile>
 	  <CodeAnalysisRuleSet>..\stylecop.analyzers.ruleset</CodeAnalysisRuleSet>
 	  <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
-    <NoWarn>1591</NoWarn>
+    <NoWarn>1591,NU1902,NU1903</NoWarn>
+    <CheckEolTargetFramework>false</CheckEolTargetFramework>
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="nunit" Version="3.11.0" />
-    <PackageReference Include="NUnit3TestAdapter" Version="3.11.0" />
+    <PackageReference Include="nunit" Version="3.14.0" />
+    <PackageReference Include="NUnit3TestAdapter" Version="3.17.0" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="15.9.0" />
-	  <PackageReference Include="Moq" Version="4.7.63" />
-	  <PackageReference Include="StyleCop.Analyzers" Version="1.1.1-rc.94">
+	<PackageReference Include="Moq" Version="4.7.63" />
+	<PackageReference Include="StyleCop.Analyzers" Version="1.1.1-rc.94">
       <PrivateAssets>All</PrivateAssets>
     </PackageReference>
   </ItemGroup>
@@ -34,5 +35,14 @@
     <Reference Include="System.Management" />
     <Reference Include="System.ServiceModel" />
   </ItemGroup>
+
+  <ItemGroup Condition="'$(TargetFramework)' == 'netcoreapp2.2'">
+    <PackageReference Include="Microsoft.Bcl.AsyncInterfaces">
+      <Version>8.0.0</Version>
+    </PackageReference>
+    <PackageReference Include="System.Linq.Async">
+      <Version>6.0.1</Version>
+    </PackageReference>
+  </ItemGroup>
 
 </Project>

MedallionShell.Tests/Streams/MergedLinesEnumerableTestAsync.cs

@@ -0,0 +1,136 @@
+#if NETCOREAPP2_2_OR_GREATER
+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 class MergedLinesEnumerableTestAsync
+    {
+        [Test]
+        public async Task TestOneIsEmpty()
+        {
+            var empty1 = new StringReader(string.Empty);
+            var nonEmpty1 = new StringReader("abc\r\ndef\r\nghi\r\njkl");
+
+            IAsyncEnumerable<string> asyncEnumerable1 = new MergedLinesEnumerable(empty1, nonEmpty1);
+            var list1 = await asyncEnumerable1.ToListAsync();
+            list1.SequenceEqual(["abc", "def", "ghi", "jkl"])
+                .ShouldEqual(true, string.Join(", ", list1));
+
+            var empty2 = new StringReader(string.Empty);
+            var nonEmpty2 = new StringReader("a\nbb\nccc\n");
+            var asyncEnumerable2 = new MergedLinesEnumerable(nonEmpty2, empty2);
+            var list2 = await asyncEnumerable2.ToListAsync();
+            list2.SequenceEqual(["a", "bb", "ccc"])
+                .ShouldEqual(true, string.Join(", ", list2));
+        }
+
+        [Test]
+        public async Task TestBothAreEmpty()
+        {
+            var list = await new MergedLinesEnumerable(new StringReader(string.Empty), new StringReader(string.Empty)).ToListAsync();
+            list.Count.ShouldEqual(0, string.Join(", ", list));
+        }
+
+        [Test]
+        public async Task TestBothArePopulatedEqualSizes()
+        {
+            var list = await new MergedLinesEnumerable(
+                    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", ["x", "y", "z"]);
+            var lines2 = string.Join("\n", ["1", "2", "3", "4", "5"]);
+
+            var list1 = await new MergedLinesEnumerable(new StringReader(lines1), new StringReader(lines2))
+                .ToListAsync();
+            string.Join(", ", list1).ShouldEqual("x, 1, y, 2, z, 3, 4, 5");
+
+            var list2 = await new MergedLinesEnumerable(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 = new MergedLinesEnumerable(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 new MergedLinesEnumerable(
+                            reverse ? mockReader.Object : reader1,
+                            reverse ? reader1 : mockReader.Object
+                        )
+                        .ToListAsync()
+                );
+            }
+
+            TestOneThrows(reverse: false);
+            TestOneThrows(reverse: true);
+        }
+
+        [Test]
+        public async Task FuzzTest()
+        {
+            var pipe1 = new Pipe();
+            var pipe2 = new Pipe();
+
+            var asyncEnumerable = new MergedLinesEnumerable(new StreamReader(pipe1.OutputStream), new StreamReader(pipe2.OutputStream));
+
+            var strings1 = await AsyncEnumerable.Range(0, 2000).Select(_ => Guid.NewGuid().ToString()).ToArrayAsync();
+            var strings2 = await AsyncEnumerable.Range(0, 2300).Select(_ => Guid.NewGuid().ToString()).ToArrayAsync();
+
+            static void WriteStrings(IReadOnlyList<string> strings, Pipe pipe)
+            {
+                var spinWait = default(SpinWait);
+                var random = new Random(Guid.NewGuid().GetHashCode());
+                using var writer = new StreamWriter(pipe.InputStream);
+                foreach (var line in strings)
+                {
+                    if (random.Next(4) == 1)
+                    {
+                        spinWait.SpinOnce();
+                    }
+
+                    writer.WriteLine(line);
+                }
+            }
+
+            var task1 = Task.Run(() => WriteStrings(strings1, pipe1));
+            var task2 = Task.Run(() => WriteStrings(strings2, pipe2));
+            var consumeTask = Task.Run(async () => await asyncEnumerable.ToListAsync());
+            Task.WaitAll(task1, task2, consumeTask);
+
+            CollectionAssert.AreEquivalent(strings1.Concat(strings2).ToList(), await consumeTask);
+        }
+    }
+}
+#endif

MedallionShell/MedallionShell.csproj

@@ -66,4 +66,7 @@
       <LogicalName>MedallionShell.ProcessSignaler.exe</LogicalName>
     </EmbeddedResource>
   </ItemGroup>
+  <ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.0'">
+    <PackageReference Include="Microsoft.Bcl.AsyncInterfaces" Version="8.0.0" />
+  </ItemGroup>
 </Project>