[56-74] MaybeSourceSpec (#6604)

* [56-74] `MaybeSourceSpec`

* Changes to `async`

Author: eaba

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

@@ -29,10 +29,9 @@ public MaybeSourceSpec(ITestOutputHelper output) : base(output)
         }
 
         [Fact(DisplayName = "The Maybe Source must complete materialized promise with None when stream cancels")]
-        public void CompleteMaterializedPromiseWithNoneWhenCancelled()
+        public async Task CompleteMaterializedPromiseWithNoneWhenCancelled()
         {
-            this.AssertAllStagesStopped(() =>
-            {
+            await this.AssertAllStagesStoppedAsync(async() => {
                 var neverSource = Source.Maybe<int>();
                 var pubSink = Sink.AsPublisher<int>(false);
 
@@ -42,11 +41,11 @@ public void CompleteMaterializedPromiseWithNoneWhenCancelled()
 
                 var c = this.CreateManualSubscriberProbe<int>();
                 neverPub.Subscribe(c);
-                var subs = c.ExpectSubscription();
+                var subs = await c.ExpectSubscriptionAsync();
 
                 subs.Request(1000);
-                c.ExpectNoMsg(100.Milliseconds());
-                
+                await c.ExpectNoMsgAsync(100.Milliseconds());
+
                 subs.Cancel();
 
                 tcs.Task.Wait(3.Seconds()).Should().BeTrue();
@@ -55,39 +54,39 @@ public void CompleteMaterializedPromiseWithNoneWhenCancelled()
         }
 
         [Fact(DisplayName = "The Maybe Source must complete materialized promise with 0 when stream cancels with a failure cause")]
-        public void CompleteMaterializedTaskWithNoneWhenStreamCancelsWithFailure()
+        public async Task CompleteMaterializedTaskWithNoneWhenStreamCancelsWithFailure()
         {
-            this.AssertAllStagesStopped(() =>
-            {
-                var (tcs, killSwitch) = Source.Maybe<int>()
-                    .ViaMaterialized(KillSwitches.Single<int>(), Keep.Both)
-                    .To(Sink.Ignore<int>())
-                    .Run(_materializer);
+            await this.AssertAllStagesStoppedAsync(() => {
+                var (tcs, killSwitch) = Source.Maybe<int>()                                                                             
+                .ViaMaterialized(KillSwitches.Single<int>(), Keep.Both)                                                                             
+                .To(Sink.Ignore<int>())                                                                             
+                .Run(_materializer);
 
                 var boom = new TestException("Boom");
                 killSwitch.Abort(boom);
                 // Could make sense to fail it with the propagated exception instead but that breaks
                 // the assumptions in the CoupledTerminationFlowSpec
                 tcs.Task.Wait(3.Seconds()).Should().BeTrue();
                 tcs.Task.Result.Should().Be(0);
+                return Task.CompletedTask;
             }, _materializer);
         }
 
         [Fact(DisplayName = "The Maybe Source must allow external triggering of empty completion")]
-        public void AllowExternalTriggeringOfEmptyCompletion()
+        public async Task AllowExternalTriggeringOfEmptyCompletion()
         {
-            this.AssertAllStagesStopped(() =>
-            {
+            await this.AssertAllStagesStoppedAsync(() => {
                 var neverSource = Source.Maybe<int>().Where(_ => false);
                 var counterSink = Sink.Aggregate<int, int>(0, (acc, _) => acc + 1);
                 var (neverPromise, counterFuture) = neverSource
                     .ToMaterialized(counterSink, Keep.Both)
                     .Run(_materializer);
-            
+
                 // external cancellation
                 neverPromise.TrySetResult(0).Should().BeTrue();
                 counterFuture.Wait(3.Seconds()).Should().BeTrue();
                 counterFuture.Result.Should().Be(0);
+                return Task.CompletedTask;
             }, _materializer);
         }
 
@@ -96,54 +95,52 @@ public void AllowExternalTriggeringOfEmptyCompletion()
         [Fact(
             DisplayName = "The Maybe Source must allow external triggering of empty completion when there was no demand",
             Skip = "Not working, check Maybe<T> source.")]
-        public void AllowExternalTriggerOfEmptyCompletionWhenNoDemand()
+        public async Task AllowExternalTriggerOfEmptyCompletionWhenNoDemand()
         {
-            this.AssertAllStagesStopped(() =>
-            {
+            await this.AssertAllStagesStoppedAsync(async() => {
                 var probe = this.CreateSubscriberProbe<int>();
                 var promise = Source
                     .Maybe<int>()
                     .To(Sink.FromSubscriber(probe))
                     .Run(_materializer);
-            
+
                 // external cancellation
-                probe.EnsureSubscription();
+                await probe.EnsureSubscriptionAsync();
                 promise.TrySetResult(0).Should().BeTrue();
-                probe.ExpectComplete();
+                await probe.ExpectCompleteAsync();
             }, _materializer);
         }
 
         [Fact(DisplayName = "The Maybe Source must allow external triggering of non-empty completion")]
-        public void AllowExternalTriggerNonEmptyCompletion()
+        public async Task AllowExternalTriggerNonEmptyCompletion()
         {
-            this.AssertAllStagesStopped(() =>
-            {
+            await this.AssertAllStagesStoppedAsync(() => {
                 var neverSource = Source.Maybe<int>();
                 var counterSink = Sink.First<int>();
 
                 var (neverPromise, counterFuture) = neverSource
                     .ToMaterialized(counterSink, Keep.Both)
                     .Run(_materializer);
-            
+
                 // external cancellation
                 neverPromise.TrySetResult(6).Should().BeTrue();
                 counterFuture.Wait(3.Seconds()).Should().BeTrue();
                 counterFuture.Result.Should().Be(6);
+                return Task.CompletedTask;
             }, _materializer);
         }
 
         [Fact(DisplayName = "The Maybe Source must allow external triggering of onError")]
-        public void AllowExternalTriggerOnError()
+        public async Task AllowExternalTriggerOnError()
         {
-            this.AssertAllStagesStopped(() =>
-            {
+            await this.AssertAllStagesStoppedAsync(() => {
                 var neverSource = Source.Maybe<int>();
                 var counterSink = Sink.Aggregate<int, int>(0, (acc, _) => acc + 1);
 
                 var (neverPromise, counterFuture) = neverSource
                     .ToMaterialized(counterSink, Keep.Both)
                     .Run(_materializer);
-            
+
                 // external cancellation
                 neverPromise.TrySetException(new TestException("Boom")).Should().BeTrue();
 
@@ -152,6 +149,7 @@ public void AllowExternalTriggerOnError()
                     .WithInnerException<AggregateException>()
                     .WithInnerException<TestException>()
                     .WithMessage("Boom");
+                return Task.CompletedTask;
             }, _materializer);
         }