ProcessingPlatform.Persistence.Sqlite 10.0.0.2

Processing Platform

Version: 10.0.0.1 Framework: .NET 10.0 License: LGPL-3.0-or-later

A high-performance flow processing engine for .NET. Define complex flows with branching, looping, parallelism, events, sagas, sub-flows, scheduling, and more — all with a fluent builder API and pluggable persistence/queue/lock providers.

Prerequisites

• .NET 10.0 SDK or later

PRELIMINARY

Activities

public void Build(IFlowBuilder<MyData> builder)
    {
        builder
                .StartWith(context =>
                {
                    Console.WriteLine("Starting flow...");
                    return ExecutionResult.Next();
                })
                .Then<GreetYourself>()
                .Activity("ExampleActivity", (data) => data.Value1)
                .Output(data => data.Value2, step => step.Result)
                .Then<PrintMessage>()
                .Input(step => step.Message, data => data.Value2.ToString())
                .Then(context =>
                {
                    Console.WriteLine("Flow complete");
                    return ExecutionResult.Next();
                });
    }
             
}
...

var activity = host.GetPendingActivity("ExampleActivity", "worker1", TimeSpan.FromMinutes(1)).Result;

if (activity != null)
{
    Console.WriteLine(activity.Parameters);
    host.SubmitActivitySuccess(activity.Token, "Response");
}

Injecting dependencies

• Interface

public interface IExampleService
{
    void PerformSomeActions();
}
...
public class ExampleService : IExampleService
{
    public void PerformSomeActions()
    {
        Console.WriteLine("Performance in progress...");
    }
}

• Step

public class FlowInExample : StepBody
{
    private IExampleService _ExampleService;

    public FlowInExample(IExampleService ExampleService)
    {
        _ExampleService = ExampleService;
    }

    public override ExecutionResult Run(IStepExecutionContext context)
    {
        _ExampleService.PerformSomeActions();
        return ExecutionResult.Next();
    }
}

• Registration

IServiceCollection services = new ServiceCollection();
services.AddLogging();
services.AddFlow();
            
services.AddTransient<FlowInExample>();
services.AddTransient<IExampleService, ExampleService>();

CONTROL

Flow (Then...Then)

builder.CreateConstruct()
    .StartWith<PrintMessage>()
        .Input(step => step.Message, data => "Passing first argument")
    .Then<PrintMessage>()
        .Input(step => step.Message, data => "Passing second argument");

Decide

var Construct1 = builder.CreateConstruct()
                .StartWith<PrintMessage>()
                .Input(step => step.Message, data => "Passing argument number 1")
                .Then<PrintMessage>()
                .Input(step => step.Message, data => "Passing argument number 1");

            var Construct2 = builder.CreateConstruct()
                .StartWith<PrintMessage>()
                .Input(step => step.Message, data => "Passing argument number 2")
                .Then<PrintMessage>()
                .Input(step => step.Message, data => "Passing argument number 2");

builder
    .StartWith<GreetYourself>()
    .Decide(data => data.Value1)
        .Construct((data, outcome) => data.Value1 == 1, Construct1)
        .Construct((data, outcome) => data.Value1 == 2, Construct2);

ForEach

builder
            .StartWith<GreetYourself>()
            .ForEach(data => new List<int>() { 1, 2, 3, 4 })
                .Do(x => x
                    .StartWith<PrintMessage>()
                        .Input(step => step.Message, (data, context) => context.Item)
                    .Then<FlowInExample>())
            .Then<ByeByeVegas>();

While

builder
            .StartWith<GreetYourself>()
            .While(data => data.Counter < 3)
                .Do(x => x
                    .StartWith<FlowInExample>()
                    .Then<IncrementStep>()
                        .Input(step => step.Value1, data => data.Counter)
                        .Output(data => data.Counter, step => step.Value2))
            .Then<ByeByeVegas>();

Input

builder
            .StartWith<GreetYourself>()
            .If(data => data.Counter < 3).Do(then => then
                .StartWith<PrintMessage>()
                    .Input(step => step.Message, data => "Value is less than 3")
            )
            .If(data => data.Counter < 5).Do(then => then
                .StartWith<PrintMessage>()
                    .Input(step => step.Message, data => "Value is less than 5")
            )
            .Then<ByeByeVegas>();

Parallel

