Workflow Engine tutorial

  • Updated on Sep 22, 2023
  • Published on Feb 10, 2022
  • 7 minute(s) read
Old material

The content below is deprecated.
We have rewritten this article and moved it to Coding a workflow

Introduction

This tutorial goes through the steps in creating a new workflow. The tutorial was based on version 5.2.0 of the Nexus.Link.Workflow.Sdk nuget and the code excerpts is from a proof-of-concept project, asynchronous-processes.

Setup

Install the NuGet package Nexus.Link.WorkflowEngine.Sdk for the projects with workflow logic and the NuGet package Nexus.Link.WorkflowEngine.Sdk.AspNet for the API project.

This tutorial was writting based on version 5.2.0 of the SDK.

Startup.cs

The WorkflowEngine relies on an AsyncRequestMgmtCapability as defined in the Nexus.Link.Capabilities.AsyncRequestMgmt.Abstract nuget package. For local development we advise you to use the AsyncRequestMgmtMock which you will find in the Nexus.Link.Capabilities.AsyncRequestMgmt.Mock NuGet package.

var httpSender = new HttpSender("");
var asyncRequestMgmtCapability = new AsyncRequestMgmtMock(httpSender);
services.TryAddSingleton<IAsyncRequestMgmtCapability>(asyncRequestMgmtCapability);

Add a handler for asynchronous REST calls

If you have outgoing REST calls, these should be made into asynchronous calls. This is automatically handled if you add the handler CallAsyncManagerForAsynchronousRequests to HttpClient for outgoing calls. It should be the last handler, just before the actual HTTP send method is called. If you want to use all the standard Nexus Link handlers, then you can use the following convenience method:

var handlers = Nexus.Link.WorkflowEngine.Sdk.Outbound.OutboundPipeFactory.CreateDelegatingHandlers(asyncRequestMgmtCapability);
var httpClientWithNexusLinkSupport = HttpClientFactory.Create(handlers);

Add persistence for the workflow configuration and runtime data

The workflow engine persists all configuration and runtime data. There are two prepared persistence implementations; one with memory persistence and one with persistence in SQL Server.

Example for memory persistence:

services.TryAddSingleton<IConfigurationTables>(provider => new ConfigurationTablesMemory());
services.TryAddSingleton<IRuntimeTables>(provider => new RuntimeTablesMemory());

Example for SQL Server persistence that will create the database if it doesn't exist

var masterConnectionString = "Server=localhost;Database=master;Trusted_Connection=True;";
var connectionString = "Server=localhost;Database=workflow-poc;Trusted_Connection=True;";
new DatabasePatcherHandler(connectionString, masterConnectionString)
    .PatchOrThrowAsync().Wait();
var databaseOptions = new DatabaseOptions
{
    ConnectionString =  "Server=localhost;Database=workflow-poc;Trusted_Connection=True;",
    VerboseLogging = true,
    DefaultLockTimeSpan = TimeSpan.FromSeconds(30)
};
services.TryAddSingleton<IConfigurationTables>(provider => new ConfigurationTablesSql(databaseOptions));
services.TryAddSingleton<IRuntimeTables>(provider => new RuntimeTablesSql(databaseOptions));

Add the Workflow Engine capabilities

The workflow engine needs some capabilities for persistence and for re-entry of workflows.

services.TryAddSingleton<IWorkflowEngineRequiredCapabilities, WorkflowCapabilities>();

Add Nexus Link middleware

You need to enable the Nexus Link middleware for SaveExecutionId andwWhen a client calls an API method that is implemented by a workflow, we need to reply with a 202 Accepted and refer to the AsyncManager for result polling or registering a callback. This is done by the RedirectAsynchronousRequests middleware.

using Nexus.Link.Misc.AspNet.Sdk.Inbound;
using Nexus.Link.Misc.AspNet.Sdk.Inbound.Options;

