WorkflowRb

WorkflowRb is a light weight workflow engine for Ruby. It supports pluggable persistence and concurrency providers to allow for multi-node clusters.

Installation

Add this line to your application's Gemfile:

gem 'workflow_rb'

And then execute:

$ bundle

Or install it yourself as:

$ gem install workflow_rb

Basic Concepts

Steps

A workflow consists of a series of connected steps. Each step produces an outcome value and subsequent steps are triggered by subscribing to a particular outcome of a preceeding step. The default outcome of nil can be used for a basic linear workflow. Steps are usually defined by inheriting from the StepBody abstract class and implementing the run method. They can also be created inline while defining the workflow structure.

First we define some steps

class HelloWorld < WorkflowRb::StepBody
  def run(context)
    puts 'Hello world'
    WorkflowRb::ExecutionResult.NextStep
  end
end

Then we define the workflow structure by composing a chain of steps.

class HelloWorld_Workflow
  ID = 'hello world'
  VERSION = 1
  DATA_CLASS = nil

  def build(builder)
    builder
        .start_with(HelloWorld)
        .then(GoodbyeWorld)
  end
end

The ID and VERSION constants are used to identify the workflow definition.

You can also define your steps inline

class HelloWorld_Workflow
  ID = 'hello world'
  VERSION = 1
  DATA_CLASS = nil

  def build(builder)
    builder
        .start_step do |context|
          puts 'Hello world'
          WorkflowRb::ExecutionResult.NextStep
        end
        .then_step do |context|
          puts 'Goodbye world'
          WorkflowRb::ExecutionResult.NextStep
        end
  end
end

Each running workflow is persisted to the chosen persistence provider between each step, where it can be picked up at a later point in time to continue execution. The outcome result of your step can instruct the workflow host to defer further execution of the workflow until a future point in time or in response to an external event.

The first time a particular step within the workflow is called, the persistence_data property on the context object is nil. The ExecutionResult produced by the run method can either cause the workflow to proceed to the next step by providing an outcome value, instruct the workflow to sleep for a defined period or simply not move the workflow forward. If no outcome value is produced, then the step becomes re-entrant by setting persistence_data, so the workflow host will call this step again in the future buy will populate the persistence_data with it's previous value.

For example, this step will initially run with nil persistence_data and put the workflow to sleep for 12 hours, while setting the persistence_data to 'something'. 12 hours later, the step will be called again but context.persistence_data will now contain the object constructed in the previous iteration, and will now produce an outcome value of nil, causing the workflow to move forward.

public class SleepStep : StepBody
{
    public override ExecutionResult Run(IStepExecutionContext context)
    {
        if (context.PersistenceData == null)
            return ExecutionResult.Sleep(Timespan.FromHours(12), new Object());
        else
            return ExecutionResult.Next();
    }
}

Passing data between steps

Each step is intented to be a blackbox, therefore they support inputs and outputs. These inputs and outputs can be mapped to a data class that defines the custom data relevant to each workflow instance.

The following sample shows how to define inputs and outputs on a step, it then shows how define a workflow with a typed class for internal data and how to map the inputs and outputs to properties on the custom data class.

//Our workflow step with inputs and outputs
public class AddNumbers : StepBody
{
    public int Input1 { get; set; }

    public int Input2 { get; set; }

    public int Output { get; set; }

    public override ExecutionResult Run(IStepExecutionContext context)
    {
        Output = (Input1 + Input2);
        return ExecutionResult.Next();
    }
}

//Our class to define the internal data of our workflow
public class MyDataClass
{
    public int Value1 { get; set; }
    public int Value2 { get; set; }
    public int Value3 { get; set; }
}

//Our workflow definition with strongly typed internal data and mapped inputs & outputs
public class PassingDataWorkflow : IWorkflow<MyDataClass>
{  
    public void Build(IWorkflowBuilder<MyDataClass> builder)
    {
        builder            
            .StartWith<AddNumbers>()
                .Input(step => step.Input1, data => data.Value1)
                .Input(step => step.Input2, data => data.Value2)
                .Output(data => data.Value3, step => step.Output)
            .Then<CustomMessage>()
                .Input(step => step.Message, data => "The answer is " + data.Value3.ToString());
    }
    ...
}