builder
                    .StartWith<GreetYourself>()
                    .Parallel()
                    .Do(then => 
                        then.StartWith<PrintMessage>()
                            .Input(step => step.Message, data => "Item 1.1")
                            .Then<PrintMessage>()
                            .Input(step => step.Message, data => "Item 1.2"))
                    .Do(then =>
                        then.StartWith<PrintMessage>()
                            .Input(step => step.Message, data => "Item 2.1")
                            .Then<PrintMessage>()
                            .Input(step => step.Message, data => "Item 2.2")
                            .Then<PrintMessage>()
                            .Input(step => step.Message, data => "Item 2.3"))
                    .Do(then =>
                        then.StartWith<PrintMessage>()
                            .Input(step => step.Message, data => "Item 3.1")
                            .Then<PrintMessage>()
                            .Input(step => step.Message, data => "Item 3.2"))
                    .Join()
                    .Then<ByeByeVegas>();

SCHEDULING

builder
    .StartWith(context => Console.WriteLine("Hello"))
    .Schedule(data => TimeSpan.FromSeconds(5)).Do(schedule => schedule
        .StartWith(context => Console.WriteLine("Doing scheduled tasks"))
    )
    .Then(context => Console.WriteLine("Doing normal tasks"));

SUB-FLOWS

Sub-flows allow you to start a child flow from within a parent flow and wait for it to complete before continuing.

builder
    .StartWith(context => Console.WriteLine("Parent starts"))
    .SubFlow("ChildFlowDefinition")
    .Then(context => Console.WriteLine("Child completed, parent continues"));

You can also pass data to the child flow and specify a version:

builder
    .StartWith<PrepareData>()
    .SubFlow("ChildFlow", version: 2, data: d => d.ChildInput)
    .Then<ProcessResult>();

Recurrency - Resurge

builder
    .StartWith(context => Console.WriteLine("Hello"))
    .Resurge(data => TimeSpan.FromSeconds(5), data => data.Counter > 5).Do(Resurge => Resurge
        .StartWith(context => Console.WriteLine("Doing Resurging task"))
    )
    .Then(context => Console.WriteLine("Carry on"));

ERROR Handling

builder                
        .StartWith<GreetYourself>()
            .OnError(FlowErrorHandling.Retry, TimeSpan.FromMinutes(10))
        .Then<ByeByeVegas>();

EVENTS (External - out of flows)

public class EventSampleFlow : IFlow<ExampleData>
{
    public void Build(IFlowBuilder<ExampleData> builder)
    {
        builder
            .StartWith(context => ExecutionResult.Next())
            .WaitFor("ExampleEvent", data => "0")
                .Output(data => data.Value, step => step.EventData)
            .Then<CustomMessage>()
                .Input(step => step.Message, data => "The data from the event is " + data.Value);
    }
}
...
//External events are published via the host
//All Flows that have subscribed to MyEvent 0, will be passed "hello"
host.PublishEvent("ExampleEvent", "0", "hello");

SAGAS

Simple Saga

builder
    .StartWith(context => Console.WriteLine("Begin"))
    .Saga(saga => saga
        .StartWith<Task1>()
            .CompensateWith<UndoTask1>()
        .Then<Task2>()
            .CompensateWith<UndoTask2>()
        .Then<Task3>()
            .CompensateWith<UndoTask3>()
    )
        .CompensateWith<CleanUp>()
    .Then(context => Console.WriteLine("End"));

Retrials in Sagas

builder
    .StartWith(context => Console.WriteLine("Begin"))
    .Saga(saga => saga
        .StartWith<Task1>()
            .CompensateWith<UndoTask1>()
        .Then<Task2>()
            .CompensateWith<UndoTask2>()
        .Then<Task3>()
            .CompensateWith<UndoTask3>()
    )
    .OnError(Models.FlowErrorHandling.Retry, TimeSpan.FromSeconds(5))
    .Then(context => Console.WriteLine("End"));

Compensate in Sagas

builder
    .StartWith(context => Console.WriteLine("Begin"))
        .Saga(saga => saga
            .StartWith<Task1>()
            .Then<Task2>()
            .Then<Task3>()
    )
        .CompensateWith<UndoEverything>()
    .Then(context => Console.WriteLine("End"));

Passing Sagas to Steps

builder
    .StartWith<GreetYourself>()
        .CompensateWith<PrintMessage>(compensate => 
        {
            compensate.Input(step => step.Message, data => "undoing...");
        })

CONSTRUCTS (Division/Branch Isolation)

var Construct1 = builder.CreateConstruct()
                    .StartWith<PrintMessage>()
                    .Input(step => step.Message, data => "Passing first argument #1")
                    .Then<PrintMessage>()
                    .Input(step => step.Message, data => "Passing second argument #1");

                var Construct2 = builder.CreateConstruct()
                    .StartWith<PrintMessage>()
                    .Input(step => step.Message, data => "Passing first argument #2")
                    .Then<PrintMessage>()
                    .Input(step => step.Message, data => "Passing second argument #2");


                builder
                    .StartWith<GreetYourself>()
                    .Decide(data => data.Value1)
                    .Construct(1, Construct1)
                    .Construct(2, Construct2);