var linkOptions = new NexusLinkMiddlewareOptions
{
    Features = new MiddlewareFeatures
    {
        SaveExecutionId = {Enabled = true},
        SaveReentryAuthentication = { Enabled = true},
        RedirectAsynchronousRequests =
        {
            Enabled = true,
            RequestService = asyncRequestMgmtCapability!.Request
        }
    }
};
app.UseNexusLinkMiddleware(linkOptions);

Add ReentryAuthentication

As a workflow can be called many times over a minutes, days and even weeks and months a normal authentication token will eventually expire. To handle this, the AsyncManager and WorkflowEngine has support for something we call reentry authentication. When WorkflowEngine redirects a request to AsyncManager, the WE sends a long a secret token to AM in a header. As long as AM sends along that token when calling BE, BE will accept the reentry request, even if the authentication token has expired. The authentication token must still be valid from other perspectives, though.

Add the following to the NexusLinkMiddlewareOptions part above:

    Features = new MiddlewareFeatures
    {
        ...
        SaveReentryAuthentication = { Enabled = true},
        ...
    }

You will need to dependency inject two more services:

services.TryAddSingleton<Nexus.Link.Contracts.Misc.Sdk.Authentication.IHashService,
    Nexus.Link.Authentication.Sdk.Logic.HashService>();
services.TryAddSingleton<Nexus.Link.Contracts.Misc.AspNet.Sdk.IReentryAuthenticationService,
    Nexus.Link.Authentication.AspNet.Sdk.Logic.ReentryAuthenticationService>();

Nexus Link provides a token validation handler: Nexus.Link.Authentication.AspNet.Sdk.Handlers.TokenValidationHandler. Its constructor has an overload that accepts an IReentryAuthenticationService. If you use that handler and an incoming request contains a valid reentry authentication, then the request is accepted even if the authentication token has expired.

If you have written your own token validation handler, you must modify it to accepted ReentryAuthentication. Feel free to see how we solved it in our handler: GitHub. Look for the ShouldWeIgnoreExpirationAsync() method and how it is used in the handler.

Code the workflow

Create a workflow container

A workflow container contains the information that is common to all implementations, such as the name of the workflow. It also has a list of all the implementations. It always inherits from WorkflowContainer.

What you need to do is to define a number of constants for the activities that you are going to use in your first workflow implementation.

This is boilerplate for a workflow container:

public class MyWorkflowContainer : WorkflowContainer
{
    /// <inheritdoc />
    public MyWorkflowContainer(IWorkflowEngineRequiredCapabilities workflowCapabilities)
        : base("CapabilityName", "WorkflowTitle", "056AAB43-934D-4E8B-94A0-474B1F7F3811", workflowCapabilities)
    {
        DefineActivity(Activities.ActionA,nameof(Activities.ActionA), ActivityTypeEnum.Action);
    }

    /// <summary>
    /// Constants for the activities
    /// </summary>
    public static class Activities
    {
        public const string ActionA = "F5B5D5F1-577E-4014-BCF0-FBF483DF9E6A";
    }

    /// <summary>
    /// Constants for the activities
    /// </summary>
    public static class ParameterNames
    {
        public const string ParameterA = nameof(ParameterA);
    }
}

This is an excerpt from the container for a customer on-boarding workflow. We kept three activities; The action "Get person", the condition "Person exists" and the action "Send email".

namespace PoC.Workflows.OnBoardCustomer;
/// <summary>
/// The container for the implementations of "On-board customer"
/// </summary>
public class OnBoardCustomerWorkflowContainer : WorkflowContainer
{
    /// <summary>
    /// Constants for the activities
    /// </summary>
    public static class Activities
    {
        public const string GetPerson = "A4D6F17F-ED40-4318-A08B-482302E53063";
        public const string PersonExists = "C5A628AC-5BAD-4DF9-BA46-B878C06D47CE";
        ...
        public const string SendEmail = "2CBE2645-436B-4386-9A15-EAF6BFD82D9F";
    }

