Merge pull request #18 from danielgerlag/saga-feature

Saga feature

Author: danielgerlag

README.md

@@ -2,7 +2,7 @@
 
 [![Build Status](https://travis-ci.org/danielgerlag/workflow-es.svg?branch=master)](https://travis-ci.org/danielgerlag/workflow-es)
 
-Workflow ES is a workflow / durable task library for Node.js (or modern browsers).  It supports pluggable persistence and concurrency providers to allow for multi-node clusters.
+Workflow ES is a workflow / saga library for Node.js (or modern browsers).  It supports pluggable persistence and concurrency providers to allow for multi-node clusters.
 
 ## Installing
 

core/README.md

@@ -2,7 +2,7 @@
 
 [![Build Status](https://travis-ci.org/danielgerlag/workflow-es.svg?branch=master)](https://travis-ci.org/danielgerlag/workflow-es)
 
-Workflow ES is a workflow / durable task library for Node.js (or modern browsers).  It supports pluggable persistence and concurrency providers to allow for multi-node clusters.
+Workflow ES is a workflow / saga library for Node.js (or modern browsers).  It supports pluggable persistence and concurrency providers to allow for multi-node clusters.
 
 ## Installing
 

core/package.json

@@ -1,6 +1,6 @@
 {
   "name": "workflow-es",
-  "version": "2.2.1",
+  "version": "2.3.0",
   "description": "A lightweight workflow engine for Node.js",
   "main": "./build/src/index.js",
   "typings": "./build/src/index.d.ts",
@@ -10,7 +10,7 @@
     "test": "jasmine"
   },
   "keywords": [
-    "workflow"
+    "workflow", "saga"
   ],
   "author": {
     "email": "[email protected]",
@@ -30,7 +30,6 @@
   },
   "dependencies": {
     "inversify": "^4.1.0",
-    "moment": "^2.21.0",
     "reflect-metadata": "^0.1.10"
   }
 }

core/src/abstractions/execution-pointer-factory.ts

@@ -4,4 +4,5 @@ export interface IExecutionPointerFactory {
     buildGenesisPointer(defintion: WorkflowDefinition): ExecutionPointer;
     buildNextPointer(pointer: ExecutionPointer, outcomeTarget: StepOutcome): ExecutionPointer;
     buildChildPointer(pointer: ExecutionPointer, childId: number, branch: any): ExecutionPointer;
+    buildCompensationPointer(pointer: ExecutionPointer, exceptionPointer: ExecutionPointer, compensationStepId: number): ExecutionPointer;
 }

core/src/abstractions/execution-result-processor.ts

@@ -1,5 +1,6 @@
-import { WorkflowInstance, WorkflowExecutorResult, WorkflowStepBase, ExecutionPointer, ExecutionResult } from "../models";
+import { WorkflowInstance, WorkflowDefinition, WorkflowExecutorResult, WorkflowStepBase, ExecutionPointer, ExecutionResult } from "../models";
 
 export interface IExecutionResultProcessor {
     processExecutionResult(stepResult: ExecutionResult, pointer: ExecutionPointer, instance: WorkflowInstance, step: WorkflowStepBase, workflowResult: WorkflowExecutorResult);
+    handleStepException(workflow: WorkflowInstance, definition: WorkflowDefinition, pointer: ExecutionPointer, step: WorkflowStepBase);
 }

core/src/fluent-builders/step-builder.ts

@@ -1,5 +1,5 @@
 import { StepBody, InlineStepBody } from "../abstractions";
-import { WorkflowDefinition, WorkflowStepBase, WorkflowStep, StepOutcome, StepExecutionContext, ExecutionResult, WorkflowErrorHandling } from "../models";
+import { WorkflowDefinition, WorkflowStepBase, WorkflowStep, StepOutcome, StepExecutionContext, ExecutionResult, WorkflowErrorHandling, SagaContainer } from "../models";
 import { WaitFor, Foreach, While, If, Delay, Schedule, Sequence } from "../primitives";
 import { WorkflowBuilder } from "./workflow-builder";
 import { ReturnStepBuilder } from "./return-step-builder";
@@ -193,6 +193,20 @@ export class StepBuilder<TStepBody extends StepBody, TData> {
         return stepBuilder;
     }
 
+    public saga(builder: (then: WorkflowBuilder<TData>) => void): StepBuilder<Sequence, TData> {
+        var newStep = new SagaContainer<Sequence>();
+        newStep.body = Sequence;
+        this.workflowBuilder.addStep(newStep);
+        let stepBuilder = new StepBuilder<Sequence, TData>(this.workflowBuilder, newStep);
+        let outcome = new StepOutcome();
+        outcome.nextStep = newStep.id;
+        this.step.outcomes.push(outcome);
+        builder(this.workflowBuilder);
+        stepBuilder.step.children.push(stepBuilder.step.id + 1); //TODO: make more elegant
+
+        return stepBuilder;
+    }
+
     public schedule(interval: (data :TData) => number): ReturnStepBuilder<TData, Schedule, TStepBody> {
         let newStep = new WorkflowStep<Schedule>();
         newStep.body = Schedule;
@@ -223,11 +237,39 @@ export class StepBuilder<TStepBody extends StepBody, TData> {
         return stepBuilder;
     }
 
+    public compensateWith<TNewStepBody extends StepBody>(body: { new(): TNewStepBody; }, setup: (step: StepBuilder<TNewStepBody, TData>) => void = null): StepBuilder<TStepBody, TData> {
+        let newStep = new WorkflowStep<TNewStepBody>();
+        newStep.body = body;
+        this.workflowBuilder.addStep(newStep);
+        let stepBuilder = new StepBuilder<TNewStepBody, TData>(this.workflowBuilder, newStep);
+
+        //setup
+        if (setup) {
+            setup(stepBuilder);
+        }
+        
+        this.step.compensationStepId = newStep.id;
+                
+        return this;
+    }
+
+    public compensateWithSequence(sequence: (then: WorkflowBuilder<TData>) => void): StepBuilder<TStepBody, TData> {
+        let newStep = new WorkflowStep<Sequence>();
+        newStep.body = Sequence;
+        this.workflowBuilder.addStep(newStep);
+        let stepBuilder = new StepBuilder<Sequence, TData>(this.workflowBuilder, newStep);
+        this.step.compensationStepId = newStep.id;
+        sequence(this.workflowBuilder);
+        stepBuilder.step.children.push(stepBuilder.step.id + 1); //TODO: make more elegant
+
+        return this;
+    }
+
     public do(builder: (then: WorkflowBuilder<TData>) => void): StepBuilder<TStepBody, TData> {
         builder(this.workflowBuilder);
-        this.step.children.push(this.step.id + 1); //TODO: make more elegant                        
+        this.step.children.push(this.step.id + 1); //TODO: make more elegant
 
         return this;
     }
 
-}
+}

core/src/models.ts

@@ -13,4 +13,5 @@ export * from "./models/workflow-error-handling";
 export * from "./models/execution-pipeline-directive";
 export * from "./models/workflow-executor-result";
 export * from "./models/container-data";
-export * from "./models/schedule-persistence-data";
+export * from "./models/schedule-persistence-data";
+export * from "./models/saga-container";

core/src/models/execution-pointer.ts

@@ -14,9 +14,21 @@ export class ExecutionPointer {
     public eventData: any;
     public outcome: any;
     public stepName: string;
-    public retryCount: number;
+    public retryCount: number = 0;
     public children: string[] = [];
     public contextItem: any;
     public predecessorId: string;
-    
+    public scope: string[] = [];
+    public status: number = 0;    
+}
+
+export var PointerStatus = {
+    Legacy: 0,
+    Pending: 1,
+    Running: 2,
+    Complete: 3,
+    Sleeping: 4,
+    WaitingForEvent: 5,
+    Failed: 6,
+    Compensated: 7
 }

core/src/models/saga-container.ts

@@ -0,0 +1,20 @@
+import { StepBody } from "../abstractions";
+import { ExecutionPointer } from "./execution-pointer";
+import { WorkflowStep } from "./workflow-step";
+
+export class SagaContainer<T extends StepBody> extends WorkflowStep<T> {
+    
+    public resumeChildrenAfterCompensation(): boolean {
+        return false;
+    } 
+
+    public revertChildrenAfterCompensation(): boolean {
+        return true;
+    }
+
+    public primeForRetry(pointer: ExecutionPointer) {
+        super.primeForRetry(pointer);
+        pointer.persistenceData = null;
+    }
+    
+}

core/src/models/workflow-error-handling.ts

@@ -1,5 +1,6 @@
 export var WorkflowErrorHandling = {
-    Retry : 0,
-    Suspend : 1,
-    Terminate : 2
+    Retry : 1,
+    Suspend : 2,
+    Terminate : 3,
+    Compensate: 4
 }

core/src/models/workflow-step.ts

@@ -16,7 +16,8 @@ export abstract class WorkflowStepBase {
     public outcomes: Array<StepOutcome> = [];
     public children: Array<number> = [];
     public errorBehavior : number;
-    public retryInterval : number;
+    public retryInterval : number = 60000;
+    public compensationStepId : number;
 
     public inputs: Array<(step: StepBody, data: any) => void> = [];
     public outputs: Array<(step: StepBody, data: any) => void> = [];
@@ -31,6 +32,20 @@ export abstract class WorkflowStepBase {
 
     public afterExecute(executorResult: WorkflowExecutorResult, context: StepExecutionContext, stepResult: ExecutionResult, executionPointer: ExecutionPointer) {            
     }
+
+    public afterWorkflowIteration(executorResult: WorkflowExecutorResult, defintion: WorkflowDefinition, workflow: WorkflowInstance, executionPointer: ExecutionPointer) {
+    }
+
+    public resumeChildrenAfterCompensation(): boolean {
+        return true;
+    } 
+
+    public revertChildrenAfterCompensation(): boolean {
+        return false;
+    }
+
+    public primeForRetry(pointer: ExecutionPointer) {
+    }
 
 }
 

core/src/services/execution-pointer-factory.ts

@@ -1,7 +1,7 @@
 import { injectable, inject } from "inversify";
 import { IExecutionPointerFactory, ILogger, IWorkflowRegistry, IWorkflowExecutor, TYPES, IExecutionResultProcessor } from "../abstractions";
 import { WorkflowHost } from "./workflow-host";
-import { WorkflowDefinition, ExecutionPointer, EventSubscription, StepOutcome, ExecutionResult, StepExecutionContext, WorkflowStepBase, WorkflowStatus, ExecutionError, WorkflowErrorHandling, ExecutionPipelineDirective, WorkflowExecutorResult } from "../models";
+import { WorkflowDefinition, ExecutionPointer, PointerStatus, EventSubscription, StepOutcome, ExecutionResult, StepExecutionContext, WorkflowStepBase, WorkflowStatus, ExecutionError, WorkflowErrorHandling, ExecutionPipelineDirective, WorkflowExecutorResult } from "../models";
 
 @injectable()
 export class ExecutionPointerFactory implements IExecutionPointerFactory {
@@ -10,6 +10,7 @@ export class ExecutionPointerFactory implements IExecutionPointerFactory {
         let result = new ExecutionPointer();
         result.active = true;
         result.stepId = 0;
+        result.status = PointerStatus.Pending;
         result.id = this.generatePointerId();
 
         return result;
@@ -19,9 +20,12 @@ export class ExecutionPointerFactory implements IExecutionPointerFactory {
         let result = new ExecutionPointer();
         result.active = true;
         result.stepId = outcomeTarget.nextStep;
+        result.status = PointerStatus.Pending;
         result.id = this.generatePointerId();
         result.predecessorId = pointer.id;
-        result.contextItem = pointer.contextItem;        
+        result.contextItem = pointer.contextItem;
+        if (pointer.scope)
+            result.scope = pointer.scope.slice();
 
         return result;
     }
@@ -31,10 +35,28 @@ export class ExecutionPointerFactory implements IExecutionPointerFactory {
         result.active = true;
         result.id = this.generatePointerId();
         result.predecessorId = pointer.id;
+        result.status = PointerStatus.Pending;
         result.stepId = childId;        
         result.contextItem = branch;
+        if (pointer.scope)
+            result.scope = pointer.scope.slice();
 
+        result.scope.push(pointer.id);
         pointer.children.push(result.id);
+        
+        return result;
+    }
+
+    public buildCompensationPointer(pointer: ExecutionPointer, exceptionPointer: ExecutionPointer, compensationStepId: number): ExecutionPointer {
+        let result = new ExecutionPointer();
+        result.active = true;
+        result.predecessorId = exceptionPointer.id;
+        result.stepId = compensationStepId,
+        result.status = PointerStatus.Pending;
+        result.id = this.generatePointerId();        
+        result.contextItem = pointer.contextItem;
+        if (pointer.scope)
+            result.scope = pointer.scope.slice();
 
         return result;
     }