[49-74] GraphWireTapSpec (#6596)

* [49-74] `GraphWireTapSpec`

* Changes to `async` TestKit

Author: eaba

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

@@ -7,6 +7,7 @@
 
 using System.Linq;
 using System.Threading;
+using System.Threading.Tasks;
 using Akka.Streams.Dsl;
 using Akka.Streams.TestKit;
 using Akka.TestKit;
@@ -27,76 +28,70 @@ public GraphWireTapSpec(ITestOutputHelper helper)
         }
 
         [Fact]
-        public void A_WireTap_must_broadcast_to_the_tap()
+        public async Task A_WireTap_must_broadcast_to_the_tap()
         {
-            this.AssertAllStagesStopped(() =>
-            {
-                var (tps, mps) = Source.From(Enumerable.Range(1, 2))
-                    .WireTapMaterialized(this.SinkProbe<int>(), Keep.Right)
-                    .ToMaterialized(this.SinkProbe<int>(), Keep.Both)
-                    .Run(Materializer);
+            await this.AssertAllStagesStoppedAsync(async() => {
+                var (tps, mps) = Source.From(Enumerable.Range(1, 2))                                                                             
+                .WireTapMaterialized(this.SinkProbe<int>(), Keep.Right)                                                                             
+                .ToMaterialized(this.SinkProbe<int>(), Keep.Both)                                                                             
+                .Run(Materializer);
 
                 tps.Request(2);
-                mps.RequestNext(1);
-                mps.RequestNext(2);
-                tps.ExpectNext( 1, 2);
-                mps.ExpectComplete();
-                tps.ExpectComplete();
+                await mps.RequestNextAsync(1);
+                await mps.RequestNextAsync(2);
+                tps.ExpectNext(1, 2);
+                await mps.ExpectCompleteAsync();
+                await tps.ExpectCompleteAsync();
             }, Materializer);
         }
 
         [Fact]
-        public void A_WireTap_must_drop_elements_while_the_tap_has_no_demand_buffering_up_to_one_element()
+        public async Task A_WireTap_must_drop_elements_while_the_tap_has_no_demand_buffering_up_to_one_element()
         {
-            this.AssertAllStagesStopped(() =>
-            {
-                var (tps, mps) = Source.From(Enumerable.Range(1, 6))
-                    .WireTapMaterialized(this.SinkProbe<int>(), Keep.Right)
-                    .ToMaterialized(this.SinkProbe<int>(), Keep.Both)
-                    .Run(Materializer);
-
+            await this.AssertAllStagesStoppedAsync(async() => {
+                var (tps, mps) = Source.From(Enumerable.Range(1, 6))                                                                             
+                .WireTapMaterialized(this.SinkProbe<int>(), Keep.Right)                                                                             
+                .ToMaterialized(this.SinkProbe<int>(), Keep.Both)                                                                             
+                .Run(Materializer);
                 mps.Request(3);
-                mps.ExpectNext( 1, 2, 3);
+                mps.ExpectNext(1, 2, 3);
                 tps.Request(4);
-                mps.RequestNext(4);
-                mps.RequestNext(5);
-                mps.RequestNext(6);
-                tps.ExpectNext( 3, 4, 5, 6);
-                mps.ExpectComplete();
-                tps.ExpectComplete();
+                await mps.RequestNextAsync(4);
+                await mps.RequestNextAsync(5);
+                await mps.RequestNextAsync(6);
+                tps.ExpectNext(3, 4, 5, 6);
+                await mps.ExpectCompleteAsync();
+                await tps.ExpectCompleteAsync();
             }, Materializer);
         }
 
         [Fact]
-        public void A_WireTap_must_cancel_if_main_sink_cancels()
+        public async Task A_WireTap_must_cancel_if_main_sink_cancels()
         {
-            this.AssertAllStagesStopped(() =>
-            {
-                var (tps, mps) = Source.From(Enumerable.Range(1, 6))
-                    .WireTapMaterialized(this.SinkProbe<int>(), Keep.Right)
-                    .ToMaterialized(this.SinkProbe<int>(), Keep.Both)
-                    .Run(Materializer);
-                
+            await this.AssertAllStagesStoppedAsync(async() => {
+                var (tps, mps) = Source.From(Enumerable.Range(1, 6))                                                                             
+                .WireTapMaterialized(this.SinkProbe<int>(), Keep.Right)                                                                             
+                .ToMaterialized(this.SinkProbe<int>(), Keep.Both)                                                                             
+                .Run(Materializer);
+
                 tps.Request(6);
                 mps.Cancel();
-                tps.ExpectComplete();
+                await tps.ExpectCompleteAsync();
             }, Materializer);
         }
 
         [Fact]
-        public void A_WireTap_must_continue_if_tap_sink_cancels()
+        public async Task A_WireTap_must_continue_if_tap_sink_cancels()
         {
-            this.AssertAllStagesStopped(() =>
-            {
-                var (tps, mps) = Source.From(Enumerable.Range(1, 6))
-                    .WireTapMaterialized(this.SinkProbe<int>(), Keep.Right)
-                    .ToMaterialized(this.SinkProbe<int>(), Keep.Both)
-                    .Run(Materializer);
-                
+            await this.AssertAllStagesStoppedAsync(async() => {
+                var (tps, mps) = Source.From(Enumerable.Range(1, 6))                                                                             
+                .WireTapMaterialized(this.SinkProbe<int>(), Keep.Right)                                                                             
+                .ToMaterialized(this.SinkProbe<int>(), Keep.Both)                                                                             
+                .Run(Materializer);
                 tps.Cancel();
                 mps.Request(6);
-                mps.ExpectNext( 1, 2, 3, 4, 5, 6);
-                mps.ExpectComplete();
+                mps.ExpectNext(1, 2, 3, 4, 5, 6);
+                await mps.ExpectCompleteAsync();
             }, Materializer);
         }
     }