EventStream in Proto.Actor Explained: Publish-Subscribe Communication Between Actors

#actormodel #protoactor #csharp #eventdriven

Introduction

Proto.Actor provides a powerful feature called EventStream, which acts as a multi-channel publisher/subscriber dispatcher. This article explores how to use this mechanism to implement event-driven communication between actors.

Event Stream

The EventStream enables a publish/subscribe pattern in Proto.Actor, allowing actors (identified by their Process ID, or PID) to subscribe to specific event types. It also serves as a critical component for managing infrastructure events like undelivered messages (e.g., DeadLetter handling).

Requirements

.NET 8+
NuGet packages:
- Proto.Actor

Actors

For this example, we implement two actors to demonstrate event-driven interaction.

OrderActor

The OrderActor creates orders and publishes an OrderCreated event via the EventStream:

public class OrderActor : IActor
{
    private readonly List<Order> _orders = [];
    public Task ReceiveAsync(IContext context)
    {
        if (context.Message is CreateOrder createOrder)
        {
            var order = new Order(createOrder.Id, createOrder.Amount);
            _orders.Add(order);
            context.Respond(order);
            // Publish event to the EventStream
            context.System.EventStream.Publish(new OrderCreated(order)); // 
        }
        else if (context.Message is AllOrder)
        {
            context.Respond(_orders.ToList());
        }
        return Task.CompletedTask;
    }
}

PaymentActor

The PaymentActor subscribes to OrderCreated events and maintains a list of orders awaiting payment:

public class PaymentActor : IActor
{
    private List<Order> _waitingForPayment = [];
    public Task ReceiveAsync(IContext context)
    {
        if (context.Message is OrderCreated orderCreated)
        {
            _waitingForPayment.Add(orderCreated.Order); // React to event
        }
        else if (context.Message is OrderPaid orderPaid)
        {
            _waitingForPayment.RemoveAll(x => x.Id == orderPaid.Id);
        }
        else if (context.Message is IsWaitingForPayment isWaitingForPayment)
        {
            context.Respond(_waitingForPayment.Any(x => x.Id == isWaitingForPayment.Id));
        }
        return Task.CompletedTask;
    }
}

Subscription to an Event

To subscribe an actor to an event:

Spawn the actor using the actor system.
Use EventStream.Subscribe<T>() to register the actor for a specific event type:

// Creating actors
var actorOrder = system.Root.Spawn(Props.FromProducer(() => new OrderActor()));
var actorPayment = system.Root.Spawn(Props.FromProducer(() => new PaymentActor()));

// Subscribing to OrderCreated events
system.EventStream.Subscribe<OrderCreated>(system.Root, actorPayment); //

Key Concepts

EventStream Lifecycle: Events are published to all subscribers, and subscriptions persist until explicitly unsubscribed.
DeadLetter Handling: The EventStream also queues undelivered messages (e.g., when sending to a terminated actor).

Conclusion

Proto.Actor's EventStream simplifies event-driven architectures by decoupling actors through a robust pub/sub mechanism. This pattern is ideal for scenarios like event sourcing, monitoring, or cross-cutting concerns (e.g., logging).