    /// <summary>
    /// Constructor
    /// Defines all the activities and adds all known implementations of the workflow
    /// </summary>
    public OnBoardCustomerWorkflowContainer(
        IWorkflowEngineRequiredCapabilities workflowCapabilities,
        ICustomerInformationMgmtCapabilityForClient customerInformationMgmt,
        ICommunicationMgmtCapabilityForClient communicationMgmt)
        : base(
            nameof(CustomerOnboardingMgmtCapability),
            "On-board customer",
            "81B696D3-E27A-4CA8-9DC1-659B78DFE474",
            workflowCapabilities)
    {
        // Define the activities
        DefineActivity(Activities.GetPerson, "Get person", ActivityTypeEnum.Action);
        DefineActivity(Activities.PersonExists, "Person exists?", ActivityTypeEnum.Condition);
        ...
        DefineActivity(Activities.SendEmail, "Send email", ActivityTypeEnum.Action);

        // Add the implementations
        AddImplementation(new Version1.OnBoardCustomerWorkflowImplementation(this, customerInformationMgmt, communicationMgmt));
        AddImplementation(new Version2.OnBoardCustomerWorkflowImplementation(this, customerInformationMgmt, communicationMgmt));
        ...
    }
}

Create an implementation

A workflow implementation inherits from WorkflowImplementation<T>, where T is the type of the value that the workflow returns. If the workflow doesn't return a value, then it inherits from WorkflowImplementation.

The constructor should define any in parameters and set the default activity options, i.e. the options used for all created activities that doesn't specifiy any values themselves.

The workflow implementation must implement three methods:

  • CreateWorkflowInstance() is a factory method that should call the constructor and return the result.
  • GetInstanceTitle() returns a title for this specific instance, preferrably based on the information in any in parameters. This title will be used to represent the workflow instance, e.g. in lists of all the current instances of the implementation.
  • ExecuteWorkflowAsync() is a method for the core logic of the workflow. It will be called repeatedly until the workflow has completed.

This is boilerplate for a workflow implementation:

public class MyWorkflowImplementation : WorkflowImplementation<string>
{
    private readonly IWorkflowContainer _workflowContainer;
    private readonly IMyLogic _logic;

    /// <inheritdoc />
    public MyWorkflowImplementation(IWorkflowContainer workflowContainer, IMyLogic logic) 
        : base(1, 0, workflowContainer)
    {
        _workflowContainer = workflowContainer;
        _logic = logic;
        DefineParameter<int>(MyWorkflowContainer.ParameterNames.ParameterA);
        if (FulcrumApplication.IsInDevelopment)
        {
            SetDebugMode();
        }
    }

    /// <inheritdoc />
    public override string GetInstanceTitle() => "InstanceTitle";

    /// <inheritdoc />
    public override IWorkflowImplementation<string> CreateWorkflowInstance()
    {
        return new MyWorkflowImplementation(_workflowContainer, _logic);
    }

    /// <inheritdoc />
    public override async Task<string> ExecuteWorkflowAsync(CancellationToken cancellationToken = default)
    {
        var parameterA = GetWorkflowArgument<int>(MyWorkflowContainer.ParameterNames.ParameterA);
        var result = await CreateActivity<string>(1, MyWorkflowContainer.Activities.ActionA)
            .Action(a => _logic.ActionA(a, parameterA))
            .ExecuteAsync(cancellationToken);
        return result;
    }
}

This is an excerpt from an implementation of a customer on-boarding workflow. We only kept two activities; The action "Get person" and the condition "Person exists".

namespace PoC.Workflows.OnBoardCustomer.Version1;
public class OnBoardCustomerWorkflowImplementation : WorkflowImplementation<Person>
{
    private readonly ICustomerInformationMgmtCapabilityForClient _customerInformationMgmt;
    private readonly ICommunicationMgmtCapabilityForClient _communicationMgmt;

