diff --git a/src/contrib/cluster/Akka.Cluster.Tools.Tests/Singleton/ClusterSingletonLeavingSpeedSpec.cs b/src/contrib/cluster/Akka.Cluster.Tools.Tests/Singleton/ClusterSingletonLeavingSpeedSpec.cs index 450b11a142a..7b53ff4d2cc 100644 --- a/src/contrib/cluster/Akka.Cluster.Tools.Tests/Singleton/ClusterSingletonLeavingSpeedSpec.cs +++ b/src/contrib/cluster/Akka.Cluster.Tools.Tests/Singleton/ClusterSingletonLeavingSpeedSpec.cs @@ -10,6 +10,7 @@ using System.Collections.Immutable; using System.Diagnostics; using System.Linq; +using System.Threading.Tasks; using Akka.Actor; using Akka.Cluster.Tools.Singleton; using Akka.Configuration; @@ -96,24 +97,24 @@ public void Join(ActorSystem from, ActorSystem to, IActorRef probe) } [Fact] - public void ClusterSingleton_that_is_leaving_must() + public async Task ClusterSingleton_that_is_leaving_must() { - ClusterSingleton_that_is_leaving_must_join_cluster(); - ClusterSingleton_that_is_leaving_must_quickly_hand_over_to_next_oldest(); + await ClusterSingleton_that_is_leaving_must_join_cluster(); + await ClusterSingleton_that_is_leaving_must_quickly_hand_over_to_next_oldest(); } - private void ClusterSingleton_that_is_leaving_must_join_cluster() + private async Task ClusterSingleton_that_is_leaving_must_join_cluster() { - for (int i = 0; i < _systems.Length; i++) + for (var i = 0; i < _systems.Length; i++) Join(_systems[i], _systems[0], _probes[i]); // leader is most likely on system, lowest port Join(Sys, _systems[0], TestActor); - _probes[0].ExpectMsg("started"); + await _probes[0].ExpectMsgAsync("started"); } - private void ClusterSingleton_that_is_leaving_must_quickly_hand_over_to_next_oldest() + private async Task ClusterSingleton_that_is_leaving_must_quickly_hand_over_to_next_oldest() { var durations = new List<(TimeSpan stoppedDuration, TimeSpan startDuration)>(); var sw = new Stopwatch(); @@ -121,20 +122,21 @@ private void ClusterSingleton_that_is_leaving_must_quickly_hand_over_to_next_old for (var i = 0; i < _systems.Length; i++) { var leaveAddress = Cluster.Get(_systems[i]).SelfAddress; - CoordinatedShutdown.Get(_systems[i]).Run(CoordinatedShutdown.ClusterLeavingReason.Instance); - _probes[i].ExpectMsg("stopped", TimeSpan.FromSeconds(10)); + await CoordinatedShutdown.Get(_systems[i]).Run(CoordinatedShutdown.ClusterLeavingReason.Instance); + + await _probes[i].ExpectMsgAsync("stopped", TimeSpan.FromSeconds(10)); var stoppedDuration = sw.Elapsed; if (i != _systems.Length - 1) - _probes[i + 1].ExpectMsg("started", TimeSpan.FromSeconds(30)); + await _probes[i + 1].ExpectMsgAsync("started", TimeSpan.FromSeconds(30)); else - ExpectMsg("started", TimeSpan.FromSeconds(30)); + await ExpectMsgAsync("started", TimeSpan.FromSeconds(30)); var startedDuration = sw.Elapsed; - Within(TimeSpan.FromSeconds(15), () => + await WithinAsync(TimeSpan.FromSeconds(15), async () => { - AwaitAssert(() => + await AwaitAssertAsync(() => { Cluster.Get(_systems[i]).IsTerminated.Should().BeTrue(); Cluster.Get(Sys).State.Members.Select(m => m.Address).Should().NotContain(leaveAddress); diff --git a/src/core/Akka.Persistence.TestKit.Tests/JournalInterceptorsSpecs.cs b/src/core/Akka.Persistence.TestKit.Tests/JournalInterceptorsSpecs.cs index bde4e2473b3..52cbc425367 100644 --- a/src/core/Akka.Persistence.TestKit.Tests/JournalInterceptorsSpecs.cs +++ b/src/core/Akka.Persistence.TestKit.Tests/JournalInterceptorsSpecs.cs @@ -46,7 +46,8 @@ await Assert.ThrowsAsync(async () => [Fact] public async Task delay_must_call_next_interceptor_after_specified_delay() { - var duration = TimeSpan.FromMilliseconds(100); + var duration = TimeSpan.FromMilliseconds(200); + var epsilon = TimeSpan.FromMilliseconds(50); var probe = new InterceptorProbe(); var delay = new JournalInterceptors.Delay(duration, probe); @@ -54,7 +55,7 @@ public async Task delay_must_call_next_interceptor_after_specified_delay() await delay.InterceptAsync(null); probe.WasCalled.Should().BeTrue(); - probe.CalledAt.Should().BeOnOrAfter(startedAt + duration); + probe.CalledAt.Should().BeOnOrAfter(startedAt + duration - epsilon); } [Fact] diff --git a/src/core/Akka.Remote.Tests/Transport/AkkaProtocolSpec.cs b/src/core/Akka.Remote.Tests/Transport/AkkaProtocolSpec.cs index 1996822dd1b..d5d57c78cb0 100644 --- a/src/core/Akka.Remote.Tests/Transport/AkkaProtocolSpec.cs +++ b/src/core/Akka.Remote.Tests/Transport/AkkaProtocolSpec.cs @@ -237,7 +237,7 @@ public void ProtocolStateActor_must_in_outbound_mode_delay_readiness_until_hands } [Fact] - public void ProtocolStateActor_must_handle_explicit_disassociate_messages() + public async Task ProtocolStateActor_must_handle_explicit_disassociate_messages() { var collaborators = GetCollaborators(); collaborators.Transport.AssociateBehavior.PushConstant(collaborators.Handle); @@ -248,25 +248,38 @@ public void ProtocolStateActor_must_handle_explicit_disassociate_messages() statusPromise, collaborators.Transport, new AkkaProtocolSettings(config), codec, collaborators.FailureDetector)); - AwaitCondition(() => LastActivityIsAssociate(collaborators.Registry, 42), DefaultTimeout); + await AwaitConditionAsync(() => LastActivityIsAssociate(collaborators.Registry, 42), DefaultTimeout); reader.Tell(testAssociate(33), Self); - statusPromise.Task.Wait(TimeSpan.FromSeconds(3)); - statusPromise.Task.Result.Match() - .With(h => - { + var cts = new CancellationTokenSource(); + using (cts) + { + var timeoutTask = Task.Delay(TimeSpan.FromSeconds(3)); + var completeTask = await Task.WhenAny(timeoutTask, statusPromise.Task); + cts.Cancel(); + if (completeTask == timeoutTask) + throw new TimeoutException(); + } + + var result = statusPromise.Task.Result; + switch (result) + { + case AkkaProtocolHandle h: Assert.Equal(_remoteAkkaAddress, h.RemoteAddress); Assert.Equal(_localAkkaAddress, h.LocalAddress); - }) - .Default(msg => Assert.True(false, "Did not receive expected AkkaProtocolHandle from handshake")); - var wrappedHandle = statusPromise.Task.Result.AsInstanceOf(); + break; + default: + Assert.True(false, "Did not receive expected AkkaProtocolHandle from handshake"); + break; + } + var wrappedHandle = (AkkaProtocolHandle) result; wrappedHandle.ReadHandlerSource.SetResult(new ActorHandleEventListener(TestActor)); reader.Tell(testDisassociate(DisassociateInfo.Unknown), Self); - ExpectMsgPf("expected Disassociated(DisassociateInfo.Unknown", o => + await ExpectMsgOfAsync("expected Disassociated(DisassociateInfo.Unknown", o => { var disassociated = o.AsInstanceOf(); diff --git a/src/core/Akka.Tests.Shared.Internals/AkkaSpec.cs b/src/core/Akka.Tests.Shared.Internals/AkkaSpec.cs index 797eb826e56..b50088d8b11 100644 --- a/src/core/Akka.Tests.Shared.Internals/AkkaSpec.cs +++ b/src/core/Akka.Tests.Shared.Internals/AkkaSpec.cs @@ -115,34 +115,139 @@ private static string GetCallerName() public static Config AkkaSpecConfig { get { return _akkaSpecConfig; } } - protected T ExpectMsgPf(TimeSpan? timeout, string hint, Func function) - => ExpectMsgPf(timeout, hint, this, function); + protected T ExpectMsgOf( + TimeSpan? timeout, + string hint, + Func function, + CancellationToken cancellationToken = default) + => ExpectMsgOfAsync(timeout, hint, this, function, cancellationToken) + .ConfigureAwait(false).GetAwaiter().GetResult(); + + protected async Task ExpectMsgOfAsync( + TimeSpan? timeout, + string hint, + Func function, + CancellationToken cancellationToken = default) + => await ExpectMsgOfAsync(timeout, hint, this, function, cancellationToken) + .ConfigureAwait(false); + + protected async Task ExpectMsgOfAsync( + TimeSpan? timeout, + string hint, + Func> function, + CancellationToken cancellationToken = default) + => await ExpectMsgOfAsync(timeout, hint, this, function, cancellationToken) + .ConfigureAwait(false); - protected T ExpectMsgPf(TimeSpan? timeout, string hint, TestKitBase probe, Func function) + protected T ExpectMsgOf( + TimeSpan? timeout, + string hint, + TestKitBase probe, + Func function, + CancellationToken cancellationToken = default) + => ExpectMsgOfAsync(timeout, hint, probe, function, cancellationToken) + .ConfigureAwait(false).GetAwaiter().GetResult(); + + protected async Task ExpectMsgOfAsync( + TimeSpan? timeout, + string hint, + TestKitBase probe, + Func function, + CancellationToken cancellationToken = default) + => await ExpectMsgOfAsync(timeout, hint, probe, o => Task.FromResult(function(o)), cancellationToken) + .ConfigureAwait(false); + + protected async Task ExpectMsgOfAsync( + TimeSpan? timeout, + string hint, + TestKitBase probe, + Func> function, + CancellationToken cancellationToken = default) { - MessageEnvelope envelope; - var success = probe.TryReceiveOne(out envelope, timeout); + var (success, envelope) = await probe.TryReceiveOneAsync(timeout, cancellationToken) + .ConfigureAwait(false); if(!success) - Assertions.Fail(string.Format("expected message of type {0} but timed out after {1}", typeof(T), GetTimeoutOrDefault(timeout))); + Assertions.Fail($"expected message of type {typeof(T)} but timed out after {GetTimeoutOrDefault(timeout)}"); + var message = envelope.Message; - Assertions.AssertTrue(message != null, string.Format("expected {0} but got null message", hint)); + Assertions.AssertTrue(message != null, $"expected {hint} but got null message"); //TODO: Check next line. - Assertions.AssertTrue(function.GetMethodInfo().GetParameters().Any(x => x.ParameterType.IsInstanceOfType(message)), string.Format("expected {0} but got {1} instead", hint, message)); - return function.Invoke(message); + Assertions.AssertTrue( + function.GetMethodInfo().GetParameters().Any(x => x.ParameterType.IsInstanceOfType(message)), + $"expected {hint} but got {message} instead"); + + return await function(message).ConfigureAwait(false); } - protected T ExpectMsgPf(string hint, Func pf) - => ExpectMsgPf(hint, this, pf); - - protected T ExpectMsgPf(string hint, TestKitBase probe, Func pf) + protected T ExpectMsgOf( + string hint, + TestKitBase probe, + Func pf, + CancellationToken cancellationToken = default) + => ExpectMsgOfAsync(hint, probe, pf, cancellationToken) + .ConfigureAwait(false).GetAwaiter().GetResult(); + + protected async Task ExpectMsgOfAsync( + string hint, + TestKitBase probe, + Func pf, + CancellationToken cancellationToken = default) + => await ExpectMsgOfAsync(hint, probe, o => Task.FromResult(pf(o)), cancellationToken) + .ConfigureAwait(false); + + protected async Task ExpectMsgOfAsync( + string hint, + TestKitBase probe, + Func> pf, + CancellationToken cancellationToken = default) { - var t = probe.ExpectMsg(); + var t = await probe.ExpectMsgAsync(cancellationToken: cancellationToken) + .ConfigureAwait(false); + //TODO: Check if this really is needed: - Assertions.AssertTrue(pf.GetMethodInfo().GetParameters().Any(x => x.ParameterType.IsInstanceOfType(t)), string.Format("expected {0} but got {1} instead", hint, t)); - return pf.Invoke(t); + Assertions.AssertTrue(pf.GetMethodInfo().GetParameters().Any(x => x.ParameterType.IsInstanceOfType(t)), + $"expected {hint} but got {t} instead"); + return await pf(t); } + protected T ExpectMsgOf( + string hint, + Func pf, + CancellationToken cancellationToken = default) + => ExpectMsgOfAsync(hint, this, pf, cancellationToken) + .ConfigureAwait(false).GetAwaiter().GetResult(); + + protected async Task ExpectMsgOfAsync( + string hint, + Func pf, + CancellationToken cancellationToken = default) + => await ExpectMsgOfAsync(hint, this, pf, cancellationToken) + .ConfigureAwait(false); + + protected async Task ExpectMsgOfAsync( + string hint, + Func> pf, + CancellationToken cancellationToken = default) + => await ExpectMsgOfAsync(hint, this, pf, cancellationToken) + .ConfigureAwait(false); + + [Obsolete("Method name typo, please use ExpectMsgOf instead")] + protected T ExpectMsgPf(TimeSpan? timeout, string hint, Func function) + => ExpectMsgOf(timeout, hint, this, function); + + [Obsolete("Method name typo, please use ExpectMsgOf instead")] + protected T ExpectMsgPf(TimeSpan? timeout, string hint, TestKitBase probe, Func function) + => ExpectMsgOf(timeout, hint, probe, function); + + [Obsolete("Method name typo, please use ExpectMsgOf instead")] + protected T ExpectMsgPf(string hint, Func pf) + => ExpectMsgOf(hint, this, pf); + + [Obsolete("Method name typo, please use ExpectMsgOf instead")] + protected T ExpectMsgPf(string hint, TestKitBase probe, Func pf) + => ExpectMsgOf(hint, probe, pf); + /// /// Intercept and return an exception that's expected to be thrown by the passed function value. The thrown /// exception must be an instance of the type specified by the type parameter of this method. This method diff --git a/src/core/Akka.Tests/Dispatch/AsyncAwaitSpec.cs b/src/core/Akka.Tests/Dispatch/AsyncAwaitSpec.cs index 7798798dccd..843bfffa350 100644 --- a/src/core/Akka.Tests/Dispatch/AsyncAwaitSpec.cs +++ b/src/core/Akka.Tests/Dispatch/AsyncAwaitSpec.cs @@ -463,15 +463,17 @@ public AsyncPipeToDelayActor() { ReceiveAsync(async msg => { - Task.Run(() => - { - Thread.Sleep(10); - return msg; - }).PipeTo(Sender, Self); //LogicalContext is lost?!? + Delayed(msg).PipeTo(Sender, Self); - Thread.Sleep(3000); + await Task.Delay(3000); }); } + + private async Task Delayed(string msg) + { + await Task.Delay(10); + return msg; + } } public class AsyncReentrantActor : ReceiveActor @@ -491,6 +493,13 @@ public AsyncReentrantActor() Thread.Sleep(3000); }); } + + private async Task Delayed(string msg) + { + // Sleep to make sure the task is not completed when ContinueWith is called + await Task.Delay(100); + return msg; + } } [Fact]