[Async TestKit] Convert Akka.Streams.Tests to async - FlowFlattenMergeSpec FlowGroupBySpec TimeoutsSpec

build.fsx

@@ -250,8 +250,8 @@ Target "RunTests" (fun _ ->
     let runSingleProject project =
         let arguments =
             match (hasTeamCity) with
-            | true -> (sprintf "test -c Release --no-build --logger:trx --logger:\"console;verbosity=normal\" --framework %s --results-directory \"%s\" -- -parallel none -teamcity" testNetFrameworkVersion outputTests)
-            | false -> (sprintf "test -c Release --no-build --logger:trx --logger:\"console;verbosity=normal\" --framework %s --results-directory \"%s\" -- -parallel none" testNetFrameworkVersion outputTests)
+            | true -> (sprintf "test -c Release --blame-crash --blame-hang-timeout 30s --no-build --logger:trx --logger:\"console;verbosity=normal\" --framework %s --results-directory \"%s\" -- -parallel none -teamcity" testNetFrameworkVersion outputTests)
+            | false -> (sprintf "test -c Release --blame-crash --blame-hang-timeout 30s --no-build --logger:trx --logger:\"console;verbosity=normal\" --framework %s --results-directory \"%s\" -- -parallel none" testNetFrameworkVersion outputTests)
 
         let result = ExecProcess(fun info ->
             info.FileName <- "dotnet"
@@ -280,8 +280,8 @@ Target "RunTestsNetCore" (fun _ ->
         let runSingleProject project =
             let arguments =
                 match (hasTeamCity) with
-                | true -> (sprintf "test -c Release --no-build --logger:trx --logger:\"console;verbosity=normal\" --framework %s --results-directory \"%s\" -- -parallel none -teamcity" testNetCoreVersion outputTests)
-                | false -> (sprintf "test -c Release --no-build --logger:trx --logger:\"console;verbosity=normal\" --framework %s --results-directory \"%s\" -- -parallel none" testNetCoreVersion outputTests)
+                | true -> (sprintf "test -c Release --blame-crash --blame-hang-timeout 30s --no-build --logger:trx --logger:\"console;verbosity=normal\" --framework %s --results-directory \"%s\" -- -parallel none -teamcity" testNetCoreVersion outputTests)
+                | false -> (sprintf "test -c Release --blame-crash --blame-hang-timeout 30s --no-build --logger:trx --logger:\"console;verbosity=normal\" --framework %s --results-directory \"%s\" -- -parallel none" testNetCoreVersion outputTests)
 
             let result = ExecProcess(fun info ->
                 info.FileName <- "dotnet"
@@ -310,8 +310,8 @@ Target "RunTestsNet" (fun _ ->
         let runSingleProject project =
             let arguments =
                 match (hasTeamCity) with
-                | true -> (sprintf "test -c Release --no-build --logger:trx --logger:\"console;verbosity=normal\" --framework %s --results-directory \"%s\" -- -parallel none -teamcity" testNetVersion outputTests)
-                | false -> (sprintf "test -c Release --no-build --logger:trx --logger:\"console;verbosity=normal\" --framework %s --results-directory \"%s\" -- -parallel none" testNetVersion outputTests)
+                | true -> (sprintf "test -c Release --blame-crash --blame-hang-timeout 30s --no-build --logger:trx --logger:\"console;verbosity=normal\" --framework %s --results-directory \"%s\" -- -parallel none -teamcity" testNetVersion outputTests)
+                | false -> (sprintf "test -c Release --blame-crash --blame-hang-timeout 30s --no-build --logger:trx --logger:\"console;verbosity=normal\" --framework %s --results-directory \"%s\" -- -parallel none" testNetVersion outputTests)
 
             let result = ExecProcess(fun info ->
                 info.FileName <- "dotnet"

src/contrib/cluster/Akka.Cluster.Sharding.Tests/ShardEntityFailureSpec.cs

@@ -143,10 +143,10 @@ public async Task Persistent_Shard_must_recover_from_failing_entity(Props entity
             err.Cause.Should().BeOfType<ActorInitializationException>();
 
             // Need to wait for the internal state to reset, else everything we sent will go to dead letter
-            await AwaitConditionAsync(() =>
+            await AwaitConditionAsync(async () =>
             {
                 persistentShard.Tell(Shard.GetCurrentShardState.Instance);
-                var failedState = ExpectMsg<Shard.CurrentShardState>();
+                var failedState = await ExpectMsgAsync<Shard.CurrentShardState>();
                 return failedState.EntityIds.Count == 0;
             });
 

src/contrib/cluster/Akka.Cluster.Tools.Tests/PublishSubscribe/DistributedPubSubMediatorSpec.cs

@@ -80,7 +80,7 @@ public async Task DistributedPubSubMediator_should_send_messages_to_dead_letter(
 
             // assert
             await EventFilter.DeadLetter<object>().ExpectAsync(1,
-                () => { mediator.Tell(new Publish("pub-sub", $"hit")); });
+                async () => { mediator.Tell(new Publish("pub-sub", "hit")); });
         }
     }
 

src/contrib/cluster/Akka.Cluster.Tools.Tests/Singleton/ClusterSingletonProxySpec.cs

@@ -53,7 +53,7 @@ public async Task ClusterSingletonProxy_with_zero_buffering_should_work()
 
             // have to wait for cluster singleton to be ready, otherwise message will be rejected
             await AwaitConditionAsync(
-                () => Cluster.Get(testSystem.Sys).State.Members.Count(m => m.Status == MemberStatus.Up) == 2,
+                async () => Cluster.Get(testSystem.Sys).State.Members.Count(m => m.Status == MemberStatus.Up) == 2,
                 TimeSpan.FromSeconds(30));
 
             try

src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveStreams.Core.verified.txt

@@ -1935,6 +1935,7 @@ namespace Akka.Streams.Dsl
         public static Akka.Streams.Dsl.Sink<TIn, System.Threading.Tasks.Task<TIn>> First<TIn>() { }
         public static Akka.Streams.Dsl.Sink<TIn, System.Threading.Tasks.Task<TIn>> FirstOrDefault<TIn>() { }
         public static Akka.Streams.Dsl.Sink<TIn, System.Threading.Tasks.Task<Akka.Done>> ForEach<TIn>(System.Action<TIn> action) { }
+        public static Akka.Streams.Dsl.Sink<TIn, System.Threading.Tasks.Task<Akka.Done>> ForEachAsync<TIn>(int parallelism, System.Func<TIn, System.Threading.Tasks.Task> action) { }
         public static Akka.Streams.Dsl.Sink<TIn, System.Threading.Tasks.Task<Akka.Done>> ForEachParallel<TIn>(int parallelism, System.Action<TIn> action) { }
         public static Akka.Streams.Dsl.Sink<TIn, TMat> FromGraph<TIn, TMat>(Akka.Streams.IGraph<Akka.Streams.SinkShape<TIn>, TMat> graph) { }
         public static Akka.Streams.Dsl.Sink<TIn, Akka.NotUsed> FromSubscriber<TIn>(Reactive.Streams.ISubscriber<TIn> subscriber) { }

src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveStreams.DotNet.verified.txt

@@ -1935,6 +1935,7 @@ namespace Akka.Streams.Dsl
         public static Akka.Streams.Dsl.Sink<TIn, System.Threading.Tasks.Task<TIn>> First<TIn>() { }
         public static Akka.Streams.Dsl.Sink<TIn, System.Threading.Tasks.Task<TIn>> FirstOrDefault<TIn>() { }
         public static Akka.Streams.Dsl.Sink<TIn, System.Threading.Tasks.Task<Akka.Done>> ForEach<TIn>(System.Action<TIn> action) { }
+        public static Akka.Streams.Dsl.Sink<TIn, System.Threading.Tasks.Task<Akka.Done>> ForEachAsync<TIn>(int parallelism, System.Func<TIn, System.Threading.Tasks.Task> action) { }
         public static Akka.Streams.Dsl.Sink<TIn, System.Threading.Tasks.Task<Akka.Done>> ForEachParallel<TIn>(int parallelism, System.Action<TIn> action) { }
         public static Akka.Streams.Dsl.Sink<TIn, TMat> FromGraph<TIn, TMat>(Akka.Streams.IGraph<Akka.Streams.SinkShape<TIn>, TMat> graph) { }
         public static Akka.Streams.Dsl.Sink<TIn, Akka.NotUsed> FromSubscriber<TIn>(Reactive.Streams.ISubscriber<TIn> subscriber) { }

src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveStreams.Net.verified.txt

@@ -1935,6 +1935,7 @@ namespace Akka.Streams.Dsl
         public static Akka.Streams.Dsl.Sink<TIn, System.Threading.Tasks.Task<TIn>> First<TIn>() { }
         public static Akka.Streams.Dsl.Sink<TIn, System.Threading.Tasks.Task<TIn>> FirstOrDefault<TIn>() { }
         public static Akka.Streams.Dsl.Sink<TIn, System.Threading.Tasks.Task<Akka.Done>> ForEach<TIn>(System.Action<TIn> action) { }
+        public static Akka.Streams.Dsl.Sink<TIn, System.Threading.Tasks.Task<Akka.Done>> ForEachAsync<TIn>(int parallelism, System.Func<TIn, System.Threading.Tasks.Task> action) { }
         public static Akka.Streams.Dsl.Sink<TIn, System.Threading.Tasks.Task<Akka.Done>> ForEachParallel<TIn>(int parallelism, System.Action<TIn> action) { }
         public static Akka.Streams.Dsl.Sink<TIn, TMat> FromGraph<TIn, TMat>(Akka.Streams.IGraph<Akka.Streams.SinkShape<TIn>, TMat> graph) { }
         public static Akka.Streams.Dsl.Sink<TIn, Akka.NotUsed> FromSubscriber<TIn>(Reactive.Streams.ISubscriber<TIn> subscriber) { }

src/core/Akka.Cluster.Tests/ClusterLogSpec.cs

@@ -11,7 +11,9 @@
 using Akka.Actor;
 using Akka.Configuration;
 using Akka.TestKit;
+using Akka.TestKit.Extensions;
 using Akka.Util.Internal;
+using FluentAssertions.Extensions;
 using Xunit;
 using Xunit.Abstractions;
 using Xunit.Sdk;
@@ -50,26 +52,13 @@ protected async Task AwaitUpAsync()
         {
             await WithinAsync(TimeSpan.FromSeconds(10), async() =>
             {
-                await AwaitConditionAsync(() => ClusterView.IsSingletonCluster);
+                await AwaitConditionAsync(async () => ClusterView.IsSingletonCluster);
                 ClusterView.Self.Address.ShouldBe(_selfAddress);
                 ClusterView.Members.Select(m => m.Address).ShouldBe(new Address[] { _selfAddress });
                 await AwaitAssertAsync(() => ClusterView.Status.ShouldBe(MemberStatus.Up));
             });
         }
 
-
-        /// <summary>
-        /// The expected log info pattern to intercept after a <see cref="Cluster.Join(Address)"/>.
-        /// </summary>
-        protected void Join(string expected)
-        {
-            Within(TimeSpan.FromSeconds(10), () =>
-            {
-                EventFilter
-                    .Info(contains: expected)
-                    .ExpectOne(() => _cluster.Join(_selfAddress));
-            });
-        }
         /// <summary>
         /// The expected log info pattern to intercept after a <see cref="Cluster.Join(Address)"/>.
         /// </summary>
@@ -79,21 +68,7 @@ await WithinAsync(TimeSpan.FromSeconds(10), async () =>
             {
                 await EventFilter
                     .Info(contains: expected)
-                    .ExpectOneAsync(TimeSpan.FromMinutes(1), () => _cluster.Join(_selfAddress));
-            });
-        }
-
-        /// <summary>
-        /// The expected log info pattern to intercept after a <see cref="Cluster.Down(Address)"/>.
-        /// </summary>
-        /// <param name="expected"></param>
-        protected void Down(string expected)
-        {
-            Within(TimeSpan.FromSeconds(10), () =>
-            {
-                EventFilter
-                .Info(contains: expected)
-                .ExpectOne(() => _cluster.Down(_selfAddress));
+                    .ExpectOneAsync(async () => _cluster.Join(_selfAddress));
             });
         }
 
@@ -107,7 +82,7 @@ await WithinAsync(TimeSpan.FromSeconds(10), async () =>
             {
                 await EventFilter
                 .Info(contains: expected)
-                .ExpectOneAsync(() => _cluster.Down(_selfAddress));
+                .ExpectOneAsync(async () => _cluster.Down(_selfAddress));
             });
         }
     }
@@ -139,9 +114,9 @@ public ClusterLogVerboseDefaultSpec(ITestOutputHelper output)
         public async Task A_cluster_must_not_log_verbose_cluster_events_by_default()
         {
             _cluster.Settings.LogInfoVerbose.ShouldBeFalse();
-            Intercept<TrueException>(() => Join(upLogMessage));
+            await JoinAsync(upLogMessage).ShouldThrowWithin<TrueException>(10.Seconds());
             await AwaitUpAsync();
-            Intercept<TrueException>(() => Down(downLogMessage));
+            await DownAsync(downLogMessage).ShouldThrowWithin<TrueException>(10.Seconds());
         }
     }
 

src/core/Akka.Cluster.Tests/ClusterSpec.cs

@@ -79,7 +79,7 @@ public async Task A_cluster_must_initially_become_singleton_cluster_when_joining
             ClusterView.Members.Count.Should().Be(0);
             _cluster.Join(_selfAddress);
             LeaderActions(); // Joining -> Up
-            await AwaitConditionAsync(() => ClusterView.IsSingletonCluster);
+            await AwaitConditionAsync(async () => ClusterView.IsSingletonCluster);
             ClusterView.Self.Address.Should().Be(_selfAddress);
             ClusterView.Members.Select(m => m.Address).ToImmutableHashSet()
                 .Should().BeEquivalentTo(ImmutableHashSet.Create(_selfAddress));
@@ -258,7 +258,7 @@ public async Task A_cluster_must_cancel_LeaveAsync_task_if_CancellationToken_fir
 
             // Cancelling the first task
             cts.Cancel();
-            await AwaitConditionAsync(() => task1.IsCanceled, null, "Task should be cancelled");
+            await AwaitConditionAsync(async () => task1.IsCanceled, null, "Task should be cancelled");
 
             await WithinAsync(TimeSpan.FromSeconds(10), async () =>
             {
@@ -273,12 +273,12 @@ await WithinAsync(TimeSpan.FromSeconds(10), async () =>
                 ExpectMsg<ClusterEvent.MemberRemoved>().Member.Address.Should().Be(_selfAddress);
 
                 // Second task should complete (not cancelled)
-                await AwaitConditionAsync(() => task2.IsCompleted && !task2.IsCanceled, null, "Task should be completed, but not cancelled.");
+                await AwaitConditionAsync(async () => task2.IsCompleted && !task2.IsCanceled, null, "Task should be completed, but not cancelled.");
             }, cancellationToken: cts.Token);
 
             // Subsequent LeaveAsync() tasks expected to complete immediately (not cancelled)
             var task3 = _cluster.LeaveAsync();
-            await AwaitConditionAsync(() => task3.IsCompleted && !task3.IsCanceled, null, "Task should be completed, but not cancelled.");
+            await AwaitConditionAsync(async () => task3.IsCompleted && !task3.IsCanceled, null, "Task should be completed, but not cancelled.");
         }
 
         [Fact]

src/core/Akka.Remote.Tests/ActorsLeakSpec.cs

@@ -234,7 +234,7 @@ await EventFilter.Warning(contains: "Association with remote system").ExpectOneA
              * Wait for the ReliableDeliverySupervisor to receive its "TooLongIdle" message,
              * which will throw a HopelessAssociation wrapped around a TimeoutException.
              */
-            await EventFilter.Exception<TimeoutException>().ExpectOneAsync(() => { });
+            await EventFilter.Exception<TimeoutException>().ExpectOneAsync(async () => { });
 
             await AwaitAssertAsync(() =>
             {

src/core/Akka.Remote.Tests/RemotingSpec.cs

@@ -521,7 +521,7 @@ public async Task Stash_inbound_connections_until_UID_is_known_for_pending_outbo
                     (new ActorAssociationEventListener(remoteTransportProbe)));
 
                 // Hijack associations through the test transport
-                await AwaitConditionAsync(() => registry.TransportsReady(rawLocalAddress, rawRemoteAddress));
+                await AwaitConditionAsync(async () => registry.TransportsReady(rawLocalAddress, rawRemoteAddress));
                 var testTransport = registry.TransportFor(rawLocalAddress).Value.Item1;
                 testTransport.WriteBehavior.PushConstant(true);
 
@@ -601,7 +601,7 @@ public async Task Properly_quarantine_stashed_inbound_connections()
                     (new ActorAssociationEventListener(remoteTransportProbe)));
 
                 // Hijack associations through the test transport
-                await AwaitConditionAsync(() => registry.TransportsReady(rawLocalAddress, rawRemoteAddress));
+                await AwaitConditionAsync(async () => registry.TransportsReady(rawLocalAddress, rawRemoteAddress));
                 var testTransport = registry.TransportFor(rawLocalAddress).Value.Item1;
                 testTransport.WriteBehavior.PushConstant(true);
 

src/core/Akka.Remote.Tests/Transport/AkkaProtocolSpec.cs

@@ -166,7 +166,7 @@ public async Task ProtocolStateActor_must_in_inbound_mode_accept_payload_after_A
 
             reader.Tell(TestAssociate(33), TestActor);
 
-            await AwaitConditionAsync(() => collaborators.FailureDetector.IsMonitoring, DefaultTimeout);
+            await AwaitConditionAsync(async () => collaborators.FailureDetector.IsMonitoring, DefaultTimeout);
 
             var wrappedHandle = await ExpectMsgOfAsync(DefaultTimeout, "expected InboundAssociation", o =>
             {
@@ -183,7 +183,7 @@ public async Task ProtocolStateActor_must_in_inbound_mode_accept_payload_after_A
             Assert.True(collaborators.FailureDetector.IsMonitoring);
 
             // Heartbeat was sent in response to Associate
-            await AwaitConditionAsync(() => LastActivityIsHeartbeat(collaborators.Registry), DefaultTimeout);
+            await AwaitConditionAsync(async () => LastActivityIsHeartbeat(collaborators.Registry), DefaultTimeout);
 
             reader.Tell(_testPayload, TestActor);
             await ExpectMsgAsync<InboundPayload>(inbound =>
@@ -211,7 +211,7 @@ public async Task ProtocolStateActor_must_in_inbound_mode_disassociate_when_an_u
             //this associate will now be ignored
             reader.Tell(TestAssociate(33), TestActor);
 
-            await AwaitConditionAsync(() =>
+            await AwaitConditionAsync(async () =>
             {
                 var snapshots = collaborators.Registry.LogSnapshot();
                 return snapshots.Any(x => x is DisassociateAttempt);
@@ -234,12 +234,12 @@ public async Task ProtocolStateActor_must_in_outbound_mode_delay_readiness_until
                 codec: _codec,
                 failureDetector: collaborators.FailureDetector));
 
-            await AwaitConditionAsync(() => LastActivityIsAssociate(collaborators.Registry, 42), DefaultTimeout);
+            await AwaitConditionAsync(async () => LastActivityIsAssociate(collaborators.Registry, 42), DefaultTimeout);
 
-            await AwaitConditionAsync(() => collaborators.FailureDetector.IsMonitoring, DefaultTimeout);
+            await AwaitConditionAsync(async () => collaborators.FailureDetector.IsMonitoring, DefaultTimeout);
 
             //keeps sending heartbeats
-            await AwaitConditionAsync(() => LastActivityIsHeartbeat(collaborators.Registry), DefaultTimeout);
+            await AwaitConditionAsync(async () => LastActivityIsHeartbeat(collaborators.Registry), DefaultTimeout);
 
             Assert.False(statusPromise.Task.IsCompleted);
 
@@ -275,7 +275,7 @@ public async Task ProtocolStateActor_must_handle_explicit_disassociate_messages(
                 codec: _codec,
                 failureDetector: collaborators.FailureDetector));
 
-            await AwaitConditionAsync(() => LastActivityIsAssociate(collaborators.Registry, 42), DefaultTimeout);
+            await AwaitConditionAsync(async () => LastActivityIsAssociate(collaborators.Registry, 42), DefaultTimeout);
 
             reader.Tell(TestAssociate(33), TestActor);
 
@@ -323,7 +323,7 @@ public async Task ProtocolStateActor_must_handle_transport_level_disassociations
                 codec: _codec,
                 failureDetector: collaborators.FailureDetector));
 
-            await AwaitConditionAsync(() => LastActivityIsAssociate(collaborators.Registry, 42), DefaultTimeout);
+            await AwaitConditionAsync(async () => LastActivityIsAssociate(collaborators.Registry, 42), DefaultTimeout);
 
             reader.Tell(TestAssociate(33), TestActor);
 
@@ -371,7 +371,7 @@ public async Task ProtocolStateActor_must_disassociate_when_failure_detector_sig
                 codec: _codec,
                 failureDetector: collaborators.FailureDetector));
 
-            await AwaitConditionAsync(() => LastActivityIsAssociate(collaborators.Registry, 42), DefaultTimeout);
+            await AwaitConditionAsync(async () => LastActivityIsAssociate(collaborators.Registry, 42), DefaultTimeout);
 
             stateActor.Tell(TestAssociate(33), TestActor);
 
@@ -391,7 +391,7 @@ public async Task ProtocolStateActor_must_disassociate_when_failure_detector_sig
             wrappedHandle.ReadHandlerSource.SetResult(new ActorHandleEventListener(TestActor));
 
             //wait for one heartbeat
-            await AwaitConditionAsync(() => LastActivityIsHeartbeat(collaborators.Registry), DefaultTimeout);
+            await AwaitConditionAsync(async () => LastActivityIsHeartbeat(collaborators.Registry), DefaultTimeout);
 
             collaborators.FailureDetector.SetAvailable(false);
 
@@ -422,7 +422,7 @@ public async Task ProtocolStateActor_must_handle_correctly_when_the_handler_is_r
                 codec: _codec,
                 failureDetector: collaborators.FailureDetector));
 
-            await AwaitConditionAsync(() => LastActivityIsAssociate(collaborators.Registry, 42), DefaultTimeout);
+            await AwaitConditionAsync(async () => LastActivityIsAssociate(collaborators.Registry, 42), DefaultTimeout);
 
             stateActor.Tell(TestAssociate(33), TestActor);