First steps for Implementing a Saga Execution Coordinator using TheFlow

In a previous post, I wrote about how to support sagas using a Workflow Engine as Saga Execution Coordinator (if you want to have a better understanding about the Saga pattern, I recommend you to read that post first). In that moment, I shared that I am working on a lightweight .NET workflow engine called TheFlow.

TheFlow is open source, and it is available as a NuGet package.

In this post, I will explain some basic concepts that you need to understand to get TheFlow up and running in your projects (including as a Saga Execution Coordinator).

The ProcessManager

Understanding the ProcessManager is the first step for using TheFlow. It is the component responsible for taking care of creating new process instances, persisting it and loading whenever it is necessary.

You could create multiple ProcessManager instances. Anyway, in practical scenarios, it makes no sense. Because of that, I recommend following the same pattern adopted by the RavenDB team to manage a single instance of IDocumentStore.

public static class ProcessManagerHolder
{
    private static readonly Lazy<ProcessManager> LazyProcessManager = 
        new Lazy<ProcessManager>(() =>
        {
            var models = new InMemoryProcessModelsStore();
            var instances = new InMemoryProcessInstancesStore();

            return new ProcessManager(models, instances);
        });

    public static ProcessManager Instance =>
        LazyProcessManager.Value;
}

The ProcessManager uses two independent stores. The first one, for the models (description of the process components, including activities, events and other elements). The second one, for the instances (the data related with an executing process).

For long-running processes, it would be recommendable to use a persistent store for instances. Currently, there is a instances store using RavenDB as an underlying persistence mechanism under development.

The ProcessModel

After creating the ProcessManager instance, we are ready to define the models for processes that we want to support.

var model = ProcessModel.Create(Guid.Parse("a12637a3-72de-4774-b60c-d98310438c26"))
    .AddEventCatcherFor<StartEvent>()
    .AddActivity<Activity1>()
    .AddActivity<CompensatingActivity1>()
    .AttachAsCompensationActivity("CompensatingActivity1", "Activity1")
    .AddActivity<Activity2>()
    .AddActivity<CompensatingActivity2>()
    .AttachAsCompensationActivity("CompensatingActivity2", "Activity2")
    .AddActivity<Activity3>()
    .AddEventThrower<EndEventThrower>()
    .AddSequenceFlow("OnStartEvent", 
        "Activity1",
        "Activity2",
        "Activity3", 
        "End");

ProcessManagerHolder.Instance.ModelsStore.Store(model);

In the example, we are defining the process for a Saga (a sequence of activities, each activity with a corresponding compensating one [that will be executed if the process fails]).

The ProcessModel class exposes a fluent interface easy to understand and use. Currently, TheFlow has support for activities, event catchers, gateways and more.

The Starting Event

The ProcessManager will start a new process (in this example, a new Saga) instance whenever handling an event that matches with the one specified in the model.

ProcessManagerHolder.Instance.HandleEvent(new StartEvent());

The event is a plain clr object. You could retrieve it from a Message Broker (RabbitMQ, for example) and send it to the ProcessManager.

Activities

After starting a new process instance, the process manager will begin to execute all the activities in the defined sequence.

Activities are implemented by inheriting the Activity class.

public class Activity1 : Activity
{
    public override void Run(ExecutionContext context)
    {
        Console.WriteLine("Running activity 1. Is it working?");
        var response = Console.ReadKey();

        if (!(response.KeyChar == 'Y' || response.KeyChar == 'y'))
        {
            ProcessManagerHolder.Instance.HandleActivityCompletion(
                context.Instance.Id,
                context.Token.Id,
                null
            );
        }
        else
        {
            ProcessManagerHolder.Instance.HandleActivityFailure(
                context.Instance.Id,
                context.Token.Id,
                null
            );
        }
    }
}

Each activity can report success or failure to the ProcessManager. If a failure is reported, the ProcessManager will start to run the compensating flow.

The Log

Each process instance contains a detailed log describing all his events (start and end times for all the executed activities, for example).

The ProcessManager allows you to search for instances using the log information through the instances store.

Call to action

As you can see, TheFlow makes easy to define, execute, and control customized processes (including Sagas). It’s under active development and being improved every day. In this post, I shared just the basic concepts.

I invite you to use TheFlow, to improve the code, or register new issues. I would be grateful for any feedback in the comments.