TestJournal and TestSnaphostStore

src/Akka.sln

@@ -197,6 +197,10 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "SpawnBenchmark", "benchmark
 EndProject
 Project("{6EC3EE1D-3C4E-46DD-8F32-0CC8E7565705}") = "Akka.Persistence.FSharp", "core\Akka.Persistence.FSharp\Akka.Persistence.FSharp.fsproj", "{539C3EB6-FCC8-41FA-9373-364605877EE1}"
 EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Akka.Persistence.TestKit", "core\Akka.Persistence.TestKit\Akka.Persistence.TestKit.csproj", "{212A2D35-E8D1-46A7-A1D1-418CF9509D77}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Akka.Persistence.TestKit.Tests", "core\Akka.Persistence.TestKit.Tests\Akka.Persistence.TestKit.Tests.csproj", "{22F6EA86-0079-41A0-9BD3-82D2D6C34638}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -844,6 +848,30 @@ Global
 		{539C3EB6-FCC8-41FA-9373-364605877EE1}.Release|x64.Build.0 = Release|Any CPU
 		{539C3EB6-FCC8-41FA-9373-364605877EE1}.Release|x86.ActiveCfg = Release|Any CPU
 		{539C3EB6-FCC8-41FA-9373-364605877EE1}.Release|x86.Build.0 = Release|Any CPU
+		{212A2D35-E8D1-46A7-A1D1-418CF9509D77}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{212A2D35-E8D1-46A7-A1D1-418CF9509D77}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{212A2D35-E8D1-46A7-A1D1-418CF9509D77}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{212A2D35-E8D1-46A7-A1D1-418CF9509D77}.Debug|x64.Build.0 = Debug|Any CPU
+		{212A2D35-E8D1-46A7-A1D1-418CF9509D77}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{212A2D35-E8D1-46A7-A1D1-418CF9509D77}.Debug|x86.Build.0 = Debug|Any CPU
+		{212A2D35-E8D1-46A7-A1D1-418CF9509D77}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{212A2D35-E8D1-46A7-A1D1-418CF9509D77}.Release|Any CPU.Build.0 = Release|Any CPU
+		{212A2D35-E8D1-46A7-A1D1-418CF9509D77}.Release|x64.ActiveCfg = Release|Any CPU
+		{212A2D35-E8D1-46A7-A1D1-418CF9509D77}.Release|x64.Build.0 = Release|Any CPU
+		{212A2D35-E8D1-46A7-A1D1-418CF9509D77}.Release|x86.ActiveCfg = Release|Any CPU
+		{212A2D35-E8D1-46A7-A1D1-418CF9509D77}.Release|x86.Build.0 = Release|Any CPU
+		{22F6EA86-0079-41A0-9BD3-82D2D6C34638}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{22F6EA86-0079-41A0-9BD3-82D2D6C34638}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{22F6EA86-0079-41A0-9BD3-82D2D6C34638}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{22F6EA86-0079-41A0-9BD3-82D2D6C34638}.Debug|x64.Build.0 = Debug|Any CPU
+		{22F6EA86-0079-41A0-9BD3-82D2D6C34638}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{22F6EA86-0079-41A0-9BD3-82D2D6C34638}.Debug|x86.Build.0 = Debug|Any CPU
+		{22F6EA86-0079-41A0-9BD3-82D2D6C34638}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{22F6EA86-0079-41A0-9BD3-82D2D6C34638}.Release|Any CPU.Build.0 = Release|Any CPU
+		{22F6EA86-0079-41A0-9BD3-82D2D6C34638}.Release|x64.ActiveCfg = Release|Any CPU
+		{22F6EA86-0079-41A0-9BD3-82D2D6C34638}.Release|x64.Build.0 = Release|Any CPU
+		{22F6EA86-0079-41A0-9BD3-82D2D6C34638}.Release|x86.ActiveCfg = Release|Any CPU
+		{22F6EA86-0079-41A0-9BD3-82D2D6C34638}.Release|x86.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -929,6 +957,8 @@ Global
 		{A1D57384-A933-480A-9DF4-FA5E60AB1A67} = {73108242-625A-4D7B-AA09-63375DBAE464}
 		{9BEAF609-B406-4CCB-9708-6E8DFF764232} = {73108242-625A-4D7B-AA09-63375DBAE464}
 		{539C3EB6-FCC8-41FA-9373-364605877EE1} = {01167D3C-49C4-4CDE-9787-C176D139ACDD}