Multiple outcomes / forking

A workflow can take a different path depending on the outcomes of preceeding steps. The following example shows a process where first a random number of 0 or 1 is generated and is the outcome of the first step. Then, depending on the outcome value, the workflow will either fork to (TaskA + TaskB) or (TaskC + TaskD)

public class MultipleOutcomeWorkflow : IWorkflow
{
    public void Build(IWorkflowBuilder<object> builder)
    {
        builder
            .StartWith<RandomOutput>(x => x.Name("Random Step"))
                .When(0)
                    .Then<TaskA>()
                    .Then<TaskB>()                        
                    .End<RandomOutput>("Random Step")
                .When(1)
                    .Then<TaskC>()
                    .Then<TaskD>()
                    .End<RandomOutput>("Random Step");
    }
}

Events

A workflow can also wait for an external event before proceeding. In the following example, the workflow will wait for an event called "MyEvent" with a key of 0. Once an external source has fired this event, the workflow will wake up and continute processing, passing the data generated by the event onto the next step.

public class EventSampleWorkflow : IWorkflow<MyDataClass>
{
    public void Build(IWorkflowBuilder<MyDataClass> builder)
    {
        builder
            .StartWith(context =>
            {
                Console.WriteLine("workflow started");
                return ExecutionResult.Next();
            })
            .WaitFor("MyEvent", "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 workflows that have subscribed to MyEvent 0, will be passed "hello"
host.PublishEvent("MyEvent", "0", "hello");

Host

The workflow host is the service responsible for executing workflows. It does this by polling the persistence provider for workflow instances that are ready to run, executes them and then passes them back to the persistence provider to by stored for the next time they are run. It is also responsible for publishing events to any workflows that may be waiting on one.

Setup

Use the AddWorkflow extension method for IServiceCollection to configure the workflow host upon startup of your application. By default, it is configured with MemoryPersistenceProvider and SingleNodeConcurrencyProvider for testing purposes. You can also configure a DB persistence provider at this point.

services.AddWorkflow();

Usage

When your application starts, grab the workflow host from the built-in dependency injection framework IServiceProvider. Make sure you call RegisterWorkflow, so that the workflow host knows about all your workflows, and then call Start() to fire up the thread pool that executes workflows. Use the StartWorkflow method to initiate a new instance of a particular workflow.

var host = serviceProvider.GetService<IWorkflowHost>();            
host.RegisterWorkflow<HelloWorldWorkflow>();
host.Start();

host.StartWorkflow("HelloWorld", 1, null);

Console.ReadLine();
host.Stop();

Persistence

Since workflows are typically long running processes, they will need to be persisted to storage between steps. There are several persistence providers available as seperate Nuget packages.

• MemoryPersistenceProvider (Default provider, for demo and testing purposes)
• MongoDB
• SQL Server
• PostgreSQL
• Sqlite
• Redis (coming soon...)

Multi-node clusters

By default, the WorkflowHost service will run as a single node using the built-in queue and locking providers for a single node configuration. Should you wish to run a multi-node cluster, you will need to configure an external queueing mechanism and a distributed lock manager to co-ordinate the cluster. These are the providers that are currently available.

Queue Providers

• SingleNodeQueueProvider (Default built-in provider)
• RabbitMQ
• Apache ZooKeeper (coming soon...)
• 0MQ (coming soon...)

Distributed lock managers

• SingleNodeLockProvider (Default built-in provider)
• Redis Redlock
• Apache ZooKeeper (coming soon...)

Samples

Hello World

Multiple outcomes

Passing Data

Events

Deferred execution & re-entrant steps

Looping

Authors

• Daniel Gerlag - Initial work

License

This project is licensed under the MIT License - see the LICENSE.md file for details