This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 5379ab36b1 Add DurableStateBehaviorTestKit (#3360)
5379ab36b1 is described below
commit 5379ab36b18b96940c7faef0778ad39f33429e54
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Sun Jul 19 04:20:31 2026 +0800
Add DurableStateBehaviorTestKit (#3360)
Motivation:
DurableStateBehavior lacks the high-level actor-based test API available
for EventSourcedBehavior.
Modification:
Add matching Scala and Java DurableStateBehaviorTestKit APIs for commands,
replies, state inspection, serialization checks, restart, and clear. Isolate
the in-memory store between test kits, make persistence-id discovery work with
catch-all signal handlers, use generated Apache headers for new files, and add
documentation and directional tests.
Result:
Durable state behaviors can be exercised through a synchronous high-level
test API with isolated state and aligned Scala and Java DSLs.
Tests:
- Scala 2.13 and 3.3.8 focused DurableStateBehaviorTestKit suites (17
passed on each)
- sbt +mimaReportBinaryIssues
- sbt docs/paradox (passed with existing duplicate-anchor warnings)
- sbt checkCodeStyle
- sbt headerCreateAll +headerCheckAll
- JDK 17: sbt javafmtAll javafmtCheckAll
- scalafmt --list --mode diff-ref=origin/main
- git diff --check
- Qoder stdout review: No must-fix findings
References:
Fixes #3236
---
docs/src/main/paradox/typed/persistence-testing.md | 26 +++
.../internal/DurableStateBehaviorTestKitImpl.scala | 204 +++++++++++++++++
.../javadsl/DurableStateBehaviorTestKit.scala | 232 +++++++++++++++++++
.../scaladsl/DurableStateBehaviorTestKit.scala | 209 +++++++++++++++++
.../PersistenceTestKitDurableStateStore.scala | 8 +
.../DurableStateBehaviorTestKitApiTest.java | 139 ++++++++++++
.../DurableStateBehaviorTestKitJavaDslSpec.scala | 39 ++++
.../scaladsl/DurableStateBehaviorTestKitSpec.scala | 250 +++++++++++++++++++++
.../state/internal/DurableStateBehaviorImpl.scala | 19 +-
.../typed/state/internal/Recovering.scala | 20 +-
.../persistence/typed/state/internal/Running.scala | 48 ++--
11 files changed, 1170 insertions(+), 24 deletions(-)
diff --git a/docs/src/main/paradox/typed/persistence-testing.md
b/docs/src/main/paradox/typed/persistence-testing.md
index 17fe88475f..d1773fe11d 100644
--- a/docs/src/main/paradox/typed/persistence-testing.md
+++ b/docs/src/main/paradox/typed/persistence-testing.md
@@ -95,6 +95,32 @@ To test recovery the `restart` method of the
`EventSourcedBehaviorTestKit` can b
behavior, which will then recover from stored snapshot and events from
previous commands. It's also possible
to populate the storage with events or simulate failures by using the
underlying @apidoc[PersistenceTestKit].
+## Unit testing with the ActorTestKit and DurableStateBehaviorTestKit
+
+**Note!** The `DurableStateBehaviorTestKit` is a new feature: the API may have
changes breaking source compatibility in future versions.
+
+Unit testing of `DurableStateBehavior` can be done with the
@apidoc[DurableStateBehaviorTestKit]. It supports
+running one command at a time and asserting that the synchronously returned
state is as expected. It also has
+support for verifying the reply to a command.
+
+You need to configure the `ActorSystem` with the
`DurableStateBehaviorTestKit.config`. The configuration enables
+the in-memory durable state store.
+
+Scala
+: @@snip
[DurableStateBehaviorTestKitSpec.scala](/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/scaladsl/DurableStateBehaviorTestKitSpec.scala)
{ #basic-test }
+
+Java
+: @@snip
[DurableStateBehaviorTestKitApiTest.java](/persistence-testkit/src/test/java/org/apache/pekko/persistence/testkit/javadsl/DurableStateBehaviorTestKitApiTest.java)
{ #basic-test }
+
+Serialization of commands and state is verified automatically. The
serialization checks can be customized with
+the `SerializationSettings` when creating the `DurableStateBehaviorTestKit`.
By default, the serialization
+roundtrip is checked but the equality of the result of the serialization is
not checked. `equals` must be
+implemented @scala[(or using `case class`)] in the commands and state if
`verifyEquality` is enabled.
+
+To test recovery, use the `restart` method. It restarts the behavior, which
then recovers the state stored by
+previous commands. The `clear` method removes the state for the behavior's
persistence ID and restarts it
+with its empty state.
+
## Persistence TestKit
**Note!** The `PersistenceTestKit` is a new feature, api may have changes
breaking source compatibility in future versions.
diff --git
a/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/internal/DurableStateBehaviorTestKitImpl.scala
b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/internal/DurableStateBehaviorTestKitImpl.scala
new file mode 100644
index 0000000000..726027db65
--- /dev/null
+++
b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/internal/DurableStateBehaviorTestKitImpl.scala
@@ -0,0 +1,204 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.persistence.testkit.internal
+
+import scala.reflect.ClassTag
+import scala.util.control.NonFatal
+
+import org.apache.pekko
+import pekko.actor.testkit.typed.scaladsl.ActorTestKit
+import pekko.actor.testkit.typed.scaladsl.SerializationTestKit
+import pekko.actor.typed.ActorRef
+import pekko.actor.typed.ActorSystem
+import pekko.actor.typed.Behavior
+import pekko.actor.typed.scaladsl.adapter._
+import pekko.annotation.InternalApi
+import pekko.persistence.state.DurableStateStoreRegistry
+import pekko.persistence.testkit.scaladsl.DurableStateBehaviorTestKit
+import
pekko.persistence.testkit.scaladsl.DurableStateBehaviorTestKit.CommandResult
+import
pekko.persistence.testkit.scaladsl.DurableStateBehaviorTestKit.CommandResultWithReply
+import
pekko.persistence.testkit.scaladsl.DurableStateBehaviorTestKit.RestartResult
+import
pekko.persistence.testkit.scaladsl.DurableStateBehaviorTestKit.SerializationSettings
+import
pekko.persistence.testkit.state.scaladsl.PersistenceTestKitDurableStateStore
+import pekko.persistence.typed.PersistenceId
+import pekko.persistence.typed.state.internal.DurableStateBehaviorImpl
+import
pekko.persistence.typed.state.internal.DurableStateBehaviorImpl.GetStateReply
+
+/**
+ * INTERNAL API
+ */
+@InternalApi private[pekko] object DurableStateBehaviorTestKitImpl {
+
+ final case class CommandResultImpl[Command, State, Reply](
+ command: Command,
+ state: State,
+ replyOption: Option[Reply])
+ extends CommandResultWithReply[Command, State, Reply] {
+
+ override def stateOfType[S <: State: ClassTag]: S =
+ ofType(state, "state")
+
+ override def reply: Reply =
+ replyOption.getOrElse(throw new AssertionError("No reply"))
+
+ override def replyOfType[R <: Reply: ClassTag]: R =
+ ofType(reply, "reply")
+
+ override def hasNoReply: Boolean =
+ replyOption.isEmpty
+
+ private def ofType[A: ClassTag](obj: Any, errorParam: String): A =
+ obj match {
+ case a: A => a
+ case other =>
+ val expectedClass = implicitly[ClassTag[A]].runtimeClass
+ val actualClass = if (other == null) "null" else
other.getClass.getName
+ throw new AssertionError(
+ s"Expected $errorParam class [${expectedClass.getName}], " +
+ s"but was [$actualClass]")
+ }
+ }
+
+ final case class RestartResultImpl[State](state: State) extends
RestartResult[State]
+}
+
+/**
+ * INTERNAL API
+ */
+@InternalApi private[pekko] class DurableStateBehaviorTestKitImpl[Command,
State](
+ actorTestKit: ActorTestKit,
+ behavior: Behavior[Command],
+ serializationSettings: SerializationSettings)
+ extends DurableStateBehaviorTestKit[Command, State] {
+
+ import DurableStateBehaviorTestKitImpl._
+
+ private def system: ActorSystem[?] = actorTestKit.system
+
+ private val durableStateStore =
+ DurableStateStoreRegistry(system.toClassic)
+ .durableStateStoreFor[PersistenceTestKitDurableStateStore[Any]](
+ PersistenceTestKitDurableStateStore.Identifier)
+ durableStateStore.clearAll()
+
+ private val probe = actorTestKit.createTestProbe[Any]()
+ private val stateProbe = actorTestKit.createTestProbe[GetStateReply[State]]()
+ private var actor: ActorRef[Command] = actorTestKit.spawn(behavior)
+ private def internalActor: ActorRef[Any] = actor.unsafeUpcast[Any]
+ private val persistenceId = verifyDurableStateBehavior()
+
+ private val serializationTestKit = new SerializationTestKit(system)
+ private var emptyStateVerified = false
+
+ override def runCommand(command: Command): CommandResult[Command, State] = {
+ preCommandCheck(command)
+
+ actor ! command
+
+ val newState = getState()
+ postCommandCheck(newState, reply = None)
+
+ CommandResultImpl[Command, State, Nothing](command, newState, None)
+ }
+
+ override def runCommand[R](creator: ActorRef[R] => Command):
CommandResultWithReply[Command, State, R] = {
+ val replyProbe = actorTestKit.createTestProbe[R]()
+ val command = creator(replyProbe.ref)
+ preCommandCheck(command)
+
+ actor ! command
+
+ val reply =
+ try {
+ replyProbe.receiveMessage()
+ } catch {
+ case NonFatal(_) =>
+ throw new AssertionError(s"Missing expected reply for command
[$command].")
+ } finally {
+ replyProbe.stop()
+ }
+
+ val newState = getState()
+ postCommandCheck(newState, Some(reply))
+
+ CommandResultImpl(command, newState, Some(reply))
+ }
+
+ override def getState(): State = {
+ internalActor ! DurableStateBehaviorImpl.GetStateWithReply(stateProbe.ref)
+ stateProbe.receiveMessage().currentState
+ }
+
+ override def restart(): RestartResult[State] = {
+ actorTestKit.stop(actor)
+ actor = actorTestKit.spawn(behavior)
+ try {
+ RestartResultImpl(getState())
+ } catch {
+ case NonFatal(_) =>
+ throw new IllegalStateException("Could not restart
DurableStateBehavior. See logs.")
+ }
+ }
+
+ override def clear(): Unit = {
+ durableStateStore.clearByPersistenceId(persistenceId.id)
+ restart()
+ }
+
+ private def verifyDurableStateBehavior(): PersistenceId = {
+ internalActor !
DurableStateBehaviorImpl.GetPersistenceIdForTestKit(probe.ref)
+ try {
+ probe.expectMessageType[PersistenceId]
+ } catch {
+ case NonFatal(_) =>
+ throw new IllegalArgumentException("Only DurableStateBehavior, or
nested DurableStateBehavior allowed.")
+ }
+ }
+
+ private def preCommandCheck(command: Command): Unit = {
+ if (serializationSettings.enabled) {
+ if (serializationSettings.verifyCommands)
+ verifySerializationAndThrow(command, "Command")
+
+ if (serializationSettings.verifyState && !emptyStateVerified) {
+ val emptyState = getState()
+ verifySerializationAndThrow(emptyState, "Empty State")
+ emptyStateVerified = true
+ }
+ }
+ }
+
+ private def postCommandCheck(newState: State, reply: Option[Any]): Unit = {
+ if (serializationSettings.enabled) {
+ if (serializationSettings.verifyState)
+ verifySerializationAndThrow(newState, "State")
+
+ if (serializationSettings.verifyCommands)
+ reply.foreach(verifySerializationAndThrow(_, "Reply"))
+ }
+ }
+
+ private def verifySerializationAndThrow(obj: Any, errorMessagePrefix:
String): Unit = {
+ try {
+ serializationTestKit.verifySerialization(obj,
serializationSettings.verifyEquality)
+ } catch {
+ case NonFatal(exc) =>
+ throw new IllegalArgumentException(s"$errorMessagePrefix [$obj] isn't
serializable.", exc)
+ }
+ }
+}
diff --git
a/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/javadsl/DurableStateBehaviorTestKit.scala
b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/javadsl/DurableStateBehaviorTestKit.scala
new file mode 100644
index 0000000000..b0c5d59ba7
--- /dev/null
+++
b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/javadsl/DurableStateBehaviorTestKit.scala
@@ -0,0 +1,232 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.persistence.testkit.javadsl
+
+import java.util.function.{ Function => JFunction }
+
+import scala.reflect.ClassTag
+
+import org.apache.pekko
+import pekko.actor.typed.ActorRef
+import pekko.actor.typed.ActorSystem
+import pekko.actor.typed.Behavior
+import pekko.annotation.ApiMayChange
+import pekko.annotation.DoNotInherit
+import pekko.persistence.testkit.scaladsl
+
+import com.typesafe.config.Config
+
+/**
+ * Testing of [[pekko.persistence.typed.state.javadsl.DurableStateBehavior]]
implementations.
+ * It supports running one command at a time and asserting that the
synchronously returned result is as expected.
+ * The result contains the new state after the command has been processed.
+ * It also has support for verifying the reply to a command.
+ *
+ * Serialization of commands and state is verified automatically.
+ *
+ * @since 2.0.0
+ */
+@ApiMayChange
+object DurableStateBehaviorTestKit {
+
+ /**
+ * The configuration to be included in the configuration of the
`ActorSystem`. Typically used as
+ * constructor parameter to `TestKitJunitResource`. The configuration
enables the in-memory
+ * durable state store.
+ */
+ val config: Config = scaladsl.DurableStateBehaviorTestKit.config
+
+ val enabledSerializationSettings: SerializationSettings = new
SerializationSettings(
+ enabled = true,
+ verifyEquality = false,
+ verifyCommands = true,
+ verifyState = true)
+
+ val disabledSerializationSettings: SerializationSettings = new
SerializationSettings(
+ enabled = false,
+ verifyEquality = false,
+ verifyCommands = false,
+ verifyState = false)
+
+ /**
+ * Customization of which serialization checks that are performed.
+ * `equals` must be implemented when `verifyEquality` is enabled.
+ */
+ final class SerializationSettings(
+ val enabled: Boolean,
+ val verifyEquality: Boolean,
+ val verifyCommands: Boolean,
+ val verifyState: Boolean) {
+
+ def withEnabled(value: Boolean): SerializationSettings =
+ copy(enabled = value)
+
+ def withVerifyEquality(value: Boolean): SerializationSettings =
+ copy(verifyEquality = value)
+
+ def withVerifyCommands(value: Boolean): SerializationSettings =
+ copy(verifyCommands = value)
+
+ def withVerifyState(value: Boolean): SerializationSettings =
+ copy(verifyState = value)
+
+ private def copy(
+ enabled: Boolean = this.enabled,
+ verifyEquality: Boolean = this.verifyEquality,
+ verifyCommands: Boolean = this.verifyCommands,
+ verifyState: Boolean = this.verifyState): SerializationSettings =
+ new SerializationSettings(enabled, verifyEquality, verifyCommands,
verifyState)
+ }
+
+ /**
+ * Factory method to create a new DurableStateBehaviorTestKit.
+ */
+ def create[Command, State](
+ system: ActorSystem[?],
+ behavior: Behavior[Command]): DurableStateBehaviorTestKit[Command,
State] =
+ create(system, behavior, enabledSerializationSettings)
+
+ /**
+ * Factory method to create a new DurableStateBehaviorTestKit with custom
[[SerializationSettings]].
+ *
+ * Note that `equals` must be implemented in the commands and state if
`verifyEquality` is enabled.
+ */
+ def create[Command, State](
+ system: ActorSystem[?],
+ behavior: Behavior[Command],
+ serializationSettings: SerializationSettings):
DurableStateBehaviorTestKit[Command, State] = {
+ val scaladslSettings = new
scaladsl.DurableStateBehaviorTestKit.SerializationSettings(
+ enabled = serializationSettings.enabled,
+ verifyEquality = serializationSettings.verifyEquality,
+ verifyCommands = serializationSettings.verifyCommands,
+ verifyState = serializationSettings.verifyState)
+ new
DurableStateBehaviorTestKit(scaladsl.DurableStateBehaviorTestKit(system,
behavior, scaladslSettings))
+ }
+
+ /**
+ * The result of running a command.
+ */
+ @DoNotInherit class CommandResult[Command, State](
+ delegate: scaladsl.DurableStateBehaviorTestKit.CommandResult[Command,
State]) {
+
+ /**
+ * The command that was run.
+ */
+ def command: Command =
+ delegate.command
+
+ /**
+ * The state after the command has been processed.
+ */
+ def state: State =
+ delegate.state
+
+ /**
+ * The state as a given expected type. It will throw `AssertionError` if
the state is of a different type.
+ */
+ def stateOfType[S <: State](stateClass: Class[S]): S = {
+ implicit val ct: ClassTag[S] = ClassTag(stateClass)
+ delegate.stateOfType[S]
+ }
+ }
+
+ /**
+ * The result of running a command with an `ActorRef<R> replyTo`, i.e. the
`runCommand` with a
+ * `Function<ActorRef<R>, Command>` parameter.
+ */
+ final class CommandResultWithReply[Command, State, Reply](
+ delegate:
scaladsl.DurableStateBehaviorTestKit.CommandResultWithReply[Command, State,
Reply])
+ extends CommandResult[Command, State](delegate) {
+
+ /**
+ * The reply. It will throw `AssertionError` if there was no reply.
+ */
+ def reply: Reply =
+ delegate.reply
+
+ /**
+ * The reply as a given expected type. It will throw `AssertionError` if
there is no reply or
+ * if the reply is of a different type.
+ */
+ def replyOfType[R <: Reply](replyClass: Class[R]): R = {
+ implicit val ct: ClassTag[R] = ClassTag(replyClass)
+ delegate.replyOfType[R]
+ }
+
+ /**
+ * `true` if there is no reply.
+ */
+ def hasNoReply: Boolean =
+ delegate.hasNoReply
+ }
+
+ /**
+ * The result of restarting the behavior.
+ */
+ final class RestartResult[State](delegate:
scaladsl.DurableStateBehaviorTestKit.RestartResult[State]) {
+
+ /**
+ * The state after recovery.
+ */
+ def state: State =
+ delegate.state
+ }
+}
+
+/**
+ * @since 2.0.0
+ */
+@ApiMayChange
+final class DurableStateBehaviorTestKit[Command, State](
+ delegate: scaladsl.DurableStateBehaviorTestKit[Command, State]) {
+
+ import DurableStateBehaviorTestKit._
+
+ /**
+ * Run one command through the behavior. The returned result contains the
state after the command
+ * has been processed.
+ */
+ def runCommand(command: Command): CommandResult[Command, State] =
+ new CommandResult(delegate.runCommand(command))
+
+ /**
+ * Run one command with a `replyTo: ActorRef` through the behavior. The
returned result contains
+ * the state after the command has been processed, and the reply.
+ */
+ def runCommand[R](creator: JFunction[ActorRef[R], Command]):
CommandResultWithReply[Command, State, R] =
+ new CommandResultWithReply(delegate.runCommand(replyTo =>
creator.apply(replyTo)))
+
+ /**
+ * Retrieve the current state of the behavior.
+ */
+ def getState(): State =
+ delegate.getState()
+
+ /**
+ * Restart the behavior, which will then recover from the durable state
store. Can be used for testing
+ * that the recovery is correct.
+ */
+ def restart(): RestartResult[State] =
+ new RestartResult(delegate.restart())
+
+ /**
+ * Clear the state for this behavior from the in-memory durable state store
and restart the behavior.
+ */
+ def clear(): Unit =
+ delegate.clear()
+}
diff --git
a/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/scaladsl/DurableStateBehaviorTestKit.scala
b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/scaladsl/DurableStateBehaviorTestKit.scala
new file mode 100644
index 0000000000..9ebd9ddad3
--- /dev/null
+++
b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/scaladsl/DurableStateBehaviorTestKit.scala
@@ -0,0 +1,209 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.persistence.testkit.scaladsl
+
+import scala.reflect.ClassTag
+
+import org.apache.pekko
+import pekko.actor.testkit.typed.scaladsl.ActorTestKit
+import pekko.actor.typed.ActorRef
+import pekko.actor.typed.ActorSystem
+import pekko.actor.typed.Behavior
+import pekko.annotation.ApiMayChange
+import pekko.annotation.DoNotInherit
+import pekko.persistence.testkit.PersistenceTestKitDurableStateStorePlugin
+import pekko.persistence.testkit.internal.DurableStateBehaviorTestKitImpl
+
+import com.typesafe.config.Config
+
+/**
+ * Testing of [[pekko.persistence.typed.state.scaladsl.DurableStateBehavior]]
implementations.
+ * It supports running one command at a time and asserting that the
synchronously returned result is as expected.
+ * The result contains the new state after the command has been processed.
+ * It also has support for verifying the reply to a command.
+ *
+ * Serialization of commands and state is verified automatically.
+ *
+ * @since 2.0.0
+ */
+@ApiMayChange
+object DurableStateBehaviorTestKit {
+
+ /**
+ * The configuration to be included in the configuration of the
`ActorSystem`. Typically used as
+ * constructor parameter to `ScalaTestWithActorTestKit`. The configuration
enables the in-memory
+ * durable state store.
+ */
+ val config: Config = PersistenceTestKitDurableStateStorePlugin.config
+
+ object SerializationSettings {
+ val enabled: SerializationSettings = new SerializationSettings(
+ enabled = true,
+ verifyEquality = false,
+ verifyCommands = true,
+ verifyState = true)
+
+ val disabled: SerializationSettings = new SerializationSettings(
+ enabled = false,
+ verifyEquality = false,
+ verifyCommands = false,
+ verifyState = false)
+ }
+
+ /**
+ * Customization of which serialization checks that are performed.
+ * `equals` must be implemented (or using `case class`) when
`verifyEquality` is enabled.
+ */
+ final class SerializationSettings private[pekko] (
+ val enabled: Boolean,
+ val verifyEquality: Boolean,
+ val verifyCommands: Boolean,
+ val verifyState: Boolean) {
+
+ def withEnabled(value: Boolean): SerializationSettings =
+ copy(enabled = value)
+
+ def withVerifyEquality(value: Boolean): SerializationSettings =
+ copy(verifyEquality = value)
+
+ def withVerifyCommands(value: Boolean): SerializationSettings =
+ copy(verifyCommands = value)
+
+ def withVerifyState(value: Boolean): SerializationSettings =
+ copy(verifyState = value)
+
+ private def copy(
+ enabled: Boolean = this.enabled,
+ verifyEquality: Boolean = this.verifyEquality,
+ verifyCommands: Boolean = this.verifyCommands,
+ verifyState: Boolean = this.verifyState): SerializationSettings =
+ new SerializationSettings(enabled, verifyEquality, verifyCommands,
verifyState)
+ }
+
+ /**
+ * Factory method to create a new DurableStateBehaviorTestKit.
+ */
+ def apply[Command, State](
+ system: ActorSystem[?],
+ behavior: Behavior[Command]): DurableStateBehaviorTestKit[Command,
State] =
+ apply(system, behavior, SerializationSettings.enabled)
+
+ /**
+ * Factory method to create a new DurableStateBehaviorTestKit with custom
[[SerializationSettings]].
+ *
+ * Note that `equals` must be implemented (or using `case class`) in the
commands and state if
+ * `verifyEquality` is enabled.
+ */
+ def apply[Command, State](
+ system: ActorSystem[?],
+ behavior: Behavior[Command],
+ serializationSettings: SerializationSettings):
DurableStateBehaviorTestKit[Command, State] =
+ new DurableStateBehaviorTestKitImpl(ActorTestKit(system), behavior,
serializationSettings)
+
+ /**
+ * The result of running a command.
+ */
+ @DoNotInherit trait CommandResult[Command, State] {
+
+ /**
+ * The command that was run.
+ */
+ def command: Command
+
+ /**
+ * The state after the command has been processed.
+ */
+ def state: State
+
+ /**
+ * The state as a given expected type. It will throw `AssertionError` if
the state is of a different type.
+ */
+ def stateOfType[S <: State: ClassTag]: S
+ }
+
+ /**
+ * The result of running a command with a `replyTo: ActorRef[R]`, i.e. the
`runCommand` with an
+ * `ActorRef[R] => Command` parameter.
+ */
+ @DoNotInherit trait CommandResultWithReply[Command, State, Reply] extends
CommandResult[Command, State] {
+
+ /**
+ * The reply. It will throw `AssertionError` if there was no reply.
+ */
+ def reply: Reply
+
+ /**
+ * The reply as a given expected type. It will throw `AssertionError` if
there is no reply or
+ * if the reply is of a different type.
+ */
+ def replyOfType[R <: Reply: ClassTag]: R
+
+ /**
+ * `true` if there is no reply.
+ */
+ def hasNoReply: Boolean
+ }
+
+ /**
+ * The result of restarting the behavior.
+ */
+ @DoNotInherit trait RestartResult[State] {
+
+ /**
+ * The state after recovery.
+ */
+ def state: State
+ }
+}
+
+/**
+ * @since 2.0.0
+ */
+@ApiMayChange
+@DoNotInherit trait DurableStateBehaviorTestKit[Command, State] {
+
+ import DurableStateBehaviorTestKit._
+
+ /**
+ * Run one command through the behavior. The returned result contains the
state after the command
+ * has been processed.
+ */
+ def runCommand(command: Command): CommandResult[Command, State]
+
+ /**
+ * Run one command with a `replyTo: ActorRef[R]` through the behavior. The
returned result contains
+ * the state after the command has been processed, and the reply.
+ */
+ def runCommand[R](creator: ActorRef[R] => Command):
CommandResultWithReply[Command, State, R]
+
+ /**
+ * Retrieve the current state of the behavior.
+ */
+ def getState(): State
+
+ /**
+ * Restart the behavior, which will then recover from the durable state
store. Can be used for testing
+ * that the recovery is correct.
+ */
+ def restart(): RestartResult[State]
+
+ /**
+ * Clear the state for this behavior from the in-memory durable state store
and restart the behavior.
+ */
+ def clear(): Unit
+}
diff --git
a/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/state/scaladsl/PersistenceTestKitDurableStateStore.scala
b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/state/scaladsl/PersistenceTestKitDurableStateStore.scala
index 414a9f1962..f56b658e99 100644
---
a/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/state/scaladsl/PersistenceTestKitDurableStateStore.scala
+++
b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/state/scaladsl/PersistenceTestKitDurableStateStore.scala
@@ -54,6 +54,14 @@ class PersistenceTestKitDurableStateStore[A](val system:
ExtendedActorSystem)
private val persistence = Persistence(system)
private var store = Map.empty[String, Record[A]]
+ private[pekko] def clearAll(): Unit = this.synchronized {
+ store = Map.empty
+ }
+
+ private[pekko] def clearByPersistenceId(persistenceId: String): Unit =
this.synchronized {
+ store = store - persistenceId
+ }
+
private val (publisher, changesSource) =
ActorSource
.actorRef[Record[A]](PartialFunction.empty, PartialFunction.empty, 256,
OverflowStrategy.dropHead)
diff --git
a/persistence-testkit/src/test/java/org/apache/pekko/persistence/testkit/javadsl/DurableStateBehaviorTestKitApiTest.java
b/persistence-testkit/src/test/java/org/apache/pekko/persistence/testkit/javadsl/DurableStateBehaviorTestKitApiTest.java
new file mode 100644
index 0000000000..e2733b20b7
--- /dev/null
+++
b/persistence-testkit/src/test/java/org/apache/pekko/persistence/testkit/javadsl/DurableStateBehaviorTestKitApiTest.java
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.persistence.testkit.javadsl;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.pekko.Done;
+import org.apache.pekko.actor.typed.ActorRef;
+import org.apache.pekko.actor.typed.ActorSystem;
+import org.apache.pekko.actor.typed.Behavior;
+import org.apache.pekko.persistence.typed.PersistenceId;
+import org.apache.pekko.persistence.typed.state.javadsl.CommandHandler;
+import org.apache.pekko.persistence.typed.state.javadsl.DurableStateBehavior;
+import org.apache.pekko.serialization.jackson.CborSerializable;
+
+public final class DurableStateBehaviorTestKitApiTest {
+
+ interface Command extends CborSerializable {}
+
+ enum Increment implements Command {
+ INSTANCE
+ }
+
+ static final class IncrementWithReply implements Command {
+ final ActorRef<Done> replyTo;
+
+ @JsonCreator
+ IncrementWithReply(@JsonProperty("replyTo") ActorRef<Done> replyTo) {
+ this.replyTo = replyTo;
+ }
+ }
+
+ static final class State implements CborSerializable {
+ final int value;
+
+ @JsonCreator
+ State(@JsonProperty("value") int value) {
+ this.value = value;
+ }
+ }
+
+ static final class Counter extends DurableStateBehavior<Command, State> {
+
+ static Behavior<Command> create(PersistenceId persistenceId) {
+ return new Counter(persistenceId);
+ }
+
+ private Counter(PersistenceId persistenceId) {
+ super(persistenceId);
+ }
+
+ @Override
+ public State emptyState() {
+ return new State(0);
+ }
+
+ @Override
+ public CommandHandler<Command, State> commandHandler() {
+ return newCommandHandlerBuilder()
+ .forAnyState()
+ .onCommand(
+ Increment.class, (state, command) -> Effect().persist(new
State(state.value + 1)))
+ .onCommand(
+ IncrementWithReply.class,
+ (state, command) ->
+ Effect()
+ .persist(new State(state.value + 1))
+ .thenReply(command.replyTo, ignored ->
Done.getInstance()))
+ .build();
+ }
+ }
+
+ private static final AtomicInteger idCounter = new AtomicInteger();
+
+ // #basic-test
+ public static void run(ActorSystem<?> system) {
+ PersistenceId persistenceId =
+ PersistenceId.ofUniqueId("durable-state-java-test-" +
idCounter.incrementAndGet());
+ DurableStateBehaviorTestKit<Command, State> testKit =
+ DurableStateBehaviorTestKit.create(system,
Counter.create(persistenceId));
+
+ DurableStateBehaviorTestKit.CommandResult<Command, State> commandResult =
+ testKit.runCommand(Increment.INSTANCE);
+ check(commandResult.command() == Increment.INSTANCE, "command");
+ check(commandResult.state().value == 1, "first command state");
+ check(commandResult.stateOfType(State.class).value == 1, "typed state");
+
+ DurableStateBehaviorTestKit.CommandResultWithReply<Command, State, Done>
replyResult =
+ testKit.runCommand(replyTo -> new IncrementWithReply(replyTo));
+ check(replyResult.state().value == 2, "reply command state");
+ check(replyResult.reply() == Done.getInstance(), "reply");
+ check(replyResult.replyOfType(Done.class) == Done.getInstance(), "typed
reply");
+ check(!replyResult.hasNoReply(), "has reply");
+
+ check(testKit.getState().value == 2, "current state");
+ check(testKit.restart().state().value == 2, "recovered state");
+ }
+
+ // #basic-test
+
+ public static void clear(ActorSystem<?> system) {
+ PersistenceId persistenceId =
+ PersistenceId.ofUniqueId("durable-state-java-clear-" +
idCounter.incrementAndGet());
+ DurableStateBehaviorTestKit<Command, State> testKit =
+ DurableStateBehaviorTestKit.create(
+ system,
+ Counter.create(persistenceId),
+ DurableStateBehaviorTestKit.enabledSerializationSettings());
+
+ testKit.runCommand(Increment.INSTANCE);
+ testKit.clear();
+ check(testKit.getState().value == 0, "cleared state");
+ }
+
+ private static void check(boolean condition, String clue) {
+ if (!condition) {
+ throw new AssertionError(clue);
+ }
+ }
+
+ private DurableStateBehaviorTestKitApiTest() {}
+}
diff --git
a/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/javadsl/DurableStateBehaviorTestKitJavaDslSpec.scala
b/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/javadsl/DurableStateBehaviorTestKitJavaDslSpec.scala
new file mode 100644
index 0000000000..9d0fc4003c
--- /dev/null
+++
b/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/javadsl/DurableStateBehaviorTestKitJavaDslSpec.scala
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.persistence.testkit.javadsl
+
+import org.apache.pekko.actor.testkit.typed.scaladsl.LogCapturing
+import org.apache.pekko.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
+
+import org.scalatest.wordspec.AnyWordSpecLike
+
+class DurableStateBehaviorTestKitJavaDslSpec
+ extends ScalaTestWithActorTestKit(DurableStateBehaviorTestKit.config)
+ with AnyWordSpecLike
+ with LogCapturing {
+
+ "The Java DurableStateBehaviorTestKit API" must {
+ "run commands, return replies, expose state, and restart" in {
+ DurableStateBehaviorTestKitApiTest.run(system)
+ }
+
+ "clear the state" in {
+ DurableStateBehaviorTestKitApiTest.clear(system)
+ }
+ }
+}
diff --git
a/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/scaladsl/DurableStateBehaviorTestKitSpec.scala
b/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/scaladsl/DurableStateBehaviorTestKitSpec.scala
new file mode 100644
index 0000000000..4897e381fe
--- /dev/null
+++
b/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/scaladsl/DurableStateBehaviorTestKitSpec.scala
@@ -0,0 +1,250 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.persistence.testkit.scaladsl
+
+import java.io.NotSerializableException
+
+import org.apache.pekko
+import pekko.Done
+import pekko.actor.testkit.typed.scaladsl.LogCapturing
+import pekko.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
+import pekko.actor.typed.ActorRef
+import pekko.actor.typed.scaladsl.Behaviors
+import
pekko.persistence.testkit.scaladsl.DurableStateBehaviorTestKitSpec.TestCounter
+import pekko.persistence.typed.PersistenceId
+import pekko.persistence.typed.state.scaladsl.DurableStateBehavior
+import pekko.persistence.typed.state.scaladsl.Effect
+import pekko.serialization.DisabledJavaSerializer
+import pekko.serialization.jackson.CborSerializable
+
+import org.scalatest.wordspec.AnyWordSpecLike
+
+object DurableStateBehaviorTestKitSpec {
+
+ object TestCounter {
+ sealed trait Command
+ case object Increment extends Command with CborSerializable
+ final case class IncrementWithConfirmation(replyTo: ActorRef[Done])
extends Command with CborSerializable
+ final case class GetValue(replyTo: ActorRef[State]) extends Command with
CborSerializable
+ case object PersistNotSerializableState extends Command with
CborSerializable
+ case object NotSerializableCommand extends Command
+ final case class ReplyNotSerializable(replyTo:
ActorRef[NotSerializableReply])
+ extends Command
+ with CborSerializable
+
+ sealed trait State
+ final case class RealState(value: Int) extends State with CborSerializable
+ final case class NotSerializableState(value: Int) extends State
+ final class NotSerializableReply
+
+ def apply(persistenceId: PersistenceId, emptyState: State = RealState(0)):
DurableStateBehavior[Command, State] =
+ DurableStateBehavior[Command, State](
+ persistenceId,
+ emptyState,
+ commandHandler = {
+ case (RealState(value), Increment) =>
+ Effect.persist(RealState(value + 1))
+ case (null, Increment) =>
+ Effect.persist(RealState(1))
+ case (RealState(value), IncrementWithConfirmation(replyTo)) =>
+ Effect.persist(RealState(value + 1)).thenReply(replyTo)(_ => Done)
+ case (state, GetValue(replyTo)) =>
+ Effect.none[TestCounter.State].thenRun(_ => replyTo ! state)
+ case (RealState(value), PersistNotSerializableState) =>
+ Effect.persist(NotSerializableState(value + 1))
+ case (_, NotSerializableCommand) =>
+ Effect.none
+ case (_, ReplyNotSerializable(replyTo)) =>
+ Effect.none[TestCounter.State].thenRun(_ => replyTo ! new
NotSerializableReply)
+ case (state, command) =>
+ throw new IllegalArgumentException(s"Unexpected state [$state] and
command [$command]")
+ })
+ }
+}
+
+class DurableStateBehaviorTestKitSpec
+ extends ScalaTestWithActorTestKit(DurableStateBehaviorTestKit.config)
+ with AnyWordSpecLike
+ with LogCapturing {
+
+ private val persistenceId = PersistenceId.ofUniqueId("durable-state-test")
+
+ private def createTestKit(
+ emptyState: TestCounter.State = TestCounter.RealState(0),
+ settings: DurableStateBehaviorTestKit.SerializationSettings =
+ DurableStateBehaviorTestKit.SerializationSettings.enabled) =
+ DurableStateBehaviorTestKit[TestCounter.Command, TestCounter.State](
+ system,
+ TestCounter(persistenceId, emptyState),
+ settings)
+
+ "DurableStateBehaviorTestKit" must {
+
+ "run commands and expose the resulting state" in {
+ // #basic-test
+ val testKit =
+ DurableStateBehaviorTestKit[TestCounter.Command, TestCounter.State](
+ system,
+ TestCounter(PersistenceId.ofUniqueId("durable-state-test")))
+
+ val firstResult = testKit.runCommand(TestCounter.Increment)
+ firstResult.command shouldBe TestCounter.Increment
+ firstResult.state shouldBe TestCounter.RealState(1)
+
+ val secondResult = testKit.runCommand(TestCounter.Increment)
+ secondResult.stateOfType[TestCounter.RealState].value shouldBe 2
+ // #basic-test
+
+ intercept[AssertionError] {
+ secondResult.stateOfType[TestCounter.NotSerializableState]
+ }.getMessage should include("Expected state class")
+ }
+
+ "run commands with replies" in {
+ val testKit = createTestKit()
+
+ val result =
testKit.runCommand[Done](TestCounter.IncrementWithConfirmation(_))
+ result.state shouldBe TestCounter.RealState(1)
+ result.reply shouldBe Done
+ result.replyOfType[Done] shouldBe Done
+ result.hasNoReply shouldBe false
+ }
+
+ "detect a missing reply" in {
+ val testKit = createTestKit()
+
+ intercept[AssertionError] {
+ testKit.runCommand[Done](_ => TestCounter.Increment)
+ }.getMessage should include("Missing expected reply")
+ }
+
+ "expose the current state" in {
+ val testKit = createTestKit()
+
+ testKit.getState() shouldBe TestCounter.RealState(0)
+ testKit.runCommand(TestCounter.Increment)
+ testKit.getState() shouldBe TestCounter.RealState(1)
+ }
+
+ "handle a null empty state" in {
+ val testKit = createTestKit(emptyState = null)
+
+ testKit.getState() shouldBe null
+ testKit.runCommand(TestCounter.Increment).state shouldBe
TestCounter.RealState(1)
+ }
+
+ "recover state when restarted" in {
+ val testKit = createTestKit()
+
+ testKit.runCommand(TestCounter.Increment)
+ testKit.runCommand(TestCounter.Increment)
+
+ testKit.restart().state shouldBe TestCounter.RealState(2)
+ testKit.getState() shouldBe TestCounter.RealState(2)
+ }
+
+ "clear the state and restart" in {
+ val testKit = createTestKit()
+
+ testKit.runCommand(TestCounter.Increment)
+ testKit.clear()
+
+ testKit.getState() shouldBe TestCounter.RealState(0)
+ }
+
+ "start with an empty durable state store" in {
+ val firstTestKit =
+ DurableStateBehaviorTestKit[TestCounter.Command, TestCounter.State](
+ system,
+ TestCounter(persistenceId))
+ firstTestKit.runCommand(TestCounter.Increment).state shouldBe
TestCounter.RealState(1)
+
+ val secondTestKit =
+ DurableStateBehaviorTestKit[TestCounter.Command, TestCounter.State](
+ system,
+ TestCounter(persistenceId))
+ secondTestKit.getState() shouldBe TestCounter.RealState(0)
+ }
+
+ "detect a non-serializable command" in {
+ val testKit = createTestKit()
+
+ val exception = intercept[IllegalArgumentException] {
+ testKit.runCommand(TestCounter.NotSerializableCommand)
+ }
+ exception.getMessage should include("Command")
+ exception.getCause.getClass shouldBe
classOf[DisabledJavaSerializer.JavaSerializationException]
+ }
+
+ "detect a non-serializable state" in {
+ val testKit = createTestKit()
+
+ val exception = intercept[IllegalArgumentException] {
+ testKit.runCommand(TestCounter.PersistNotSerializableState)
+ }
+ exception.getMessage should include("State")
+ exception.getCause.getClass shouldBe
classOf[DisabledJavaSerializer.JavaSerializationException]
+ }
+
+ "detect a non-serializable empty state" in {
+ val testKit = createTestKit(TestCounter.NotSerializableState(0))
+
+ val exception = intercept[IllegalArgumentException] {
+ testKit.runCommand(TestCounter.Increment)
+ }
+ exception.getMessage should include("Empty State")
+ exception.getCause.getClass shouldBe
classOf[DisabledJavaSerializer.JavaSerializationException]
+ }
+
+ "detect a non-serializable reply" in {
+ val testKit = createTestKit()
+
+ val exception = intercept[IllegalArgumentException] {
+ testKit.runCommand(TestCounter.ReplyNotSerializable(_))
+ }
+ exception.getMessage should include("Reply")
+ exception.getCause.getClass shouldBe classOf[NotSerializableException]
+ }
+
+ "allow serialization verification to be disabled" in {
+ val testKit = createTestKit(
+ settings = DurableStateBehaviorTestKit.SerializationSettings.disabled)
+
+ testKit.runCommand(TestCounter.NotSerializableCommand).state shouldBe
TestCounter.RealState(0)
+ }
+
+ "only allow DurableStateBehavior" in {
+ intercept[IllegalArgumentException] {
+ DurableStateBehaviorTestKit[TestCounter.Command, TestCounter.State](
+ system,
+ Behaviors.empty[TestCounter.Command])
+ }
+ }
+
+ "support a catch-all signal handler" in {
+ val behavior = TestCounter(persistenceId).receiveSignal {
+ case (_, _) => ()
+ }
+
+ val testKit =
+ DurableStateBehaviorTestKit[TestCounter.Command,
TestCounter.State](system, behavior)
+
+ testKit.runCommand(TestCounter.Increment).state shouldBe
TestCounter.RealState(1)
+ }
+ }
+}
diff --git
a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/internal/DurableStateBehaviorImpl.scala
b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/internal/DurableStateBehaviorImpl.scala
index bb6662d07d..a56a22b38f 100644
---
a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/internal/DurableStateBehaviorImpl.scala
+++
b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/internal/DurableStateBehaviorImpl.scala
@@ -35,10 +35,10 @@ import pekko.persistence.typed.PersistenceId
import pekko.persistence.typed.SnapshotAdapter
import pekko.persistence.typed.state.scaladsl._
-import com.typesafe.config.Config
-
import org.slf4j.LoggerFactory
+import com.typesafe.config.Config
+
@InternalApi
private[pekko] object DurableStateBehaviorImpl {
@@ -48,11 +48,24 @@ private[pekko] object DurableStateBehaviorImpl {
final case class GetPersistenceId(replyTo: ActorRef[PersistenceId]) extends
Signal
/**
- * Used by DurableStateBehaviorTestKit to retrieve the state.
+ * Used by DurableStateBehaviorTestKit to retrieve the `persistenceId`
without invoking user signal handlers.
+ */
+ final case class GetPersistenceIdForTestKit(replyTo:
ActorRef[PersistenceId]) extends InternalProtocol
+
+ /**
+ * Used to retrieve a non-null state.
* Can't be a Signal because those are not stashed.
*/
final case class GetState[State](replyTo: ActorRef[State]) extends
InternalProtocol
+ /**
+ * Used by DurableStateBehaviorTestKit to retrieve the state.
+ * The response envelope supports a `null` state.
+ */
+ final case class GetStateWithReply[State](replyTo:
ActorRef[GetStateReply[State]]) extends InternalProtocol
+
+ final case class GetStateReply[State](currentState: State)
+
}
@InternalApi
diff --git
a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/internal/Recovering.scala
b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/internal/Recovering.scala
index a81a666899..89d90e0390 100644
---
a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/internal/Recovering.scala
+++
b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/internal/Recovering.scala
@@ -30,7 +30,11 @@ import pekko.persistence._
import pekko.persistence.state.scaladsl.GetObjectResult
import pekko.persistence.typed.state.RecoveryCompleted
import pekko.persistence.typed.state.RecoveryFailed
-import pekko.persistence.typed.state.internal.DurableStateBehaviorImpl.GetState
+import pekko.persistence.typed.state.internal.DurableStateBehaviorImpl.{
+ GetPersistenceIdForTestKit,
+ GetState,
+ GetStateWithReply
+}
import pekko.persistence.typed.state.internal.Running.WithRevisionAccessible
import pekko.util.PrettyDuration._
@@ -94,12 +98,14 @@ private[pekko] class Recovering[C, S](
Behaviors.unhandled
} else
onCommand(cmd)
- case get: GetState[S @unchecked] => stashInternal(get)
- case RecoveryPermitGranted => Behaviors.unhandled // should not
happen, we already have the permit
- case UpsertSuccess => Behaviors.unhandled
- case _: UpsertFailure => Behaviors.unhandled
- case DeleteSuccess => Behaviors.unhandled
- case _: DeleteFailure => Behaviors.unhandled
+ case get: GetPersistenceIdForTestKit => stashInternal(get)
+ case get: GetState[S @unchecked] => stashInternal(get)
+ case get: GetStateWithReply[S @unchecked] => stashInternal(get)
+ case RecoveryPermitGranted => Behaviors.unhandled //
should not happen, we already have the permit
+ case UpsertSuccess => Behaviors.unhandled
+ case _: UpsertFailure => Behaviors.unhandled
+ case DeleteSuccess => Behaviors.unhandled
+ case _: DeleteFailure => Behaviors.unhandled
}
}
diff --git
a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/internal/Running.scala
b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/internal/Running.scala
index 7d8d005acf..1371ce3515 100644
---
a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/internal/Running.scala
+++
b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/internal/Running.scala
@@ -28,7 +28,12 @@ import pekko.actor.typed.scaladsl.Behaviors
import pekko.actor.typed.scaladsl.LoggerOps
import pekko.annotation.InternalApi
import pekko.annotation.InternalStableApi
-import pekko.persistence.typed.state.internal.DurableStateBehaviorImpl.GetState
+import pekko.persistence.typed.state.internal.DurableStateBehaviorImpl.{
+ GetPersistenceIdForTestKit,
+ GetState,
+ GetStateReply,
+ GetStateWithReply
+}
import pekko.persistence.typed.state.scaladsl.Effect
/**
@@ -109,8 +114,10 @@ private[pekko] object Running {
} else {
onCommand(state, c)
}
- case get: GetState[S @unchecked] => onGetState(get)
- case _ => Behaviors.unhandled
+ case get: GetPersistenceIdForTestKit => onGetPersistenceId(get)
+ case get: GetState[S @unchecked] => onGetState(get)
+ case get: GetStateWithReply[S @unchecked] => onGetStateWithReply(get)
+ case _ => Behaviors.unhandled
}
override def onSignal: PartialFunction[Signal, Behavior[InternalProtocol]]
= {
@@ -174,12 +181,23 @@ private[pekko] object Running {
}
}
- // Used by DurableStateBehaviorTestKit to retrieve the state.
+ // Used by DurableStateBehaviorTestKit to retrieve the persistence id.
+ def onGetPersistenceId(get: GetPersistenceIdForTestKit):
Behavior[InternalProtocol] = {
+ get.replyTo ! setup.persistenceId
+ this
+ }
+
def onGetState(get: GetState[S]): Behavior[InternalProtocol] = {
get.replyTo ! state.state
this
}
+ // Used by DurableStateBehaviorTestKit to retrieve the state.
+ def onGetStateWithReply(get: GetStateWithReply[S]):
Behavior[InternalProtocol] = {
+ get.replyTo ! GetStateReply(state.state)
+ this
+ }
+
private def handlePersist(
newState: S,
cmd: Any,
@@ -276,16 +294,18 @@ private[pekko] object Running {
override def onMessage(msg: InternalProtocol): Behavior[InternalProtocol]
= {
msg match {
- case UpsertSuccess => onUpsertSuccess()
- case UpsertFailure(exc) => onUpsertFailed(exc)
- case in: IncomingCommand[C @unchecked] => onCommand(in)
- case get: GetState[S @unchecked] => stashInternal(get)
- case DeleteSuccess => onDeleteSuccess()
- case DeleteFailure(exc) => onDeleteFailed(exc)
- case RecoveryTimeout => Behaviors.unhandled
- case RecoveryPermitGranted => Behaviors.unhandled
- case _: GetSuccess[?] => Behaviors.unhandled
- case _: GetFailure => Behaviors.unhandled
+ case UpsertSuccess => onUpsertSuccess()
+ case UpsertFailure(exc) => onUpsertFailed(exc)
+ case in: IncomingCommand[C @unchecked] => onCommand(in)
+ case get: GetPersistenceIdForTestKit => stashInternal(get)
+ case get: GetState[S @unchecked] => stashInternal(get)
+ case get: GetStateWithReply[S @unchecked] => stashInternal(get)
+ case DeleteSuccess => onDeleteSuccess()
+ case DeleteFailure(exc) => onDeleteFailed(exc)
+ case RecoveryTimeout => Behaviors.unhandled
+ case RecoveryPermitGranted => Behaviors.unhandled
+ case _: GetSuccess[?] => Behaviors.unhandled
+ case _: GetFailure => Behaviors.unhandled
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
