From f99fd970c964d895cc12bbda0c7840f8dce47aaf Mon Sep 17 00:00:00 2001 From: Gregorius Soedharmo Date: Wed, 9 Mar 2022 04:45:31 +0700 Subject: [PATCH 1/4] Convert racy unit tests to async --- .../ClusterSingletonLeavingSpeedSpec.cs | 22 +-- .../JournalInterceptorsSpecs.cs | 5 +- .../Transport/AkkaProtocolSpec.cs | 33 +++-- src/core/Akka.TestKit/TestKitBase_Expect.cs | 2 +- .../Akka.Tests.Shared.Internals/AkkaSpec.cs | 137 ++++++++++++++++-- .../Akka.Tests/Dispatch/AsyncAwaitSpec.cs | 27 ++-- 6 files changed, 178 insertions(+), 48 deletions(-) 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..92c2c943521 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; @@ -102,18 +103,18 @@ public void ClusterSingleton_that_is_leaving_must() 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 04db2e29152..90131770ea9 100644 --- a/src/core/Akka.Persistence.TestKit.Tests/JournalInterceptorsSpecs.cs +++ b/src/core/Akka.Persistence.TestKit.Tests/JournalInterceptorsSpecs.cs @@ -42,7 +42,8 @@ public void rejection_must_throw_specific_exception() [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); @@ -50,7 +51,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.TestKit/TestKitBase_Expect.cs b/src/core/Akka.TestKit/TestKitBase_Expect.cs index 07bd3f864fc..9912ebd8d88 100644 --- a/src/core/Akka.TestKit/TestKitBase_Expect.cs +++ b/src/core/Akka.TestKit/TestKitBase_Expect.cs @@ -596,7 +596,7 @@ public IReadOnlyCollection ExpectMsgAllOf( public async IAsyncEnumerable ExpectMsgAllOfAsync( IReadOnlyCollection messages, - CancellationToken cancellationToken = default) + [EnumeratorCancellation] CancellationToken cancellationToken = default) { var enumerable = InternalExpectMsgAllOfAsync(RemainingOrDefault, messages, cancellationToken: cancellationToken) .ConfigureAwait(false).WithCancellation(cancellationToken); diff --git a/src/core/Akka.Tests.Shared.Internals/AkkaSpec.cs b/src/core/Akka.Tests.Shared.Internals/AkkaSpec.cs index b58345a7b53..289ddbecf5f 100644 --- a/src/core/Akka.Tests.Shared.Internals/AkkaSpec.cs +++ b/src/core/Akka.Tests.Shared.Internals/AkkaSpec.cs @@ -114,34 +114,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 50ff04a276d..32d13443e0c 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 @@ -483,7 +485,7 @@ public AsyncReentrantActor() var sender = Sender; Task.Run(() => { - //Sleep to make sure the task is not completed when ContinueWith is called + Thread.Sleep(100); return msg; }).ContinueWith(_ => sender.Tell(msg)); // ContinueWith will schedule with the implicit ActorTaskScheduler @@ -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] @@ -503,12 +512,12 @@ public void ActorTaskScheduler_reentrancy_should_not_be_possible() } [Fact] - public void Actor_PipeTo_should_not_be_delayed_by_async_receive() + public async Task Actor_PipeTo_should_not_be_delayed_by_async_receive() { var actor = Sys.ActorOf(); actor.Tell("hello"); - ExpectMsg(m => "hello".Equals(m), TimeSpan.FromMilliseconds(1000)); + await ExpectMsgAsync(m => "hello".Equals(m), TimeSpan.FromMilliseconds(1000)); } [Fact] From 8ed4e7a77ee2dc4273ac514faad87a1d74b3f9f3 Mon Sep 17 00:00:00 2001 From: Gregorius Soedharmo Date: Wed, 9 Mar 2022 21:55:35 +0700 Subject: [PATCH 2/4] Code fixes --- .../Singleton/ClusterSingletonLeavingSpeedSpec.cs | 6 +++--- src/core/Akka.TestKit/TestKitBase_Expect.cs | 2 +- src/core/Akka.Tests/Dispatch/AsyncAwaitSpec.cs | 2 +- 3 files changed, 5 insertions(+), 5 deletions(-) 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 92c2c943521..7b53ff4d2cc 100644 --- a/src/contrib/cluster/Akka.Cluster.Tools.Tests/Singleton/ClusterSingletonLeavingSpeedSpec.cs +++ b/src/contrib/cluster/Akka.Cluster.Tools.Tests/Singleton/ClusterSingletonLeavingSpeedSpec.cs @@ -97,10 +97,10 @@ 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 async Task ClusterSingleton_that_is_leaving_must_join_cluster() diff --git a/src/core/Akka.TestKit/TestKitBase_Expect.cs b/src/core/Akka.TestKit/TestKitBase_Expect.cs index 9912ebd8d88..437d2fdc1d9 100644 --- a/src/core/Akka.TestKit/TestKitBase_Expect.cs +++ b/src/core/Akka.TestKit/TestKitBase_Expect.cs @@ -475,7 +475,7 @@ public void ExpectNoMsg(CancellationToken cancellationToken = default) } /// - public async ValueTask ExpectNoMsgAsync(CancellationToken cancellationToken) + public async ValueTask ExpectNoMsgAsync(CancellationToken cancellationToken = default) { await InternalExpectNoMsgAsync(RemainingOrDefault, cancellationToken) .ConfigureAwait(false); diff --git a/src/core/Akka.Tests/Dispatch/AsyncAwaitSpec.cs b/src/core/Akka.Tests/Dispatch/AsyncAwaitSpec.cs index 32d13443e0c..23bb5c61749 100644 --- a/src/core/Akka.Tests/Dispatch/AsyncAwaitSpec.cs +++ b/src/core/Akka.Tests/Dispatch/AsyncAwaitSpec.cs @@ -485,7 +485,7 @@ public AsyncReentrantActor() var sender = Sender; Task.Run(() => { - + //Sleep to make sure the task is not completed when ContinueWith is called Thread.Sleep(100); return msg; }).ContinueWith(_ => sender.Tell(msg)); // ContinueWith will schedule with the implicit ActorTaskScheduler From 91bf88c1dbfd7efbd4a389d3348bf3a5f0c034b5 Mon Sep 17 00:00:00 2001 From: Gregorius Soedharmo Date: Sat, 12 Mar 2022 00:48:20 +0700 Subject: [PATCH 3/4] Convert Akka.Persistence.TestKit.Tests to async --- .../Actors/CounterActor.cs | 4 +- .../Bug4762FixSpec.cs | 16 ++--- .../JournalInterceptorsSpecs.cs | 28 +++++---- .../SnapshotStoreInterceptorsSpec.cs | 28 +++++---- .../TestJournalSpec.cs | 60 +++++++++---------- .../TestSnapshotStoreSpec.cs | 34 +++++------ 6 files changed, 91 insertions(+), 79 deletions(-) diff --git a/src/core/Akka.Persistence.TestKit.Tests/Actors/CounterActor.cs b/src/core/Akka.Persistence.TestKit.Tests/Actors/CounterActor.cs index 8d7e3724b3f..494f1b47106 100644 --- a/src/core/Akka.Persistence.TestKit.Tests/Actors/CounterActor.cs +++ b/src/core/Akka.Persistence.TestKit.Tests/Actors/CounterActor.cs @@ -77,13 +77,13 @@ await WithJournalWrite(write => write.Fail(), async () => Watch(actor); actor.Tell("inc", TestActor); - ExpectMsg(TimeSpan.FromSeconds(3)); + await ExpectMsgAsync(TimeSpan.FromSeconds(3)); // need to restart actor actor = ActorOf(counterProps, "counter1"); actor.Tell("read", TestActor); - var value = ExpectMsg(TimeSpan.FromSeconds(3)); + var value = await ExpectMsgAsync(TimeSpan.FromSeconds(3)); value.ShouldBe(0); }); } diff --git a/src/core/Akka.Persistence.TestKit.Tests/Bug4762FixSpec.cs b/src/core/Akka.Persistence.TestKit.Tests/Bug4762FixSpec.cs index 02d9f1fbbda..dab23cd7719 100644 --- a/src/core/Akka.Persistence.TestKit.Tests/Bug4762FixSpec.cs +++ b/src/core/Akka.Persistence.TestKit.Tests/Bug4762FixSpec.cs @@ -21,13 +21,13 @@ namespace Akka.Persistence.TestKit.Tests /// public class Bug4762FixSpec : PersistenceTestKit { - class WriteMessage + private class WriteMessage { } - class TestEvent + private class TestEvent { } - class TestActor2 : UntypedPersistentActor + private class TestActor2 : UntypedPersistentActor { private readonly IActorRef _probe; private readonly ILoggingAdapter _log; @@ -68,7 +68,7 @@ protected override void OnRecover(object message) public async Task TestJournal_PersistAll_should_only_count_each_event_exceptions_once() { var probe = CreateTestProbe(); - await WithJournalWrite(write => write.Pass(), () => + await WithJournalWrite(write => write.Pass(), async () => { var actor = ActorOf(() => new TestActor2(probe)); Watch(actor); @@ -76,10 +76,10 @@ await WithJournalWrite(write => write.Pass(), () => var command = new WriteMessage(); actor.Tell(command, actor); - probe.ExpectMsg(); - probe.ExpectMsg(); - probe.ExpectMsg(); - probe.ExpectNoMsg(3000); + await probe.ExpectMsgAsync(); + await probe.ExpectMsgAsync(); + await probe.ExpectMsgAsync(); + await probe.ExpectNoMsgAsync(3000); }); } } diff --git a/src/core/Akka.Persistence.TestKit.Tests/JournalInterceptorsSpecs.cs b/src/core/Akka.Persistence.TestKit.Tests/JournalInterceptorsSpecs.cs index 90131770ea9..a1e6c263b52 100644 --- a/src/core/Akka.Persistence.TestKit.Tests/JournalInterceptorsSpecs.cs +++ b/src/core/Akka.Persistence.TestKit.Tests/JournalInterceptorsSpecs.cs @@ -12,31 +12,35 @@ namespace Akka.Persistence.TestKit.Tests using Akka.Persistence.TestKit; using FluentAssertions; using Xunit; + using static FluentAssertions.FluentActions; public class JournalInterceptorsSpecs { [Fact] - public void noop_immediately_returns_without_exception() + public async Task noop_immediately_returns_without_exception() { - JournalInterceptors.Noop.Instance - .Awaiting(x => x.InterceptAsync(null)) - .Should().NotThrow(); + await Awaiting(async () => + { + await JournalInterceptors.Noop.Instance.InterceptAsync(null); + }).Should().NotThrowAsync(); } [Fact] - public void failure_must_throw_specific_exception() + public async Task failure_must_throw_specific_exception() { - JournalInterceptors.Failure.Instance - .Awaiting(x => x.InterceptAsync(null)) - .Should().ThrowExactly(); + await Awaiting(async () => + { + await JournalInterceptors.Failure.Instance.InterceptAsync(null); + }).Should().ThrowExactlyAsync(); } [Fact] - public void rejection_must_throw_specific_exception() + public async Task rejection_must_throw_specific_exception() { - JournalInterceptors.Rejection.Instance - .Awaiting(x => x.InterceptAsync(null)) - .Should().ThrowExactly(); + await Awaiting(async () => + { + await JournalInterceptors.Rejection.Instance.InterceptAsync(null); + }).Should().ThrowExactlyAsync(); } [Fact] diff --git a/src/core/Akka.Persistence.TestKit.Tests/SnapshotStoreInterceptorsSpec.cs b/src/core/Akka.Persistence.TestKit.Tests/SnapshotStoreInterceptorsSpec.cs index abce8d55286..f3b8df781c4 100644 --- a/src/core/Akka.Persistence.TestKit.Tests/SnapshotStoreInterceptorsSpec.cs +++ b/src/core/Akka.Persistence.TestKit.Tests/SnapshotStoreInterceptorsSpec.cs @@ -11,25 +11,33 @@ namespace Akka.Persistence.TestKit.Tests using System.Threading.Tasks; using FluentAssertions; using Xunit; + using static FluentAssertions.FluentActions; public class SnapshotStoreInterceptorsSpec { [Fact] - public void noop_must_do_nothing() - => SnapshotStoreInterceptors.Noop.Instance - .Awaiting(x => x.InterceptAsync(null, null)) - .Should().NotThrow(); + public async Task noop_must_do_nothing() + { + await Awaiting(async () => + { + await SnapshotStoreInterceptors.Noop.Instance.InterceptAsync(null, null); + }).Should().NotThrowAsync(); + } [Fact] - public void failure_must_always_throw_exception() - => SnapshotStoreInterceptors.Failure.Instance - .Awaiting(x => x.InterceptAsync(null, null)) - .Should().ThrowExactly(); + public async Task failure_must_always_throw_exception() + { + await Awaiting(async () => + { + await SnapshotStoreInterceptors.Failure.Instance.InterceptAsync(null, null); + }).Should().ThrowExactlyAsync(); + } [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 SnapshotStoreInterceptors.Delay(duration, probe); @@ -37,7 +45,7 @@ public async Task delay_must_call_next_interceptor_after_specified_delay() await delay.InterceptAsync(null, 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.Persistence.TestKit.Tests/TestJournalSpec.cs b/src/core/Akka.Persistence.TestKit.Tests/TestJournalSpec.cs index b32b1fa0286..f66189c7c1d 100644 --- a/src/core/Akka.Persistence.TestKit.Tests/TestJournalSpec.cs +++ b/src/core/Akka.Persistence.TestKit.Tests/TestJournalSpec.cs @@ -30,23 +30,23 @@ public TestJournalSpec() : base(DefaultTimeoutConfig) private readonly TestProbe _probe; [Fact] - public void must_return_ack_after_new_write_interceptor_is_set() + public async Task must_return_ack_after_new_write_interceptor_is_set() { JournalActorRef.Tell(new TestJournal.UseWriteInterceptor(null), TestActor); - ExpectMsg(TimeSpan.FromSeconds(3)); + await ExpectMsgAsync(TimeSpan.FromSeconds(3)); } [Fact] public async Task works_as_memory_journal_by_default() { var actor = ActorOf(() => new PersistActor(_probe)); - _probe.ExpectMsg(); + await _probe.ExpectMsgAsync(); await Journal.OnWrite.Pass(); actor.Tell(new PersistActor.WriteMessage("write"), TestActor); - _probe.ExpectMsg("ack"); + await _probe.ExpectMsgAsync("ack"); } [Fact] @@ -54,21 +54,21 @@ public async Task must_recover_restarted_actor() { var actor = ActorOf(() => new PersistActor(_probe)); Watch(actor); - _probe.ExpectMsg(); + await _probe.ExpectMsgAsync(); await Journal.OnRecovery.Pass(); actor.Tell(new PersistActor.WriteMessage("1"), TestActor); - _probe.ExpectMsg("ack"); + await _probe.ExpectMsgAsync("ack"); actor.Tell(new PersistActor.WriteMessage("2"), TestActor); - _probe.ExpectMsg("ack"); + await _probe.ExpectMsgAsync("ack"); await actor.GracefulStop(TimeSpan.FromSeconds(1)); - ExpectTerminated(actor); + await ExpectTerminatedAsync(actor); ActorOf(() => new PersistActor(_probe)); - _probe.ExpectMsg("1"); - _probe.ExpectMsg("2"); - _probe.ExpectMsg(); + await _probe.ExpectMsgAsync("1"); + await _probe.ExpectMsgAsync("2"); + await _probe.ExpectMsgAsync(); } [Fact] @@ -76,13 +76,13 @@ public async Task when_fail_on_write_is_set_all_writes_to_journal_will_fail() { var actor = ActorOf(() => new PersistActor(_probe)); Watch(actor); - _probe.ExpectMsg(); + await _probe.ExpectMsgAsync(); await Journal.OnWrite.Fail(); actor.Tell(new PersistActor.WriteMessage("write"), TestActor); - _probe.ExpectMsg("failure"); - ExpectTerminated(actor); + await _probe.ExpectMsgAsync("failure"); + await ExpectTerminatedAsync(actor); } [Fact] @@ -90,24 +90,24 @@ public async Task must_recover_failed_actor() { var actor = ActorOf(() => new PersistActor(_probe)); Watch(actor); - _probe.ExpectMsg(); + await _probe.ExpectMsgAsync(); await Journal.OnRecovery.Pass(); actor.Tell(new PersistActor.WriteMessage("1"), TestActor); - _probe.ExpectMsg("ack"); + await _probe.ExpectMsgAsync("ack"); actor.Tell(new PersistActor.WriteMessage("2"), TestActor); - _probe.ExpectMsg("ack"); + await _probe.ExpectMsgAsync("ack"); await Journal.OnWrite.Fail(); actor.Tell(new PersistActor.WriteMessage("3"), TestActor); - _probe.ExpectMsg("failure"); - ExpectTerminated(actor); + await _probe.ExpectMsgAsync("failure"); + await ExpectTerminatedAsync(actor); ActorOf(() => new PersistActor(_probe)); - _probe.ExpectMsg("1"); - _probe.ExpectMsg("2"); - _probe.ExpectMsg(); + await _probe.ExpectMsgAsync("1"); + await _probe.ExpectMsgAsync("2"); + await _probe.ExpectMsgAsync(); } [Fact] @@ -115,34 +115,34 @@ public async Task when_reject_on_write_is_set_all_writes_to_journal_will_be_reje { var actor = ActorOf(() => new PersistActor(_probe)); Watch(actor); - _probe.ExpectMsg(); + await _probe.ExpectMsgAsync(); await Journal.OnWrite.Reject(); actor.Tell(new PersistActor.WriteMessage("write"), TestActor); - _probe.ExpectMsg("rejected"); + await _probe.ExpectMsgAsync("rejected"); } [Fact] public async Task journal_must_reset_state_to_pass() { - await WithJournalWrite(write => write.Fail(), () => + await WithJournalWrite(write => write.Fail(), async () => { var actor = ActorOf(() => new PersistActor(_probe)); Watch(actor); - _probe.ExpectMsg(); + await _probe.ExpectMsgAsync(); actor.Tell(new PersistActor.WriteMessage("write"), TestActor); - _probe.ExpectMsg("failure"); - ExpectTerminated(actor); + await _probe.ExpectMsgAsync("failure"); + await ExpectTerminatedAsync(actor); }); var actor2 = ActorOf(() => new PersistActor(_probe)); Watch(actor2); - _probe.ExpectMsg(); + await _probe.ExpectMsgAsync(); actor2.Tell(new PersistActor.WriteMessage("write"), TestActor); - _probe.ExpectMsg("ack"); + await _probe.ExpectMsgAsync("ack"); } } } diff --git a/src/core/Akka.Persistence.TestKit.Tests/TestSnapshotStoreSpec.cs b/src/core/Akka.Persistence.TestKit.Tests/TestSnapshotStoreSpec.cs index 37497714232..84c1777684d 100644 --- a/src/core/Akka.Persistence.TestKit.Tests/TestSnapshotStoreSpec.cs +++ b/src/core/Akka.Persistence.TestKit.Tests/TestSnapshotStoreSpec.cs @@ -23,24 +23,24 @@ public TestSnapshotStoreSpec() private readonly TestProbe _probe; [Fact] - public void send_ack_after_load_interceptor_is_set() + public async Task send_ack_after_load_interceptor_is_set() { SnapshotsActorRef.Tell(new TestSnapshotStore.UseLoadInterceptor(null), TestActor); - ExpectMsg(); + await ExpectMsgAsync(); } [Fact] - public void send_ack_after_save_interceptor_is_set() + public async Task send_ack_after_save_interceptor_is_set() { SnapshotsActorRef.Tell(new TestSnapshotStore.UseSaveInterceptor(null), TestActor); - ExpectMsg(); + await ExpectMsgAsync(); } [Fact] - public void send_ack_after_delete_interceptor_is_set() + public async Task send_ack_after_delete_interceptor_is_set() { SnapshotsActorRef.Tell(new TestSnapshotStore.UseDeleteInterceptor(null), TestActor); - ExpectMsg(); + await ExpectMsgAsync(); } [Fact] @@ -49,17 +49,17 @@ public async Task after_load_behavior_was_executed_store_is_back_to_pass_mode() // create snapshot var actor = ActorOf(() => new SnapshotActor(_probe)); actor.Tell("save"); - _probe.ExpectMsg(); + await _probe.ExpectMsgAsync(); await actor.GracefulStop(TimeSpan.FromSeconds(3)); - await WithSnapshotLoad(load => load.Fail(), () => + await WithSnapshotLoad(load => load.Fail(), async () => { ActorOf(() => new SnapshotActor(_probe)); - _probe.ExpectMsg(); + await _probe.ExpectMsgAsync(); }); ActorOf(() => new SnapshotActor(_probe)); - _probe.ExpectMsg(); + await _probe.ExpectMsgAsync(); } [Fact] @@ -68,14 +68,14 @@ public async Task after_save_behavior_was_executed_store_is_back_to_pass_mode() // create snapshot var actor = ActorOf(() => new SnapshotActor(_probe)); - await WithSnapshotSave(save => save.Fail(), () => + await WithSnapshotSave(save => save.Fail(), async () => { actor.Tell("save"); - _probe.ExpectMsg(); + await _probe.ExpectMsgAsync(); }); actor.Tell("save"); - _probe.ExpectMsg(); + await _probe.ExpectMsgAsync(); } [Fact] @@ -85,17 +85,17 @@ public async Task after_delete_behavior_was_executed_store_is_back_to_pass_mode( var actor = ActorOf(() => new SnapshotActor(_probe)); actor.Tell("save"); - var success = _probe.ExpectMsg(); + var success = await _probe.ExpectMsgAsync(); var nr = success.Metadata.SequenceNr; - await WithSnapshotDelete(del => del.Fail(), () => + await WithSnapshotDelete(del => del.Fail(), async () => { actor.Tell(new SnapshotActor.DeleteOne(nr), TestActor); - _probe.ExpectMsg(); + await _probe.ExpectMsgAsync(); }); actor.Tell(new SnapshotActor.DeleteOne(nr), TestActor); - _probe.ExpectMsg(); + await _probe.ExpectMsgAsync(); } } } From bd02f235d400735ed32e88c605a4508f910b1cd1 Mon Sep 17 00:00:00 2001 From: Gregorius Soedharmo Date: Mon, 14 Mar 2022 21:48:05 +0700 Subject: [PATCH 4/4] Revert "Convert Akka.Persistence.TestKit.Tests to async" This reverts commit 91bf88c1dbfd7efbd4a389d3348bf3a5f0c034b5. --- .../Actors/CounterActor.cs | 4 +- .../Bug4762FixSpec.cs | 16 ++--- .../JournalInterceptorsSpecs.cs | 28 ++++----- .../SnapshotStoreInterceptorsSpec.cs | 28 ++++----- .../TestJournalSpec.cs | 60 +++++++++---------- .../TestSnapshotStoreSpec.cs | 34 +++++------ 6 files changed, 79 insertions(+), 91 deletions(-) diff --git a/src/core/Akka.Persistence.TestKit.Tests/Actors/CounterActor.cs b/src/core/Akka.Persistence.TestKit.Tests/Actors/CounterActor.cs index 494f1b47106..8d7e3724b3f 100644 --- a/src/core/Akka.Persistence.TestKit.Tests/Actors/CounterActor.cs +++ b/src/core/Akka.Persistence.TestKit.Tests/Actors/CounterActor.cs @@ -77,13 +77,13 @@ await WithJournalWrite(write => write.Fail(), async () => Watch(actor); actor.Tell("inc", TestActor); - await ExpectMsgAsync(TimeSpan.FromSeconds(3)); + ExpectMsg(TimeSpan.FromSeconds(3)); // need to restart actor actor = ActorOf(counterProps, "counter1"); actor.Tell("read", TestActor); - var value = await ExpectMsgAsync(TimeSpan.FromSeconds(3)); + var value = ExpectMsg(TimeSpan.FromSeconds(3)); value.ShouldBe(0); }); } diff --git a/src/core/Akka.Persistence.TestKit.Tests/Bug4762FixSpec.cs b/src/core/Akka.Persistence.TestKit.Tests/Bug4762FixSpec.cs index dab23cd7719..02d9f1fbbda 100644 --- a/src/core/Akka.Persistence.TestKit.Tests/Bug4762FixSpec.cs +++ b/src/core/Akka.Persistence.TestKit.Tests/Bug4762FixSpec.cs @@ -21,13 +21,13 @@ namespace Akka.Persistence.TestKit.Tests /// public class Bug4762FixSpec : PersistenceTestKit { - private class WriteMessage + class WriteMessage { } - private class TestEvent + class TestEvent { } - private class TestActor2 : UntypedPersistentActor + class TestActor2 : UntypedPersistentActor { private readonly IActorRef _probe; private readonly ILoggingAdapter _log; @@ -68,7 +68,7 @@ protected override void OnRecover(object message) public async Task TestJournal_PersistAll_should_only_count_each_event_exceptions_once() { var probe = CreateTestProbe(); - await WithJournalWrite(write => write.Pass(), async () => + await WithJournalWrite(write => write.Pass(), () => { var actor = ActorOf(() => new TestActor2(probe)); Watch(actor); @@ -76,10 +76,10 @@ await WithJournalWrite(write => write.Pass(), async () => var command = new WriteMessage(); actor.Tell(command, actor); - await probe.ExpectMsgAsync(); - await probe.ExpectMsgAsync(); - await probe.ExpectMsgAsync(); - await probe.ExpectNoMsgAsync(3000); + probe.ExpectMsg(); + probe.ExpectMsg(); + probe.ExpectMsg(); + probe.ExpectNoMsg(3000); }); } } diff --git a/src/core/Akka.Persistence.TestKit.Tests/JournalInterceptorsSpecs.cs b/src/core/Akka.Persistence.TestKit.Tests/JournalInterceptorsSpecs.cs index a1e6c263b52..90131770ea9 100644 --- a/src/core/Akka.Persistence.TestKit.Tests/JournalInterceptorsSpecs.cs +++ b/src/core/Akka.Persistence.TestKit.Tests/JournalInterceptorsSpecs.cs @@ -12,35 +12,31 @@ namespace Akka.Persistence.TestKit.Tests using Akka.Persistence.TestKit; using FluentAssertions; using Xunit; - using static FluentAssertions.FluentActions; public class JournalInterceptorsSpecs { [Fact] - public async Task noop_immediately_returns_without_exception() + public void noop_immediately_returns_without_exception() { - await Awaiting(async () => - { - await JournalInterceptors.Noop.Instance.InterceptAsync(null); - }).Should().NotThrowAsync(); + JournalInterceptors.Noop.Instance + .Awaiting(x => x.InterceptAsync(null)) + .Should().NotThrow(); } [Fact] - public async Task failure_must_throw_specific_exception() + public void failure_must_throw_specific_exception() { - await Awaiting(async () => - { - await JournalInterceptors.Failure.Instance.InterceptAsync(null); - }).Should().ThrowExactlyAsync(); + JournalInterceptors.Failure.Instance + .Awaiting(x => x.InterceptAsync(null)) + .Should().ThrowExactly(); } [Fact] - public async Task rejection_must_throw_specific_exception() + public void rejection_must_throw_specific_exception() { - await Awaiting(async () => - { - await JournalInterceptors.Rejection.Instance.InterceptAsync(null); - }).Should().ThrowExactlyAsync(); + JournalInterceptors.Rejection.Instance + .Awaiting(x => x.InterceptAsync(null)) + .Should().ThrowExactly(); } [Fact] diff --git a/src/core/Akka.Persistence.TestKit.Tests/SnapshotStoreInterceptorsSpec.cs b/src/core/Akka.Persistence.TestKit.Tests/SnapshotStoreInterceptorsSpec.cs index f3b8df781c4..abce8d55286 100644 --- a/src/core/Akka.Persistence.TestKit.Tests/SnapshotStoreInterceptorsSpec.cs +++ b/src/core/Akka.Persistence.TestKit.Tests/SnapshotStoreInterceptorsSpec.cs @@ -11,33 +11,25 @@ namespace Akka.Persistence.TestKit.Tests using System.Threading.Tasks; using FluentAssertions; using Xunit; - using static FluentAssertions.FluentActions; public class SnapshotStoreInterceptorsSpec { [Fact] - public async Task noop_must_do_nothing() - { - await Awaiting(async () => - { - await SnapshotStoreInterceptors.Noop.Instance.InterceptAsync(null, null); - }).Should().NotThrowAsync(); - } + public void noop_must_do_nothing() + => SnapshotStoreInterceptors.Noop.Instance + .Awaiting(x => x.InterceptAsync(null, null)) + .Should().NotThrow(); [Fact] - public async Task failure_must_always_throw_exception() - { - await Awaiting(async () => - { - await SnapshotStoreInterceptors.Failure.Instance.InterceptAsync(null, null); - }).Should().ThrowExactlyAsync(); - } + public void failure_must_always_throw_exception() + => SnapshotStoreInterceptors.Failure.Instance + .Awaiting(x => x.InterceptAsync(null, null)) + .Should().ThrowExactly(); [Fact] public async Task delay_must_call_next_interceptor_after_specified_delay() { - var duration = TimeSpan.FromMilliseconds(200); - var epsilon = TimeSpan.FromMilliseconds(50); + var duration = TimeSpan.FromMilliseconds(100); var probe = new InterceptorProbe(); var delay = new SnapshotStoreInterceptors.Delay(duration, probe); @@ -45,7 +37,7 @@ public async Task delay_must_call_next_interceptor_after_specified_delay() await delay.InterceptAsync(null, null); probe.WasCalled.Should().BeTrue(); - probe.CalledAt.Should().BeOnOrAfter(startedAt + duration - epsilon); + probe.CalledAt.Should().BeOnOrAfter(startedAt + duration); } [Fact] diff --git a/src/core/Akka.Persistence.TestKit.Tests/TestJournalSpec.cs b/src/core/Akka.Persistence.TestKit.Tests/TestJournalSpec.cs index f66189c7c1d..b32b1fa0286 100644 --- a/src/core/Akka.Persistence.TestKit.Tests/TestJournalSpec.cs +++ b/src/core/Akka.Persistence.TestKit.Tests/TestJournalSpec.cs @@ -30,23 +30,23 @@ public TestJournalSpec() : base(DefaultTimeoutConfig) private readonly TestProbe _probe; [Fact] - public async Task must_return_ack_after_new_write_interceptor_is_set() + public void must_return_ack_after_new_write_interceptor_is_set() { JournalActorRef.Tell(new TestJournal.UseWriteInterceptor(null), TestActor); - await ExpectMsgAsync(TimeSpan.FromSeconds(3)); + ExpectMsg(TimeSpan.FromSeconds(3)); } [Fact] public async Task works_as_memory_journal_by_default() { var actor = ActorOf(() => new PersistActor(_probe)); - await _probe.ExpectMsgAsync(); + _probe.ExpectMsg(); await Journal.OnWrite.Pass(); actor.Tell(new PersistActor.WriteMessage("write"), TestActor); - await _probe.ExpectMsgAsync("ack"); + _probe.ExpectMsg("ack"); } [Fact] @@ -54,21 +54,21 @@ public async Task must_recover_restarted_actor() { var actor = ActorOf(() => new PersistActor(_probe)); Watch(actor); - await _probe.ExpectMsgAsync(); + _probe.ExpectMsg(); await Journal.OnRecovery.Pass(); actor.Tell(new PersistActor.WriteMessage("1"), TestActor); - await _probe.ExpectMsgAsync("ack"); + _probe.ExpectMsg("ack"); actor.Tell(new PersistActor.WriteMessage("2"), TestActor); - await _probe.ExpectMsgAsync("ack"); + _probe.ExpectMsg("ack"); await actor.GracefulStop(TimeSpan.FromSeconds(1)); - await ExpectTerminatedAsync(actor); + ExpectTerminated(actor); ActorOf(() => new PersistActor(_probe)); - await _probe.ExpectMsgAsync("1"); - await _probe.ExpectMsgAsync("2"); - await _probe.ExpectMsgAsync(); + _probe.ExpectMsg("1"); + _probe.ExpectMsg("2"); + _probe.ExpectMsg(); } [Fact] @@ -76,13 +76,13 @@ public async Task when_fail_on_write_is_set_all_writes_to_journal_will_fail() { var actor = ActorOf(() => new PersistActor(_probe)); Watch(actor); - await _probe.ExpectMsgAsync(); + _probe.ExpectMsg(); await Journal.OnWrite.Fail(); actor.Tell(new PersistActor.WriteMessage("write"), TestActor); - await _probe.ExpectMsgAsync("failure"); - await ExpectTerminatedAsync(actor); + _probe.ExpectMsg("failure"); + ExpectTerminated(actor); } [Fact] @@ -90,24 +90,24 @@ public async Task must_recover_failed_actor() { var actor = ActorOf(() => new PersistActor(_probe)); Watch(actor); - await _probe.ExpectMsgAsync(); + _probe.ExpectMsg(); await Journal.OnRecovery.Pass(); actor.Tell(new PersistActor.WriteMessage("1"), TestActor); - await _probe.ExpectMsgAsync("ack"); + _probe.ExpectMsg("ack"); actor.Tell(new PersistActor.WriteMessage("2"), TestActor); - await _probe.ExpectMsgAsync("ack"); + _probe.ExpectMsg("ack"); await Journal.OnWrite.Fail(); actor.Tell(new PersistActor.WriteMessage("3"), TestActor); - await _probe.ExpectMsgAsync("failure"); - await ExpectTerminatedAsync(actor); + _probe.ExpectMsg("failure"); + ExpectTerminated(actor); ActorOf(() => new PersistActor(_probe)); - await _probe.ExpectMsgAsync("1"); - await _probe.ExpectMsgAsync("2"); - await _probe.ExpectMsgAsync(); + _probe.ExpectMsg("1"); + _probe.ExpectMsg("2"); + _probe.ExpectMsg(); } [Fact] @@ -115,34 +115,34 @@ public async Task when_reject_on_write_is_set_all_writes_to_journal_will_be_reje { var actor = ActorOf(() => new PersistActor(_probe)); Watch(actor); - await _probe.ExpectMsgAsync(); + _probe.ExpectMsg(); await Journal.OnWrite.Reject(); actor.Tell(new PersistActor.WriteMessage("write"), TestActor); - await _probe.ExpectMsgAsync("rejected"); + _probe.ExpectMsg("rejected"); } [Fact] public async Task journal_must_reset_state_to_pass() { - await WithJournalWrite(write => write.Fail(), async () => + await WithJournalWrite(write => write.Fail(), () => { var actor = ActorOf(() => new PersistActor(_probe)); Watch(actor); - await _probe.ExpectMsgAsync(); + _probe.ExpectMsg(); actor.Tell(new PersistActor.WriteMessage("write"), TestActor); - await _probe.ExpectMsgAsync("failure"); - await ExpectTerminatedAsync(actor); + _probe.ExpectMsg("failure"); + ExpectTerminated(actor); }); var actor2 = ActorOf(() => new PersistActor(_probe)); Watch(actor2); - await _probe.ExpectMsgAsync(); + _probe.ExpectMsg(); actor2.Tell(new PersistActor.WriteMessage("write"), TestActor); - await _probe.ExpectMsgAsync("ack"); + _probe.ExpectMsg("ack"); } } } diff --git a/src/core/Akka.Persistence.TestKit.Tests/TestSnapshotStoreSpec.cs b/src/core/Akka.Persistence.TestKit.Tests/TestSnapshotStoreSpec.cs index 84c1777684d..37497714232 100644 --- a/src/core/Akka.Persistence.TestKit.Tests/TestSnapshotStoreSpec.cs +++ b/src/core/Akka.Persistence.TestKit.Tests/TestSnapshotStoreSpec.cs @@ -23,24 +23,24 @@ public TestSnapshotStoreSpec() private readonly TestProbe _probe; [Fact] - public async Task send_ack_after_load_interceptor_is_set() + public void send_ack_after_load_interceptor_is_set() { SnapshotsActorRef.Tell(new TestSnapshotStore.UseLoadInterceptor(null), TestActor); - await ExpectMsgAsync(); + ExpectMsg(); } [Fact] - public async Task send_ack_after_save_interceptor_is_set() + public void send_ack_after_save_interceptor_is_set() { SnapshotsActorRef.Tell(new TestSnapshotStore.UseSaveInterceptor(null), TestActor); - await ExpectMsgAsync(); + ExpectMsg(); } [Fact] - public async Task send_ack_after_delete_interceptor_is_set() + public void send_ack_after_delete_interceptor_is_set() { SnapshotsActorRef.Tell(new TestSnapshotStore.UseDeleteInterceptor(null), TestActor); - await ExpectMsgAsync(); + ExpectMsg(); } [Fact] @@ -49,17 +49,17 @@ public async Task after_load_behavior_was_executed_store_is_back_to_pass_mode() // create snapshot var actor = ActorOf(() => new SnapshotActor(_probe)); actor.Tell("save"); - await _probe.ExpectMsgAsync(); + _probe.ExpectMsg(); await actor.GracefulStop(TimeSpan.FromSeconds(3)); - await WithSnapshotLoad(load => load.Fail(), async () => + await WithSnapshotLoad(load => load.Fail(), () => { ActorOf(() => new SnapshotActor(_probe)); - await _probe.ExpectMsgAsync(); + _probe.ExpectMsg(); }); ActorOf(() => new SnapshotActor(_probe)); - await _probe.ExpectMsgAsync(); + _probe.ExpectMsg(); } [Fact] @@ -68,14 +68,14 @@ public async Task after_save_behavior_was_executed_store_is_back_to_pass_mode() // create snapshot var actor = ActorOf(() => new SnapshotActor(_probe)); - await WithSnapshotSave(save => save.Fail(), async () => + await WithSnapshotSave(save => save.Fail(), () => { actor.Tell("save"); - await _probe.ExpectMsgAsync(); + _probe.ExpectMsg(); }); actor.Tell("save"); - await _probe.ExpectMsgAsync(); + _probe.ExpectMsg(); } [Fact] @@ -85,17 +85,17 @@ public async Task after_delete_behavior_was_executed_store_is_back_to_pass_mode( var actor = ActorOf(() => new SnapshotActor(_probe)); actor.Tell("save"); - var success = await _probe.ExpectMsgAsync(); + var success = _probe.ExpectMsg(); var nr = success.Metadata.SequenceNr; - await WithSnapshotDelete(del => del.Fail(), async () => + await WithSnapshotDelete(del => del.Fail(), () => { actor.Tell(new SnapshotActor.DeleteOne(nr), TestActor); - await _probe.ExpectMsgAsync(); + _probe.ExpectMsg(); }); actor.Tell(new SnapshotActor.DeleteOne(nr), TestActor); - await _probe.ExpectMsgAsync(); + _probe.ExpectMsg(); } } }