+		{212A2D35-E8D1-46A7-A1D1-418CF9509D77} = {01167D3C-49C4-4CDE-9787-C176D139ACDD}
+		{22F6EA86-0079-41A0-9BD3-82D2D6C34638} = {01167D3C-49C4-4CDE-9787-C176D139ACDD}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {03AD8E21-7507-4E68-A4E9-F4A7E7273164}

src/core/Akka.Persistence.TestKit.Tests/Akka.Persistence.TestKit.Tests.csproj

@@ -0,0 +1,35 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <Import Project="..\..\common.props" />
+
+  <PropertyGroup>
+    <AssemblyTitle>Akka.Persistence.TestKit.Tests</AssemblyTitle>
+    <TargetFrameworks>$(NetFrameworkTestVersion);$(NetCoreTestVersion)</TargetFrameworks>
+    <GenerateAssemblyInfo>false</GenerateAssemblyInfo>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="$(TestSdkVersion)" />
+    <PackageReference Include="xunit" Version="$(XunitVersion)" />
+    <PackageReference Include="xunit.runner.utility" Version="$(XunitVersion)" />
+    <PackageReference Include="xunit.runner.visualstudio" Version="$(XunitVersion)" />
+    <PackageReference Include="TeamCity.ServiceMessages" Version="3.0.8" />
+    <PackageReference Include="FluentAssertions" Version="$(FluentAssertionsVersion)" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\Akka.Persistence.TestKit\Akka.Persistence.TestKit.csproj" />
+    <ProjectReference Include="..\Akka.Tests.Shared.Internals\Akka.Tests.Shared.Internals.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <EmbeddedResource Include="test-journal.conf" />
+  </ItemGroup>
+
+  <PropertyGroup Condition=" '$(TargetFramework)' == '$(NetCoreTestVersion)' ">
+    <DefineConstants>$(DefineConstants);CORECLR</DefineConstants>
+  </PropertyGroup>
+
+  <PropertyGroup Condition=" '$(Configuration)' == 'Release' ">
+    <DefineConstants>$(DefineConstants);RELEASE</DefineConstants>
+  </PropertyGroup>
+</Project>

src/core/Akka.Persistence.TestKit.Tests/JournalWriteInterceptorsSpecs.cs

