You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by he...@apache.org on 2025/11/09 12:43:18 UTC
(pekko) branch 1.3.x-unpersistence created (now 554a13c831)
This is an automated email from the ASF dual-hosted git repository.
hepin pushed a change to branch 1.3.x-unpersistence
in repository https://gitbox.apache.org/repos/asf/pekko.git
at 554a13c831 feat: Add unpersistent versions of persistent behaviors (#2456)
This branch includes the following new commits:
new 554a13c831 feat: Add unpersistent versions of persistent behaviors (#2456)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org
(pekko) 01/01: feat: Add unpersistent versions of persistent behaviors (#2456)
Posted by he...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hepin pushed a commit to branch 1.3.x-unpersistence
in repository https://gitbox.apache.org/repos/asf/pekko.git
commit 554a13c831998f217ca159805f2ae97286d9ed41
Author: He-Pin(kerr) <he...@gmail.com>
AuthorDate: Sun Nov 9 19:48:33 2025 +0800
feat: Add unpersistent versions of persistent behaviors (#2456)
(cherry picked from commit 523c05f7f6b9ab7cc8b4e91e80b877055221e405)
---
.../typed/AccountExampleUnpersistentDocTest.java | 161 +++++++
.../typed/AccountExampleUnpersistentDocSpec.scala | 97 ++++
docs/src/main/paradox/typed/persistence-testing.md | 44 +-
docs/src/main/paradox/typed/testing-sync.md | 2 +-
.../testkit/internal/Unpersistent.scala | 488 +++++++++++++++++++++
.../testkit/javadsl/UnpersistentBehavior.scala | 162 +++++++
.../testkit/scaladsl/UnpersistentBehavior.scala | 144 ++++++
.../scaladsl/UnpersistentDurableStateSpec.scala | 263 +++++++++++
.../scaladsl/UnpersistentEventSourcedSpec.scala | 336 ++++++++++++++
9 files changed, 1694 insertions(+), 3 deletions(-)
diff --git a/cluster-sharding-typed/src/test/java/jdocs/org/apache/pekko/cluster/sharding/typed/AccountExampleUnpersistentDocTest.java b/cluster-sharding-typed/src/test/java/jdocs/org/apache/pekko/cluster/sharding/typed/AccountExampleUnpersistentDocTest.java
new file mode 100644
index 0000000000..759408f65e
--- /dev/null
+++ b/cluster-sharding-typed/src/test/java/jdocs/org/apache/pekko/cluster/sharding/typed/AccountExampleUnpersistentDocTest.java
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2022 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package jdocs.org.apache.pekko.cluster.sharding.typed;
+
+import org.apache.pekko.Done;
+import org.scalatestplus.junit.JUnitSuite;
+
+import static jdocs.org.apache.pekko.cluster.sharding.typed.AccountExampleWithEventHandlersInState.AccountEntity;
+import static org.junit.Assert.*;
+
+// #test
+import java.math.BigDecimal;
+
+import org.apache.pekko.actor.testkit.typed.javadsl.BehaviorTestKit;
+import org.apache.pekko.actor.testkit.typed.javadsl.ReplyInbox;
+import org.apache.pekko.actor.testkit.typed.javadsl.StatusReplyInbox;
+import org.apache.pekko.persistence.testkit.javadsl.UnpersistentBehavior;
+import org.apache.pekko.persistence.testkit.javadsl.PersistenceEffect;
+import org.apache.pekko.persistence.typed.PersistenceId;
+import org.junit.Test;
+
+public class AccountExampleUnpersistentDocTest
+ // #test
+ extends JUnitSuite
+// #test
+{
+ @Test
+ public void createWithEmptyBalance() {
+ UnpersistentBehavior<AccountEntity.Command, AccountEntity.Event, AccountEntity.Account>
+ unpersistent = emptyAccount();
+
+ BehaviorTestKit<AccountEntity.Command> testkit = unpersistent.getBehaviorTestKit();
+
+ StatusReplyInbox<Done> ackInbox = testkit.runAskWithStatus(AccountEntity.CreateAccount::new);
+
+ ackInbox.expectValue(Done.getInstance());
+ unpersistent.getEventProbe().expectPersisted(AccountEntity.AccountCreated.INSTANCE);
+
+ // internal state is only exposed by the behavior via responses to messages or if it happens
+ // to snapshot. This particular behavior never snapshots, so we query within the actor's
+ // protocol
+ assertFalse(unpersistent.getSnapshotProbe().hasEffects());
+
+ ReplyInbox<AccountEntity.CurrentBalance> currentBalanceInbox =
+ testkit.runAsk(AccountEntity.GetBalance::new);
+
+ assertEquals(BigDecimal.ZERO, currentBalanceInbox.receiveReply().balance);
+ }
+
+ @Test
+ public void handleDepositAndWithdraw() {
+ UnpersistentBehavior<AccountEntity.Command, AccountEntity.Event, AccountEntity.Account>
+ unpersistent = openedAccount();
+
+ BehaviorTestKit<AccountEntity.Command> testkit = unpersistent.getBehaviorTestKit();
+ BigDecimal currentBalance;
+
+ testkit
+ .runAskWithStatus(
+ Done.class, replyTo -> new AccountEntity.Deposit(BigDecimal.valueOf(100), replyTo))
+ .expectValue(Done.getInstance());
+
+ assertEquals(
+ BigDecimal.valueOf(100),
+ unpersistent
+ .getEventProbe()
+ .expectPersistedClass(AccountEntity.Deposited.class)
+ .persistedObject()
+ .amount);
+
+ currentBalance =
+ testkit
+ .runAsk(AccountEntity.CurrentBalance.class, AccountEntity.GetBalance::new)
+ .receiveReply()
+ .balance;
+
+ assertEquals(BigDecimal.valueOf(100), currentBalance);
+
+ testkit
+ .runAskWithStatus(
+ Done.class, replyTo -> new AccountEntity.Withdraw(BigDecimal.valueOf(10), replyTo))
+ .expectValue(Done.getInstance());
+
+ // can save the persistence effect for in-depth inspection
+ PersistenceEffect<AccountEntity.Withdrawn> withdrawEffect =
+ unpersistent.getEventProbe().expectPersistedClass(AccountEntity.Withdrawn.class);
+ assertEquals(BigDecimal.valueOf(10), withdrawEffect.persistedObject().amount);
+ assertEquals(3L, withdrawEffect.sequenceNr());
+ assertTrue(withdrawEffect.tags().isEmpty());
+
+ currentBalance =
+ testkit
+ .runAsk(AccountEntity.CurrentBalance.class, AccountEntity.GetBalance::new)
+ .receiveReply()
+ .balance;
+
+ assertEquals(BigDecimal.valueOf(90), currentBalance);
+ }
+
+ @Test
+ public void rejectWithdrawOverdraft() {
+ UnpersistentBehavior<AccountEntity.Command, AccountEntity.Event, AccountEntity.Account>
+ unpersistent = accountWithBalance(BigDecimal.valueOf(100));
+
+ BehaviorTestKit<AccountEntity.Command> testkit = unpersistent.getBehaviorTestKit();
+
+ testkit
+ .runAskWithStatus(
+ Done.class, replyTo -> new AccountEntity.Withdraw(BigDecimal.valueOf(110), replyTo))
+ .expectErrorMessage("not enough funds to withdraw 110");
+
+ assertFalse(unpersistent.getEventProbe().hasEffects());
+ }
+
+ // #test
+ private UnpersistentBehavior<AccountEntity.Command, AccountEntity.Event, AccountEntity.Account>
+ emptyAccount() {
+ return
+ // #unpersistent-behavior
+ UnpersistentBehavior.fromEventSourced(
+ AccountEntity.create("1", PersistenceId.of("Account", "1")),
+ null, // use the initial state
+ 0 // initial sequence number
+ );
+ // #unpersistent-behavior
+ }
+
+ private UnpersistentBehavior<AccountEntity.Command, AccountEntity.Event, AccountEntity.Account>
+ openedAccount() {
+ return
+ // #unpersistent-behavior-provided-state
+ UnpersistentBehavior.fromEventSourced(
+ AccountEntity.create("1", PersistenceId.of("Account", "1")),
+ new AccountEntity.EmptyAccount()
+ .openedAccount(), // duplicate the event handler for AccountCreated on an EmptyAccount
+ 1 // assume that CreateAccount was the first command
+ );
+ // #unpersistent-behavior-provided-state
+ }
+
+ private UnpersistentBehavior<AccountEntity.Command, AccountEntity.Event, AccountEntity.Account>
+ accountWithBalance(BigDecimal balance) {
+ return UnpersistentBehavior.fromEventSourced(
+ AccountEntity.create("1", PersistenceId.of("Account", "1")),
+ new AccountEntity.OpenedAccount(balance),
+ 2);
+ }
+ // #test
+}
+// #test
diff --git a/cluster-sharding-typed/src/test/scala/docs/org/apache/pekko/cluster/sharding/typed/AccountExampleUnpersistentDocSpec.scala b/cluster-sharding-typed/src/test/scala/docs/org/apache/pekko/cluster/sharding/typed/AccountExampleUnpersistentDocSpec.scala
new file mode 100644
index 0000000000..865e936a50
--- /dev/null
+++ b/cluster-sharding-typed/src/test/scala/docs/org/apache/pekko/cluster/sharding/typed/AccountExampleUnpersistentDocSpec.scala
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2022 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package docs.org.apache.pekko.cluster.sharding.typed
+
+import org.apache.pekko
+import pekko.Done
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpecLike
+
+// #test
+import pekko.persistence.testkit.scaladsl.UnpersistentBehavior
+import pekko.persistence.typed.PersistenceId
+
+class AccountExampleUnpersistentDocSpec
+ extends AnyWordSpecLike
+ // #test
+ with Matchers
+ // #test
+ {
+ // #test
+ import AccountExampleWithEventHandlersInState.AccountEntity
+ // #test
+ "Account" must {
+ "be created with zero balance" in {
+ onAnEmptyAccount { (testkit, eventProbe, snapshotProbe) =>
+ testkit.runAskWithStatus[Done](AccountEntity.CreateAccount(_)).expectDone()
+
+ eventProbe.expectPersisted(AccountEntity.AccountCreated)
+
+ // internal state is only exposed by the behavior via responses to messages or if it happens
+ // to snapshot. This particular behavior never snapshots, so we query within the actor's
+ // protocol
+ snapshotProbe.hasEffects shouldBe false
+
+ testkit.runAsk[AccountEntity.CurrentBalance](AccountEntity.GetBalance(_)).receiveReply().balance shouldBe 0
+ }
+ }
+
+ "handle Deposit and Withdraw" in {
+ onAnOpenedAccount { (testkit, eventProbe, _) =>
+ testkit.runAskWithStatus[Done](AccountEntity.Deposit(100, _)).expectDone()
+
+ eventProbe.expectPersisted(AccountEntity.Deposited(100))
+
+ testkit.runAskWithStatus[Done](AccountEntity.Withdraw(10, _)).expectDone()
+
+ eventProbe.expectPersisted(AccountEntity.Withdrawn(10))
+
+ testkit.runAsk[AccountEntity.CurrentBalance](AccountEntity.GetBalance(_)).receiveReply().balance shouldBe 90
+ }
+ }
+
+ "reject Withdraw overdraft" in {
+ onAnAccountWithBalance(100) { (testkit, eventProbe, _) =>
+ testkit.runAskWithStatus(AccountEntity.Withdraw(110, _)).receiveStatusReply().isError shouldBe true
+
+ eventProbe.hasEffects shouldBe false
+ }
+ }
+ }
+ // #test
+
+ // #unpersistent-behavior
+ private def onAnEmptyAccount
+ : UnpersistentBehavior.EventSourced[AccountEntity.Command, AccountEntity.Event, AccountEntity.Account] =
+ UnpersistentBehavior.fromEventSourced(AccountEntity("1", PersistenceId("Account", "1")))
+ // #unpersistent-behavior
+
+ // #unpersistent-behavior-provided-state
+ private def onAnOpenedAccount
+ : UnpersistentBehavior.EventSourced[AccountEntity.Command, AccountEntity.Event, AccountEntity.Account] =
+ UnpersistentBehavior.fromEventSourced(
+ AccountEntity("1", PersistenceId("Account", "1")),
+ Some(
+ AccountEntity.EmptyAccount.applyEvent(AccountEntity.AccountCreated) -> // reuse the event handler
+ 1L // assume that CreateAccount was the first command
+ ))
+ // #unpersistent-behavior-provided-state
+
+ private def onAnAccountWithBalance(balance: BigDecimal) =
+ UnpersistentBehavior.fromEventSourced(
+ AccountEntity("1", PersistenceId("Account", "1")),
+ Some(AccountEntity.OpenedAccount(balance) -> 2L))
+ // #test
+}
+// #test
diff --git a/docs/src/main/paradox/typed/persistence-testing.md b/docs/src/main/paradox/typed/persistence-testing.md
index 46d828c153..12eea0336f 100644
--- a/docs/src/main/paradox/typed/persistence-testing.md
+++ b/docs/src/main/paradox/typed/persistence-testing.md
@@ -19,9 +19,49 @@ To use Pekko Persistence TestKit, add the module to your project:
@@project-info{ projectId="persistence-testkit" }
-## Unit testing
+## Unit testing with the BehaviorTestKit
-**Note!** The `EventSourcedBehaviorTestKit` is a new feature, api may have changes breaking source compatibility in future versions.
+**Note!** The `UnpersistentBehavior` is a new feature: the API may have changes breaking source compatibility in future versions.
+
+Unit testing of `EventSourcedBehavior` can be performed by converting it into an @apidoc[UnpersistentBehavior]. Instead of
+persisting events and snapshots, the `UnpersistentBehavior` exposes @apidoc[PersistenceProbe]s for events and snapshots which
+can be asserted on.
+
+Scala
+: @@snip [AccountExampleUnpersistentDocSpec.scala](/cluster-sharding-typed/src/test/scala/docs/org/apache/pekko/cluster/sharding/typed/AccountExampleUnpersistentDocSpec.scala) { #unpersistent-behavior }
+
+Java
+: @@snip [AccountExampleUnpersistentDocTest.java](/cluster-sharding-typed/src/test/java/jdocs/org/apache/pekko/cluster/sharding/typed/AccountExampleUnpersistentDocTest.java) { #unpersistent-behavior }
+
+The `UnpersistentBehavior` can be initialized with arbitrary states:
+
+Scala
+: @@snip [AccountExampleUnpersistentDocSpec.scala](/cluster-sharding-typed/src/test/scala/docs/org/apache/pekko/cluster/sharding/typed/AccountExampleUnpersistentDocSpec.scala) { #unpersistent-behavior-provided-state }
+
+Java
+: @@snip [AccountExampleUnpersistentDocTest.java](/cluster-sharding-typed/src/test/java/jdocs/org/apache/pekko/cluster/sharding/typed/AccountExampleUnpersistentDocTest.java) { #unpersistent-behavior-provided-state }
+
+The `UnpersistentBehavior` is especially well-suited to the synchronous @ref:[`BehaviorTestKit`](testing-sync.md#synchronous-behavior-testing):
+the `UnpersistentBehavior` can directly construct a `BehaviorTestKit` wrapping the behavior. When commands are run by `BehaviorTestKit`,
+they are processed in the calling thread (viz. the test suite), so when the run returns, the suite can be sure that the message has been
+fully processed. The internal state of the `EventSourcedBehavior` is not exposed to the suite except to the extent that it affects how
+the behavior responds to commands or the events it persists (in addition, any snapshots made by the behavior are available through a
+`PersistenceProbe`).
+
+A full test for the `AccountEntity`, which is shown in the @ref:[Persistence Style Guide](persistence-style.md) might look like:
+
+Scala
+: @@snip [AccountExampleUnpersistentDocSpec.scala](/cluster-sharding-typed/src/test/scala/docs/org/apache/pekko/cluster/sharding/typed/AccountExampleUnpersistentDocSpec.scala) { #test }
+
+Java
+: @@snip [AccountExampleUnpersistentDocTest.java](/cluster-sharding-typed/src/test/java/jdocs/org/apache/pekko/cluster/sharding/typed/AccountExampleUnpersistentDocTest.java) { #test }
+
+`UnpersistentBehavior` does not require any configuration. It therefore does not verify the serialization of commands, events, or state.
+If using this style, it is advised to independently test serialization for those classes.
+
+## Unit testing with the the ActorTestKit and EventSourcedBehaviorTestKit
+
+**Note!** The `EventSourcedBehaviorTestKit` is a new feature: the API may have changes breaking source compatibility in future versions.
Unit testing of `EventSourcedBehavior` can be done with the @apidoc[EventSourcedBehaviorTestKit]. It supports running
one command at a time and you can assert that the synchronously returned result is as expected. The result contains the
diff --git a/docs/src/main/paradox/typed/testing-sync.md b/docs/src/main/paradox/typed/testing-sync.md
index 3e9ab222f7..a96828e66c 100644
--- a/docs/src/main/paradox/typed/testing-sync.md
+++ b/docs/src/main/paradox/typed/testing-sync.md
@@ -12,7 +12,7 @@ limitations:
* Spawning of @scala[`Future`]@java[`CompletionStage`] or other asynchronous task and you rely on a callback to
complete before observing the effect you want to test.
* Usage of scheduler is not supported.
-* `EventSourcedBehavior` can't be tested.
+* `EventSourcedBehavior` can't be fully tested, but it is possible to @ref:[test the core functionality](persistence-testing.md#unit-testing-with-the-behaviortestkit)
* Interactions with other actors must be stubbed.
* Blackbox testing style.
* Supervision is not supported.
diff --git a/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/internal/Unpersistent.scala b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/internal/Unpersistent.scala
new file mode 100644
index 0000000000..adecc6bcdc
--- /dev/null
+++ b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/internal/Unpersistent.scala
@@ -0,0 +1,488 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2022 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.testkit.internal
+
+import java.util.concurrent.ConcurrentLinkedQueue
+
+import scala.annotation.tailrec
+import scala.collection.immutable
+import scala.collection.mutable.ListBuffer
+import scala.reflect.ClassTag
+
+import org.apache.pekko
+import pekko.actor.typed.Behavior
+import pekko.actor.typed.internal.BehaviorImpl.DeferredBehavior
+import pekko.actor.typed.scaladsl.{ AbstractBehavior, ActorContext, Behaviors }
+import pekko.annotation.InternalApi
+import pekko.persistence.testkit.{ javadsl, scaladsl }
+import pekko.persistence.typed.internal.EventSourcedBehaviorImpl
+import pekko.persistence.typed.internal.Running.WithSeqNrAccessible
+import pekko.persistence.typed.state.internal.DurableStateBehaviorImpl
+import pekko.persistence.typed.state.internal.Running.WithRevisionAccessible
+import pekko.util.ConstantFun.{ scalaAnyToUnit => doNothing }
+import pekko.util.ccompat.JavaConverters._
+
+/**
+ * INTERNAL API
+ */
+@InternalApi
+private[pekko] object Unpersistent {
+
+ def eventSourced[Command, Event, State](behavior: Behavior[Command], fromStateAndSequenceNr: Option[(State, Long)])(
+ onEvent: (Event, Long, Set[String]) => Unit)(onSnapshot: (State, Long) => Unit): Behavior[Command] = {
+ @tailrec
+ def findEventSourcedBehavior(
+ b: Behavior[Command],
+ context: ActorContext[Command]): Option[EventSourcedBehaviorImpl[Command, Event, State]] = {
+ b match {
+ case es: EventSourcedBehaviorImpl[Command, _, _] =>
+ Some(es.asInstanceOf[EventSourcedBehaviorImpl[Command, Event, State]])
+
+ case deferred: DeferredBehavior[Command] =>
+ findEventSourcedBehavior(deferred(context), context)
+
+ case _ => None
+ }
+ }
+
+ Behaviors.setup[Command] { context =>
+ findEventSourcedBehavior(behavior, context).fold {
+ throw new AssertionError("Did not find the expected EventSourcedBehavior")
+ } { esBehavior =>
+ val (initialState, initialSequenceNr) = fromStateAndSequenceNr.getOrElse(esBehavior.emptyState -> 0L)
+ new WrappedEventSourcedBehavior(context, esBehavior, initialState, initialSequenceNr, onEvent, onSnapshot)
+ }
+ }
+ }
+
+ def durableState[Command, State](behavior: Behavior[Command], fromState: Option[State])(
+ onPersist: (State, Long, String) => Unit): Behavior[Command] = {
+
+ @tailrec
+ def findDurableStateBehavior(
+ b: Behavior[Command],
+ context: ActorContext[Command]): Option[DurableStateBehaviorImpl[Command, State]] =
+ b match {
+ case ds: DurableStateBehaviorImpl[Command, _] =>
+ Some(ds.asInstanceOf[DurableStateBehaviorImpl[Command, State]])
+
+ case deferred: DeferredBehavior[Command] => findDurableStateBehavior(deferred(context), context)
+ case _ => None
+ }
+
+ Behaviors.setup[Command] { context =>
+ findDurableStateBehavior(behavior, context).fold {
+ throw new AssertionError("Did not find the expected DurableStateBehavior")
+ } { dsBehavior =>
+ val initialState = fromState.getOrElse(dsBehavior.emptyState)
+ new WrappedDurableStateBehavior(context, dsBehavior, initialState, onPersist)
+ }
+ }
+ }
+
+ private class WrappedEventSourcedBehavior[Command, Event, State](
+ context: ActorContext[Command],
+ esBehavior: EventSourcedBehaviorImpl[Command, Event, State],
+ initialState: State,
+ initialSequenceNr: Long,
+ onEvent: (Event, Long, Set[String]) => Unit,
+ onSnapshot: (State, Long) => Unit)
+ extends AbstractBehavior[Command](context)
+ with WithSeqNrAccessible {
+ import pekko.persistence.typed.{ EventSourcedSignal, RecoveryCompleted, SnapshotCompleted, SnapshotMetadata }
+ import pekko.persistence.typed.internal._
+
+ override def currentSequenceNumber: Long = sequenceNr
+
+ private def commandHandler = esBehavior.commandHandler
+ private def eventHandler = esBehavior.eventHandler
+ private def tagger = esBehavior.tagger
+ private def snapshotWhen = esBehavior.snapshotWhen
+ private def retention = esBehavior.retention
+ private def signalHandler = esBehavior.signalHandler
+
+ private var sequenceNr: Long = initialSequenceNr
+ private var state: State = initialState
+ private val stashedCommands = ListBuffer.empty[Command]
+
+ private def snapshotMetadata() =
+ SnapshotMetadata(esBehavior.persistenceId.toString, sequenceNr, System.currentTimeMillis())
+ private def sendSignal(signal: EventSourcedSignal): Unit =
+ signalHandler.applyOrElse(state -> signal, doNothing)
+
+ sendSignal(RecoveryCompleted)
+
+ override def onMessage(cmd: Command): Behavior[Command] = {
+ var shouldSnapshot = false
+ var shouldUnstash = false
+ var shouldStop = false
+
+ def snapshotRequested(evt: Event): Boolean = {
+ val snapshotFromRetention = retention match {
+ case DisabledRetentionCriteria => false
+ case s: SnapshotCountRetentionCriteriaImpl => s.snapshotWhen(sequenceNr)
+ case unexpected => throw new IllegalStateException(s"Unexpected retention criteria: $unexpected")
+ }
+
+ snapshotFromRetention || snapshotWhen(state, evt, sequenceNr)
+ }
+
+ @tailrec
+ def applyEffects(curEffect: EffectImpl[Event, State], sideEffects: immutable.Seq[SideEffect[State]]): Unit =
+ curEffect match {
+ case CompositeEffect(eff: EffectImpl[Event, State], se) =>
+ applyEffects(eff, se ++ sideEffects)
+
+ case Persist(event) =>
+ sequenceNr += 1
+ state = eventHandler(state, event)
+ onEvent(event, sequenceNr, tagger(event))
+ shouldSnapshot = shouldSnapshot || snapshotRequested(event)
+ sideEffect(sideEffects)
+
+ case PersistAll(events) =>
+ val eventsWithSeqNrs =
+ events.map { event =>
+ sequenceNr += 1
+ state = eventHandler(state, event)
+ event -> sequenceNr
+ }
+
+ eventsWithSeqNrs.foreach {
+ case (event, seqNr) =>
+ // technically doesn't persist them atomically, but in tests that shouldn't matter
+ onEvent(event, seqNr, tagger(event))
+ shouldSnapshot = shouldSnapshot || snapshotRequested(event)
+ }
+
+ sideEffect(sideEffects)
+
+ // From outside of the behavior, these are equivalent: no state update
+ case _: PersistNothing.type | _: Unhandled.type =>
+ sideEffect(sideEffects)
+
+ case _: Stash.type =>
+ stashedCommands.append(cmd)
+ sideEffect(sideEffects)
+
+ case _ =>
+ context.log.error("Unexpected effect {}, stopping", curEffect)
+ Behaviors.stopped
+ }
+
+ def sideEffect(sideEffects: immutable.Seq[SideEffect[State]]): Unit =
+ sideEffects.iterator.foreach { effect =>
+ effect match {
+ case _: Stop.type => shouldStop = true
+ case _: UnstashAll.type => shouldUnstash = true
+ case cb: Callback[_] => cb.sideEffect(state)
+ }
+ }
+
+ applyEffects(commandHandler(state, cmd).asInstanceOf[EffectImpl[Event, State]], Nil)
+
+ if (shouldSnapshot) {
+ onSnapshot(state, sequenceNr)
+ sendSignal(SnapshotCompleted(snapshotMetadata()))
+ }
+
+ if (shouldStop) Behaviors.stopped
+ else if (shouldUnstash && stashedCommands.nonEmpty) {
+ val numStashed = stashedCommands.length
+ val thisWrappedBehavior = this
+
+ Behaviors.setup { _ =>
+ Behaviors.withStash(numStashed) { stash =>
+ stashedCommands.foreach { sc =>
+ stash.stash(sc)
+ ()
+ }
+
+ stashedCommands.remove(0, numStashed)
+ stash.unstashAll(thisWrappedBehavior)
+ }
+ }
+ } else this
+ }
+ }
+
+ private class WrappedDurableStateBehavior[Command, State](
+ context: ActorContext[Command],
+ dsBehavior: DurableStateBehaviorImpl[Command, State],
+ initialState: State,
+ onPersist: (State, Long, String) => Unit)
+ extends AbstractBehavior[Command](context)
+ with WithRevisionAccessible {
+
+ import pekko.persistence.typed.state.{ DurableStateSignal, RecoveryCompleted }
+ import pekko.persistence.typed.state.internal._
+
+ override def currentRevision: Long = sequenceNr
+
+ private def commandHandler = dsBehavior.commandHandler
+ private def signalHandler = dsBehavior.signalHandler
+ private val tag = dsBehavior.tag
+
+ private var sequenceNr: Long = 0
+ private var state: State = initialState
+ private val stashedCommands = ListBuffer.empty[Command]
+
+ private def sendSignal(signal: DurableStateSignal): Unit =
+ signalHandler.applyOrElse(state -> signal, doNothing)
+
+ sendSignal(RecoveryCompleted)
+
+ override def onMessage(cmd: Command): Behavior[Command] = {
+ var shouldUnstash = false
+ var shouldStop = false
+
+ def persistState(st: State): Unit = {
+ sequenceNr += 1
+ onPersist(st, sequenceNr, tag)
+ state = st
+ }
+
+ @tailrec
+ def applyEffects(curEffect: EffectImpl[State], sideEffects: immutable.Seq[SideEffect[State]]): Unit =
+ curEffect match {
+ case CompositeEffect(eff: EffectImpl[_], se) =>
+ applyEffects(eff.asInstanceOf[EffectImpl[State]], se ++ sideEffects)
+
+ case Persist(st) =>
+ persistState(st)
+ sideEffect(sideEffects)
+
+ case _: PersistNothing.type | _: Unhandled.type =>
+ sideEffect(sideEffects)
+
+ case _: Stash.type =>
+ stashedCommands.append(cmd)
+ sideEffect(sideEffects)
+
+ case _ =>
+ context.log.error("Unexpected effect, stopping")
+ Behaviors.stopped
+ }
+
+ def sideEffect(sideEffects: immutable.Seq[SideEffect[State]]): Unit =
+ sideEffects.iterator.foreach { effect =>
+ effect match {
+ case _: Stop.type => shouldStop = true
+ case _: UnstashAll.type => shouldUnstash = true
+ case cb: Callback[_] => cb.sideEffect(state)
+ }
+ }
+
+ applyEffects(commandHandler(state, cmd).asInstanceOf[EffectImpl[State]], Nil)
+
+ if (shouldStop) Behaviors.stopped
+ else if (shouldUnstash && stashedCommands.nonEmpty) {
+ val numStashed = stashedCommands.length
+ val thisWrappedBehavior = this
+
+ Behaviors.setup { _ =>
+ Behaviors.withStash(numStashed) { stash =>
+ stashedCommands.foreach { sc =>
+ stash.stash(sc)
+ () // explicit discard
+ }
+ stashedCommands.remove(0, numStashed)
+ stash.unstashAll(thisWrappedBehavior)
+ }
+ }
+ } else this
+ }
+ }
+}
+
+/**
+ * INTERNAL API
+ */
+@InternalApi
+private[pekko] class PersistenceProbeImpl[T] {
+ type Element = (T, Long, Set[String])
+
+ val queue = new ConcurrentLinkedQueue[Element]()
+
+ def persist(elem: Element): Unit = { queue.offer(elem); () }
+
+ def rawExtract(): Element =
+ queue.poll() match {
+ case null => throw new AssertionError("No persistence effects in probe")
+ case elem => elem
+ }
+
+ def asScala: scaladsl.PersistenceProbe[T] =
+ new scaladsl.PersistenceProbe[T] {
+ import scaladsl.{ PersistenceEffect, PersistenceProbe }
+
+ def drain(): Seq[PersistenceEffect[T]] = {
+ @annotation.tailrec
+ def iter(acc: List[PersistenceEffect[T]]): List[PersistenceEffect[T]] = {
+ val elem = queue.poll()
+ if (elem == null) acc else iter(persistenceEffect(elem) :: acc)
+ }
+
+ iter(Nil).reverse
+ }
+
+ def extract(): PersistenceEffect[T] = persistenceEffect(rawExtract())
+
+ def expectPersistedType[S <: T: ClassTag](): PersistenceEffect[S] =
+ rawExtract() match {
+ case (obj: S, sequenceNr, tags) => PersistenceEffect(obj, sequenceNr, tags)
+ case (extracted, _, _) =>
+ throw new AssertionError(
+ s"Expected object of type [${implicitly[ClassTag[S]].runtimeClass.getName}] to be persisted, " +
+ s"but actual was of type [${extracted.getClass.getName}]")
+ }
+
+ def hasEffects: Boolean = !queue.isEmpty
+
+ def expectPersisted(obj: T): PersistenceProbe[T] =
+ rawExtract() match {
+ case (persistedObj, _, _) if obj == persistedObj => this
+ case (persistedObj, _, _) =>
+ throw new AssertionError(s"Expected object [$obj] to be persisted, but actual was [$persistedObj]")
+ }
+
+ def expectPersisted(obj: T, tag: String): PersistenceProbe[T] =
+ rawExtract() match {
+ case (persistedObj, _, tags) if (obj == persistedObj) && (tags(tag)) => this
+
+ case (persistedObj, _, tags) if obj == persistedObj =>
+ throw new AssertionError(
+ s"Expected persistence with tag [$tag], but actual tags were [${tags.mkString(",")}]")
+
+ case (persistedObj, _, tags) if tags(tag) =>
+ throw new AssertionError(s"Expected object [$obj] to be persisted, but actual was [$persistedObj]")
+
+ case (persistedObj, _, tags) =>
+ throw new AssertionError(
+ s"Expected object [$obj] to be persisted with tag [$tag], " +
+ s"but actual object was [$persistedObj] with tags [${tags.mkString(",")}]")
+ }
+
+ def expectPersisted(obj: T, tags: Set[String]): PersistenceProbe[T] =
+ rawExtract() match {
+ case (persistedObj, _, persistedTags) if (obj == persistedObj) && (tags == persistedTags) => this
+ case (persistedObj, _, persistedTags) if obj == persistedObj =>
+ val unexpected = persistedTags.diff(tags)
+ val notPersistedWith = tags.diff(persistedTags)
+
+ throw new AssertionError(
+ s"Expected persistence with [${tags.mkString(",")}], " +
+ s"but saw unexpected actual tags [${unexpected.mkString(",")}] and " +
+ s"did not see actual tags [${notPersistedWith.mkString(",")}]")
+
+ case (persistedObj, _, persistedTags) if tags == persistedTags =>
+ throw new AssertionError(s"Expected object [$obj] to be persisted, but actual was [$persistedObj}]")
+
+ case (persistedObj, _, persistedTags) =>
+ throw new AssertionError(
+ s"Expected object [$obj] to be persisted with tags [${tags.mkString(",")}], " +
+ s"but actual object was [$persistedObj] with tags [${persistedTags.mkString(",")}]")
+ }
+
+ private def persistenceEffect(elem: Element): PersistenceEffect[T] =
+ PersistenceEffect(elem._1, elem._2, elem._3)
+ }
+
+ def asJava: javadsl.PersistenceProbe[T] =
+ new javadsl.PersistenceProbe[T] {
+ import java.util.{ List => JList, Set => JSet }
+
+ import javadsl.{ PersistenceEffect, PersistenceProbe }
+
+ def drain(): JList[PersistenceEffect[T]] = {
+ @annotation.tailrec
+ def iter(acc: List[PersistenceEffect[T]]): List[PersistenceEffect[T]] = {
+ val elem = queue.poll()
+ if (elem == null) acc else iter(persistenceEffect(elem) :: acc)
+ }
+ iter(Nil).reverse.asJava
+ }
+
+ def extract(): PersistenceEffect[T] = persistenceEffect(rawExtract())
+
+ def expectPersistedClass[S <: T](clazz: Class[S]): PersistenceEffect[S] =
+ rawExtract() match {
+ case (obj, sequenceNr, tags) if clazz.isInstance(obj) =>
+ PersistenceEffect(clazz.cast(obj), sequenceNr, tags.asJava)
+
+ case (extracted, _, _) =>
+ throw new AssertionError(
+ s"Expected object of type [${clazz.getName}] to be persisted, " +
+ s"but actual was of type [${extracted.getClass.getName}]")
+ }
+
+ def hasEffects: Boolean = !queue.isEmpty
+
+ def expectPersisted(obj: T): PersistenceProbe[T] =
+ rawExtract() match {
+ case (persistedObj, _, _) if obj == persistedObj => this
+ case (persistedObj, _, _) =>
+ throw new AssertionError(s"Expected object [$obj] to be persisted, but actual was [$persistedObj]")
+ }
+
+ def expectPersisted(obj: T, tag: String): PersistenceProbe[T] =
+ rawExtract() match {
+ case (persistedObj, _, tags) if (obj == persistedObj) && (tags(tag)) => this
+
+ case (persistedObj, _, tags) if obj == persistedObj =>
+ throw new AssertionError(
+ s"Expected persistence with tag [$tag], but actual tags were [${tags.mkString(",")}]")
+
+ case (persistedObj, _, tags) if tags(tag) =>
+ throw new AssertionError(s"Expected object [$obj] to be persisted, but actual was [$persistedObj]")
+
+ case (persistedObj, _, tags) =>
+ throw new AssertionError(
+ s"Expected object [$obj] to be persisted with tag [$tag], " +
+ s"but actual object was [$persistedObj] with tags [${tags.mkString(",")}]")
+ }
+
+ def expectPersisted(obj: T, tags: JSet[String]): PersistenceProbe[T] = {
+ val sTags = tags.asScala
+
+ // Not sure if a Java Set after asScala-ing will compare equal to a Scala Set...
+ def sameTags(persistedTags: Set[String]): Boolean =
+ sTags.forall(persistedTags) && persistedTags.forall(sTags)
+
+ rawExtract() match {
+ case (persistedObj, _, persistedTags) if (obj == persistedObj) && sameTags(persistedTags) => this
+
+ case (persistedObj, _, persistedTags) if obj == persistedObj =>
+ val unexpected = persistedTags.diff(sTags)
+ val notPersistedWith = sTags.diff(persistedTags)
+
+ throw new AssertionError(
+ s"Expected persistence with [${sTags.mkString(",")}], " +
+ s"but saw unexpected actual tags [${unexpected.mkString(",")}] and " +
+ s"did not see actual tags [${notPersistedWith.mkString(",")}]")
+
+ case (persistedObj, _, persistedTags) if sameTags(persistedTags) =>
+ throw new AssertionError(s"Expected object [$obj] to be persisted, but actual was [$persistedObj}]")
+
+ case (persistedObj, _, persistedTags) =>
+ throw new AssertionError(
+ s"Expected object [$obj] to be persisted with tags [${sTags.mkString(",")}], " +
+ s"but actual object was [$persistedObj] with tags [${persistedTags.mkString(",")}]")
+ }
+ }
+
+ private def persistenceEffect(element: Element): PersistenceEffect[T] =
+ PersistenceEffect(element._1, element._2, element._3.asJava)
+ }
+}
diff --git a/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/javadsl/UnpersistentBehavior.scala b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/javadsl/UnpersistentBehavior.scala
new file mode 100644
index 0000000000..71c78195fc
--- /dev/null
+++ b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/javadsl/UnpersistentBehavior.scala
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2022 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.testkit.javadsl
+
+import java.util.{ List, Set }
+
+import scala.collection.immutable.{ Set => ScalaSet }
+
+import org.apache.pekko
+import pekko.actor.testkit.typed.javadsl.BehaviorTestKit
+import pekko.actor.typed.Behavior
+import pekko.annotation.DoNotInherit
+import pekko.persistence.testkit.internal.{ PersistenceProbeImpl, Unpersistent }
+
+/**
+ * Factory methods to create UnpersistentBehavior instances for testing.
+ *
+ * @since 1.3.0
+ */
+object UnpersistentBehavior {
+
+ /**
+ * Given an EventSourcedBehavior, produce a non-persistent Behavior which synchronously publishes events and snapshots
+ * for inspection. State is updated as in the EventSourcedBehavior, and side effects are performed synchronously. The
+ * resulting Behavior is, contingent on the command handling, event handling, and side effects being compatible with the
+ * BehaviorTestKit, testable with the BehaviorTestKit.
+ *
+ * The returned Behavior does not intrinsically depend on configuration: it therefore does not serialize and
+ * assumes an unbounded stash for commands.
+ *
+ * @param behavior a (possibly wrapped) EventSourcedBehavior to serve as the basis for the unpersistent behavior
+ * @param initialState start the unpersistent behavior with this state; if null, behavior's initialState will be used
+ * @param initialSequenceNr start the unpersistent behavior with this sequence number; only applies if initialState is non-null
+ * @return an UnpersistentBehavior based on an EventSourcedBehavior
+ */
+ def fromEventSourced[Command, Event, State](
+ behavior: Behavior[Command],
+ initialState: State,
+ initialSequenceNr: Long): UnpersistentBehavior[Command, Event, State] = {
+ require(initialSequenceNr >= 0, "initialSequenceNr must be at least zero")
+
+ val initialStateAndSequenceNr = Option(initialState).map(_ -> initialSequenceNr)
+ val eventProbe = new PersistenceProbeImpl[Event]
+ val snapshotProbe = new PersistenceProbeImpl[State]
+
+ val b =
+ Unpersistent.eventSourced(behavior, initialStateAndSequenceNr) {
+ (event: Event, sequenceNr: Long, tags: ScalaSet[String]) =>
+ eventProbe.persist((event, sequenceNr, tags))
+ } { (snapshot, sequenceNr) =>
+ snapshotProbe.persist((snapshot, sequenceNr, ScalaSet.empty))
+ }
+
+ new UnpersistentBehavior(b, eventProbe.asJava, snapshotProbe.asJava)
+ }
+
+ def fromEventSourced[Command, Event, State](
+ behavior: Behavior[Command]): UnpersistentBehavior[Command, Event, State] =
+ fromEventSourced(behavior, null.asInstanceOf[State], 0)
+
+ def fromDurableState[Command, State](
+ behavior: Behavior[Command],
+ initialState: State): UnpersistentBehavior[Command, Void, State] = {
+ val probe = new PersistenceProbeImpl[State]
+ val b =
+ Unpersistent.durableState(behavior, Option(initialState)) { (state, version, tag) =>
+ probe.persist((state, version, if (tag == "") ScalaSet.empty else ScalaSet(tag)))
+ }
+
+ new UnpersistentBehavior(b, noEventProbe, probe.asJava)
+ }
+
+ def fromDurableState[Command, State](behavior: Behavior[Command]): UnpersistentBehavior[Command, Void, State] =
+ fromDurableState(behavior, null.asInstanceOf[State])
+
+ private val noEventProbe: PersistenceProbe[Void] =
+ new PersistenceProbe[Void] {
+ def drain(): List[PersistenceEffect[Void]] =
+ // could return an empty list, but the intent is that any use of this probe should fail the test
+ boom()
+
+ def extract(): PersistenceEffect[Void] = boom()
+ def expectPersistedClass[S <: Void](clazz: Class[S]): PersistenceEffect[S] = boom()
+ def hasEffects: Boolean = boom()
+ def expectPersisted(obj: Void): PersistenceProbe[Void] = boom()
+ def expectPersisted(obj: Void, tag: String): PersistenceProbe[Void] = boom()
+ def expectPersisted(obj: Void, tags: Set[String]): PersistenceProbe[Void] = boom()
+
+ private def boom() = throw new AssertionError("No events were persisted")
+ }
+}
+
+final class UnpersistentBehavior[Command, Event, State] private (
+ behavior: Behavior[Command],
+ eventProbe: PersistenceProbe[Event],
+ stateProbe: PersistenceProbe[State]) {
+ def getBehavior(): Behavior[Command] = behavior
+ def getBehaviorTestKit(): BehaviorTestKit[Command] = btk
+
+ /** Note: durable state behaviors will not publish events to this probe */
+ def getEventProbe(): PersistenceProbe[Event] = eventProbe
+
+ def getStateProbe(): PersistenceProbe[State] = stateProbe
+ def getSnapshotProbe(): PersistenceProbe[State] = stateProbe
+
+ private lazy val btk = BehaviorTestKit.create(behavior)
+}
+
+final case class PersistenceEffect[T](persistedObject: T, sequenceNr: Long, tags: Set[String])
+
+/**
+ * Not for user extension
+ */
+@DoNotInherit
+trait PersistenceProbe[T] {
+
+ /** Collect all persistence effects from the probe and empty the probe */
+ def drain(): List[PersistenceEffect[T]]
+
+ /** Get and remove the oldest persistence effect from the probe */
+ def extract(): PersistenceEffect[T]
+
+ /**
+ * Get and remove the oldest persistence effect from the probe, failing if the
+ * persisted object is not of the requested type
+ */
+ def expectPersistedClass[S <: T](clazz: Class[S]): PersistenceEffect[S]
+
+ /** Are there any persistence effects */
+ def hasEffects: Boolean
+
+ /**
+ * Assert that the given object was persisted in the oldest persistence effect and
+ * remove that persistence effect
+ */
+ def expectPersisted(obj: T): PersistenceProbe[T]
+
+ /**
+ * Assert that the given object was persisted with the given tag in the oldest persistence
+ * effect and remove that persistence effect. If the persistence effect has multiple tags,
+ * only one of them has to match in order for the assertion to succeed.
+ */
+ def expectPersisted(obj: T, tag: String): PersistenceProbe[T]
+
+ /**
+ * Assert that the given object was persisted with the given tag in the oldest persistence
+ * effect and remove that persistence effect. If the persistence effect has tags which are
+ * not given, the assertion fails.
+ */
+ def expectPersisted(obj: T, tags: Set[String]): PersistenceProbe[T]
+}
diff --git a/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/scaladsl/UnpersistentBehavior.scala b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/scaladsl/UnpersistentBehavior.scala
new file mode 100644
index 0000000000..72405f0e27
--- /dev/null
+++ b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/scaladsl/UnpersistentBehavior.scala
@@ -0,0 +1,144 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2022 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.testkit.scaladsl
+
+import scala.reflect.ClassTag
+
+import org.apache.pekko
+import pekko.actor.testkit.typed.scaladsl.BehaviorTestKit
+import pekko.actor.typed.Behavior
+import pekko.annotation.DoNotInherit
+import pekko.persistence.testkit.internal.{ PersistenceProbeImpl, Unpersistent }
+
+sealed trait UnpersistentBehavior[Command, State] {
+ val behavior: Behavior[Command]
+ lazy val behaviorTestKit = BehaviorTestKit(behavior)
+
+ def stateProbe: PersistenceProbe[State]
+}
+
+/**
+ * Factory methods to create UnpersistentBehavior instances for testing.
+ *
+ * @since 1.3.0
+ */
+object UnpersistentBehavior {
+
+ /**
+ * Given an EventSourcedBehavior, produce a non-persistent Behavior which synchronously publishes events and snapshots
+ * for inspection. State is updated as in the EventSourcedBehavior, and side effects are performed synchronously. The
+ * resulting Behavior is, contingent on the command handling, event handling, and side effects being compatible with the
+ * BehaviorTestKit, testable with the BehaviorTestKit.
+ *
+ * The returned Behavior does not intrinsically depend on configuration: it therefore does not serialize and assumes an
+ * unbounded stash for commands.
+ */
+ def fromEventSourced[Command, Event, State](
+ behavior: Behavior[Command],
+ initialStateAndSequenceNr: Option[(State, Long)] = None): EventSourced[Command, Event, State] = {
+ val eventProbe = new PersistenceProbeImpl[Event]
+ val snapshotProbe = new PersistenceProbeImpl[State]
+ val resultingBehavior =
+ Unpersistent.eventSourced(behavior, initialStateAndSequenceNr) {
+ (event: Event, sequenceNr: Long, tags: Set[String]) =>
+ eventProbe.persist((event, sequenceNr, tags))
+ } { (snapshot, sequenceNr) =>
+ snapshotProbe.persist((snapshot, sequenceNr, Set.empty))
+ }
+
+ EventSourced(resultingBehavior, eventProbe.asScala, snapshotProbe.asScala)
+ }
+
+ def fromEventSourced[Command, Event, State](
+ behavior: Behavior[Command],
+ initialState: State): EventSourced[Command, Event, State] =
+ fromEventSourced(behavior, Some(initialState -> 0L))
+
+ def fromDurableState[Command, State](
+ behavior: Behavior[Command],
+ initialState: Option[State] = None): DurableState[Command, State] = {
+ val probe = new PersistenceProbeImpl[State]
+
+ val resultingBehavior =
+ Unpersistent.durableState(behavior, initialState) { (state, version, tag) =>
+ probe.persist((state, version, if (tag.isEmpty) Set.empty else Set(tag)))
+ }
+
+ DurableState(resultingBehavior, probe.asScala)
+ }
+
+ final case class EventSourced[Command, Event, State](
+ override val behavior: Behavior[Command],
+ val eventProbe: PersistenceProbe[Event],
+ override val stateProbe: PersistenceProbe[State])
+ extends UnpersistentBehavior[Command, State] {
+ def apply(f: (BehaviorTestKit[Command], PersistenceProbe[Event], PersistenceProbe[State]) => Unit): Unit =
+ f(behaviorTestKit, eventProbe, stateProbe)
+
+ def snapshotProbe: PersistenceProbe[State] = stateProbe
+ }
+
+ final case class DurableState[Command, State](
+ override val behavior: Behavior[Command],
+ override val stateProbe: PersistenceProbe[State])
+ extends UnpersistentBehavior[Command, State] {
+ def apply(f: (BehaviorTestKit[Command], PersistenceProbe[State]) => Unit): Unit =
+ f(behaviorTestKit, stateProbe)
+ }
+}
+
+final case class PersistenceEffect[T](persistedObject: T, sequenceNr: Long, tags: Set[String])
+
+/**
+ * Not for user extension
+ */
+@DoNotInherit
+trait PersistenceProbe[T] {
+
+ /** Collect all persistence effects from the probe and empty the probe */
+ def drain(): Seq[PersistenceEffect[T]]
+
+ /** Get and remove the oldest persistence effect from the probe */
+ def extract(): PersistenceEffect[T]
+
+ /**
+ * Get and remove the oldest persistence effect from the probe, failing if the
+ * persisted object is not of the requested type
+ */
+ def expectPersistedType[S <: T: ClassTag](): PersistenceEffect[S]
+
+ /** Are there any persistence effects? */
+ def hasEffects: Boolean
+
+ /**
+ * Assert that the given object was persisted in the oldest persistence effect and
+ * remove that persistence effect
+ */
+ def expectPersisted(obj: T): PersistenceProbe[T]
+
+ /**
+ * Assert that the given object was persisted with the given tag in the oldest
+ * persistence effect and remove that persistence effect. If the persistence
+ * effect has multiple tags, only one of them has to match in order for the
+ * assertion to succeed.
+ */
+ def expectPersisted(obj: T, tag: String): PersistenceProbe[T]
+
+ /**
+ * Assert that the given object was persisted with the given tags in the oldest
+ * persistence effect and remove that persistence effect. If the persistence
+ * effect has tags which are not given, the assertion fails.
+ */
+ def expectPersisted(obj: T, tags: Set[String]): PersistenceProbe[T]
+}
diff --git a/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/scaladsl/UnpersistentDurableStateSpec.scala b/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/scaladsl/UnpersistentDurableStateSpec.scala
new file mode 100644
index 0000000000..33efc52814
--- /dev/null
+++ b/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/scaladsl/UnpersistentDurableStateSpec.scala
@@ -0,0 +1,263 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2022 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.testkit.scaladsl
+
+import org.apache.pekko
+import pekko.Done
+import pekko.actor.typed.{ Behavior, RecipientRef }
+import pekko.actor.typed.scaladsl.{ ActorContext, Behaviors }
+import pekko.persistence.typed.PersistenceId
+import pekko.persistence.typed.state.RecoveryCompleted
+import pekko.persistence.typed.state.scaladsl._
+
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+object UnpersistentDurableStateSpec {
+ object BehaviorUnderTest {
+ sealed trait Command
+
+ case class Add(n: Int, replyTo: RecipientRef[Done]) extends Command
+ case class AddIfLessThan(toAdd: Int, ifLessThan: Int, replyTo: RecipientRef[Boolean]) extends Command
+ case class AddWhenAtLeast(toAdd: Int, whenAtLeast: Int, replyTo: RecipientRef[Done]) extends Command
+ case class NotifyIfAtLeast(n: Int, notifyTo: RecipientRef[Done], replyTo: RecipientRef[Boolean]) extends Command
+ case class GetRevisionNumber(replyTo: RecipientRef[Long]) extends Command
+
+ case class State(count: Int, notifyAfter: Map[Int, RecipientRef[Done]], nextNotifyAt: Int) {
+ def processAdd(n: Int): State = {
+ val nextCount = count + n
+
+ if (nextCount < nextNotifyAt) copy(count = nextCount)
+ else {
+ import scala.collection.mutable
+
+ val (nextNNA, nextNotifyAfter) = {
+ var lowestNotifyAt = Int.MaxValue
+ val inProgress = mutable.Map.empty[Int, RecipientRef[Done]]
+
+ notifyAfter.keysIterator.foreach { at =>
+ if (at > nextCount) {
+ lowestNotifyAt = lowestNotifyAt.min(at)
+ inProgress += (at -> notifyAfter(at))
+ }
+ }
+
+ lowestNotifyAt -> inProgress.toMap
+ }
+
+ copy(count = nextCount, notifyAfter = nextNotifyAfter, nextNotifyAt = nextNNA)
+ }
+ }
+
+ def addObserver(at: Int, notifyTo: RecipientRef[Done]): State = {
+ val nextNNA = nextNotifyAt.min(at)
+ val nextNotifyAfter = notifyAfter.updated(at, notifyTo)
+
+ copy(notifyAfter = nextNotifyAfter, nextNotifyAt = nextNNA)
+ }
+ }
+
+ def apply(id: String, recoveryDone: RecipientRef[Done]): Behavior[Command] =
+ Behaviors.setup { context =>
+ context.setLoggerName(s"entity-$id")
+
+ DurableStateBehavior[Command, State](
+ persistenceId = PersistenceId.ofUniqueId(id),
+ emptyState = State(0, Map.empty, Int.MaxValue),
+ commandHandler = applyCommand(_, _, context))
+ .receiveSignal {
+ case (_, RecoveryCompleted) =>
+ recoveryDone ! Done
+ }
+ .withTag("count")
+ }
+
+ private def applyCommand(state: State, cmd: Command, context: ActorContext[Command]): Effect[State] = {
+ def persistAdd[Reply](n: Int, replyTo: RecipientRef[Reply], reply: Reply): Effect[State] = {
+ val newState = state.processAdd(n)
+
+ Effect
+ .persist(newState)
+ .thenRun { nextState => // should be the same as newState, but...
+ state.notifyAfter.keysIterator
+ .filter { at =>
+ (at <= nextState.nextNotifyAt) && !nextState.notifyAfter.isDefinedAt(at)
+ }
+ .foreach { at =>
+ state.notifyAfter(at) ! Done
+ }
+
+ replyTo ! reply
+ }
+ .thenUnstashAll()
+ }
+
+ cmd match {
+ case Add(n, replyTo) => persistAdd(n, replyTo, Done)
+
+ case AddIfLessThan(toAdd, ifLessThan, replyTo) =>
+ if (state.count >= ifLessThan) {
+ context.log.info("Rejecting AddIfLessThan as count = {}", state.count)
+ Effect.none[State].thenRun(_ => replyTo ! false)
+ } else persistAdd(toAdd, replyTo, true)
+
+ case AddWhenAtLeast(toAdd, whenAtLeast, replyTo) =>
+ if (state.count < whenAtLeast) Effect.stash()
+ else persistAdd(toAdd, replyTo, Done)
+
+ case NotifyIfAtLeast(n, notifyTo, replyTo) =>
+ if (state.count >= n) {
+ Effect.none[State].thenRun { _ =>
+ notifyTo ! Done
+ replyTo ! true
+ }
+ } else if (state.notifyAfter.isDefinedAt(n)) {
+ Effect.none[State].thenRun(_ => replyTo ! false)
+ } else {
+ Effect.persist(state.addObserver(n, notifyTo)).thenRun(_ => replyTo ! true)
+ }
+
+ case GetRevisionNumber(replyTo) =>
+ Effect.none[State].thenRun(_ => replyTo ! DurableStateBehavior.lastSequenceNumber(context))
+ }
+ }
+ }
+}
+
+class UnpersistentDurableStateSpec extends AnyWordSpec with Matchers {
+ import UnpersistentDurableStateSpec._
+
+ import pekko.actor.testkit.typed.scaladsl._
+
+ import org.slf4j.event.Level
+
+ "Unpersistent DurableStateBehavior" must {
+ "generate a fail-fast behavior from a non-DurableStateBehavior" in {
+ val notDurableState =
+ Behaviors.receive[Any] { (context, msg) =>
+ context.log.info("Got message {}", msg)
+ Behaviors.same
+ }
+
+ val unpersistent = UnpersistentBehavior.fromDurableState[Any, Any](notDurableState)
+ an[AssertionError] shouldBe thrownBy { unpersistent.behaviorTestKit }
+ assert(!unpersistent.stateProbe.hasEffects, "should be no persistence effects")
+ }
+
+ "generate a Behavior from a DurableStateBehavior and process RecoveryCompleted" in {
+ import BehaviorUnderTest._
+
+ val recoveryDone = TestInbox[Done]()
+ val behavior = BehaviorUnderTest("test-1", recoveryDone.ref)
+
+ // accessor-style API
+ val unpersistent = UnpersistentBehavior.fromDurableState[Command, State](behavior)
+ val probe = unpersistent.stateProbe
+ val testkit = unpersistent.behaviorTestKit
+
+ assert(!probe.hasEffects, "should not be persistence yet")
+ recoveryDone.expectMessage(Done)
+ val logs = testkit.logEntries()
+ logs.size shouldBe 1
+ logs.head.level shouldBe Level.DEBUG
+ logs.head.message shouldBe s"Recovered state for id [test-1] is [${State(0, Map.empty, Int.MaxValue)}]"
+ }
+
+ "publish state changes in response to commands" in {
+ import BehaviorUnderTest._
+
+ val behavior = BehaviorUnderTest("test-1", TestInbox[Done]().ref)
+ val replyTo = TestInbox[Done]()
+
+ // and the more functional-style API
+ UnpersistentBehavior.fromDurableState[Command, State](behavior) { (testkit, probe) =>
+ testkit.run(Add(1, replyTo.ref))
+ replyTo.expectMessage(Done)
+ probe.expectPersisted(State(1, Map.empty, Int.MaxValue), tag = "count")
+ assert(!testkit.hasEffects(), "should have no actor effects")
+ }
+ }
+
+ "allow a state to be injected" in {
+ import BehaviorUnderTest._
+
+ val behavior = BehaviorUnderTest("test-1", TestInbox[Done]().ref)
+ val notify3 = TestInbox[Done]()
+ val initialState = State(1, Map(3 -> notify3.ref), 3)
+
+ UnpersistentBehavior.fromDurableState[Command, State](behavior, Some(initialState)) { (testkit, probe) =>
+ val logs = testkit.logEntries()
+
+ logs.size shouldBe 1
+ logs.head.level shouldBe Level.DEBUG
+ logs.head.message shouldBe s"Recovered state for id [test-1] is [$initialState]"
+ assert(!probe.hasEffects, "should be no persistence effect")
+ assert(!notify3.hasMessages, "no messages should be sent to notify3")
+
+ val replyTo = TestInbox[Done]()
+ testkit.run(AddWhenAtLeast(2, 2, replyTo.ref))
+ assert(!replyTo.hasMessages, "no messages should be sent now")
+ assert(!notify3.hasMessages, "no messages should be sent to notify3")
+ assert(!probe.hasEffects, "should be no persistence effect")
+ assert(!testkit.hasEffects(), "should be no testkit effects")
+
+ testkit.run(Add(3, TestInbox[Done]().ref))
+ replyTo.expectMessage(Done)
+ notify3.expectMessage(Done)
+ assert(!testkit.hasEffects(), "should be no testkit effects")
+ probe.drain() should contain theSameElementsInOrderAs Seq(
+ PersistenceEffect(State(4, Map.empty, Int.MaxValue), 1, Set("count")),
+ PersistenceEffect(State(6, Map.empty, Int.MaxValue), 2, Set("count")))
+ }
+ }
+
+ "stash and unstash properly" in {
+ import BehaviorUnderTest._
+
+ val behavior = BehaviorUnderTest("test-1", TestInbox[Done]().ref)
+ val replyTo1 = TestInbox[Done]()
+ val add = Add(1, TestInbox[Done]().ref)
+
+ UnpersistentBehavior.fromDurableState[Command, State](behavior) { (testkit, probe) =>
+ // stashes
+ testkit.run(AddWhenAtLeast(1, 1, replyTo1.ref))
+ assert(!probe.hasEffects, "should be no persistence effect")
+ assert(!replyTo1.hasMessages, "count is not yet 1")
+
+ // unstashes
+ testkit.run(add)
+ replyTo1.expectMessage(Done)
+ probe.drain() shouldNot be(empty)
+
+ // unstash but nothing in the stash
+ testkit.run(add)
+ assert(!replyTo1.hasMessages, "should not send again")
+ probe.drain() shouldNot be(empty)
+ }
+ }
+
+ "retrieve revision number" in {
+ import BehaviorUnderTest._
+
+ val behavior = BehaviorUnderTest("test-1", TestInbox[Done]().ref)
+
+ val replyTo = TestInbox[Long]()
+ UnpersistentBehavior.fromDurableState[Command, State](behavior) { (testkit, probe) =>
+ testkit.run(GetRevisionNumber(replyTo.ref))
+ (the[AssertionError] thrownBy (probe.extract())).getMessage shouldBe "No persistence effects in probe"
+ replyTo.expectMessage(0)
+ }
+ }
+ }
+}
diff --git a/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/scaladsl/UnpersistentEventSourcedSpec.scala b/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/scaladsl/UnpersistentEventSourcedSpec.scala
new file mode 100644
index 0000000000..900aea4b4c
--- /dev/null
+++ b/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/scaladsl/UnpersistentEventSourcedSpec.scala
@@ -0,0 +1,336 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2022 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.testkit.scaladsl
+
+import org.apache.pekko
+import pekko.Done
+import pekko.actor.typed.{ Behavior, RecipientRef }
+import pekko.actor.typed.scaladsl.{ ActorContext, Behaviors }
+import pekko.persistence.typed.{ PersistenceId, RecoveryCompleted }
+import pekko.persistence.typed.scaladsl._
+
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+object UnpersistentEventSourcedSpec {
+ object BehaviorUnderTest {
+ sealed trait Command
+
+ case class PersistDomainEvent(replyTo: RecipientRef[Done]) extends Command
+ case class PersistDomainEventIfBefore(count: Int, replyTo: RecipientRef[Boolean]) extends Command
+ case class PersistDomainEventAfter(count: Int, replyTo: RecipientRef[Done]) extends Command
+ case class NotifyAfter(count: Int, notifyTo: RecipientRef[Done], replyTo: RecipientRef[Boolean]) extends Command
+ case class GetSequenceNumber(replyTo: RecipientRef[Long]) extends Command
+ case object SnapshotNow extends Command
+
+ sealed trait Event
+ case object DomainEvent extends Event
+ case object SnapshotMade extends Event
+ case class ObserverAdded(count: Int, notifyTo: RecipientRef[Done]) extends Event
+
+ case class State(domainEvtCount: Int, notifyAfter: Map[Int, RecipientRef[Done]], nextNotifyAt: Int)
+
+ val initialState = State(0, Map.empty, Int.MaxValue)
+
+ def apply(id: String, recoveryDone: RecipientRef[Done]): Behavior[Command] =
+ Behaviors.setup { context =>
+ context.setLoggerName(s"entity-$id")
+
+ EventSourcedBehavior[Command, Event, State](
+ persistenceId = PersistenceId.ofUniqueId(id),
+ emptyState = initialState,
+ commandHandler = applyCommand(_, _, context),
+ eventHandler = applyEvent(_, _))
+ .receiveSignal {
+ case (_, RecoveryCompleted) =>
+ recoveryDone ! Done
+ }
+ .snapshotWhen {
+ case (_, SnapshotMade, _) => true
+ case _ => false
+ }
+ .withRetention(RetentionCriteria.snapshotEvery(numberOfEvents = 3, keepNSnapshots = 2))
+ .withTagger {
+ case DomainEvent => Set("domain")
+ case _ => Set.empty
+ }
+ }
+
+ private def applyCommand(state: State, cmd: Command, context: ActorContext[Command]): Effect[Event, State] = {
+ def persistDomainEvent[Reply](replyTo: RecipientRef[Reply], reply: Reply): Effect[Event, State] =
+ Effect
+ .persist[Event, State](DomainEvent)
+ .thenRun { newState =>
+ state.notifyAfter.keysIterator
+ .filter { at =>
+ (at <= newState.nextNotifyAt) && !newState.notifyAfter.isDefinedAt(at)
+ }
+ .foreach { at =>
+ state.notifyAfter(at) ! Done
+ }
+
+ replyTo ! reply
+ }
+ .thenUnstashAll()
+
+ cmd match {
+ case PersistDomainEvent(replyTo) => persistDomainEvent(replyTo, Done)
+
+ case PersistDomainEventIfBefore(count, replyTo) =>
+ if (state.domainEvtCount >= count) {
+ context.log.info("Rejecting PersistDomainEventIfBefore as domainEvtCount = {}", state.domainEvtCount)
+ Effect.none.thenRun(_ => replyTo ! false)
+ } else persistDomainEvent(replyTo, true)
+
+ case PersistDomainEventAfter(count, replyTo) =>
+ if (state.domainEvtCount < count) Effect.stash()
+ else persistDomainEvent(replyTo, Done)
+
+ case NotifyAfter(count, notifyTo, replyTo) =>
+ if (state.domainEvtCount >= count) {
+ Effect.none.thenRun { _ =>
+ notifyTo ! Done
+ replyTo ! true
+ }
+ } else if (state.notifyAfter.isDefinedAt(count)) {
+ Effect.none.thenRun(_ => replyTo ! false)
+ } else {
+ Effect.persist(ObserverAdded(count, notifyTo)).thenRun(_ => replyTo ! true)
+ }
+
+ case SnapshotNow => Effect.persist(SnapshotMade)
+
+ case GetSequenceNumber(replyTo) =>
+ Effect.none.thenRun(_ => replyTo ! EventSourcedBehavior.lastSequenceNumber(context))
+ }
+ }
+
+ private[scaladsl] def applyEvent(state: State, evt: Event): State =
+ evt match {
+ case DomainEvent =>
+ val nextDomainEvtCount = state.domainEvtCount + 1
+
+ if (nextDomainEvtCount < state.nextNotifyAt) state.copy(domainEvtCount = nextDomainEvtCount)
+ else {
+ import scala.collection.mutable
+
+ val (nextNNA, nextNotifyAfter) = {
+ var lowestNotifyAt = Int.MaxValue
+ val inProgress = mutable.Map.empty[Int, RecipientRef[Done]]
+
+ state.notifyAfter.keysIterator.foreach { at =>
+ if (at > nextDomainEvtCount) {
+ lowestNotifyAt = lowestNotifyAt.min(at)
+ inProgress += (at -> state.notifyAfter(at))
+ }
+ // else ()
+ }
+
+ lowestNotifyAt -> inProgress.toMap
+ }
+
+ state.copy(domainEvtCount = nextDomainEvtCount, notifyAfter = nextNotifyAfter, nextNotifyAt = nextNNA)
+ }
+
+ case SnapshotMade => state
+ case ObserverAdded(count, notifyTo) =>
+ val nextNNA = state.nextNotifyAt.min(count)
+ val nextNotifyAfter = state.notifyAfter.updated(count, notifyTo)
+
+ state.copy(notifyAfter = nextNotifyAfter, nextNotifyAt = nextNNA)
+ }
+ }
+}
+
+class UnpersistentEventSourcedSpec extends AnyWordSpec with Matchers {
+ import UnpersistentEventSourcedSpec._
+
+ import pekko.actor.testkit.typed.scaladsl._
+
+ import org.slf4j.event.Level
+
+ "Unpersistent EventSourcedBehavior" must {
+ "generate a failing behavior from a non-EventSourcedBehavior" in {
+ val notEventSourced =
+ Behaviors.receive[Any] { (context, msg) =>
+ context.log.info("Got message {}", msg)
+ Behaviors.same
+ }
+
+ val unpersistent = UnpersistentBehavior.fromEventSourced[Any, Any, Any](notEventSourced)
+ an[AssertionError] shouldBe thrownBy { unpersistent.behaviorTestKit }
+ an[AssertionError] shouldBe thrownBy { unpersistent.eventProbe.extract() }
+ an[AssertionError] shouldBe thrownBy { unpersistent.snapshotProbe.extract() }
+ }
+
+ "generate a Behavior from an EventSourcedBehavior and process RecoveryCompleted" in {
+ import BehaviorUnderTest._
+
+ val recoveryDone = TestInbox[Done]()
+ val behavior = BehaviorUnderTest("test-1", recoveryDone.ref)
+
+ // accessor API
+ val unpersistent = UnpersistentBehavior.fromEventSourced[Command, Event, State](behavior)
+ val testkit = unpersistent.behaviorTestKit
+ val eventProbe = unpersistent.eventProbe
+ val snapshotProbe = unpersistent.snapshotProbe
+
+ assert(!eventProbe.hasEffects, "should not be events")
+ assert(!snapshotProbe.hasEffects, "should not be snapshots")
+ recoveryDone.expectMessage(Done)
+ val logs = testkit.logEntries()
+ logs.size shouldBe 1
+ logs.head.level shouldBe Level.DEBUG
+ logs.head.message shouldBe s"Recovered state for id [test-1] is [${State(0, Map.empty, Int.MaxValue)}]"
+ }
+
+ "publish events and evolve observed state in response to commands" in {
+ import BehaviorUnderTest._
+
+ val behavior = BehaviorUnderTest("test-1", TestInbox[Done]().ref)
+ val replyTo = TestInbox[Done]()
+
+ // resource-style API
+ UnpersistentBehavior.fromEventSourced[Command, Event, State](behavior) { (testkit, eventProbe, snapshotProbe) =>
+ testkit.clearLog()
+
+ testkit.run(PersistDomainEvent(replyTo.ref))
+ replyTo.expectMessage(Done)
+ eventProbe.expectPersisted(DomainEvent, Set("domain"))
+ snapshotProbe.drain() shouldBe empty
+ assert(!testkit.hasEffects(), "should have no effects")
+ testkit.clearLog()
+
+ testkit.run(SnapshotNow)
+ assert(!replyTo.hasMessages, "should not be a reply")
+
+ val PersistenceEffect(_, seqNr, tags) = eventProbe.expectPersistedType[SnapshotMade.type]()
+ seqNr shouldBe 2
+ tags shouldBe empty
+
+ snapshotProbe.expectPersisted(State(1, Map.empty, Int.MaxValue))
+ }
+ }
+
+ "not publish events if the event handler fails" in {
+ val behavior =
+ EventSourcedBehavior[Int, Int, Unit](
+ persistenceId = PersistenceId.ofUniqueId("degenerator"),
+ emptyState = (),
+ commandHandler = { (_, cmd: Int) =>
+ cmd match {
+ case nothing if nothing < 1 => Effect.none
+ case 1 => Effect.persist(1)
+ case n => Effect.persist(Range(n, 0, -1)) // count down to 1
+ }
+ },
+ eventHandler = { (_, evt) =>
+ if (evt == 1) throw new RuntimeException("Kaboom!")
+ })
+
+ UnpersistentBehavior.fromEventSourced[Int, Int, Unit](behavior) { (testkit, eventProbe, _) =>
+ val oneException = the[RuntimeException] thrownBy testkit.run(1)
+ oneException.getMessage shouldBe "Kaboom!"
+ eventProbe.hasEffects shouldBe false
+
+ val twoException = the[RuntimeException] thrownBy testkit.run(2)
+ twoException.getMessage shouldBe "Kaboom!"
+ eventProbe.hasEffects shouldBe false
+ }
+ }
+
+ "allow a state and starting offset to be injected" in {
+ import BehaviorUnderTest._
+
+ val recoveryDone = TestInbox[Done]()
+ val behavior = BehaviorUnderTest("test-1", recoveryDone.ref)
+
+ val notify3 = TestInbox[Done]()
+ val initialState =
+ Seq(ObserverAdded(3, notify3.ref), SnapshotMade, DomainEvent, DomainEvent)
+ .foldLeft(State(0, Map.empty, Int.MaxValue))(applyEvent _)
+
+ UnpersistentBehavior.fromEventSourced[Command, Event, State](behavior, Some(initialState -> 41L)) {
+ (testkit, eventProbe, snapshotProbe) =>
+ recoveryDone.expectMessage(Done)
+ val logs = testkit.logEntries()
+ logs.size shouldBe 1
+ logs.head.level shouldBe Level.DEBUG
+ logs.head.message shouldBe s"Recovered state for id [test-1] is [$initialState]"
+ eventProbe.drain() shouldBe empty
+ snapshotProbe.drain() shouldBe empty
+ assert(!notify3.hasMessages, "no messages should be sent to notify3")
+
+ val replyTo = TestInbox[Done]()
+ testkit.run(PersistDomainEventAfter(2, replyTo.ref))
+ assert(!testkit.hasEffects(), "should be no testkit effects")
+ eventProbe.extract() shouldBe PersistenceEffect(DomainEvent, 42, Set("domain"))
+ snapshotProbe.expectPersisted(State(3, Map.empty, Int.MaxValue))
+
+ notify3.expectMessage(Done)
+ replyTo.expectMessage(Done)
+ }
+ }
+
+ "stash and unstash properly" in {
+ import BehaviorUnderTest._
+
+ val behavior = BehaviorUnderTest("test-1", TestInbox[Done]().ref)
+
+ UnpersistentBehavior.fromEventSourced[Command, Event, State](behavior, None) {
+ (testkit, eventProbe, snapshotProbe) =>
+ val replyTo1 = TestInbox[Done]()
+ val pde = PersistDomainEvent(TestInbox[Done]().ref)
+
+ // stashes
+ testkit.run(PersistDomainEventAfter(1, replyTo1.ref))
+ eventProbe.drain() shouldBe empty
+ snapshotProbe.drain() shouldBe empty
+ assert(!replyTo1.hasMessages, "have not persisted first domain event")
+
+ // unstashes
+ testkit.run(pde)
+ replyTo1.expectMessage(Done)
+
+ // unstash, but nothing in the stash
+ testkit.run(pde)
+ assert(!replyTo1.hasMessages, "should not send again")
+ }
+ }
+
+ "retrieve sequence number properly" in {
+ import BehaviorUnderTest._
+
+ val behavior = BehaviorUnderTest("test-1", TestInbox[Done]().ref)
+
+ val randomStartingOffset =
+ scala.util.Random.nextLong() match {
+ case Long.MinValue => Long.MaxValue
+ case x if x < 0 => -x
+ case x => x
+ }
+
+ UnpersistentBehavior.fromEventSourced[Command, Event, State](behavior,
+ Some(initialState -> randomStartingOffset)) {
+ (testkit, eventProbe, snapshotProbe) =>
+ val replyTo = TestInbox[Long]()
+
+ testkit.run(GetSequenceNumber(replyTo.ref))
+ eventProbe.drain() shouldBe empty
+ snapshotProbe.drain() shouldBe empty
+ replyTo.expectMessage(randomStartingOffset)
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org