improvements of the PersistenceTestKit doc samples

* and actually run the tests

akka-docs/src/main/paradox/typed/persistence-testing.md

@@ -107,10 +107,10 @@ Java
 A typical scenario is to create a persistent actor, send commands to it and check that it persists events as it is expected:
 
 Scala
-:  @@snip [TestKitExamples.scala](/akka-docs/src/test/scala/docs/persistence/testkit/TestKitExamples.scala) { #testkit-typed-usecase }
+:  @@snip [TestKitExamples.scala](/akka-docs/src/test/scala/docs/persistence/testkit/TestKitExamples.scala) { #test }
 
 Java
-:  @@snip [TestKitExamples.java](/akka-docs/src/test/java/jdocs/persistence/testkit/TestKitExamples.java) { #testkit-typed-usecase }
+:  @@snip [TestKitExamples.java](/akka-docs/src/test/java/jdocs/persistence/testkit/PersistenceTestKitSampleTest.java) { #test }
 
 You can safely use persistence testkit in combination with main akka testkit.
 
@@ -188,7 +188,7 @@ For tests that involve more than one Cluster node you have to use another journa
 While it's possible to use the @ref:[Persistence Plugin Proxy](../persistence-plugins.md#persistence-plugin-proxy)
 it's often better and more realistic to use a real database.
 
-See [akka-samples issue #128](https://github.com/akka/akka-samples/issues/128).
+The @ref:[CQRS example](../project/examples.md#cqrs) includes tests that are using Akka Persistence Cassandra.
 
 ### Plugin initialization
 

akka-docs/src/test/java/jdocs/persistence/testkit/PersistenceTestKitSampleTest.java

@@ -0,0 +1,125 @@
+/*
+ * Copyright (C) 2009-2020 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package jdocs.persistence.testkit;
+
+import akka.actor.testkit.typed.javadsl.TestKitJunitResource;
+import akka.actor.typed.ActorRef;
+import akka.actor.typed.Behavior;
+import akka.actor.typed.javadsl.Behaviors;
+import akka.persistence.testkit.PersistenceTestKitPlugin;
+import akka.persistence.testkit.javadsl.PersistenceTestKit;
+import akka.persistence.typed.PersistenceId;
+import akka.persistence.typed.javadsl.CommandHandler;
+import akka.persistence.typed.javadsl.EventHandler;
+import akka.persistence.typed.javadsl.EventSourcedBehavior;
+import akka.serialization.jackson.CborSerializable;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.typesafe.config.ConfigFactory;
+import jdocs.AbstractJavaTest;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+// #test
+public class PersistenceTestKitSampleTest extends AbstractJavaTest {
+
+  @ClassRule
+  public static final TestKitJunitResource testKit =
+      new TestKitJunitResource(
+          PersistenceTestKitPlugin.getInstance()
+              .config()
+              .withFallback(ConfigFactory.defaultApplication()));
+
+  PersistenceTestKit persistenceTestKit = PersistenceTestKit.create(testKit.system());
+
+  @Before
+  public void beforeEach() {
+    persistenceTestKit.clearAll();
+  }
+
+  @Test
+  public void test() {
+    PersistenceId persistenceId = PersistenceId.ofUniqueId("some-id");
+    ActorRef<YourPersistentBehavior.Cmd> ref =
+        testKit.spawn(YourPersistentBehavior.create(persistenceId));
+
+    YourPersistentBehavior.Cmd cmd = new YourPersistentBehavior.Cmd("data");
+    ref.tell(cmd);
+    YourPersistentBehavior.Evt expectedEventPersisted = new YourPersistentBehavior.Evt(cmd.data);
+
+    persistenceTestKit.expectNextPersisted(persistenceId.id(), expectedEventPersisted);
+  }
+}
+
+class YourPersistentBehavior
+    extends EventSourcedBehavior<
+        YourPersistentBehavior.Cmd, YourPersistentBehavior.Evt, YourPersistentBehavior.State> {
+
+  static final class Cmd implements CborSerializable {
+
+    public final String data;
+
+    @JsonCreator
+    public Cmd(String data) {
+      this.data = data;
+    }
+  }
+
+  static final class Evt implements CborSerializable {
+
+    public final String data;
+
+    @JsonCreator
+    public Evt(String data) {
+      this.data = data;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) return true;
+      if (o == null || getClass() != o.getClass()) return false;
+
+      Evt evt = (Evt) o;
+
+      return data.equals(evt.data);
+    }
+
+    @Override
+    public int hashCode() {
+      return data.hashCode();
+    }
+  }
+
+  static final class State implements CborSerializable {}
+
+  static Behavior<Cmd> create(PersistenceId persistenceId) {
+    return Behaviors.setup(context -> new YourPersistentBehavior(persistenceId));
+  }
+
+  private YourPersistentBehavior(PersistenceId persistenceId) {
+    super(persistenceId);
+  }
+
+  @Override
+  public State emptyState() {
+    // some state
+    return new State();
+  }
+
+  @Override
+  public CommandHandler<Cmd, Evt, State> commandHandler() {
+    return newCommandHandlerBuilder()
+        .forAnyState()
+        .onCommand(Cmd.class, command -> Effect().persist(new Evt(command.data)))
+        .build();
+  }
+
+  @Override
+  public EventHandler<State, Evt> eventHandler() {
+    // TODO handle events
+    return newEventHandlerBuilder().forAnyState().onEvent(Evt.class, (state, evt) -> state).build();
+  }
+}
+// #test

akka-docs/src/test/java/jdocs/persistence/testkit/TestKitExamples.java

@@ -4,13 +4,10 @@
 
 package jdocs.persistence.testkit;
 
-import akka.actor.testkit.typed.javadsl.TestKitJunitResource;
-import akka.actor.typed.ActorRef;
 import akka.persistence.testkit.DeleteEvents;
 import akka.persistence.testkit.DeleteSnapshotByMeta;
 import akka.persistence.testkit.DeleteSnapshotsByCriteria;
 import akka.persistence.testkit.JournalOperation;
-import akka.persistence.testkit.PersistenceTestKitPlugin;
 import akka.persistence.testkit.ProcessingPolicy;
 import akka.persistence.testkit.ProcessingResult;
 import akka.persistence.testkit.ProcessingSuccess;
@@ -22,15 +19,6 @@ import akka.persistence.testkit.SnapshotOperation;
 import akka.persistence.testkit.StorageFailure;
 import akka.persistence.testkit.WriteEvents;
 import akka.persistence.testkit.WriteSnapshot;
-import akka.persistence.testkit.javadsl.PersistenceTestKit;
-import akka.persistence.typed.PersistenceId;
-import akka.persistence.typed.javadsl.CommandHandler;
-import akka.persistence.typed.javadsl.EventHandler;
-import akka.persistence.typed.javadsl.EventSourcedBehavior;
-import com.typesafe.config.ConfigFactory;
-import org.junit.Before;
-import org.junit.ClassRule;
-import org.junit.Test;
 
 public class TestKitExamples {
 
@@ -105,81 +93,3 @@ public class TestKitExamples {
   // #set-snapshot-storage-policy
 
 }
-
-// #testkit-typed-usecase
-class SampleTest {
-
-  @ClassRule
-  public static final TestKitJunitResource testKit =
-      new TestKitJunitResource(
-          PersistenceTestKitPlugin.getInstance()
-              .config()
-              .withFallback(ConfigFactory.defaultApplication()));
-
-  PersistenceTestKit persistenceTestKit = PersistenceTestKit.create(testKit.system());
-
-  @Before
-  void beforeAll() {
-    persistenceTestKit.clearAll();
-  }
-
-  @Test
-  void test() {
-    ActorRef<Cmd> ref =
-        testKit.spawn(new YourPersistentBehavior(PersistenceId.ofUniqueId("some-id")));
-
-    Cmd cmd = new Cmd("data");
-    ref.tell(cmd);
-    Evt expectedEventPersisted = new Evt(cmd.data);
-
-    persistenceTestKit.expectNextPersisted("your-persistence-id", expectedEventPersisted);
-  }
-}
-
-final class Cmd {
-
-  public final String data;
-
-  public Cmd(String data) {
-    this.data = data;
-  }
-}
-
-final class Evt {
-
-  public final String data;
-
-  public Evt(String data) {
-    this.data = data;
-  }
-}
-
-final class State {}
-
-class YourPersistentBehavior extends EventSourcedBehavior<Cmd, Evt, State> {
-
-  public YourPersistentBehavior(PersistenceId persistenceId) {
-    super(persistenceId);
-  }
-
-  @Override
-  public State emptyState() {
-    // some state
-    return new State();
-  }
-
-  @Override
-  public CommandHandler<Cmd, Evt, State> commandHandler() {
-    return newCommandHandlerBuilder()
-        .forAnyState()
-        .onCommand(Cmd.class, command -> Effect().persist(new Evt(command.data)))
-        .build();
-  }
-
-  @Override
-  public EventHandler<State, Evt> eventHandler() {
-    // TODO handle events
-    return newEventHandlerBuilder().build();
-  }
-}
-// #testkit-typed-usecase

akka-docs/src/test/scala/docs/persistence/testkit/TestKitExamples.scala

@@ -4,104 +4,116 @@
 
 package docs.persistence.testkit
 
-import akka.actor.typed.ActorSystem
+import akka.persistence.typed.PersistenceId
-import akka.persistence.testkit._
+import akka.persistence.typed.scaladsl.Effect
-import akka.persistence.testkit.scaladsl.PersistenceTestKit
+import akka.persistence.typed.scaladsl.EventSourcedBehavior
-import akka.persistence.typed.scaladsl.{ Effect, EventSourcedBehavior }
+import akka.serialization.jackson.CborSerializable
 import com.typesafe.config.ConfigFactory
-import org.scalatest.BeforeAndAfterAll
+import docs.persistence.testkit.PersistenceTestKitSampleSpec._
+import org.scalatest.BeforeAndAfterEach
 import org.scalatest.wordspec.AnyWordSpecLike
 
-class TestKitExamples {
+object PersistenceTestKitSampleSpec {
-
+  final case class Cmd(data: String) extends CborSerializable
-  //#testkit-typed-usecase
+  final case class Evt(data: String) extends CborSerializable
-  class TypedSampleSpec extends AnyWordSpecLike with BeforeAndAfterAll {
+  object State {
-
+    val empty: State = new State
-    val system: ActorSystem[Cmd] = ActorSystem(
-      EventSourcedBehavior[Cmd, Evt, State](
-        persistenceId = ???,
-        eventHandler = ???,
-        commandHandler = (_, cmd) => Effect.persist(Evt(cmd.data)),
-        emptyState = ???),
-      "name",
-      PersistenceTestKitPlugin.config.withFallback(ConfigFactory.defaultApplication()))
-    val persistenceTestKit = PersistenceTestKit(system)
-
-    override def beforeAll(): Unit =
-      persistenceTestKit.clearAll()
-
-    "Persistent actor" should {
-
-      "persist all events" in {
-
-        val persistentActor = system
-        val cmd = Cmd("data")
-
-        persistentActor ! cmd
-
-        val expectedPersistedEvent = Evt(cmd.data)
-        persistenceTestKit.expectNextPersisted("your-persistence-id", expectedPersistedEvent)
-      }
-
-    }
   }
-  //#testkit-typed-usecase
+  final class State extends CborSerializable {
-
+    def updated(event: Evt): State = this
-  //#set-event-storage-policy
-  class SampleEventStoragePolicy extends EventStorage.JournalPolicies.PolicyType {
-
-    //you can use internal state, it does not need to be thread safe
-    var count = 1
-
-    override def tryProcess(persistenceId: String, processingUnit: JournalOperation): ProcessingResult =
-      if (count < 10) {
-        count += 1
-        //check the type of operation and react with success or with reject or with failure.
-        //if you return ProcessingSuccess the operation will be performed, otherwise not.
-        processingUnit match {
-          case ReadEvents(batch) if batch.nonEmpty => ProcessingSuccess
-          case WriteEvents(batch) if batch.size > 1 =>
-            ProcessingSuccess
-          case ReadSeqNum      => StorageFailure()
-          case DeleteEvents(_) => Reject()
-          case _               => StorageFailure()
-        }
-      } else {
-        ProcessingSuccess
-      }
-
   }
-  //#set-event-storage-policy
-
-  //#set-snapshot-storage-policy
-  class SampleSnapshotStoragePolicy extends SnapshotStorage.SnapshotPolicies.PolicyType {
-
-    //you can use internal state, it does not need to be thread safe
-    var count = 1
-
-    override def tryProcess(persistenceId: String, processingUnit: SnapshotOperation): ProcessingResult =
-      if (count < 10) {
-        count += 1
-        //check the type of operation and react with success or with reject or with failure.
-        //if you return ProcessingSuccess the operation will be performed, otherwise not.
-        processingUnit match {
-          case ReadSnapshot(_, payload) if payload.nonEmpty =>
-            ProcessingSuccess
-          case WriteSnapshot(meta, payload) if meta.sequenceNr > 10 =>
-            ProcessingSuccess
-          case DeleteSnapshotsByCriteria(_) => StorageFailure()
-          case DeleteSnapshotByMeta(meta) if meta.sequenceNr < 10 =>
-            ProcessingSuccess
-          case _ => StorageFailure()
-        }
-      } else {
-        ProcessingSuccess
-      }
-  }
-  //#set-snapshot-storage-policy
-
 }
 
-case class Cmd(data: String)
+//#test
-case class Evt(data: String)
+import akka.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
-trait State
+import akka.persistence.testkit.PersistenceTestKitPlugin
+import akka.persistence.testkit.scaladsl.PersistenceTestKit
+
+class PersistenceTestKitSampleSpec
+    extends ScalaTestWithActorTestKit(PersistenceTestKitPlugin.config.withFallback(ConfigFactory.defaultApplication()))
+    with AnyWordSpecLike
+    with BeforeAndAfterEach {
+
+  val persistenceTestKit = PersistenceTestKit(system)
+
+  override def beforeEach(): Unit = {
+    persistenceTestKit.clearAll()
+  }
+
+  "Persistent actor" should {
+
+    "persist all events" in {
+
+      val persistenceId = PersistenceId.ofUniqueId("your-persistence-id")
+      val persistentActor = spawn(
+        EventSourcedBehavior[Cmd, Evt, State](
+          persistenceId,
+          emptyState = State.empty,
+          commandHandler = (_, cmd) => Effect.persist(Evt(cmd.data)),
+          eventHandler = (state, evt) => state.updated(evt)))
+      val cmd = Cmd("data")
+
+      persistentActor ! cmd
+
+      val expectedPersistedEvent = Evt(cmd.data)
+      persistenceTestKit.expectNextPersisted(persistenceId.id, expectedPersistedEvent)
+    }
+
+  }
+}
+//#test
+
+//#set-event-storage-policy
+import akka.persistence.testkit._
+
+class SampleEventStoragePolicy extends EventStorage.JournalPolicies.PolicyType {
+
+  //you can use internal state, it does not need to be thread safe
+  var count = 1
+
+  override def tryProcess(persistenceId: String, processingUnit: JournalOperation): ProcessingResult =
+    if (count < 10) {
+      count += 1
+      //check the type of operation and react with success or with reject or with failure.
+      //if you return ProcessingSuccess the operation will be performed, otherwise not.
+      processingUnit match {
+        case ReadEvents(batch) if batch.nonEmpty => ProcessingSuccess
+        case WriteEvents(batch) if batch.size > 1 =>
+          ProcessingSuccess
+        case ReadSeqNum      => StorageFailure()
+        case DeleteEvents(_) => Reject()
+        case _               => StorageFailure()
+      }
+    } else {
+      ProcessingSuccess
+    }
+
+}
+//#set-event-storage-policy
+
+//#set-snapshot-storage-policy
+class SampleSnapshotStoragePolicy extends SnapshotStorage.SnapshotPolicies.PolicyType {
+
+  //you can use internal state, it does not need to be thread safe
+  var count = 1
+
+  override def tryProcess(persistenceId: String, processingUnit: SnapshotOperation): ProcessingResult =
+    if (count < 10) {
+      count += 1
+      //check the type of operation and react with success or with reject or with failure.
+      //if you return ProcessingSuccess the operation will be performed, otherwise not.
+      processingUnit match {
+        case ReadSnapshot(_, payload) if payload.nonEmpty =>
+          ProcessingSuccess
+        case WriteSnapshot(meta, payload) if meta.sequenceNr > 10 =>
+          ProcessingSuccess
+        case DeleteSnapshotsByCriteria(_) => StorageFailure()
+        case DeleteSnapshotByMeta(meta) if meta.sequenceNr < 10 =>
+          ProcessingSuccess
+        case _ => StorageFailure()
+      }
+    } else {
+      ProcessingSuccess
+    }
+}
+//#set-snapshot-storage-policy

build.sbt

@@ -201,6 +201,7 @@ lazy val docs = akkaModule("akka-docs")
   .settings(Dependencies.docs)
   .settings(Paradox.settings)
   .settings(ParadoxSupport.paradoxWithCustomDirectives)
+  .settings(javacOptions += "-parameters") // for Jackson
   .enablePlugins(
     AkkaParadoxPlugin,
     DeployRsync,