@@ -0,0 +1,184 @@
+namespace Akka.Persistence.TestKit.Tests
+{
+    using System;
+    using System.Threading.Tasks;
+    using Akka.Persistence.TestKit;
+    using FluentAssertions;
+    using Xunit;
+
+    public class JournalWriteInterceptorsSpecs
+    {
+        [Fact]
+        public void noop_immediately_returns_without_exception()
+        {
+            JournalInterceptors.Noop.Instance
+                .Awaiting(x => x.InterceptAsync(null))
+                .ShouldNotThrow();
+        }
+
+        [Fact]
+        public void failure_must_throw_specific_exception()
+        {
+            JournalInterceptors.Failure.Instance
+                .Awaiting(x => x.InterceptAsync(null))
+                .ShouldThrowExactly<TestJournalFailureException>();
+        }
+
+        [Fact]
+        public void rejection_must_throw_specific_exception()
+        {
+            JournalInterceptors.Rejection.Instance
+                .Awaiting(x => x.InterceptAsync(null))
+                .ShouldThrowExactly<TestJournalRejectionException>();
+        }
+
+        [Fact]
+        public async Task delay_must_call_next_interceptor_after_specified_delay()
+        {
+            var duration = TimeSpan.FromMilliseconds(100);
+            var probe = new InterceptorProbe();
+            var delay = new JournalInterceptors.Delay(duration, probe);
+
+            var startedAt = DateTime.Now;
+            await delay.InterceptAsync(null);
+
+            probe.WasCalled.Should().BeTrue();
+            probe.CalledAt.Should().BeOnOrAfter(startedAt + duration);
+        }
+
+        [Fact]
+        public async Task on_type_must_call_next_interceptor_when_message_is_exactly_awaited_type()
+        {
+            var probe = new InterceptorProbe();
+            var onType = new JournalInterceptors.OnType(typeof(SpecificMessage), probe);
+            var message = new Persistent(new SpecificMessage());
+
+            await onType.InterceptAsync(message);
+
+            probe.WasCalled.Should().BeTrue();
+            probe.Message.Should().BeSameAs(message);
+        }
+
+        [Fact]
+        public async Task on_type_must_call_next_interceptor_when_message_is_subclass_of_awaited_type()
+        {
+            var probe = new InterceptorProbe();
+            var onType = new JournalInterceptors.OnType(typeof(SpecificMessage), probe);
+            var message = new Persistent(new SubclassMessage());
+
+            await onType.InterceptAsync(message);
+
+            probe.WasCalled.Should().BeTrue();
+            probe.Message.Should().BeSameAs(message);
+        }
+
+        [Fact]
+        public async Task on_type_must_call_next_interceptor_when_message_is_implements_awaited_interface_type()
+        {
+            var probe = new InterceptorProbe();
+            var onType = new JournalInterceptors.OnType(typeof(IMessageWithInterface), probe);
+            var message = new Persistent(new MessageWithInterface());
+
+            await onType.InterceptAsync(message);
+
+            probe.WasCalled.Should().BeTrue();
+            probe.Message.Should().BeSameAs(message);
+        }
+
+        [Fact]
+        public async Task on_type_must_not_call_next_interceptor_when_message_does_not_correspond_to_described_rules()
+        {
+            var probe = new InterceptorProbe();
+            var onType = new JournalInterceptors.OnType(typeof(SubclassMessage), probe);
+            var message = new Persistent(new SpecificMessage());
+
+            await onType.InterceptAsync(message);
+
+            probe.WasCalled.Should().BeFalse();
+        }
+
+        [Fact]
+        public async Task on_condition_must_accept_sync_lambda()
+        {
+            var probe = new InterceptorProbe();
+            var onCondition = new JournalInterceptors.OnCondition(_ => true, probe);
+
+            await onCondition.InterceptAsync(null);
+
+            probe.WasCalled.Should().BeTrue();
+        }
+
+        [Fact]
+        public async Task on_condition_must_accept_async_lambda()
+        {
+            var probe = new InterceptorProbe();
+            var onCondition = new JournalInterceptors.OnCondition(_ => Task.FromResult(true), probe);
+
+            await onCondition.InterceptAsync(null);
+
+            probe.WasCalled.Should().BeTrue();
+        }
+
+        [Fact]
+        public async Task on_condition_must_call_next_interceptor_unless_predicate_returns_false()
+        {
+            var probe = new InterceptorProbe();
+            var onCondition = new JournalInterceptors.OnCondition(_ => false, probe);
+
+            await onCondition.InterceptAsync(null);
+
+            probe.WasCalled.Should().BeFalse();
+        }
+
+        [Fact]
+        public async Task on_condition_with_negation_must_call_next_interceptor_unless_predicate_returns_true()
+        {
+            var probe = new InterceptorProbe();
+            var onCondition = new JournalInterceptors.OnCondition(_ => false, probe, negate: true);
+
+            await onCondition.InterceptAsync(null);
+
+            probe.WasCalled.Should().BeTrue();
+        }
+
+        [Fact]
+        public async Task on_condition_must_pass_the_same_message_to_predicate()
+        {
+            var probe = new InterceptorProbe();
+            var expectedMessage = new Persistent("test");
+
+            var onCondition = new JournalInterceptors.OnCondition(message =>
+            {
+                message.Should().BeSameAs(expectedMessage);
+                return false;
+            }, probe);
+
+            await onCondition.InterceptAsync(expectedMessage);
+        }
+
+
+        private class SpecificMessage  { }
+
+        private class SubclassMessage : SpecificMessage { }
+
+        private interface IMessageWithInterface { }
+
+        private class MessageWithInterface : IMessageWithInterface { }
+
+        private class InterceptorProbe : IJournalInterceptor
+        {
+            public bool WasCalled { get; private set; }
+            public DateTime CalledAt { get; private set; }
+            public IPersistentRepresentation Message { get; private set; }
+
+            public Task InterceptAsync(IPersistentRepresentation message)
+            {
+                CalledAt = DateTime.Now;
+                WasCalled = true;
+                Message = message;
+
+                return Task.CompletedTask;
+            }
+        }
+    }
+}

