Add workflow instrumentation support

dist/index.d.ts

@@ -17,6 +17,7 @@ interface Workflow {
     nodes: WorkflowNode[];
     outputs: Record<number, string>;
     binding: Record<number, number[]>;
+    nodeNames?: Record<number, string>;
 }
 declare enum WorkflowValidationStatus {
     OK = 0,
@@ -33,7 +34,7 @@ declare enum WorkflowValidationStatus {
 declare function dumpWorkflow(workflow: Workflow): string;
 declare function validateWorkflow(workflow: Workflow): WorkflowValidationStatus;
 
-declare function buildJsxWorkflow(elementDefinition: React.ReactElement): Workflow;
+declare function buildJsxWorkflow(elementDefinition: React.ReactElement, addNodeName?: boolean): Workflow;
 
 interface WorkflowProps {
     children: JSX.Element[];
@@ -66,13 +67,23 @@ declare enum ExecutionStatus {
     Failure = 3,
     Done = 4
 }
+interface NodeExecutionStatus {
+    status: ExecutionStatus;
+    id: number;
+    start: number;
+    end: number;
+    name?: string;
+}
 interface WorkflowExecutor {
     cancel(): void;
     run(...params: any[]): Promise<any>;
     setTimeout(timeout: number): void;
     reset(): void;
     state(): ExecutionStatus;
+    inst(inst: boolean): void;
+    workflow(): Workflow;
+    stats(): NodeExecutionStatus[];
 }
 declare function createWorkflowExecutor(wf: Workflow): WorkflowExecutor;
 
-export { ExecutionStatus, InputNodeComponent, InputNodeProps, NodeComponent, NodeProps, OutputNodeComponent, OutputNodeProps, Workflow, WorkflowComponent, WorkflowExecutionNode, WorkflowExecutor, WorkflowInputProps, WorkflowNode, WorkflowProps, WorkflowValidationStatus, buildJsxWorkflow, createWorkflowExecutor, dumpWorkflow, unitNodeGenerator, validateWorkflow };
+export { ExecutionStatus, InputNodeComponent, InputNodeProps, NodeComponent, NodeExecutionStatus, NodeProps, OutputNodeComponent, OutputNodeProps, Workflow, WorkflowComponent, WorkflowExecutionNode, WorkflowExecutor, WorkflowInputProps, WorkflowNode, WorkflowProps, WorkflowValidationStatus, buildJsxWorkflow, createWorkflowExecutor, dumpWorkflow, unitNodeGenerator, validateWorkflow };

package.json

@@ -1,6 +1,6 @@
 {
   "name": "taskflow-react",
-  "version": "0.2.3",
+  "version": "0.2.4",
   "description": "A promise and react based jsx style task flow library",
   "main": "./dist/index.js",
   "bin": {

src/ReactElementWorkflowBuilder.ts

@@ -7,6 +7,7 @@ import { WorkflowNode } from "./WorkflowNode";
 import {WorkflowProps, InputNodeProps, OutputNodeProps, NodeProps, WorkflowInputProps} from "./WorkflowComponent"
 
 let workflowNodeNameToIdMap : Record<string, number> = {}
+let workflowExecutionNodeIdToNameMap: Record<number, string> = {}
 let workflowNodeId = -1
 
 function setNodeId(name: string, id: number) {
@@ -19,6 +20,7 @@ function genNodeId(name: string) : number {
     }
 
     workflowNodeNameToIdMap[name] = ++workflowNodeId
+    workflowExecutionNodeIdToNameMap[workflowNodeId] = name
     return workflowNodeId
 }
 
@@ -98,7 +100,7 @@ function buildSubWorkflow(element: React.ReactElement, prefix: string) : Workflo
     return nodes
 }
 
-export function buildJsxWorkflow(elementDefinition: React.ReactElement) : Workflow{
+export function buildJsxWorkflow(elementDefinition: React.ReactElement, addNodeName: boolean = false) : Workflow{ // eslint-disable-line
     const element: React.ReactElement = resolveTillWorkflowComponent(elementDefinition)
     const props: WorkflowProps = element.props;
     const children : React.ReactElement[] = props.children
@@ -112,10 +114,6 @@ export function buildJsxWorkflow(elementDefinition: React.ReactElement) : Workfl
             const params: string[] = childProps.params
             for (const param of params) {
                 const id = genNodeId(workflowName + "." + param)
-                if (id < 0) {
-                    throw "duplicate node name found " + workflowName + "." + param
-                }
-
                 inputNodes.push({
                     id: id,
                     deps: [],
@@ -127,10 +125,6 @@ export function buildJsxWorkflow(elementDefinition: React.ReactElement) : Workfl
             const name: string = childProps.name
             const dep: string = childProps.dep
             const id = getNodeId(workflowName + "." + dep)
-            if (id < 0) {
-                throw "dependency must be defined before it's used"
-            }
-
             setNodeId(workflowName + "." + name, id)
             outputs[id] = name
         } else if (child.type == NodeComponent) {
@@ -141,19 +135,11 @@ export function buildJsxWorkflow(elementDefinition: React.ReactElement) : Workfl
             if (deps && deps.length) {
                 for (const dep of deps) {
                     const id = getNodeId(workflowName + "." + dep)
-                    if (id < 0) {
-                        throw "dependency must be defined before it's used"
-                    }
-
                     depIds.push(id)
                 }
             }
 
             const id = genNodeId(workflowName + "." + name)
-            if (id < 0) {
-                throw "duplicate node name found " + workflowName + "." + name
-            }
-
             innerNodes.push({
                 id: id,
                 deps: depIds,
@@ -172,5 +158,10 @@ export function buildJsxWorkflow(elementDefinition: React.ReactElement) : Workfl
     .next(innerNodes)
     .output(outputs);
 
+    if (addNodeName) {
+        workflowBuilder.nodeNames(Object.assign({}, workflowExecutionNodeIdToNameMap))
+    }
+
+    workflowExecutionNodeIdToNameMap = {}
     return workflowBuilder.build()
 }

src/Workflow.ts

@@ -6,6 +6,7 @@ export interface Workflow {
     nodes: WorkflowNode[];
     outputs: Record<number, string>;
     binding: Record<number, number[]>;
+    nodeNames?: Record<number, string>;
 }
 
 export enum WorkflowValidationStatus {

src/WorkflowBuilder.ts

@@ -6,6 +6,7 @@ export interface WorkflowBuilder {
     input(nodes: WorkflowNode[]) : WorkflowBuilder;
     next(nodes: WorkflowNode[]) : WorkflowBuilder;
     output(output: Record<number, string>) : WorkflowBuilder;
+    nodeNames(namesMap: Record<number, string>) : WorkflowBuilder;
 }
 
 export function createWorkflowBuilder() : WorkflowBuilder {
@@ -14,6 +15,7 @@ export function createWorkflowBuilder() : WorkflowBuilder {
     let outputs: Record<number, string> = {}
     let nodes: WorkflowNode[] = []
     const binding: Record<number, number[]> = {}
+    let idToNameMap: Record<number, string> | undefined = undefined
     const builder : WorkflowBuilder = {
         build: () => {
             nodes.sort((a: WorkflowNode, b: WorkflowNode) => {
@@ -25,7 +27,8 @@ export function createWorkflowBuilder() : WorkflowBuilder {
                 outputs: outputs,
                 nodes: nodes,
                 binding: binding,
-                zeroDepNodes: zeroDepNodes
+                zeroDepNodes: zeroDepNodes,
+                nodeNames: idToNameMap
             }
 
             return workflow
@@ -63,6 +66,11 @@ export function createWorkflowBuilder() : WorkflowBuilder {
 
             nodes = nodes.concat(nextNodes)    
             return builder;
+        },
+
+        nodeNames: (nodeMap: Record<number, string>) => {
+            idToNameMap = nodeMap
+            return builder
         }
     }
 

src/WorkflowExecutor.ts

@@ -9,12 +9,23 @@ export enum ExecutionStatus {
     Done = 4,
 }
 
+export interface NodeExecutionStatus {
+    status: ExecutionStatus;
+    id: number;
+    start: number;
+    end: number;
+    name?: string;
+}
+
 export interface WorkflowExecutor {
     cancel() : void;
     run(...params: any[]) : Promise<any>;
     setTimeout(timeout: number) : void;
     reset() : void;
     state() : ExecutionStatus;
+    inst(inst: boolean) : void;
+    workflow() : Workflow;
+    stats(): NodeExecutionStatus[];
 }
 
 export function createWorkflowExecutor(wf: Workflow) {
@@ -26,6 +37,7 @@ export function createWorkflowExecutor(wf: Workflow) {
     let runningNodes: WorkflowRunTimeNode[] = [];
     let pendingPromises: Promise<any>[] = []
     function update(id: number, oriResult: any) : WorkflowRunTimeNode[] {
+        updateNodeExecutionStatus(id, ExecutionStatus.Done)
         if (typeof oriResult === 'undefined') {
             return []
         }
@@ -71,17 +83,55 @@ export function createWorkflowExecutor(wf: Workflow) {
         return newNodesToRun
     }
 
+    let enableInst = false
+    let nodeExecutionStatus: NodeExecutionStatus[] = []
+    function updateNodeExecutionStatus(id: number, status: ExecutionStatus) {
+        if (enableInst) {
+            if (status == ExecutionStatus.Running) {
+                nodeExecutionStatus.push({
+                    id: id,
+                    start: new Date().getTime(),
+                    end: 0,
+                    status: status,
+                    name: workflow.nodeNames ? ((id in workflow.nodeNames) ? workflow.nodeNames[id] : undefined) : undefined
+                })
+            } else {
+                // search from last one
+                for (let inx = nodeExecutionStatus.length - 1; inx >= 0; inx--) {
+                    if (nodeExecutionStatus[inx].id == id) {
+                        nodeExecutionStatus[inx].status = status
+                        nodeExecutionStatus[inx].end = new Date().getTime()
+                        break
+                    }
+                }
+            }
+        }
+    }
+
     const executor: WorkflowExecutor = {
         state() : ExecutionStatus {
             return status
         },
 
+        inst(inst: boolean) : void {
+            enableInst = inst
+        },
+
+        stats() : NodeExecutionStatus[] {
+            return nodeExecutionStatus
+        },
+
+        workflow() : Workflow {
+            return wf
+        },
+
         reset() : void {
             status = ExecutionStatus.NotStarted;
             intermediateResults = {}
             output = {}
             runningNodes = [];
             pendingPromises = []
+            nodeExecutionStatus = []
         },
 
         setTimeout(timeout: number) : void {
@@ -96,6 +146,7 @@ export function createWorkflowExecutor(wf: Workflow) {
             status = ExecutionStatus.Cancelled
             for (const node of runningNodes) {
                 if (node.node.cancel) {
+                    updateNodeExecutionStatus(node.id, ExecutionStatus.Cancelled)
                     node.node.cancel()
                 }
             }
@@ -134,11 +185,13 @@ export function createWorkflowExecutor(wf: Workflow) {
                         const node: WorkflowRunTimeNode = nodesToRun.pop()!
                         let result = undefined
                         try{
+                            updateNodeExecutionStatus(node.id, ExecutionStatus.Running)
                             result = node.node.run(...node.inputs)
                         }
                         catch(err) {
                             executor.cancel()
                             status = ExecutionStatus.Failure
+                            updateNodeExecutionStatus(node.id, ExecutionStatus.Failure)
                             result = undefined
                             break
                         }
@@ -163,6 +216,7 @@ export function createWorkflowExecutor(wf: Workflow) {
                             .catch(() => {
                                 executor.cancel()
                                 status = ExecutionStatus.Failure
+                                updateNodeExecutionStatus(node.id, ExecutionStatus.Failure)
                                 result = undefined
                             })
 

test/WorkflowBuildNodeName.test.tsx

@@ -0,0 +1,87 @@
+import { buildJsxWorkflow } from "../src/ReactElementWorkflowBuilder";
+import { validateWorkflow, WorkflowValidationStatus } from "../src/Workflow";
+import { InputNodeComponent, NodeComponent, OutputNodeComponent, WorkflowComponent, WorkflowInputProps } from "../src/WorkflowComponent";
+import { unitNodeGenerator } from "../src/WorkflowNode";
+import { add, AddWithDoubleWorkflow, double, DoubleWorkflow } from "./NodeWorkflowExample";
+import React from "react";
+import { createWorkflowExecutor } from "../src/WorkflowExecutor";
+
+test("workflow node name test", async () => {
+    const workflowJsx = (<WorkflowComponent>
+        <InputNodeComponent params={["num1", "num2", "num3"]} />
+        <NodeComponent gen={add} deps={["num1", "num2"]} name="add" />
+        <NodeComponent gen={double} deps={["num3"]} name="double" />
+        <NodeComponent gen={add} deps={["add", "double"]} name="finalAdd" />
+        <OutputNodeComponent name="res" dep="finalAdd" />
+    </WorkflowComponent>)
+    const workflow = buildJsxWorkflow(workflowJsx, true)
+    expect(validateWorkflow(workflow)).toBe(WorkflowValidationStatus.OK)
+    expect(workflow.nodes.length).toBe(6)
+    const ndoeNameMap = workflow.nodeNames
+    expect(Object.keys(ndoeNameMap).length).toBe(6)
+    expect(ndoeNameMap[0]).toBe(".num1")
+    expect(ndoeNameMap[1]).toBe(".num2")
+    expect(ndoeNameMap[2]).toBe(".num3")
+    expect(ndoeNameMap[3]).toBe(".add")
+    expect(ndoeNameMap[4]).toBe(".double")
+    expect(ndoeNameMap[5]).toBe(".finalAdd")
+})
+
+function ComputationWorkflow(props: WorkflowInputProps) {
+    return (<WorkflowComponent {...props}>
+        <InputNodeComponent params={["num1", "num2", "num3"]} />
+        <NodeComponent gen={add} deps={["num1", "num2"]} name="add" />
+        <NodeComponent gen={double} deps={["num3"]} name="double" />
+        <NodeComponent gen={add} deps={["add", "double"]} name="finalAdd" />
+        <OutputNodeComponent name="res" dep="finalAdd" />
+    </WorkflowComponent>)
+}
+
+test("nested workflow node name test", async () => {
+    const workflowJsx = (<WorkflowComponent>
+        <InputNodeComponent params={["num1", "num2", "num3"]} />
+        <NodeComponent gen={add} deps={["num1", "num2"]} name="add" />
+        <NodeComponent gen={double} deps={["num3"]} name="double" />
+        <NodeComponent gen={add} deps={["add", "double"]} name="add1" />
+        <ComputationWorkflow name="comp" params={["num1", "num2", "num3"]} />
+        <NodeComponent gen={add} deps={["add1", "comp.res"]} name="finalAdd" />
+        <OutputNodeComponent name="res" dep="finalAdd" />
+    </WorkflowComponent>)
+    const workflow = buildJsxWorkflow(workflowJsx, true)
+    expect(validateWorkflow(workflow)).toBe(WorkflowValidationStatus.OK)
+    expect(workflow.nodes.length).toBe(10)
+    const ndoeNameMap = workflow.nodeNames
+    expect(Object.keys(ndoeNameMap).length).toBe(10)
+    expect(ndoeNameMap[0]).toBe(".num1")
+    expect(ndoeNameMap[1]).toBe(".num2")
+    expect(ndoeNameMap[2]).toBe(".num3")
+    expect(ndoeNameMap[3]).toBe(".add")
+    expect(ndoeNameMap[4]).toBe(".double")
+    expect(ndoeNameMap[5]).toBe(".add1")
+    expect(ndoeNameMap[6]).toBe(".comp.add")
+    expect(ndoeNameMap[7]).toBe(".comp.double")
+    expect(ndoeNameMap[8]).toBe(".comp.finalAdd")
+    expect(ndoeNameMap[9]).toBe(".finalAdd")
+})
+
+test("nested nested workflow node name test", async () => {
+    let workflowJsx = (<WorkflowComponent>
+        <InputNodeComponent params={["num1", "num2", "num3"]} />
+        <AddWithDoubleWorkflow name="add" params={["num1", "num2"]} />
+        <DoubleWorkflow name="double" params={["num3"]} />
+        <NodeComponent gen={add} name="finalAdd" deps={["add.output", "double.output"]} />
+        <OutputNodeComponent name="res" dep="finalAdd" />
+    </WorkflowComponent>)
+    let workflow = buildJsxWorkflow(workflowJsx, true)
+    expect(validateWorkflow(workflow)).toBe(WorkflowValidationStatus.OK)
+    expect(workflow.nodes.length).toBe(7)
+    const ndoeNameMap = workflow.nodeNames
+    expect(Object.keys(ndoeNameMap).length).toBe(7)
+    expect(ndoeNameMap[0]).toBe(".num1")
+    expect(ndoeNameMap[1]).toBe(".num2")
+    expect(ndoeNameMap[2]).toBe(".num3")
+    expect(ndoeNameMap[3]).toBe(".add.double.double")
+    expect(ndoeNameMap[4]).toBe(".add.add")
+    expect(ndoeNameMap[5]).toBe(".double.double")
+    expect(ndoeNameMap[6]).toBe(".finalAdd")   
+})