[46-74] GraphStageTimersSpec (#6593)

* [46-74] `GraphStageTimersSpec`

* Changes to `async` TestKit

Author: eaba

src/core/Akka.Streams.Tests/Dsl/GraphStageTimersSpec.cs

@@ -50,60 +50,59 @@ private SideChannel SetupIsolatedStage()
         public async Task GraphStage_timer_support_must_receive_single_shot_timer()
         {
             var driver = SetupIsolatedStage();
-            await AwaitAssertAsync(() =>
+            await AwaitAssertAsync(async() =>
             {
                 driver.Tell(TestSingleTimer.Instance);
-                ExpectMsg(new Tick(1), TimeSpan.FromSeconds(10));
-                ExpectNoMsg(TimeSpan.FromSeconds(1));
+                await ExpectMsgAsync(new Tick(1), TimeSpan.FromSeconds(10));
+                await ExpectNoMsgAsync(TimeSpan.FromSeconds(1));
             });
             driver.StopStage();
         }
 
         [LocalFact(SkipLocal = "Racy on Azure DevOps")]
-        public void GraphStage_timer_support_must_resubmit_single_shot_timer()
+        public async Task GraphStage_timer_support_must_resubmit_single_shot_timer()
         {
             var driver = SetupIsolatedStage();
-            Within(TimeSpan.FromSeconds(2.5), () =>
+            await WithinAsync(TimeSpan.FromSeconds(2.5), async() =>
             {
-                Within(TimeSpan.FromMilliseconds(500), TimeSpan.FromSeconds(1), () =>
+                await WithinAsync(TimeSpan.FromMilliseconds(500), TimeSpan.FromSeconds(1), async() =>
                 {
                     driver.Tell(TestSingleTimerResubmit.Instance);
-                    ExpectMsg(new Tick(1));
+                    await ExpectMsgAsync(new Tick(1));
                 });
-                Within(TimeSpan.FromSeconds(1), () => ExpectMsg(new Tick(2)));
+                await WithinAsync(TimeSpan.FromSeconds(1), async() => await ExpectMsgAsync(new Tick(2)));
 
-                ExpectNoMsg(TimeSpan.FromSeconds(1));
+                await ExpectNoMsgAsync(TimeSpan.FromSeconds(1));
             });
             driver.StopStage();
         }
 
         [Fact]
-        public void GraphStage_timer_support_must_correctly_cancel_a_named_timer()
+        public async Task GraphStage_timer_support_must_correctly_cancel_a_named_timer()
         {
             var driver = SetupIsolatedStage();
             driver.Tell(TestCancelTimer.Instance);
-            Within(TimeSpan.FromMilliseconds(5000), () => ExpectMsg<TestCancelTimerAck>());
-            Within(TimeSpan.FromMilliseconds(200), TimeSpan.FromSeconds(3000), () => ExpectMsg(new Tick(1)));
-            ExpectNoMsg(TimeSpan.FromSeconds(1));
+            await WithinAsync(TimeSpan.FromMilliseconds(5000), async() => await ExpectMsgAsync<TestCancelTimerAck>());
+            await WithinAsync(TimeSpan.FromMilliseconds(200), TimeSpan.FromSeconds(3000), async() => await ExpectMsgAsync(new Tick(1)));
+            await ExpectNoMsgAsync(TimeSpan.FromSeconds(1));
             driver.StopStage();
         }
 
         [Fact]
-        public void GraphStage_timer_support_must_receive_and_cancel_a_repeated_timer()
+        public async Task GraphStage_timer_support_must_receive_and_cancel_a_repeated_timer()
         {
             var driver = SetupIsolatedStage();
             driver.Tell(TestRepeatedTimer.Instance);
             var seq = ReceiveWhile(TimeSpan.FromSeconds(30), o => (Tick)o, msgs: 5);
             seq.Should().HaveCount(5);
-            ExpectNoMsg(TimeSpan.FromSeconds(1));
+            await ExpectNoMsgAsync(TimeSpan.FromSeconds(1));
             driver.StopStage();
         }
 
         [Fact]
-        public void GraphStage_timer_support_must_produce_scheduled_ticks_as_expected()
+        public async Task GraphStage_timer_support_must_produce_scheduled_ticks_as_expected()
         {
-            this.AssertAllStagesStopped(() =>
-            {
+            await this.AssertAllStagesStoppedAsync(async() => {
                 var upstream = this.CreatePublisherProbe<int>();
                 var downstream = this.CreateSubscriberProbe<int>();
 
@@ -112,20 +111,19 @@ public void GraphStage_timer_support_must_produce_scheduled_ticks_as_expected()
                     .RunWith(Sink.FromSubscriber(downstream), Materializer);
 
                 downstream.Request(5);
-                downstream.ExpectNext( 1, 2, 3);
+                downstream.ExpectNext(1, 2, 3);
 
-                downstream.ExpectNoMsg(TimeSpan.FromSeconds(1));
+                await downstream.ExpectNoMsgAsync(TimeSpan.FromSeconds(1));
 
-                upstream.SendComplete();
-                downstream.ExpectComplete();
+                await upstream.SendCompleteAsync();
+                await downstream.ExpectCompleteAsync();
             }, Materializer);
         }
 
         [Fact]
-        public void GraphStage_timer_support_must_propagate_error_if_OnTimer_throws_an_Exception()
+        public async Task GraphStage_timer_support_must_propagate_error_if_OnTimer_throws_an_Exception()
         {
-            this.AssertAllStagesStopped(() =>
-            {
+            await this.AssertAllStagesStoppedAsync(() => {
                 var exception = new TestException("Expected exception to the rule");
                 var upstream = this.CreatePublisherProbe<int>();
                 var downstream = this.CreateSubscriberProbe<int>();
@@ -136,6 +134,7 @@ public void GraphStage_timer_support_must_propagate_error_if_OnTimer_throws_an_E
 
                 downstream.Request(1);
                 downstream.ExpectError().Should().Be(exception);
+                return Task.CompletedTask;
             }, Materializer);
         }