src/core/Akka.Persistence.TestKit.Tests/TestJournalSpec.cs

@@ -0,0 +1,107 @@
+namespace Akka.Persistence.TestKit.Tests
+{
+    using System;
+    using Actor;
+    using Akka.Persistence.TestKit;
+    using Akka.TestKit;
+    using Configuration;
+    using Xunit;
+    using Xunit.Abstractions;
+
+    public class TestJournalSpec : AkkaSpec
+    {
+        public TestJournalSpec(ITestOutputHelper output = null)
+            : base(GetConfig().ToString(), output)
+        {
+        }
+
+        [Fact]
+        public void must_return_ack_after_new_write_interceptor_is_set()
+        {
+            var journalActor = GetJournalRef(Sys);
+
+            journalActor.Tell(new TestJournal.UseWriteInterceptor(null), TestActor);
+
+            ExpectMsg<TestJournal.Ack>(TimeSpan.FromSeconds(3));
+        }
+
+        [Fact]
+        public void works_as_memory_journal_by_default()
+        {
+            var journal = TestJournal.FromRef(GetJournalRef(Sys));
+
+            var actor = Sys.ActorOf<PersistActor>();
+
+            // should pass
+            journal.OnWrite.Pass();
+            actor.Tell("write", TestActor);
+            ExpectMsg("ack", TimeSpan.FromSeconds(3));
+        }
+
+        [Fact]
+        public void when_fail_on_write_is_set_all_writes_to_journal_will_fail()
+        {
+            var journal = TestJournal.FromRef(GetJournalRef(Sys));
+            var actor = Sys.ActorOf<PersistActor>();
+            Watch(actor);
+
+            journal.OnWrite.Fail();
+            actor.Tell("write", TestActor);
+
+            ExpectTerminated(actor, TimeSpan.FromSeconds(3));
+        }
+
+        [Fact]
+        public void when_reject_on_write_is_set_all_writes_to_journal_will_be_rejected()
+        {
+            var journal = TestJournal.FromRef(GetJournalRef(Sys));
+            var actor = Sys.ActorOf<PersistActor>();
+            Watch(actor);
+
+            journal.OnWrite.Reject();
+            actor.Tell("write", TestActor);
+
+            ExpectMsg("rejected", TimeSpan.FromSeconds(3));
+        }
+
+        static IActorRef GetJournalRef(ActorSystem sys) => Persistence.Instance.Apply(sys).JournalFor(null);
+
+        static Config GetConfig()
+            => ConfigurationFactory.FromResource<TestJournalSpec>("Akka.Persistence.TestKit.Tests.test-journal.conf");
+    }
+
+    public class PersistActor : UntypedPersistentActor
+    {
+        public override string PersistenceId  => "foo";
+
+        protected override void OnCommand(object message)
+        {
+            var sender = Sender;
+            switch (message as string)
+            {
+                case "write":
+                    Persist(message, _ =>
+                    {
+                        sender.Tell("ack");
+                    });
+
+                    break;
+
+                default:
+                    return;
+            }
+        }
+
+        protected override void OnRecover(object message)
+        {
+            // noop
+        }
+
+        protected override void OnPersistRejected(Exception cause, object @event, long sequenceNr)
+        {
+            Sender.Tell("rejected");
+
+            base.OnPersistRejected(cause, @event, sequenceNr);
+        }
+    }
+}

src/core/Akka.Persistence.TestKit.Tests/test-journal.conf

@@ -0,0 +1,13 @@
+akka {
+  persistence {
+    journal {
+      plugin = "akka.persistence.journal.test"
+      auto-start-journals = ["akka.persistence.journal.test"]
+
+      test {
+        class = "Akka.Persistence.TestKit.TestJournal, Akka.Persistence.TestKit"
+        plugin-dispatcher = "akka.actor.default-dispatcher"
+      }
+    }
+  }
+}