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
const workflow_es = require("workflow-es");
class HelloWorld extends workflow_es.StepBody {
run(context) {
console.log("Hello World");
return workflow_es.ExecutionResult.next();
}
}Then we define the workflow structure by composing a chain of steps.
class HelloWorld_Workflow {
constructor() {
this.id = "hello-world";
this.version = 1;
}
build(builder) {
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 {
constructor() {
this.id = "hello-world";
this.version = 1;
}
build(builder) {
builder
.startWith(HelloWorld)
.thenRun((context) => {
console.log("Goodbye world");
return workflow_es.ExecutionResult.next();
});
}
}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 workflow_es.StepBody {
run(context) {
if (!context.persistenceData) {
console.log("going to sleep...");
return workflow_es.ExecutionResult.sleep(new Date(Date.now() + (1000 * 60 * 60))), true);
}
else {
console.log("waking up...");
return workflow_es.ExecutionResult.next();
}
}
}Each step is intended to be a black-box, therefore they support inputs and outputs. Each workflow instance carries a data property for holding 'workflow wide' data that the steps can use to communicate.
The following sample shows how to define inputs and outputs on a step, it then shows how to map the inputs and outputs to properties on the workflow data property.
//Our workflow step with inputs and outputs
class AddNumbers extends workflow_es.StepBody {
run(context) {
this.result = this.number1 + this.number2;
return workflow_es.ExecutionResult.next();
}
}
//Our workflow definition with mapped inputs & outputs
class DataSample_Workflow {
constructor() {
this.id = "data-sample";
this.version = 1;
}
build(builder) {
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 {
constructor() {
this.id = "event-sample";
this.version = 1;
}
build(builder) {
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 Sample_Workflow {
constructor() {
this.id = "sample";
this.version = 1;
}
build(builder) {
builder
.startWith(SayHello)
.foreach((data) => ["one", "two", "three"]).do((then) => then
.startWith(DisplayContext)
.then(DoSomething))
.then(SayGoodbye);
}
}
...
class DisplayContext extends workflow_es.StepBody {
run(context) {
console.log(`Working on ${context.item}`);
return workflow_es.ExecutionResult.next();
}
}class Sample_Workflow {
constructor() {
this.id = "sample";
this.version = 1;
}
build(builder) {
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 Sample_Workflow {
constructor() {
this.id = "sample";
this.version = 1;
}
build(builder) {
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.
const workflow_es = require("workflow-es");
...
let config = workflow_es.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);