EventStream
You are viewing the documentation for the new actor APIs, to view the Akka Classic documentation, see Classic Event Stream.
Dependency
The Akka dependencies are available from Akka’s library repository. To access them there, you need to configure the URL for this repository.
- sbt
resolvers += "Akka library repository".at("https://repo.akka.io/maven")
- Maven
<project> ... <repositories> <repository> <id>akka-repository</id> <name>Akka library repository</name> <url>https://repo.akka.io/maven</url> </repository> </repositories> </project>
- Gradle
repositories { mavenCentral() maven { url "https://repo.akka.io/maven" } }
To use Akka Actor Typed, you must add the following dependency in your project:
- sbt
val AkkaVersion = "2.10.0+14-bc29c0c3-SNAPSHOT" libraryDependencies += "com.typesafe.akka" %% "akka-actor-typed" % AkkaVersion
- Maven
<properties> <scala.binary.version>2.13</scala.binary.version> </properties> <dependencyManagement> <dependencies> <dependency> <groupId>com.typesafe.akka</groupId> <artifactId>akka-bom_${scala.binary.version}</artifactId> <version>2.10.0+14-bc29c0c3-SNAPSHOT</version> <type>pom</type> <scope>import</scope> </dependency> </dependencies> </dependencyManagement> <dependencies> <dependency> <groupId>com.typesafe.akka</groupId> <artifactId>akka-actor-typed_${scala.binary.version}</artifactId> </dependency> </dependencies>
- Gradle
def versions = [ ScalaBinary: "2.13" ] dependencies { implementation platform("com.typesafe.akka:akka-bom_${versions.ScalaBinary}:2.10.0+14-bc29c0c3-SNAPSHOT") implementation "com.typesafe.akka:akka-actor-typed_${versions.ScalaBinary}" }
Introduction
The event stream is the main Event Bus of each actor system: it is used for carrying log messages and Dead Letters and may be used by the user code for other purposes as well.
It uses Subchannel Classification which enables registering to related sets of channels.
How to use
The following example demonstrates how a subscription works. Given an actor:
- Scala
-
source
import akka.actor.typed.Behavior import akka.actor.typed.eventstream.EventStream.Subscribe import akka.actor.typed.scaladsl.Behaviors object DeadLetterListener { def apply(): Behavior[String] = Behaviors.setup { context => // subscribe DeadLetter at startup. val adapter = context.messageAdapter[DeadLetter](d => d.message.toString) context.system.eventStream ! Subscribe(adapter) Behaviors.receiveMessage { case msg: String => println(msg) Behaviors.same } } } ActorSystem(Behaviors.setup[Void] { context => context.spawn(DeadLetterListener(), "DeadLetterListener", Props.empty) Behaviors.empty }, "System")
- Java
-
source
import akka.actor.DeadLetter; import akka.actor.typed.ActorRef; import akka.actor.typed.ActorSystem;
the actor definition like this:
sourcestatic class DeadLetterActor extends AbstractBehavior<String> {
public static Behavior<String> create() {
return Behaviors.setup(DeadLetterActor::new);
}
public DeadLetterActor(ActorContext<String> context) {
super(context);
ActorRef<DeadLetter> messageAdapter = context.messageAdapter(
DeadLetter.class,
d -> d.message().toString()
);
context.getSystem().eventStream()
.tell(new Subscribe<>(DeadLetter.class, messageAdapter));
}
@Override
public Receive<String> createReceive() {
return newReceiveBuilder().onMessage(String.class, msg -> {
System.out.println(msg);
return Behaviors.same();
}).build();
}
}
it can be subscribed like this:
sourceActorSystem<DeadLetter> system = ActorSystem.create(Behaviors.empty(), "DeadLetters");
system.eventStream().tell(new Subscribe<>(DeadLetter.class, system));
It is also worth pointing out that thanks to the way the subchannel classification is implemented in the event stream, it is possible to subscribe to a group of events, by subscribing to their common superclass as demonstrated in the following example:
- Scala
-
source
object ListenerActor { abstract class AllKindsOfMusic { def artist: String } case class Jazz(artist: String) extends AllKindsOfMusic case class Electronic(artist: String) extends AllKindsOfMusic def apply(): Behavior[ListenerActor.AllKindsOfMusic] = Behaviors.receive { (context, msg) => msg match { case m: Jazz => println(s"${context.self.path.name} is listening to: ${m.artist}") Behaviors.same case m: Electronic => println(s"${context.self.path.name} is listening to: ${m.artist}") Behaviors.same case _ => Behaviors.same } } } implicit val system: ActorSystem[SpawnProtocol.Command] = ActorSystem(SpawnProtocol(), "SpawnProtocol") implicit val ec: ExecutionContext = system.executionContext val jazzListener: Future[ActorRef[Jazz]] = system.ask(Spawn(behavior = ListenerActor(), name = "jazz", props = Props.empty, _)) val musicListener: Future[ActorRef[AllKindsOfMusic]] = system.ask(Spawn(behavior = ListenerActor(), name = "music", props = Props.empty, _)) for (jazzListenerRef <- jazzListener; musicListenerRef <- musicListener) { system.eventStream ! Subscribe(jazzListenerRef) system.eventStream ! Subscribe(musicListenerRef) } // only musicListener gets this message, since it listens to *all* kinds of music: system.eventStream ! Publish(Electronic("Parov Stelar")) // jazzListener and musicListener will be notified about Jazz: system.eventStream ! Publish(Jazz("Sonny Rollins"))
- Java
-
source
interface AllKindsOfMusic { } class Jazz implements AllKindsOfMusic { public final String artist; public Jazz(String artist) { this.artist = artist; } } class Electronic implements AllKindsOfMusic { public final String artist; public Electronic(String artist) { this.artist = artist; } } static class Listener extends AbstractBehavior<AllKindsOfMusic> { public static Behavior<AllKindsOfMusic> create() { return Behaviors.setup(Listener::new); } public Listener(ActorContext<AllKindsOfMusic> context) { super(context); } @Override public Receive<AllKindsOfMusic> createReceive() { return newReceiveBuilder() .onMessage(Jazz.class, msg -> { System.out.printf("%s is listening to: %s%n", getContext().getSelf().path().name(), msg); return Behaviors.same(); }) .onMessage(Electronic.class, msg -> { System.out.printf("%s is listening to: %s%n", getContext().getSelf().path().name(), msg); return Behaviors.same(); }).build(); } } ActorSystem<SpawnProtocol.Command> system = ActorSystem.create(SpawnProtocol.create(), "Subclassification"); final Duration timeout = Duration.ofSeconds(3); CompletionStage<ActorRef<Jazz>> jazzListener = AskPattern.ask( system, replyTo -> new Spawn<>(Listener.create().narrow(), "jazzListener", Props.empty(), replyTo), timeout, system.scheduler() ); CompletionStage<ActorRef<AllKindsOfMusic>> musicListener = AskPattern.ask( system, replyTo -> new Spawn<>(Listener.create(), "musicListener", Props.empty(), replyTo), timeout, system.scheduler() ); ActorRef<Jazz> jazzListenerActorRef = jazzListener.toCompletableFuture().join(); ActorRef<AllKindsOfMusic> musicListenerActorRef = musicListener.toCompletableFuture() .join(); system.eventStream().tell(new Subscribe<>(Jazz.class, jazzListenerActorRef)); system.eventStream().tell(new Subscribe<>(AllKindsOfMusic.class, musicListenerActorRef)); // only musicListener gets this message, since it listens to *all* kinds of music: system.eventStream().tell(new Publish<>(new Electronic("Parov Stelar"))); // jazzListener and musicListener will be notified about Jazz: system.eventStream().tell(new Publish<>(new Jazz("Sonny Rollins")));
Similarly to Actor Classification, EventStream
EventStream
will automatically remove subscribers when they terminate.
The event stream is a local facility, meaning that it will not distribute events to other nodes in a clustered environment (unless you subscribe a Remote Actor to the stream explicitly). If you need to broadcast events in an Akka cluster, without knowing your recipients explicitly (i.e. obtaining their ActorRefs), you may want to look into: The Receptionist, Group Routers or Distributed Publish Subscribe in Cluster.
Dead Letters
As described at Stopping actors, messages queued when an actor terminates or sent after its death are re-routed to the dead letter mailbox, which by default will publish the messages wrapped in DeadLetter
DeadLetter
. This wrapper holds the original sender, receiver and message of the envelope which was redirected.
Some internal messages (marked with the DeadLetterSuppression
DeadLetterSuppression
traitinterface) will not end up as dead letters like normal messages. These are by design safe and expected to sometimes arrive at a terminated actor and since they are nothing to worry about, they are suppressed from the default dead letters logging mechanism.
However, in case you find yourself in need of debugging these kinds of low level suppressed dead letters, it’s still possible to subscribe to them explicitly:
- Scala
-
source
import akka.actor.SuppressedDeadLetter system.eventStream ! Subscribe[SuppressedDeadLetter](listener)
- Java
-
source
system.eventStream().tell(new Subscribe<>(SuppressedDeadLetter.class, actor));
or all dead letters (including the suppressed ones):
- Scala
-
source
import akka.actor.AllDeadLetters system.eventStream ! Subscribe[AllDeadLetters](listener)
- Java
-
source
system.eventStream().tell(new Subscribe<>(AllDeadLetters.class, actor));
Other Uses
The event stream is always there and ready to be used, you can publish your own events (it accepts AnyRef
Object
) and subscribe listeners to the corresponding JVM classes.
EventBus
Originally conceived as a way to send messages to groups of actors, the EventBus
EventBus
has been generalized into a set of composable traits abstract base classes implementing a simple interface:
- Scala
-
source
/** * Attempts to register the subscriber to the specified Classifier * @return true if successful and false if not (because it was already * subscribed to that Classifier, or otherwise) */ def subscribe(subscriber: Subscriber, to: Classifier): Boolean /** * Attempts to deregister the subscriber from the specified Classifier * @return true if successful and false if not (because it wasn't subscribed * to that Classifier, or otherwise) */ def unsubscribe(subscriber: Subscriber, from: Classifier): Boolean /** * Attempts to deregister the subscriber from all Classifiers it may be subscribed to */ def unsubscribe(subscriber: Subscriber): Unit /** * Publishes the specified Event to this bus */ def publish(event: Event): Unit
- Java
-
source
/** * Attempts to register the subscriber to the specified Classifier * * @return true if successful and false if not (because it was already subscribed to that * Classifier, or otherwise) */ public boolean subscribe(Subscriber subscriber, Classifier to); /** * Attempts to deregister the subscriber from the specified Classifier * * @return true if successful and false if not (because it wasn't subscribed to that Classifier, * or otherwise) */ public boolean unsubscribe(Subscriber subscriber, Classifier from); /** Attempts to deregister the subscriber from all Classifiers it may be subscribed to */ public void unsubscribe(Subscriber subscriber); /** Publishes the specified Event to this bus */ public void publish(Event event);
Please note that the EventBus does not preserve the sender of the published messages. If you need a reference to the original sender you have to provide it inside the message.
This mechanism is used in different places within Akka, e.g. the EventStream. Implementations can make use of the specific building blocks presented below.
An event bus must define the following three abstract typestype parameters:
Event
is the type of all events published on that busSubscriber
is the type of subscribers allowed to register on that event busClassifier
defines the classifier to be used in selecting subscribers for dispatching events
The traits below are still generic in these types, but they need to be defined for any concrete implementation.
Classifiers
The classifiers presented here are part of the Akka distribution, but rolling your own in case you do not find a perfect match is not difficult, check the implementation of the existing ones on github
Lookup Classification
The simplest classification is just to extract an arbitrary classifier from each event and maintaining a set of subscribers for each possible classifier. This can be compared to tuning in on a radio station. The trait LookupClassification
abstract class LookupEventBus
is still generic in that it abstracts over how to compare subscribers and how exactly to classify them.
The necessary methods to be implemented are illustrated with the following example:
- Scala
-
source
import akka.event.EventBus import akka.event.LookupClassification final case class MsgEnvelope(topic: String, payload: Any) /** * Publishes the payload of the MsgEnvelope when the topic of the * MsgEnvelope equals the String specified when subscribing. */ class LookupBusImpl extends EventBus with LookupClassification { type Event = MsgEnvelope type Classifier = String type Subscriber = ActorRef // is used for extracting the classifier from the incoming events override protected def classify(event: Event): Classifier = event.topic // will be invoked for each event for all subscribers which registered themselves // for the event’s classifier override protected def publish(event: Event, subscriber: Subscriber): Unit = { subscriber ! event.payload } // must define a full order over the subscribers, expressed as expected from // `java.lang.Comparable.compare` override protected def compareSubscribers(a: Subscriber, b: Subscriber): Int = a.compareTo(b) // determines the initial size of the index data structure // used internally (i.e. the expected number of different classifiers) override protected def mapSize(): Int = 128 }
- Java
-
source
import akka.event.japi.LookupEventBus; static class MsgEnvelope { public final String topic; public final Object payload; public MsgEnvelope(String topic, Object payload) { this.topic = topic; this.payload = payload; } } /** * Publishes the payload of the MsgEnvelope when the topic of the MsgEnvelope equals the String * specified when subscribing. */ static class LookupBusImpl extends LookupEventBus<MsgEnvelope, ActorRef, String> { // is used for extracting the classifier from the incoming events @Override public String classify(MsgEnvelope event) { return event.topic; } // will be invoked for each event for all subscribers which registered themselves // for the event’s classifier @Override public void publish(MsgEnvelope event, ActorRef subscriber) { subscriber.tell(event.payload, ActorRef.noSender()); } // must define a full order over the subscribers, expressed as expected from // `java.lang.Comparable.compare` @Override public int compareSubscribers(ActorRef a, ActorRef b) { return a.compareTo(b); } // determines the initial size of the index data structure // used internally (i.e. the expected number of different classifiers) @Override public int mapSize() { return 128; } }
A test for this implementation may look like this:
- Scala
-
source
val lookupBus = new LookupBusImpl lookupBus.subscribe(testActor, "greetings") lookupBus.publish(MsgEnvelope("time", System.currentTimeMillis())) lookupBus.publish(MsgEnvelope("greetings", "hello")) expectMsg("hello")
- Java
-
source
LookupBusImpl lookupBus = new LookupBusImpl(); lookupBus.subscribe(getTestActor(), "greetings"); lookupBus.publish(new MsgEnvelope("time", System.currentTimeMillis())); lookupBus.publish(new MsgEnvelope("greetings", "hello")); expectMsgEquals("hello");
This classifier is efficient in case no subscribers exist for a particular event.
Subchannel Classification
If classifiers form a hierarchy and it is desired that subscription be possible not only at the leaf nodes, this classification may be just the right one. It can be compared to tuning in on (possibly multiple) radio channels by genre. This classification has been developed for the case where the classifier is just the JVM class of the event and subscribers may be interested in subscribing to all subclasses of a certain class, but it may be used with any classifier hierarchy.
The necessary methods to be implemented are illustrated with the following example:
- Scala
-
source
import akka.util.Subclassification class StartsWithSubclassification extends Subclassification[String] { override def isEqual(x: String, y: String): Boolean = x == y override def isSubclass(x: String, y: String): Boolean = x.startsWith(y) } import akka.event.SubchannelClassification /** * Publishes the payload of the MsgEnvelope when the topic of the * MsgEnvelope starts with the String specified when subscribing. */ class SubchannelBusImpl extends EventBus with SubchannelClassification { type Event = MsgEnvelope type Classifier = String type Subscriber = ActorRef // Subclassification is an object providing `isEqual` and `isSubclass` // to be consumed by the other methods of this classifier override protected val subclassification: Subclassification[Classifier] = new StartsWithSubclassification // is used for extracting the classifier from the incoming events override protected def classify(event: Event): Classifier = event.topic // will be invoked for each event for all subscribers which registered // themselves for the event’s classifier override protected def publish(event: Event, subscriber: Subscriber): Unit = { subscriber ! event.payload } }
- Java
-
source
import akka.event.japi.SubchannelEventBus; static class StartsWithSubclassification implements Subclassification<String> { @Override public boolean isEqual(String x, String y) { return x.equals(y); } @Override public boolean isSubclass(String x, String y) { return x.startsWith(y); } } /** * Publishes the payload of the MsgEnvelope when the topic of the MsgEnvelope starts with the * String specified when subscribing. */ static class SubchannelBusImpl extends SubchannelEventBus<MsgEnvelope, ActorRef, String> { // Subclassification is an object providing `isEqual` and `isSubclass` // to be consumed by the other methods of this classifier @Override public Subclassification<String> subclassification() { return new StartsWithSubclassification(); } // is used for extracting the classifier from the incoming events @Override public String classify(MsgEnvelope event) { return event.topic; } // will be invoked for each event for all subscribers which registered themselves // for the event’s classifier @Override public void publish(MsgEnvelope event, ActorRef subscriber) { subscriber.tell(event.payload, ActorRef.noSender()); } }
A test for this implementation may look like this:
- Scala
-
source
val subchannelBus = new SubchannelBusImpl subchannelBus.subscribe(testActor, "abc") subchannelBus.publish(MsgEnvelope("xyzabc", "x")) subchannelBus.publish(MsgEnvelope("bcdef", "b")) subchannelBus.publish(MsgEnvelope("abc", "c")) expectMsg("c") subchannelBus.publish(MsgEnvelope("abcdef", "d")) expectMsg("d")
- Java
-
source
SubchannelBusImpl subchannelBus = new SubchannelBusImpl(); subchannelBus.subscribe(getTestActor(), "abc"); subchannelBus.publish(new MsgEnvelope("xyzabc", "x")); subchannelBus.publish(new MsgEnvelope("bcdef", "b")); subchannelBus.publish(new MsgEnvelope("abc", "c")); expectMsgEquals("c"); subchannelBus.publish(new MsgEnvelope("abcdef", "d")); expectMsgEquals("d");
This classifier is also efficient in case no subscribers are found for an event, but it uses conventional locking to synchronize an internal classifier cache, hence it is not well-suited to use cases in which subscriptions change with very high frequency (keep in mind that “opening” a classifier by sending the first message will also have to re-check all previous subscriptions).
Scanning Classification
The previous classifier was built for multi-classifier subscriptions which are strictly hierarchical, this classifier is useful if there are overlapping classifiers which cover various parts of the event space without forming a hierarchy. It can be compared to tuning in on (possibly multiple) radio stations by geographical reachability (for old-school radio-wave transmission).
The necessary methods to be implemented are illustrated with the following example:
- Scala
-
source
import akka.event.ScanningClassification /** * Publishes String messages with length less than or equal to the length * specified when subscribing. */ class ScanningBusImpl extends EventBus with ScanningClassification { type Event = String type Classifier = Int type Subscriber = ActorRef // is needed for determining matching classifiers and storing them in an // ordered collection override protected def compareClassifiers(a: Classifier, b: Classifier): Int = if (a < b) -1 else if (a == b) 0 else 1 // is needed for storing subscribers in an ordered collection override protected def compareSubscribers(a: Subscriber, b: Subscriber): Int = a.compareTo(b) // determines whether a given classifier shall match a given event; it is invoked // for each subscription for all received events, hence the name of the classifier override protected def matches(classifier: Classifier, event: Event): Boolean = event.length <= classifier // will be invoked for each event for all subscribers which registered themselves // for a classifier matching this event override protected def publish(event: Event, subscriber: Subscriber): Unit = { subscriber ! event } }
- Java
-
source
import akka.event.japi.ScanningEventBus; /** * Publishes String messages with length less than or equal to the length specified when * subscribing. */ static class ScanningBusImpl extends ScanningEventBus<String, ActorRef, Integer> { // is needed for determining matching classifiers and storing them in an // ordered collection @Override public int compareClassifiers(Integer a, Integer b) { return a.compareTo(b); } // is needed for storing subscribers in an ordered collection @Override public int compareSubscribers(ActorRef a, ActorRef b) { return a.compareTo(b); } // determines whether a given classifier shall match a given event; it is invoked // for each subscription for all received events, hence the name of the classifier @Override public boolean matches(Integer classifier, String event) { return event.length() <= classifier; } // will be invoked for each event for all subscribers which registered themselves // for the event’s classifier @Override public void publish(String event, ActorRef subscriber) { subscriber.tell(event, ActorRef.noSender()); } }
A test for this implementation may look like this:
- Scala
-
source
val scanningBus = new ScanningBusImpl scanningBus.subscribe(testActor, 3) scanningBus.publish("xyzabc") scanningBus.publish("ab") expectMsg("ab") scanningBus.publish("abc") expectMsg("abc")
- Java
-
source
ScanningBusImpl scanningBus = new ScanningBusImpl(); scanningBus.subscribe(getTestActor(), 3); scanningBus.publish("xyzabc"); scanningBus.publish("ab"); expectMsgEquals("ab"); scanningBus.publish("abc"); expectMsgEquals("abc");
This classifier takes always a time which is proportional to the number of subscriptions, independent of how many actually match.
Actor Classification
This classification was originally developed specifically for implementing DeathWatch: subscribers as well as classifiers are of type ActorRef
ActorRef
.
This classification requires an ActorSystem
ActorSystem
in order to perform book-keeping operations related to the subscribers being Actors, which can terminate without first unsubscribing from the EventBus. ManagedActorClassification maintains a system Actor which takes care of unsubscribing terminated actors automatically.
The necessary methods to be implemented are illustrated with the following example:
- Scala
-
source
import akka.event.ActorEventBus import akka.event.ManagedActorClassification import akka.event.ActorClassifier final case class Notification(ref: ActorRef, id: Int) class ActorBusImpl(val system: ActorSystem) extends ActorEventBus with ActorClassifier with ManagedActorClassification { type Event = Notification // is used for extracting the classifier from the incoming events override protected def classify(event: Event): ActorRef = event.ref // determines the initial size of the index data structure // used internally (i.e. the expected number of different classifiers) override protected def mapSize: Int = 128 }
- Java
-
source
import akka.event.japi.ManagedActorEventBus; static class Notification { public final ActorRef ref; public final int id; public Notification(ActorRef ref, int id) { this.ref = ref; this.id = id; } } static class ActorBusImpl extends ManagedActorEventBus<Notification> { // the ActorSystem will be used for book-keeping operations, such as subscribers terminating public ActorBusImpl(ActorSystem system) { super(system); } // is used for extracting the classifier from the incoming events @Override public ActorRef classify(Notification event) { return event.ref; } // determines the initial size of the index data structure // used internally (i.e. the expected number of different classifiers) @Override public int mapSize() { return 128; } }
A test for this implementation may look like this:
- Scala
-
source
val observer1 = TestProbe().ref val observer2 = TestProbe().ref val probe1 = TestProbe() val probe2 = TestProbe() val subscriber1 = probe1.ref val subscriber2 = probe2.ref val actorBus = new ActorBusImpl(system) actorBus.subscribe(subscriber1, observer1) actorBus.subscribe(subscriber2, observer1) actorBus.subscribe(subscriber2, observer2) actorBus.publish(Notification(observer1, 100)) probe1.expectMsg(Notification(observer1, 100)) probe2.expectMsg(Notification(observer1, 100)) actorBus.publish(Notification(observer2, 101)) probe2.expectMsg(Notification(observer2, 101)) probe1.expectNoMessage(500.millis)
- Java
-
source
ActorRef observer1 = new TestKit(system).getRef(); ActorRef observer2 = new TestKit(system).getRef(); TestKit probe1 = new TestKit(system); TestKit probe2 = new TestKit(system); ActorRef subscriber1 = probe1.getRef(); ActorRef subscriber2 = probe2.getRef(); ActorBusImpl actorBus = new ActorBusImpl(system); actorBus.subscribe(subscriber1, observer1); actorBus.subscribe(subscriber2, observer1); actorBus.subscribe(subscriber2, observer2); Notification n1 = new Notification(observer1, 100); actorBus.publish(n1); probe1.expectMsgEquals(n1); probe2.expectMsgEquals(n1); Notification n2 = new Notification(observer2, 101); actorBus.publish(n2); probe2.expectMsgEquals(n2); probe1.expectNoMessage(Duration.ofMillis(500));
This classifier is still is generic in the event type, and it is efficient for all use cases.