Durable Entities natively support Pub/Sub

[mpaul31]

Now with the new Netherite backend in the works, it would be awesome if durable entities supported pub/sub between entities (maybe orchestrations as well?).

I had a use case for this on one of my current projects and it was trivial to add but the number of subscribers per entity was low with a very naïve implementation. Would be awesome to see this baked in and too throw in a twist using Reactive Extensions. :)

@sebastianburckhardt thoughts?

[sebastianburckhardt]

I like the idea of using entities for reactive programming a lot, definitely. That always was one of the goals.
Exactly what we should "bake in" is not so clear though.

To get a better idea I would probably want to collect some concrete examples. @mpaul31, would it be possible for you to share details about your use case? Code would be great, or otherwise just a description. Same applies to anyone else reading this post - if you have a use case let us know.

One of the interesting questions is how we should handle subscription management. Sometimes, subscriptions are static (e.g. can be computed purely from on entity ID), but sometimes they need to be managed dynamically using explicit subscribe and unsubscribe operations.

[mpaul31]

We use durable entities to implement the aggregator pattern. Each event that comes in from an external system has what we call an entity store that stores the data (current state) and list of the subscribers (entities).

Here is an example of one of our stores:

public partial class VehicleStore : IVehicleStore
{
    public const string FunctionName = nameof(VehicleStore);

    [FunctionName(FunctionName)]
    public static Task Run([EntityTrigger] IDurableEntityContext context, ILogger log) => context.DispatchAsync<VehicleStore>(log);

    [JsonIgnore]
    private readonly ILogger log;

    public VehicleStore(ILogger log)
    {
        this.log = log;
    }

    public void Update(IVehicleChangeEvent eventData)
    {
        using (BeginReceiveScope(eventData?.MessageId))
        {
            if (eventData.CompareTo(State.EventData) < 1)
            {
                LogOnReceivedDuplicate();

                return;
            }

            if (VehicleChangeEventValueComparer.Instance.Equals(State.EventData, eventData))
            {
                LogOnReceivedSameEventData();

                return;
            }

            LogOnUpdate();

            State.EventData = eventData;

            NotifySubscribers();
        }
    }

    public void Subscribe(EntityId entityId)
    {
        using (BeginReceiveScope(entityId))
        {
            if (State.Subscriptions.Contains(entityId))
            {
                return;
            }

            LogOnSubscribe();

            State.Subscriptions.Add(entityId);

            if (State.EventData is null)
            {
                return;
            }

            Notify(entityId);
        }
    }

    public void Unsubscribe(EntityId entityId)
    {
        using (BeginReceiveScope(entityId))
        {
            bool result = State.Subscriptions.Remove(entityId);

            LogOnUnsubscribe(result);
        }
    }

    private void NotifySubscribers()
    {
        State.Subscriptions.ForEach(Notify);
    }

    private void Notify(EntityId entityId)
    {
        LogNotifySubscriberOnChange(entityId);

        Entity.Current.SignalEntity<IVehicleStateChangeHandler>(entityId, o => o.OnVehicleChanged(State.EventData));
    }

    public VehicleState State { get; set; } = new VehicleState();
}

Then any entities interested in receiving notifications from the entity store send a 'Subscribe' signal and implement the entity stores state change handler.

Entity.Current.SignalEntity<IVehicleStore>(vehicleEntityKey, e => e.Subscribe(Entity.Current.EntityId));

public interface IVehicleStateChangeHandler
{
    Task OnVehicleChanged(IVehicleChangeEvent eventData);
}

The code is meant to be simple and we didn't want to introduce a bunch of abstractions so there is some boilerplate each new store needs to adhere too, but it has worked out thus far very well.

Hope this helps!