[4-74]FlowAggregateAsyncSpec: Change AssertAllStagesStopped to AssertAllStagesStoppedAsync (#6543)

eaba · Aaronontheweb · web-flow · commit 678429697849 · 2023-03-23T10:49:10.000-05:00
Co-authored-by: Aaron Stannard &lt;aaron@petabridge.com&gt;
diff --git a/src/core/Akka.Streams.Tests/Dsl/FlowAggregateAsyncSpec.cs b/src/core/Akka.Streams.Tests/Dsl/FlowAggregateAsyncSpec.cs
@@ -67,9 +67,9 @@ await this.AssertAllStagesStoppedAsync(async() =>
         }
 
         [Fact]
-        public void A_AggregateAsync_must_work_when_using_Sink_AggregateAsync()
+        public async Task A_AggregateAsync_must_work_when_using_Sink_AggregateAsync()
         {
-            this.AssertAllStagesStopped(async() =>
+            await this.AssertAllStagesStoppedAsync(async() =>
             {
                 var task = InputSource.RunWith(AggregateSink, Materializer);
                 var complete = await task.ShouldCompleteWithin(3.Seconds());
@@ -78,10 +78,10 @@ public void A_AggregateAsync_must_work_when_using_Sink_AggregateAsync()
         }
 
         [LocalFact(SkipLocal = "Racy on Azure DevOps")]
-        public void A_AggregateAsync_must_work_when_using_Flow_AggregateAsync()
+        public async Task A_AggregateAsync_must_work_when_using_Flow_AggregateAsync()
         {
             var flowTimeout = TimeSpan.FromMilliseconds(FlowDelayInMs*Input.Count()) + TimeSpan.FromSeconds(3);
-            this.AssertAllStagesStopped(async() =>
+            await this.AssertAllStagesStoppedAsync(async() =>
             {
                 var task = InputSource.Via(AggregateFlow).RunWith(Sink.First<int>(), Materializer);
                 var complete = await task.ShouldCompleteWithin(flowTimeout);
@@ -90,9 +90,9 @@ public void A_AggregateAsync_must_work_when_using_Flow_AggregateAsync()
         }
 
         [Fact]
-        public void A_AggregateAsync_must_work_when_using_Source_AggregateAsync_and_Flow_AggregateAsync_and_Sink_AggregateAsync()
+        public async Task A_AggregateAsync_must_work_when_using_Source_AggregateAsync_and_Flow_AggregateAsync_and_Sink_AggregateAsync()
         {
-            this.AssertAllStagesStopped(async() =>
+            await this.AssertAllStagesStoppedAsync(async() =>
             {
                 var task = AggregateSource.Via(AggregateFlow).RunWith(AggregateSink, Materializer);
                 var complete = await task.ShouldCompleteWithin(3.Seconds());
@@ -101,10 +101,9 @@ public void A_AggregateAsync_must_work_when_using_Source_AggregateAsync_and_Flow
         }
 
         [Fact]
-        public void A_AggregateAsync_must_propagate_an_error()
+        public async Task A_AggregateAsync_must_propagate_an_error()
         {
-            this.AssertAllStagesStopped(() =>
-            {
+            await this.AssertAllStagesStoppedAsync(() => {
                 var error = new TestException("buh");
                 var future = InputSource.Select(x =>
                 {
@@ -116,14 +115,14 @@ public void A_AggregateAsync_must_propagate_an_error()
                 future.Invoking(f => f.Wait(TimeSpan.FromSeconds(3)))
                     .Should().Throw<TestException>()
                     .And.Should().Be(error);
+                return Task.CompletedTask;
             }, Materializer);
         }
 
         [Fact]
-        public void A_AggregateAsync_must_complete_task_with_failure_when_Aggregating_functions_throws()
+        public async Task A_AggregateAsync_must_complete_task_with_failure_when_Aggregating_functions_throws()
         {
-            this.AssertAllStagesStopped(() =>
-            {
+            await this.AssertAllStagesStoppedAsync(() => {
                 var error = new TestException("buh");
                 var future = InputSource.RunAggregateAsync(0, (x, y) =>
                 {
@@ -139,6 +138,7 @@ public void A_AggregateAsync_must_complete_task_with_failure_when_Aggregating_fu
                 future.Invoking(f => f.Wait(TimeSpan.FromSeconds(3)))
                     .Should().Throw<TestException>()
                     .And.Should().Be(error);
+                return Task.CompletedTask;
             }, Materializer);
         }
 
@@ -159,10 +159,9 @@ public void A_AggregateAsync_must_not_blow_up_with_high_request_count()
         }
 
         [Fact]
-        public void A_AggregateAsync_must_signal_task_failure()
+        public async Task A_AggregateAsync_must_signal_task_failure()
         {
-            this.AssertAllStagesStopped(() =>
-            {
+            await this.AssertAllStagesStoppedAsync(() => {
                 var probe = this.CreateSubscriberProbe<int>();
                 Source.From(Enumerable.Range(1, 5)).AggregateAsync(0, (_, n) => Task.Run(() =>
                 {
@@ -174,14 +173,14 @@ public void A_AggregateAsync_must_signal_task_failure()
                 var subscription = probe.ExpectSubscription();
                 subscription.Request(100);
                 probe.ExpectError().InnerException.Message.Should().Be("err1");
+                return Task.CompletedTask;
             }, Materializer);
         }
 
         [Fact]
-        public void A_AggregateAsync_must_signal_error_from_AggregateAsync()
+        public async Task A_AggregateAsync_must_signal_error_from_AggregateAsync()
         {
-            this.AssertAllStagesStopped(() =>
-            {
+            await this.AssertAllStagesStoppedAsync(() => {
                 var c = this.CreateManualSubscriberProbe<int>();
 
                 Source.From(Enumerable.Range(1, 5)).AggregateAsync(0, (_, n) =>
@@ -195,14 +194,14 @@ public void A_AggregateAsync_must_signal_error_from_AggregateAsync()
                 var subscription = c.ExpectSubscription();
                 subscription.Request(10);
                 c.ExpectError().Message.Should().Be("err2");
+                return Task.CompletedTask;
             }, Materializer);
         }
 
         [Fact]
-        public void A_AggregateAsync_must_resume_after_task_failure()
+        public async Task A_AggregateAsync_must_resume_after_task_failure()
         {
-            this.AssertAllStagesStopped(() =>
-            {
+            await this.AssertAllStagesStoppedAsync(() => {
                 var probe = this.CreateSubscriberProbe<(int, int)>();
                 Source.From(Enumerable.Range(1, 5)).AggregateAsync((0, 1), (t, n) =>
                     {
@@ -213,7 +212,7 @@ public void A_AggregateAsync_must_resume_after_task_failure()
                             if (n == 3)
                                 throw new Exception("err3");
 
-                            return (n, i + res*n);
+                            return (n, i + res * n);
                         });
                     })
                     .WithAttributes(ActorAttributes.CreateSupervisionStrategy(Deciders.ResumingDecider))
@@ -224,14 +223,14 @@ public void A_AggregateAsync_must_resume_after_task_failure()
                 subscription.Request(10);
                 probe.ExpectNext((5, 74));
                 probe.ExpectComplete();
+                return Task.CompletedTask;
             }, Materializer);
         }
 
         [Fact]
-        public void A_AggregateAsync_must_restart_after_task_failure()
+        public async Task A_AggregateAsync_must_restart_after_task_failure()
         {
-            this.AssertAllStagesStopped(() =>
-            {
+            await this.AssertAllStagesStoppedAsync(() => {
                 var probe = this.CreateSubscriberProbe<(int, int)>();
                 Source.From(Enumerable.Range(1, 5)).AggregateAsync((0, 1), (t, n) =>
                 {
@@ -253,6 +252,7 @@ public void A_AggregateAsync_must_restart_after_task_failure()
                 subscription.Request(10);
                 probe.ExpectNext((5, 24));
                 probe.ExpectComplete();
+                return Task.CompletedTask;
             }, Materializer);
         }
 
@@ -399,10 +399,9 @@ public void A_AggregateAsync_must_restart_when_task_is_completed_with_null()
         }
 
         [Fact]
-        public void A_AggregateAsync_must_handle_cancel_properly()
+        public async Task A_AggregateAsync_must_handle_cancel_properly()
         {
-            this.AssertAllStagesStopped(() =>
-            {
+            await this.AssertAllStagesStoppedAsync(() => {
                 var pub = this.CreateManualPublisherProbe<int>();
                 var sub = this.CreateSubscriberProbe<int>();
 
@@ -414,15 +413,16 @@ public void A_AggregateAsync_must_handle_cancel_properly()
                 upstream.ExpectRequest();
 
                 sub.ExpectSubscription().Cancel();
-                
+
                 upstream.ExpectCancellation();
+                return Task.CompletedTask;
             }, Materializer);
         }
 
         [Fact]
-        public void A_AggregateAsync_must_complete_task_and_return_zero_given_an_empty_stream()
+        public async Task A_AggregateAsync_must_complete_task_and_return_zero_given_an_empty_stream()
         {
-            this.AssertAllStagesStopped(async() =>
+            await this.AssertAllStagesStoppedAsync(async() =>
             {
                 var task = Source.From(Enumerable.Empty<int>())
                     .RunAggregateAsync(0, (acc, element) => Task.FromResult(acc + element), Materializer);
@@ -432,9 +432,9 @@ public void A_AggregateAsync_must_complete_task_and_return_zero_given_an_empty_s
         }
 
         [Fact]
-        public void A_AggregateAsync_must_complete_task_and_return_zero_and_item_given_a_stream_of_one_item()
+        public async Task A_AggregateAsync_must_complete_task_and_return_zero_and_item_given_a_stream_of_one_item()
         {
-            this.AssertAllStagesStopped(async() =>
+            await this.AssertAllStagesStoppedAsync(async() =>
             {
                 var task = Source.Single(100)
                     .RunAggregateAsync(5, (acc, element) => Task.FromResult(acc + element), Materializer);

Original file line number	Diff line number	Diff line change
`@@ -67,9 +67,9 @@ await this.AssertAllStagesStoppedAsync(async() =>`
`67`	`67`	`}`
`68`	`68`
`69`	`69`	`[Fact]`
`70`		`- public void A_AggregateAsync_must_work_when_using_Sink_AggregateAsync()`
	`70`	`+ public async Task A_AggregateAsync_must_work_when_using_Sink_AggregateAsync()`
`71`	`71`	`{`
`72`		`- this.AssertAllStagesStopped(async() =>`
	`72`	`+ await this.AssertAllStagesStoppedAsync(async() =>`
`73`	`73`	`{`
`74`	`74`	`var task = InputSource.RunWith(AggregateSink, Materializer);`
`75`	`75`	`var complete = await task.ShouldCompleteWithin(3.Seconds());`
`@@ -78,10 +78,10 @@ public void A_AggregateAsync_must_work_when_using_Sink_AggregateAsync()`
`78`	`78`	`}`
`79`	`79`
`80`	`80`	`[LocalFact(SkipLocal = "Racy on Azure DevOps")]`
`81`		`- public void A_AggregateAsync_must_work_when_using_Flow_AggregateAsync()`
	`81`	`+ public async Task A_AggregateAsync_must_work_when_using_Flow_AggregateAsync()`
`82`	`82`	`{`
`83`	`83`	`var flowTimeout = TimeSpan.FromMilliseconds(FlowDelayInMs*Input.Count()) + TimeSpan.FromSeconds(3);`
`84`		`- this.AssertAllStagesStopped(async() =>`
	`84`	`+ await this.AssertAllStagesStoppedAsync(async() =>`
`85`	`85`	`{`
`86`	`86`	`var task = InputSource.Via(AggregateFlow).RunWith(Sink.First<int>(), Materializer);`
`87`	`87`	`var complete = await task.ShouldCompleteWithin(flowTimeout);`
`@@ -90,9 +90,9 @@ public void A_AggregateAsync_must_work_when_using_Flow_AggregateAsync()`
`90`	`90`	`}`
`91`	`91`
`92`	`92`	`[Fact]`
`93`		`- public void A_AggregateAsync_must_work_when_using_Source_AggregateAsync_and_Flow_AggregateAsync_and_Sink_AggregateAsync()`
	`93`	`+ public async Task A_AggregateAsync_must_work_when_using_Source_AggregateAsync_and_Flow_AggregateAsync_and_Sink_AggregateAsync()`
`94`	`94`	`{`
`95`		`- this.AssertAllStagesStopped(async() =>`
	`95`	`+ await this.AssertAllStagesStoppedAsync(async() =>`
`96`	`96`	`{`
`97`	`97`	`var task = AggregateSource.Via(AggregateFlow).RunWith(AggregateSink, Materializer);`
`98`	`98`	`var complete = await task.ShouldCompleteWithin(3.Seconds());`
`@@ -101,10 +101,9 @@ public void A_AggregateAsync_must_work_when_using_Source_AggregateAsync_and_Flow`
`101`	`101`	`}`
`102`	`102`
`103`	`103`	`[Fact]`
`104`		`- public void A_AggregateAsync_must_propagate_an_error()`
	`104`	`+ public async Task A_AggregateAsync_must_propagate_an_error()`
`105`	`105`	`{`
`106`		`- this.AssertAllStagesStopped(() =>`
`107`		`- {`
	`106`	`+ await this.AssertAllStagesStoppedAsync(() => {`
`108`	`107`	`var error = new TestException("buh");`
`109`	`108`	`var future = InputSource.Select(x =>`
`110`	`109`	`{`
`@@ -116,14 +115,14 @@ public void A_AggregateAsync_must_propagate_an_error()`
`116`	`115`	`future.Invoking(f => f.Wait(TimeSpan.FromSeconds(3)))`
`117`	`116`	`.Should().Throw<TestException>()`
`118`	`117`	`.And.Should().Be(error);`
	`118`	`+ return Task.CompletedTask;`
`119`	`119`	`}, Materializer);`
`120`	`120`	`}`
`121`	`121`
`122`	`122`	`[Fact]`
`123`		`- public void A_AggregateAsync_must_complete_task_with_failure_when_Aggregating_functions_throws()`
	`123`	`+ public async Task A_AggregateAsync_must_complete_task_with_failure_when_Aggregating_functions_throws()`
`124`	`124`	`{`
`125`		`- this.AssertAllStagesStopped(() =>`
`126`		`- {`
	`125`	`+ await this.AssertAllStagesStoppedAsync(() => {`
`127`	`126`	`var error = new TestException("buh");`
`128`	`127`	`var future = InputSource.RunAggregateAsync(0, (x, y) =>`
`129`	`128`	`{`
`@@ -139,6 +138,7 @@ public void A_AggregateAsync_must_complete_task_with_failure_when_Aggregating_fu`
`139`	`138`	`future.Invoking(f => f.Wait(TimeSpan.FromSeconds(3)))`
`140`	`139`	`.Should().Throw<TestException>()`
`141`	`140`	`.And.Should().Be(error);`
	`141`	`+ return Task.CompletedTask;`
`142`	`142`	`}, Materializer);`
`143`	`143`	`}`
`144`	`144`
`@@ -159,10 +159,9 @@ public void A_AggregateAsync_must_not_blow_up_with_high_request_count()`
`159`	`159`	`}`
`160`	`160`
`161`	`161`	`[Fact]`
`162`		`- public void A_AggregateAsync_must_signal_task_failure()`
	`162`	`+ public async Task A_AggregateAsync_must_signal_task_failure()`
`163`	`163`	`{`
`164`		`- this.AssertAllStagesStopped(() =>`
`165`		`- {`
	`164`	`+ await this.AssertAllStagesStoppedAsync(() => {`
`166`	`165`	`var probe = this.CreateSubscriberProbe<int>();`
`167`	`166`	`Source.From(Enumerable.Range(1, 5)).AggregateAsync(0, (_, n) => Task.Run(() =>`
`168`	`167`	`{`
`@@ -174,14 +173,14 @@ public void A_AggregateAsync_must_signal_task_failure()`
`174`	`173`	`var subscription = probe.ExpectSubscription();`
`175`	`174`	`subscription.Request(100);`
`176`	`175`	`probe.ExpectError().InnerException.Message.Should().Be("err1");`
	`176`	`+ return Task.CompletedTask;`
`177`	`177`	`}, Materializer);`
`178`	`178`	`}`
`179`	`179`
`180`	`180`	`[Fact]`
`181`		`- public void A_AggregateAsync_must_signal_error_from_AggregateAsync()`
	`181`	`+ public async Task A_AggregateAsync_must_signal_error_from_AggregateAsync()`
`182`	`182`	`{`
`183`		`- this.AssertAllStagesStopped(() =>`
`184`		`- {`
	`183`	`+ await this.AssertAllStagesStoppedAsync(() => {`
`185`	`184`	`var c = this.CreateManualSubscriberProbe<int>();`
`186`	`185`
`187`	`186`	`Source.From(Enumerable.Range(1, 5)).AggregateAsync(0, (_, n) =>`
`@@ -195,14 +194,14 @@ public void A_AggregateAsync_must_signal_error_from_AggregateAsync()`
`195`	`194`	`var subscription = c.ExpectSubscription();`
`196`	`195`	`subscription.Request(10);`
`197`	`196`	`c.ExpectError().Message.Should().Be("err2");`
	`197`	`+ return Task.CompletedTask;`
`198`	`198`	`}, Materializer);`
`199`	`199`	`}`
`200`	`200`
`201`	`201`	`[Fact]`
`202`		`- public void A_AggregateAsync_must_resume_after_task_failure()`
	`202`	`+ public async Task A_AggregateAsync_must_resume_after_task_failure()`
`203`	`203`	`{`
`204`		`- this.AssertAllStagesStopped(() =>`
`205`		`- {`
	`204`	`+ await this.AssertAllStagesStoppedAsync(() => {`
`206`	`205`	`var probe = this.CreateSubscriberProbe<(int, int)>();`
`207`	`206`	`Source.From(Enumerable.Range(1, 5)).AggregateAsync((0, 1), (t, n) =>`
`208`	`207`	`{`
`@@ -213,7 +212,7 @@ public void A_AggregateAsync_must_resume_after_task_failure()`
`213`	`212`	`if (n == 3)`
`214`	`213`	`throw new Exception("err3");`
`215`	`214`
`216`		`- return (n, i + res*n);`
	`215`	`+ return (n, i + res * n);`
`217`	`216`	`});`
`218`	`217`	`})`
`219`	`218`	`.WithAttributes(ActorAttributes.CreateSupervisionStrategy(Deciders.ResumingDecider))`
`@@ -224,14 +223,14 @@ public void A_AggregateAsync_must_resume_after_task_failure()`
`224`	`223`	`subscription.Request(10);`
`225`	`224`	`probe.ExpectNext((5, 74));`
`226`	`225`	`probe.ExpectComplete();`
	`226`	`+ return Task.CompletedTask;`
`227`	`227`	`}, Materializer);`
`228`	`228`	`}`
`229`	`229`
`230`	`230`	`[Fact]`
`231`		`- public void A_AggregateAsync_must_restart_after_task_failure()`
	`231`	`+ public async Task A_AggregateAsync_must_restart_after_task_failure()`
`232`	`232`	`{`
`233`		`- this.AssertAllStagesStopped(() =>`
`234`		`- {`
	`233`	`+ await this.AssertAllStagesStoppedAsync(() => {`
`235`	`234`	`var probe = this.CreateSubscriberProbe<(int, int)>();`
`236`	`235`	`Source.From(Enumerable.Range(1, 5)).AggregateAsync((0, 1), (t, n) =>`
`237`	`236`	`{`
`@@ -253,6 +252,7 @@ public void A_AggregateAsync_must_restart_after_task_failure()`
`253`	`252`	`subscription.Request(10);`
`254`	`253`	`probe.ExpectNext((5, 24));`
`255`	`254`	`probe.ExpectComplete();`
	`255`	`+ return Task.CompletedTask;`
`256`	`256`	`}, Materializer);`
`257`	`257`	`}`
`258`	`258`
`@@ -399,10 +399,9 @@ public void A_AggregateAsync_must_restart_when_task_is_completed_with_null()`
`399`	`399`	`}`
`400`	`400`
`401`	`401`	`[Fact]`
`402`		`- public void A_AggregateAsync_must_handle_cancel_properly()`
	`402`	`+ public async Task A_AggregateAsync_must_handle_cancel_properly()`
`403`	`403`	`{`
`404`		`- this.AssertAllStagesStopped(() =>`
`405`		`- {`
	`404`	`+ await this.AssertAllStagesStoppedAsync(() => {`
`406`	`405`	`var pub = this.CreateManualPublisherProbe<int>();`
`407`	`406`	`var sub = this.CreateSubscriberProbe<int>();`
`408`	`407`
`@@ -414,15 +413,16 @@ public void A_AggregateAsync_must_handle_cancel_properly()`
`414`	`413`	`upstream.ExpectRequest();`
`415`	`414`
`416`	`415`	`sub.ExpectSubscription().Cancel();`
`417`		`-`
	`416`	`+`
`418`	`417`	`upstream.ExpectCancellation();`
	`418`	`+ return Task.CompletedTask;`
`419`	`419`	`}, Materializer);`
`420`	`420`	`}`
`421`	`421`
`422`	`422`	`[Fact]`
`423`		`- public void A_AggregateAsync_must_complete_task_and_return_zero_given_an_empty_stream()`
	`423`	`+ public async Task A_AggregateAsync_must_complete_task_and_return_zero_given_an_empty_stream()`
`424`	`424`	`{`
`425`		`- this.AssertAllStagesStopped(async() =>`
	`425`	`+ await this.AssertAllStagesStoppedAsync(async() =>`
`426`	`426`	`{`
`427`	`427`	`var task = Source.From(Enumerable.Empty<int>())`
`428`	`428`	`.RunAggregateAsync(0, (acc, element) => Task.FromResult(acc + element), Materializer);`
`@@ -432,9 +432,9 @@ public void A_AggregateAsync_must_complete_task_and_return_zero_given_an_empty_s`
`432`	`432`	`}`
`433`	`433`
`434`	`434`	`[Fact]`
`435`		`- public void A_AggregateAsync_must_complete_task_and_return_zero_and_item_given_a_stream_of_one_item()`
	`435`	`+ public async Task A_AggregateAsync_must_complete_task_and_return_zero_and_item_given_a_stream_of_one_item()`
`436`	`436`	`{`
`437`		`- this.AssertAllStagesStopped(async() =>`
	`437`	`+ await this.AssertAllStagesStoppedAsync(async() =>`
`438`	`438`	`{`
`439`	`439`	`var task = Source.Single(100)`
`440`	`440`	`.RunAggregateAsync(5, (acc, element) => Task.FromResult(acc + element), Materializer);`