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 63d22b60fe Clean up stopped FSM transition listeners (#3096)
63d22b60fe is described below
commit 63d22b60fee9f2ca1fd5c61f6ea37860201bac09
Author: Minh Vu <[email protected]>
AuthorDate: Wed Jun 24 14:36:10 2026 +0200
Clean up stopped FSM transition listeners (#3096)
* Clean up stopped FSM transition listeners
* mima issues
---------
Co-authored-by: PJ Fanning <[email protected]>
---
.../org/apache/pekko/actor/FSMTransitionSpec.scala | 28 +++++++++++++--
.../fsm-transition-listeners.excludes | 19 ++++++++++
.../main/scala/org/apache/pekko/actor/FSM.scala | 40 +++++++++++++++++-----
docs/src/main/paradox/fsm.md | 5 ++-
.../fsm-transition-listeners.excludes | 19 ++++++++++
.../pekko/persistence/fsm/PersistentFSM.scala | 1 +
.../pekko/persistence/fsm/PersistentFSMBase.scala | 39 ++++++++++++++++-----
.../pekko/persistence/fsm/PersistentFSMSpec.scala | 30 ++++++++++++++--
8 files changed, 158 insertions(+), 23 deletions(-)
diff --git
a/actor-tests/src/test/scala/org/apache/pekko/actor/FSMTransitionSpec.scala
b/actor-tests/src/test/scala/org/apache/pekko/actor/FSMTransitionSpec.scala
index aaca6064a3..65c11e4dc4 100644
--- a/actor-tests/src/test/scala/org/apache/pekko/actor/FSMTransitionSpec.scala
+++ b/actor-tests/src/test/scala/org/apache/pekko/actor/FSMTransitionSpec.scala
@@ -20,6 +20,8 @@ import pekko.testkit._
object FSMTransitionSpec {
+ case object ListenerCount
+
class Supervisor extends Actor {
def receive = { case _ => }
}
@@ -44,7 +46,8 @@ object FSMTransitionSpec {
case Event("tick", _) => goto(0)
}
whenUnhandled {
- case Event("reply", _) => stay().replying("reply")
+ case Event("reply", _) => stay().replying("reply")
+ case Event(ListenerCount, _) => stay().replying(listeners.size)
}
initialize()
override def preRestart(reason: Throwable, msg: Option[Any]): Unit = {
target ! "restarted" }
@@ -107,11 +110,32 @@ class FSMTransitionSpec extends PekkoSpec with
ImplicitSender {
within(1.second) {
fsm ! FSM.SubscribeTransitionCallBack(forward)
expectMsg(FSM.CurrentState(fsm, 0))
- pekko.pattern.gracefulStop(forward, 5.seconds)
+ watch(forward)
+ system.stop(forward)
+ expectTerminated(forward)
fsm ! "tick"
expectNoMessage()
}
}
+
+ "remove stopped listeners" in {
+ val forward = system.actorOf(Props(new Forwarder(testActor)))
+ val fsm = system.actorOf(Props(new MyFSM(testActor)))
+
+ within(3.seconds) {
+ fsm ! FSM.SubscribeTransitionCallBack(forward)
+ expectMsg(FSM.CurrentState(fsm, 0))
+
+ watch(forward)
+ system.stop(forward)
+ expectTerminated(forward)
+
+ awaitAssert {
+ fsm ! ListenerCount
+ expectMsg(0)
+ }
+ }
+ }
}
"A FSM" must {
diff --git
a/actor/src/main/mima-filters/2.0.x.backwards.excludes/fsm-transition-listeners.excludes
b/actor/src/main/mima-filters/2.0.x.backwards.excludes/fsm-transition-listeners.excludes
new file mode 100644
index 0000000000..2c6aaef9bf
--- /dev/null
+++
b/actor/src/main/mima-filters/2.0.x.backwards.excludes/fsm-transition-listeners.excludes
@@ -0,0 +1,19 @@
+# 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.
+
+# Clean up stopped FSM transition listeners
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.actor.FSM.org$apache$pekko$actor$FSM$$watchedListeners*")
diff --git a/actor/src/main/scala/org/apache/pekko/actor/FSM.scala
b/actor/src/main/scala/org/apache/pekko/actor/FSM.scala
index fd93a273b7..e7a2a10805 100644
--- a/actor/src/main/scala/org/apache/pekko/actor/FSM.scala
+++ b/actor/src/main/scala/org/apache/pekko/actor/FSM.scala
@@ -99,6 +99,7 @@ object FSM {
* INTERNAL API
*/
private final case class TimeoutMarker(generation: Long)
+ private final case class ListenerTerminated(actorRef: ActorRef)
/** INTERNAL API */
@InternalApi
@@ -406,9 +407,8 @@ object FSM {
*
* Another feature is that other actors may subscribe for transition events by
* sending a <code>SubscribeTransitionCallback</code> message to this actor.
- * Stopping a listener without unregistering will not remove the listener from
the
- * subscription list; use <code>UnsubscribeTransitionCallback</code> before
stopping
- * the listener.
+ * Stopping a listener will remove it from the subscription list. Subscriptions
+ * may also be removed explicitly with
<code>UnsubscribeTransitionCallback</code>.
*
* State timeouts set an upper bound to the time which may pass before another
* message is received in the current state. If no external message is
@@ -779,11 +779,34 @@ trait FSM[S, D] extends Actor with Listeners with
ActorLogging {
* transition handling
*/
private var transitionEvent: List[TransitionHandler] = Nil
+ private var watchedListeners: Set[ActorRef] = Set.empty
private def handleTransition(prev: S, next: S): Unit = {
val tuple = (prev, next)
for (te <- transitionEvent) { if (te.isDefinedAt(tuple)) te(tuple) }
}
+ private def addListener(actorRef: ActorRef): Unit = {
+ if (listeners.add(actorRef)) {
+ context.asInstanceOf[ActorCell].isWatching(actorRef) match {
+ case false =>
+ context.watchWith(actorRef, ListenerTerminated(actorRef))
+ watchedListeners += actorRef
+ case true =>
+ log.warning(
+ "Listener [{}] is already watched by this FSM; it will not be
automatically removed on termination. " +
+ "Use UnsubscribeTransitionCallBack explicitly.",
+ actorRef)
+ }
+ }
+ }
+
+ private def removeListener(actorRef: ActorRef): Unit = {
+ if (listeners.remove(actorRef) && watchedListeners.contains(actorRef)) {
+ watchedListeners -= actorRef
+ context.unwatch(actorRef)
+ }
+ }
+
/*
* *******************************************
* Main actor receive() method
@@ -807,19 +830,20 @@ trait FSM[S, D] extends Actor with Listeners with
ActorLogging {
processMsg(msg, t)
}
case SubscribeTransitionCallBack(actorRef) =>
- // TODO Use context.watch(actor) and receive Terminated(actor) to clean
up list
- listeners.add(actorRef)
+ addListener(actorRef)
// send current state back as reference point
actorRef ! CurrentState(self, currentState.stateName)
case Listen(actorRef) =>
- // TODO Use context.watch(actor) and receive Terminated(actor) to clean
up list
- listeners.add(actorRef)
+ addListener(actorRef)
// send current state back as reference point
actorRef ! CurrentState(self, currentState.stateName)
case UnsubscribeTransitionCallBack(actorRef) =>
- listeners.remove(actorRef)
+ removeListener(actorRef)
case Deafen(actorRef) =>
+ removeListener(actorRef)
+ case ListenerTerminated(actorRef) =>
listeners.remove(actorRef)
+ watchedListeners -= actorRef
case value =>
if (timeoutFuture.isDefined) {
timeoutFuture.get.cancel()
diff --git a/docs/src/main/paradox/fsm.md b/docs/src/main/paradox/fsm.md
index c97df9f214..65f44037f0 100644
--- a/docs/src/main/paradox/fsm.md
+++ b/docs/src/main/paradox/fsm.md
@@ -417,9 +417,8 @@ transitions use `stay()` instead of `goto(S)`.
External monitors may be unregistered by sending
`UnsubscribeTransitionCallBack(actorRef)` to the `FSM` actor.
-Stopping a listener without unregistering will not remove the listener from the
-subscription list; use `UnsubscribeTransitionCallback` before stopping
-the listener.
+Stopping a listener will remove it from the subscription list. Subscriptions
may
+also be removed explicitly with `UnsubscribeTransitionCallback`.
@@@ div { .group-scala }
diff --git
a/persistence/src/main/mima-filters/2.0.x.backwards.excludes/fsm-transition-listeners.excludes
b/persistence/src/main/mima-filters/2.0.x.backwards.excludes/fsm-transition-listeners.excludes
new file mode 100644
index 0000000000..3f1be9e8f1
--- /dev/null
+++
b/persistence/src/main/mima-filters/2.0.x.backwards.excludes/fsm-transition-listeners.excludes
@@ -0,0 +1,19 @@
+# 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.
+
+# Clean up stopped FSM transition listeners
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.persistence.fsm.PersistentFSMBase.org$apache$pekko$persistence$fsm$PersistentFSMBase$$watchedListeners*")
diff --git
a/persistence/src/main/scala/org/apache/pekko/persistence/fsm/PersistentFSM.scala
b/persistence/src/main/scala/org/apache/pekko/persistence/fsm/PersistentFSM.scala
index 78de37df71..18c8cbaac4 100644
---
a/persistence/src/main/scala/org/apache/pekko/persistence/fsm/PersistentFSM.scala
+++
b/persistence/src/main/scala/org/apache/pekko/persistence/fsm/PersistentFSM.scala
@@ -310,6 +310,7 @@ object PersistentFSM {
/** INTERNAL API */
@InternalApi
private[persistence] final case class TimeoutMarker(generation: Long)
+ private[persistence] final case class ListenerTerminated(actorRef: ActorRef)
/** INTERNAL API */
@InternalApi
diff --git
a/persistence/src/main/scala/org/apache/pekko/persistence/fsm/PersistentFSMBase.scala
b/persistence/src/main/scala/org/apache/pekko/persistence/fsm/PersistentFSMBase.scala
index 448f00ef52..a26c28db3e 100644
---
a/persistence/src/main/scala/org/apache/pekko/persistence/fsm/PersistentFSMBase.scala
+++
b/persistence/src/main/scala/org/apache/pekko/persistence/fsm/PersistentFSMBase.scala
@@ -85,9 +85,8 @@ import pekko.routing.{ Deafen, Listen, Listeners }
*
* Another feature is that other actors may subscribe for transition events by
* sending a <code>SubscribeTransitionCallback</code> message to this actor.
- * Stopping a listener without unregistering will not remove the listener from
the
- * subscription list; use <code>UnsubscribeTransitionCallback</code> before
stopping
- * the listener.
+ * Stopping a listener will remove it from the subscription list. Subscriptions
+ * may also be removed explicitly with
<code>UnsubscribeTransitionCallback</code>.
*
* State timeouts set an upper bound to the time which may pass before another
* message is received in the current state. If no external message is
@@ -482,11 +481,34 @@ trait PersistentFSMBase[S, D, E] extends Actor with
Listeners with ActorLogging
* transition handling
*/
private var transitionEvent: List[TransitionHandler] = Nil
+ private var watchedListeners: Set[ActorRef] = Set.empty
private def handleTransition(prev: S, next: S): Unit = {
val tuple = (prev, next)
for (te <- transitionEvent) { if (te.isDefinedAt(tuple)) te(tuple) }
}
+ private def addListener(actorRef: ActorRef): Unit = {
+ if (listeners.add(actorRef)) {
+ context.asInstanceOf[ActorCell].isWatching(actorRef) match {
+ case false =>
+ context.watchWith(actorRef, ListenerTerminated(actorRef))
+ watchedListeners += actorRef
+ case true =>
+ log.warning(
+ "Listener [{}] is already watched by this FSM; it will not be
automatically removed on termination. " +
+ "Use UnsubscribeTransitionCallBack explicitly.",
+ actorRef)
+ }
+ }
+ }
+
+ private def removeListener(actorRef: ActorRef): Unit = {
+ if (listeners.remove(actorRef) && watchedListeners.contains(actorRef)) {
+ watchedListeners -= actorRef
+ context.unwatch(actorRef)
+ }
+ }
+
/*
* *******************************************
* Main actor receive() method
@@ -510,19 +532,20 @@ trait PersistentFSMBase[S, D, E] extends Actor with
Listeners with ActorLogging
processMsg(msg, t)
}
case SubscribeTransitionCallBack(actorRef) =>
- // TODO Use context.watch(actor) and receive Terminated(actor) to clean
up list
- listeners.add(actorRef)
+ addListener(actorRef)
// send current state back as reference point
actorRef ! CurrentState(self, currentState.stateName,
currentState.timeout)
case Listen(actorRef) =>
- // TODO Use context.watch(actor) and receive Terminated(actor) to clean
up list
- listeners.add(actorRef)
+ addListener(actorRef)
// send current state back as reference point
actorRef ! CurrentState(self, currentState.stateName,
currentState.timeout)
case UnsubscribeTransitionCallBack(actorRef) =>
- listeners.remove(actorRef)
+ removeListener(actorRef)
case Deafen(actorRef) =>
+ removeListener(actorRef)
+ case ListenerTerminated(actorRef) =>
listeners.remove(actorRef)
+ watchedListeners -= actorRef
case value =>
if (timeoutFuture.isDefined) {
timeoutFuture.get.cancel()
diff --git
a/persistence/src/test/scala/org/apache/pekko/persistence/fsm/PersistentFSMSpec.scala
b/persistence/src/test/scala/org/apache/pekko/persistence/fsm/PersistentFSMSpec.scala
index c2d2708670..57f26babfb 100644
---
a/persistence/src/test/scala/org/apache/pekko/persistence/fsm/PersistentFSMSpec.scala
+++
b/persistence/src/test/scala/org/apache/pekko/persistence/fsm/PersistentFSMSpec.scala
@@ -73,6 +73,30 @@ abstract class PersistentFSMSpec(config: Config) extends
PersistenceSpec(config)
expectTerminated(fsmRef)
}
+ "remove stopped transition listeners" in {
+ val persistenceId = name
+ val listener = TestProbe()
+ val fsmRef = system.actorOf(SimpleTransitionFSM.props(persistenceId,
TestProbe().ref))
+
+ fsmRef ! SubscribeTransitionCallBack(listener.ref)
+ listener.expectMsg(CurrentState(fsmRef, LookingAround, None))
+ fsmRef ! ListenerCount
+ expectMsg(1)
+
+ watch(listener.ref)
+ system.stop(listener.ref)
+ expectTerminated(listener.ref)
+
+ awaitAssert {
+ fsmRef ! ListenerCount
+ expectMsg(0)
+ }
+
+ watch(fsmRef)
+ fsmRef ! PoisonPill
+ expectTerminated(fsmRef)
+ }
+
"function as a regular FSM on state timeout" taggedAs TimingTest in {
val persistenceId = name
val fsmRef = system.actorOf(WebStoreCustomerFSM.props(persistenceId,
dummyReportActorRef))
@@ -450,6 +474,7 @@ object PersistentFSMSpec {
case object Leave extends Command
case object GetCurrentCart extends Command
// #customer-commands
+ case object ListenerCount
// #customer-domain-events
sealed trait DomainEvent
@@ -472,8 +497,9 @@ object PersistentFSMSpec {
startWith(LookingAround, EmptyShoppingCart)
when(LookingAround) {
- case Event("stay", _) => stay()
- case Event(_, _) => goto(LookingAround)
+ case Event("stay", _) => stay()
+ case Event(ListenerCount, _) => stay().replying(listeners.size)
+ case Event(_, _) => goto(LookingAround)
}
onTransition {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
