Update process compiler

package.json

@@ -30,7 +30,7 @@
     "is-git-url": "^1.0.0",
     "js-yaml": "^3.13.1",
     "lodash.pick": "^4.4.0",
-    "mesg-js": "^4.3.0-beta.3",
+    "mesg-js": "^4.3.0-beta.4",
     "node-docker-api": "^1.1.22",
     "rimraf": "^2.6.3",
     "tar": "^4.4.8",
@@ -113,4 +113,4 @@
     "lint": "tslint -p ./ -t stylish --fix"
   },
   "types": "lib/index.d.ts"
-}
+}

src/utils/compiler.ts

@@ -33,78 +33,92 @@ export const service = async (content: Buffer): Promise<Service> => {
   }
 }
 
-const nodeCompiler = (instanceResolver: (object: any) => Promise<hash>) => {
+const nodeCompiler = async (
+  type: 'result' | 'event' | 'task' | 'map' | 'filter',
+  def: any,
+  key: string,
+  opts: {
+    defaultNodeKey?: string | null,
+    instanceResolver?(object: any): Promise<hash>
+  }
+) => {
   const nodes = {
-    result: async (def: any, key: string): Promise<ProcessType.types.Process.Node.IResult> => ({
+    result: async (def: any, key: string, opts: any): Promise<ProcessType.types.Process.Node.IResult> => ({
       key,
       taskKey: def.taskKey,
-      instanceHash: await instanceResolver(def)
+      instanceHash: opts.instanceResolver ? await opts.instanceResolver(def) : def.instanceHash
     }),
-    event: async (def: any, key: string): Promise<ProcessType.types.Process.Node.IEvent> => ({
+    event: async (def: any, key: string, opts: any): Promise<ProcessType.types.Process.Node.IEvent> => ({
       key,
       eventKey: def.eventKey,
-      instanceHash: await instanceResolver(def)
+      instanceHash: opts.instanceResolver ? await opts.instanceResolver(def) : def.instanceHash
     }),
-    task: async (def: any, key: string): Promise<ProcessType.types.Process.Node.ITask> => ({
+    task: async (def: any, key: string, opts: any): Promise<ProcessType.types.Process.Node.ITask> => ({
       key,
       taskKey:
       def.taskKey,
-      instanceHash: await instanceResolver(def)
+      instanceHash: opts.instanceResolver ? await opts.instanceResolver(def) : def.instanceHash
     }),
-    map: async (def: any, key: string): Promise<ProcessType.types.Process.Node.IMap> => ({
+    map: async (def: any, key: string, opts: any): Promise<ProcessType.types.Process.Node.IMap> => ({
       key,
-      outputs: Object.keys(def.inputs).map(key => ({
+      outputs: Object.keys(def).map(key => ({
         key,
-        ref: pick(def.inputs[key], ['key', 'nodeKey'])
+        ref: {
+          key: def[key].key,
+          nodeKey: def[key].stepKey || opts.defaultNodeKey,
+        }
       }))
     }),
-    filter: async (def: any, key: string): Promise<ProcessType.types.Process.Node.IFilter> => ({
+    filter: async (def: any, key: string, opts: any): Promise<ProcessType.types.Process.Node.IFilter> => ({
       key,
-      conditions: Object.keys(def).map(key => ({
+      conditions: Object.keys(def.conditions).map(key => ({
         key,
         predicate: 1, // EQ
-        value: def[key]
+        value: def.conditions[key]
       }))
     })
   }
-  return async (type: 'result' | 'event' | 'task' | 'map' | 'filter', def: any, key: string) => ({
-    [type]: await nodes[type](def, key)
-  })
+  return {
+    [type]: await nodes[type](def, key, opts)
+  }
 }
 
 export const process = async (content: Buffer, instanceResolver: (object: any) => Promise<hash>): Promise<Process> => {
   const definition = decode(content)
-  const compileNode = nodeCompiler(instanceResolver)
 
   let nodes = []
   let edges = []
 
-  let trigger = await (definition.trigger.eventKey
-    ? compileNode('event', definition.trigger, definition.trigger.key)
-    : compileNode('result', definition.trigger, definition.trigger.key))
-
-  nodes.push(trigger)
-
-  let previousKey = definition.trigger.key
-  for (const task of definition.tasks) {
-    if (task.filter) {
-      const filterKey = `${previousKey}-filter`
-      const filterNode = await compileNode('filter', task.filter, filterKey)
-      nodes.push(filterNode)
-      edges.push({src: previousKey, dst: filterKey})
-      previousKey = filterKey
-    }
-    if (task.inputs) {
-      const mapKey = `${previousKey}-map`
-      const mapNode = await compileNode('map', task, mapKey)
+  let previousKey: string | null = null
+  let previousKeyWithOutputs: string | null = null
+  let i = 0
+  for (const step of definition.steps) {
+    step.key = step.key || `node-${i}`
+    if (step.inputs) {
+      const mapKey = `${step.key}-inputs`
+      const mapNode = await nodeCompiler('map', step.inputs, mapKey, {defaultNodeKey: previousKeyWithOutputs})
       nodes.push(mapNode)
-      edges.push({src: previousKey, dst: mapKey})
-      previousKey = mapKey
+      if (previousKey) {
+        edges.push({src: previousKey, dst: mapKey})
+      }
+      previousKeyWithOutputs = previousKey = mapKey
+      i++
+    }
+    const type = step.type !== 'trigger'
+      ? step.type
+      : step.eventKey
+        ? 'event'
+        : 'result'
+    const stepNode = await nodeCompiler(type, step, step.key, {instanceResolver})
+    nodes.push(stepNode)
+    if (previousKey) {
+      edges.push({src: previousKey, dst: step.key})
+    }
+    if (type !== 'filter') {
+      previousKeyWithOutputs = step.key
     }
-    const taskNode = await compileNode('task', task, task.key)
-    nodes.push(taskNode)
-    edges.push({src: previousKey, dst: task.key})
-    previousKey = task.key
+    previousKey = step.key
+    i++
   }
 
   return {