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 25f8b12436 Document and test EventStream type-based subscription
semantics #3658 (#3658)
25f8b12436 is described below
commit 25f8b1243672a25807930fd89661da2cbada438a
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Fri Oct 23 22:49:31 2026 +0800
Document and test EventStream type-based subscription semantics #3658
(#3658)
Motivation:
EventStream supports subscriptions to multiple event types, with different
class-specific and all-subscription unsubscribe operations.
Modification:
Document the Classic and Typed subscription contracts, add focused
assertions for both APIs, clarify dead-letter superclass subscriptions, and
link both guides from the general delivery documentation.
Result:
The documented subscription behavior has focused coverage without changing
event dispatch.
Tests:
Validation used a local integration checkout with current main and the
other pending PRs.
- Classic and Typed `EventStreamSpec`: passed.
- `docs/paradox`, `headerCreateAll`, `+headerCheckAll`, native scalafmt,
and `git diff --check`: passed.
- Full validation is delegated to remote CI.
References:
- Refs [akka/akka-core#32094](https://github.com/akka/akka-core/pull/32094).
- Refs [akka/akka-core#32100](https://github.com/akka/akka-core/pull/32100).
---
.../org/apache/pekko/event/EventStreamSpec.scala | 22 +++++++++++++++++++
.../actor/typed/eventstream/EventStreamSpec.scala | 25 +++++++++++++++++++++-
.../actor/typed/eventstream/EventStream.scala | 7 +++---
.../scala/org/apache/pekko/event/EventStream.scala | 12 ++++++++++-
.../general/message-delivery-reliability.md | 4 ++--
docs/src/main/paradox/typed/event-stream.md | 15 ++++++++++++-
docs/src/main/paradox/typed/logging.md | 2 +-
7 files changed, 78 insertions(+), 9 deletions(-)
diff --git
a/actor-tests/src/test/scala/org/apache/pekko/event/EventStreamSpec.scala
b/actor-tests/src/test/scala/org/apache/pekko/event/EventStreamSpec.scala
index 476135829e..80ebb117d8 100644
--- a/actor-tests/src/test/scala/org/apache/pekko/event/EventStreamSpec.scala
+++ b/actor-tests/src/test/scala/org/apache/pekko/event/EventStreamSpec.scala
@@ -83,6 +83,28 @@ class EventStreamSpec extends
PekkoSpec(EventStreamSpec.config) {
"An EventStream" must {
+ "remove one unrelated subscription or all subscriptions for the same
actor" in {
+ val bus = new EventStream(system, false)
+ val listener = TestProbe()
+ bus.subscribe(listener.ref, classOf[M]) shouldBe true
+ bus.subscribe(listener.ref, classOf[String]) shouldBe true
+ bus.publish(M(1))
+ bus.publish("both subscriptions")
+ listener.expectMsg(M(1))
+ listener.expectMsg("both subscriptions")
+
+ bus.unsubscribe(listener.ref, classOf[M]) shouldBe true
+ bus.publish(M(2))
+ bus.publish("remaining subscription")
+ listener.expectMsg("remaining subscription")
+ listener.expectNoMessage(100.millis)
+
+ bus.unsubscribe(listener.ref)
+ bus.publish(M(3))
+ bus.publish("no subscriptions")
+ listener.expectNoMessage(100.millis)
+ }
+
"manage subscriptions" in {
// #event-bus-start-unsubscriber-scala
val bus = new EventStream(system, true)
diff --git
a/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/eventstream/EventStreamSpec.scala
b/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/eventstream/EventStreamSpec.scala
index 922768cc5d..8d33add6f9 100644
---
a/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/eventstream/EventStreamSpec.scala
+++
b/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/eventstream/EventStreamSpec.scala
@@ -8,7 +8,7 @@
*/
/*
- * Copyright (C) 2019-2022 Lightbend Inc. <https://www.lightbend.com>
+ * Copyright (C) 2019-2023 Lightbend Inc. <https://www.lightbend.com>
*/
package org.apache.pekko.actor.typed.eventstream
@@ -33,6 +33,29 @@ class EventStreamSpec extends ScalaTestWithActorTestKit with
AnyWordSpecLike wit
private final val ShortWait = 100.millis
+ "a subscriber to unrelated event types" must {
+ "receive both types and remove every subscription with Unsubscribe" in {
+ val listener = createTestProbe[AnyRef]()
+ val barrier = createTestProbe[String]()
+ val stream = system.eventStream
+ stream ! Subscribe[EventClass](listener.ref)
+ stream ! Subscribe[String](listener.ref)
+ stream ! Publish(EventClass())
+ stream ! Publish("before unsubscribe")
+ listener.expectMessage(EventClass())
+ listener.expectMessage("before unsubscribe")
+
+ stream ! Subscribe(barrier.ref)
+ stream ! Unsubscribe(listener.ref)
+ stream ! Publish(EventClass())
+ stream ! Publish("after unsubscribe")
+ // This publication is processed after both Unsubscribe and the
preceding Publish.
+ barrier.expectMessage("after unsubscribe")
+ listener.expectNoMessage(ShortWait)
+ stream ! Unsubscribe(barrier.ref)
+ }
+ }
+
"system event stream".can {
val eventObjListener: TestProbe[EventObj.type] = testKit.createTestProbe()
val eventClassListener: TestProbe[EventClass] = testKit.createTestProbe()
diff --git
a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/eventstream/EventStream.scala
b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/eventstream/EventStream.scala
index 07190a07c4..106001a4eb 100644
---
a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/eventstream/EventStream.scala
+++
b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/eventstream/EventStream.scala
@@ -8,7 +8,7 @@
*/
/*
- * Copyright (C) 2019-2022 Lightbend Inc. <https://www.lightbend.com>
+ * Copyright (C) 2019-2023 Lightbend Inc. <https://www.lightbend.com>
*/
package org.apache.pekko.actor.typed.eventstream
@@ -39,8 +39,9 @@ object EventStream {
}
/**
- * Subscribe a typed actor to listen for types or subtypes of E
+ * Subscribe a typed actor to listen for events of type E and its subtypes
* by sending this command to the
[[pekko.actor.typed.ActorSystem.eventStream]].
+ * The same actor can subscribe to multiple different types.
*
* ==Simple example==
* {{{
@@ -69,7 +70,7 @@ object EventStream {
}
/**
- * Unsubscribe an actor ref from the event stream
+ * Remove all subscriptions for the given actor ref from the event stream
* by sending this command to the
[[pekko.actor.typed.ActorSystem.eventStream]].
*/
final case class Unsubscribe[E](subscriber: ActorRef[E]) extends Command
diff --git a/actor/src/main/scala/org/apache/pekko/event/EventStream.scala
b/actor/src/main/scala/org/apache/pekko/event/EventStream.scala
index 8e777f1870..6ba583e0f3 100644
--- a/actor/src/main/scala/org/apache/pekko/event/EventStream.scala
+++ b/actor/src/main/scala/org/apache/pekko/event/EventStream.scala
@@ -8,7 +8,7 @@
*/
/*
- * Copyright (C) 2009-2022 Lightbend Inc. <https://www.lightbend.com>
+ * Copyright (C) 2009-2023 Lightbend Inc. <https://www.lightbend.com>
*/
package org.apache.pekko.event
@@ -53,6 +53,10 @@ class EventStream(sys: ActorSystem, private val debug:
Boolean) extends LoggingB
else subscriber ! event
}
+ /**
+ * Subscribe an actor to events of the given class and its subtypes.
+ * The same actor can subscribe to multiple different classes.
+ */
override def subscribe(subscriber: ActorRef, channel: Class[?]): Boolean = {
if (subscriber eq null) throw new IllegalArgumentException("subscriber is
null")
if (debug)
@@ -61,6 +65,9 @@ class EventStream(sys: ActorSystem, private val debug:
Boolean) extends LoggingB
super.subscribe(subscriber, channel)
}
+ /**
+ * Unsubscribe an actor from the given class and its subtypes, leaving
unrelated subscriptions intact.
+ */
override def unsubscribe(subscriber: ActorRef, channel: Class[?]): Boolean =
{
if (subscriber eq null) throw new IllegalArgumentException("subscriber is
null")
val ret = super.unsubscribe(subscriber, channel)
@@ -71,6 +78,9 @@ class EventStream(sys: ActorSystem, private val debug:
Boolean) extends LoggingB
ret
}
+ /**
+ * Remove all subscriptions for the given actor.
+ */
override def unsubscribe(subscriber: ActorRef): Unit = {
if (subscriber eq null) throw new IllegalArgumentException("subscriber is
null")
super.unsubscribe(subscriber)
diff --git a/docs/src/main/paradox/general/message-delivery-reliability.md
b/docs/src/main/paradox/general/message-delivery-reliability.md
index 7c5f3e997c..a3d7571eb5 100644
--- a/docs/src/main/paradox/general/message-delivery-reliability.md
+++ b/docs/src/main/paradox/general/message-delivery-reliability.md
@@ -339,8 +339,8 @@ guaranteed delivery.
### How do I Receive Dead Letters?
An actor can subscribe to class @apidoc[actor.DeadLetter](actor.DeadLetter) on
the event
-stream, see @ref:[Event Stream](../event-bus.md#event-stream)
-for how to do that. The subscribed actor will then receive all dead
+stream; see @ref:[Typed Event Stream](../typed/event-stream.md) or
+@ref:[Classic Event Stream](../event-bus.md#event-stream) for how to do that.
The subscribed actor will then receive all dead
letters published in the (local) system from that point onwards. Dead letters
are not propagated over the network, if you want to collect them in one place
you will have to subscribe one actor per network node and forward them
diff --git a/docs/src/main/paradox/typed/event-stream.md
b/docs/src/main/paradox/typed/event-stream.md
index c74abf413d..35eaa44849 100644
--- a/docs/src/main/paradox/typed/event-stream.md
+++ b/docs/src/main/paradox/typed/event-stream.md
@@ -17,6 +17,15 @@ version=PekkoVersion
## How to use
+Each actor system has a local event stream for publishing system events and
application events to subscribed actors.
+Events are delivered to subscribers of their class or a superclass. They are
not propagated to other actor systems.
+Subscriptions are removed automatically when the subscriber terminates.
+
+The same actor can subscribe to several event types. Sending
`EventStream.Unsubscribe` for that actor removes all of its
+subscriptions. These rules apply to both the Scala and Java APIs.
+
+Classic logging events use the event stream. Typed actors log through SLF4J,
as described in @ref:[Logging](logging.md).
+
The following example demonstrates how a subscription works. Given an actor:
Scala
@@ -37,7 +46,11 @@ It can be used as follows:
@@@
-It is possible to subscribe to common superclass:
+A subscription to `DeadLetter` receives ordinary dead letters. Messages marked
with
+`DeadLetterSuppression` are published as `SuppressedDeadLetter` instead.
Subscribe to their common superclass
+`AllDeadLetters` to receive both, as well as `Dropped` and `UnhandledMessage`,
which also extend it.
+
+It is possible to subscribe to a common superclass:
Scala
: @@snip
[EventStreamDocSpec.scala](/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/eventstream/EventStreamDocSpec.scala)
{ #listen-to-super-class }
diff --git a/docs/src/main/paradox/typed/logging.md
b/docs/src/main/paradox/typed/logging.md
index 0fb074d692..c51b79e8fa 100644
--- a/docs/src/main/paradox/typed/logging.md
+++ b/docs/src/main/paradox/typed/logging.md
@@ -329,7 +329,7 @@ pekko {
```
To customize the logging further or take other actions for dead letters you
can subscribe
-to the @ref:[Event Stream](../event-bus.md#event-stream).
+to the @ref:[Event Stream](event-stream.md).
### Auxiliary logging options
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
