Event Streams

​

Purpose

​

API

• ActorSystem.Subscribe: Create a subscriber. Subscribes to the internal events topic. Returns eventstream.Subscriber.
• ActorSystem.Unsubscribe: Release the subscriber. Call to avoid leaks.

• ID: Unique subscriber identifier.
• Topics: Topics this subscriber is subscribed to (includes the internal topic).
• Iterator: Channel of buffered messages; drains and closes. Messages are *eventstream.Message.
• Shutdown: Mark subscriber inactive.

​

Message structure

• Topic: msg.Topic() returns the topic (e.g. topic.events, the internal topic).
• Payload: msg.Payload() returns the event. Use a type assertion to handle specific events.

​

Event types

• ActorStarted: Actor has started.
• ActorStopped: Actor has stopped.
• ActorChildCreated: A child actor was spawned.
• ActorPassivated: Actor was passivated (idle timeout).
• ActorRestarted: Actor was restarted by supervisor.
• ActorSuspended: Actor was suspended after a failure.
• ActorReinstated: Actor was reinstated by supervisor.
• Deadletter: Message was not handled by its receiver (ctx.Unhandled) or could not be delivered (for example a PipeTo task that failed), or a grain failed to handle a one-way message. For a grain the receiver is named by the grain identity, kind/name. A send to a stopped actor returns ErrDead to the sender and produces no dead letter.
• NodeJoined: A node joined the cluster (cluster mode only).
• NodeLeft: A node left the cluster (cluster mode only).
• LeaderChanged: The cluster coordinator moved to a different node (cluster mode only).
• RelocationFailed: A departed node’s actors or grains could not be relocated (cluster mode only).
• RelocationStarted: The leader started relocating a departed node’s actors and grains; BestEffort is true when the set was reconstructed after a crash (cluster mode only).

​

Usage

subscriber, err := sys.Subscribe()
if err != nil {
    return err
}
defer sys.Unsubscribe(subscriber)

// Iterator() returns the events buffered so far; call it again for later ones
for msg := range subscriber.Iterator() {
    switch e := msg.Payload().(type) {
    case *actor.ActorStarted:
        log.Printf("actor started: %s", e.Address())
    case *actor.ActorStopped:
        log.Printf("actor stopped: %s", e.Address())
    case *actor.Deadletter:
        log.Printf("dead letter: sender=%s receiver=%s reason=%s", e.Sender(), e.Receiver(), e.Reason())
    case *actor.NodeJoined:
        log.Printf("node joined: %s", e.Address())
    case *actor.NodeLeft:
        log.Printf("node left: %s", e.Address())
    case *actor.LeaderChanged:
        log.Printf("cluster leader is now %s", e.Address())
    case *actor.RelocationStarted:
        log.Printf("relocation started: node=%s bestEffort=%v actors=%v grains=%v", e.Address(), e.BestEffort(), e.Actors(), e.Grains())
    case *actor.RelocationFailed:
        log.Printf("relocation failed: node=%s actors=%v grains=%v err=%v", e.Address(), e.Actors(), e.Grains(), e.Error())
    }
}

​

Implementation notes

• The event stream is backed by the eventstream package. The actor system creates an EventsStream when it is constructed (NewActorSystem), keeps it across Stop/Start cycles, and subscribes callers to the internal topic topic.events.
• Lifecycle events (ActorStarted, ActorStopped, etc.) are published by the actor’s PID as it starts, stops, restarts, passivates, is suspended or reinstated; the death watch publishes nothing to the stream. Cluster events (NodeJoined, NodeLeft) are published by the cluster events loop once the cluster’s routing table has converged on the membership change, that is once the coordinator has redistributed partitions for the new member set, so subscribers and relocation act on a stable table. If no convergence is announced within the convergence timeout, 10 seconds by default (see WithConvergenceTimeout), for example because a member fails to acknowledge the rebalance, the event is published anyway. LeaderChanged is published from the same loop, on every node, whenever a topology event reveals that the coordinator moved to a different node; the initial coordinator election on startup is not published, only subsequent transitions.
• RelocationFailed is published by the relocator when a departed node’s actors or grains cannot be re-deployed. Each item is retried a bounded number of times with backoff before it is reported, and the report is terminal: the relocation run does not revisit failed items, so subscribe to this event to drive your own recovery (for example, re-spawning the affected actors or alerting).
• RelocationStarted is published by the leader whenever it starts relocating a departed node’s actors and grains, for graceful shutdowns and crashes alike. BestEffort() distinguishes the two: false means the set comes from the departed node’s graceful-shutdown snapshot and is complete; true means the node crashed and the set was reconstructed from the replicated cluster registry, so records lost with the crashed node’s partitions cannot be listed, and an empty or unexpectedly small set signals suspected record loss. See the relocation page for the replication settings that govern completeness.
• Dead letters are published by the dead-letter actor when messages cannot be delivered.
• The event stream is always present; no option is required. It is independent of WithPubSub() and cluster mode.

​

Related

• Observability: Metrics and dead letters
• PubSub: Application-level topic-based pub/sub (separate mechanism)