Azure-functions-durable-extension: Durable Entities - Buffer Messages until 'X'

[mpaul31]

When trying to implement an aggregator pattern using durable entities I need my entity to fire off any orchestration and i would like to buffer any incoming messages until the orchestration completes (orchestration could signal back to the entity). is there any future plans to support this type of scenario?

thanks

[sebastianburckhardt]

We don't currently have any way to suspend the processing of incoming messages from within an entity (this is inherently susceptible to deadlocks, which we try to rule out by design).

One thing you can do is to lock the entity from within the orchestration, then all incoming messages are buffered until the orchestration releases the lock.

However, this still leaves a window : some incoming messages could be processed after you have started the new orchestration and before that orchestration acquires the lock.

Of course, there is always the option of buffering the messages yourself inside the entity. It's not very elegant but it should work:

public class MyEntity
{
    public bool Waiting { get; set; }

    public List<Request> Buffer { get; set; } = new List<Request>();

    public void Process(Request r)
    {
        if (Waiting)
        {
            Buffer.Add(r);
        }
        else
        {
            ProcessInternal(r);

            if (r.StartOrchestration)
            {
                Entity.Current.StartNewOrchestration("Orchestrator", Entity.Current.EntityId);
                Waiting = true;
            }
        }
    }

    public void OrchestrationIsDone()
    {
        Waiting = false;
        foreach (var request in Buffer)
        {
            ProcessInternal(request);
        }
        Buffer.Clear();
    }

    private void ProcessInternal(Request r)
    {
        // process the request
    }

    // Boilerplate (entry point for the functions runtime)
    [FunctionName(nameof(MyEntity))]
    public static Task HandleEntityOperation([EntityTrigger] IDurableEntityContext context)
    {
        return context.DispatchAsync<MyEntity>();
    }
}