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. 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 extends StepBody {
public run(context: StepExecutionContext): Promise<ExecutionResult> {
console.log("Hello World");
return ExecutionResult.next();
}
}Then we define the workflow structure by composing a chain of steps. The is done by implementing WorkflowBase<>.
class HelloWorld_Workflow implements WorkflowBase<any> {
public id: string = "hello-world";
public version: number = 1;
public build(builder: WorkflowBuilder<any>) {
builder
.startWith(HelloWorld)
.then(GoodbyeWorld);
}
}The id and version properties are used by the workflow host to identify a workflow definition.
You can also define your steps inline
class HelloWorld_Workflow implements WorkflowBase<any> {
public id: string = "hello-world";
public version: number = 1;
public build(builder: WorkflowBuilder<any>) {
builder
.startWith(HelloWorld)
.thenRun((context) => {
console.log("Goodbye world");
return ExecutionResult.next();
});
}
}The generic parameter on WorkflowBase<> is used to specify a strongly typed data class that will be persisted along with each instance of this workflow
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 persistenceData property on the context object is null. 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 persistenceData, so the workflow host will call this step again in the future buy will popluate the persistenceData with it's previous value.
For example, this step will initially run with null persistenceData and put the workflow to sleep for 1 hour, while setting the persistenceData to true. 1 hour later, the step will be called again but context.persistenceData will now contain the value from the previous iteration, and will now produce an outcome value of null, causing the workflow to move forward.
class DeferredStep extends StepBody {
public run(context: StepExecutionContext): Promise<ExecutionResult> {
if (!context.persistenceData) {
console.log("going to sleep...");
return ExecutionResult.sleep(new Date(Date.now() + (1000 * 60 * 60)), true);
}
else {
console.log("waking up...");
return ExecutionResult.next();
}
}
}Each step is intended to be a black-box, 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
class AddNumbers extends StepBody {
public number1: number;
public number2: number;
public result: number;
public run(context: StepExecutionContext): Promise<ExecutionResult> {
this.result = this.number1 + this.number2;
return ExecutionResult.next();
}
}
//Our class to define the internal data of our workflow
class MyDataClass {
public value1: number;
public value2: number;
public value3: number;
}
//Our workflow definition with strongly typed internal data and mapped inputs & outputs
class DataSample_Workflow implements WorkflowBase<MyDataClass> {
public id: string = "data-sample";
public version: number = 1;
public build(builder: WorkflowBuilder<MyDataClass>) {
builder
.startWith(AddNumbers)
.input((step, data) => step.number1 = data.value1)
.input((step, data) => step.number2 = data.value2)
.output((step, data) => data.value3 = step.result)
.then(LogMessage)
.input((step, data) => step.message = "The answer is " + data.value3)
}
}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 continue processing, passing the data generated by the event onto the next step.
class EventSample_Workflow implements WorkflowBase<MyDataClass> {
public id: string = "event-sample";
public version: number = 1;
public build(builder: WorkflowBuilder<MyDataClass>) {
builder
.startWith(LogMessage)
.input((step, data) => step.message = "Waiting for event...")
.waitFor("myEvent", data => "0")
.output((step, data) => data.externalValue = step.eventData)
.then(LogMessage)
.input((step, data) => step.message = "The event data is " + data.externalValue)
}
}
...
//External events are published via the host
//All workflows that have subscribed to MyEvent 0, will be passed "hello"
host.publishEvent("myEvent", "0", "hello");class Foreach_Workflow implements WorkflowBase<MyDataClass> {
public id: string = "foreach-sample";
public version: number = 1;
public build(builder: WorkflowBuilder<MyDataClass>) {
builder
.startWith(SayHello)
.foreach((data) => ["one", "two", "three"]).do((then) => then
.startWith(DisplayContext)
.then(DoSomething))
.then(SayGoodbye);
}
}
...
class DisplayContext extends StepBody {
run(context: StepExecutionContext): Promise<ExecutionResult> {
console.log(`Working on ${context.item}`);
return ExecutionResult.next();
}
}class While_Workflow implements WorkflowBase<MyDataClass> {
public id: string = "while-sample";
public version: number = 1;
public build(builder: WorkflowBuilder<MyDataClass>) {
builder
.startWith(SayHello)
.while((data) => data.counter < 3).do((then) => then
.startWith(GetIncrement)
.output((step, data) => data.counter += step.increment)
.then(DoSomething))
.then(SayGoodbye);
}
}class If_Workflow implements WorkflowBase<MyDataClass> {
public id: string = "if-sample";
public version: number = 1;
public build(builder: WorkflowBuilder<MyDataClass>) {
builder
.startWith(SayHello)
.if((data) => data.value > 3).do((then) => then
.startWith(PrintMessage)
.input((step, data) => step.message = "Value is greater than 3")
.then(DoSomething))
.if((data) => data.value > 6).do((then) => then
.startWith(PrintMessage)
.input((step, data) => step.message = "Value is greater than 6")
.then(DoSomething))
.if((data) => data.value == 5).do((then) => then
.startWith(PrintMessage)
.input((step, data) => step.message = "Value is 5")
.then(DoSomething))
.then(SayGoodbye);
}
}Put the workflow to sleep for a specifed number of milliseconds.
build(builder) {
builder
.startWith(HelloWorld)
.delay(data => 2000)
.then(GoodbyeWorld);
}Schedule a sequence of steps to execution asynchronously in the future.
build(builder) {
builder
.startWith(HelloWorld)
.schedule((data) => 20000).do((sequence) => sequence
.startWith(DoSomething)
.then(DoSomethingElse))
.then(ContinueWithSomething);
}Run several sequences of steps in parallel
class Parallel_Workflow {
build(builder) {
builder
.startWith(SayHello)
.parallel()
.do(branch1 => branch1
.startWith(DoSomething)
.then(WaitForSomething)
.then(DoSomethingElse)
)
.do(branch2 => branch2
.startWith(DoSomething)
.then(DoSomethingElse)
)
.do(branch3 => branch3
.startWith(DoSomething)
.then(DoSomethingElse)
)
.join()
.then(SayGoodbye);
}
}In this sample, if Task2 throws an exception, then UndoTask2 and UndoTask1 will be triggered.
builder
.startWith(SayHello)
.compensateWith(UndoHello)
.saga(saga => saga
.startWith(Task1)
.compensateWith(UndoTask1)
.then(Task2)
.compensateWith(UndoTask2)
.then(Task3)
.compensateWith(UndoTask3)
)
.then(SayGoodbye);This particular example will retry the entire saga every 5 seconds
builder
.startWith(SayHello)
.compensateWith(UndoHello)
.saga(saga => saga
.startWith(Task1)
.compensateWith(UndoTask1)
.then(Task2)
.compensateWith(UndoTask2)
.then(Task3)
.compensateWith(UndoTask3)
)
.onError(WorkflowErrorHandling.Retry, 5000)
.then(SayGoodbye);You could also only specify a master compensation step, as follows
builder
.startWith(SayHello)
.compensateWith(UndoHello)
.saga(saga => saga
.startWith(Task1)
.then(Task2)
.then(Task3)
)
.compensateWithSequence(comp => comp
.startWith(UndoTask1)
.then(UndoTask2)
.then(UndoTask3)
)
.then(SayGoodbye);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.
When your application starts, create a WorkflowHost service, call registerWorkflow, so that the workflow host knows about all your workflows, and then call start() to fire up the event loop that executes workflows. Use the startWorkflow method to initiate a new instance of a particular workflow.
let config = configureWorkflow();
let host = config.getHost();
host.registerWorkflow(HelloWorld_Workflow);
await host.start();
let id = await host.startWorkflow("hello-world", 1);
console.log("Started workflow: " + id);