    public OnBoardCustomerWorkflowImplementation(
        IWorkflowContainer workflowContainer,
        ICustomerInformationMgmtCapabilityForClient customerInformationMgmt,
        ICommunicationMgmtCapabilityForClient communicationMgmt)
    // Set the version for this workflow implementation
    : base(1, 0, workflowContainer)
    {
        _customerInformationMgmt = customerInformationMgmt;
        _communicationMgmt = communicationMgmt;
        // Define in parameters
        DefineParameter<Person>("Person");

        // Set the default activity options
        DefaultActivityOptions.LogCreateThreshold = LogSeverityLevel.Verbose;
        DefaultActivityOptions.LogPurgeStrategy = LogPurgeStrategyEnum.AfterActivitySuccess;
        DefaultActivityOptions.LogPurgeThreshold = LogSeverityLevel.Information;
    }

    /// <inheritdoc />
    public override IWorkflowImplementation<Person> CreateWorkflowInstance()
    {
        return new OnBoardCustomer(WorkflowContainer, _customerInformationMgmt, _communicationMgmt);
    }

    /// <inheritdoc />
    public override string GetInstanceTitle()
    {
        var person = GetWorkflowArgument<Person>("Person");
        return $"On-board {person.EmailAddress}";
    }

    /// <inheritdoc />
    public override async Task<Person> ExecuteWorkflowAsync(CancellationToken cancellationToken = default)
    {
        var initialPerson = GetWorkflowArgument<Person>("Person");

        // Action: 1. Get person
        var existingPerson = await CreateActivity<Person>(1, OnBoardCustomerWorkflow.Activities.GetPerson)
            .SetParameter("Person", initialPerson)
            .Action()
            .ExecuteAsync(GetPersonAsync, cancellationToken);

        // Condition: 2. Person exists?
        var exists = await CreateActivity<bool>(2, OnBoardCustomerWorkflow.Activities.PersonExists)
            .SetParameter("InitialPerson", initialPerson)
            .SetParameter("Person", existingPerson)
            .Condition()
            .ExecuteAsync(_ => existingPerson != null, cancellationToken);

        if (exists) return existingPerson;
        ...
        // The rest of the workflow will create a new customer and send a welcome message over e-mail
    }
}

Use the workflow

You would normally use the workflow in the implementation of a service. In the constructor for the service you will need to prepare the workflow container. The method that should call the workflow needs to do two things; first ask the workflow container for a suitable implementation and then execute that implementation.

public CustomerService(
    ICustomerInformationMgmtCapabilityForClient customerInformationMgmtCapability,
    ICommunicationMgmtCapabilityForClient communicationMgmtCapability,
    IWorkflowEngineRequiredCapabilities workflowCapabilities)
{
     // Prepare the container
    _onBoardCustomerWorkflowContainer = new OnBoardCustomerWorkflowContainer(
        workflowCapabilities,
        customerInformationMgmtCapability,
        communicationMgmtCapability);
}

public async Task<Person> OnBoardAsync(Person person, CancellationToken cancellationToken = default)
{
    // Get the latest implementation
    var implementation = await _onBoardCustomerWorkflowContainer
        .SelectImplementationAsync<Person>(1, 2, cancellationToken);
        
    // Execute the workflow
    return await implementation
        .SetParameter("Person", person)
        .ExecuteAsync(cancellationToken);
}

When the workflow implementation is executed it will roughly do the following:

  1. Is this the first time we run this instance?
    1. Start a new execution for the request in AsyncManager
    2. The execution throws a RequestPostponedException
    3. The middleware will return a 202 Accepted and refer to AsyncManager for result polling or registering for callback
    4. Done
  2. Fast-forward to the point where we left the workflow the last time
  3. Execute one activity (or more than one if there is parallelism)
  4. Is the activity waiting for an asynchronous response?
    1. Save the state
    2. The execution throws a RequestPostponedException
    3. The middleware will return a 202 Accepted to AsyncManager
    4. Done
  5. Return the result of the completed Workflow