feat(overmind): add parallel operator

Author: christianalfoni

packages/node_modules/overmind/src/index.test.ts

@@ -1,4 +1,4 @@
-import { EventType, Overmind, TAction, TApp, Operator, run } from './'
+import { EventType, Overmind, TAction, TApp } from './'
 import { modules } from './modules'
 
 function toJSON(obj) {

packages/node_modules/overmind/src/index.ts

@@ -587,7 +587,7 @@ function createContext(context, value, path?) {
     execution: {
       ...context.execution,
       operatorId: context.execution.operatorId + 1,
-      path: path ? context.execution.path.concat(path) : context.execution.path,
+      path: path || context.execution.path,
     },
   }
 }
@@ -671,7 +671,7 @@ export function forEach<Input extends any[]>(
 
         if (!evaluatingCount) {
           stopDebugOperator(lastContext)
-          next(null, context)
+          next(null, lastContext)
         }
       }
       startDebugOperator('forEach', '', context)
@@ -680,7 +680,7 @@ export function forEach<Input extends any[]>(
         lastContext = createContext(
           lastContext || context,
           value,
-          String(index)
+          context.execution.path.concat(String(index))
         )
         const nextWithPath = createNextPath(evaluate)
         forEachItemOperator(null, lastContext, nextWithPath)
@@ -692,6 +692,48 @@ export function forEach<Input extends any[]>(
   return instance
 }
 
+export function parallel<Input>(
+  operators: Operator<Input, any>[]
+): Operator<Input, Input> {
+  const instance = (err, context, next) => {
+    if (err) next(err)
+    else {
+      let evaluatingCount = operators.length
+      let lastContext
+      let hasErrored = false
+      const evaluate = (err) => {
+        if (hasErrored) {
+          return
+        }
+        if (err) {
+          hasErrored = true
+          return next(err, lastContext)
+        }
+        evaluatingCount--
+
+        if (!evaluatingCount) {
+          stopDebugOperator(lastContext)
+          next(null, lastContext)
+        }
+      }
+      startDebugOperator('parallel', '', context)
+
+      operators.forEach((operator, index) => {
+        lastContext = createContext(
+          lastContext || context,
+          context.value,
+          context.execution.path.concat(String(index))
+        )
+        const nextWithPath = createNextPath(evaluate)
+        operator(null, lastContext, nextWithPath)
+      })
+    }
+  }
+  instance[IS_PIPE] = true
+
+  return instance
+}
+
 export function filter<Input>(
   operation: (input: Context<Input>) => boolean
 ): Operator<Input, Input> {
@@ -751,7 +793,11 @@ export function fork<
     else {
       startDebugOperator('fork', operation, context)
       const path = operation(context)
-      const newContext = createContext(context, context.value, path)
+      const newContext = createContext(
+        context,
+        context.value,
+        context.execution.path.concat(path)
+      )
       const nextWithPaths = createNextPath((err, returnedContext) => {
         if (err) next(err)
         else {
@@ -779,12 +825,20 @@ export function when<Input, OutputA, OutputB>(
     else {
       startDebugOperator('when', operation, context)
       if (operation(context)) {
-        const newContext = createContext(context, context.value, 'true')
+        const newContext = createContext(
+          context,
+          context.value,
+          context.execution.path.concat('true')
+        )
         const nextWithPath = createNextPath(next)
         stopDebugOperator(newContext)
         paths.true(null, newContext, nextWithPath)
       } else {
-        const newContext = createContext(context, context.value)
+        const newContext = createContext(
+          context,
+          context.value,
+          context.execution.path.concat('false')
+        )
         const nextWithPath = createNextPath(next)
         stopDebugOperator(newContext)
         paths.false(null, newContext, nextWithPath)

packages/node_modules/overmind/src/pipe.test.ts

@@ -9,6 +9,7 @@ import {
   wait,
   debounce,
   run,
+  parallel,
 } from './'
 
 function createMockAction(pipe) {
@@ -84,6 +85,26 @@ describe('PIPE', () => {
         expect(runCount).toEqual(1)
       })
     })
+    test('parallel', () => {
+      expect.assertions(1)
+      let runCount = 0
+      const test: Operator<string, string> = pipe(
+        parallel([
+          (_, value, next) => {
+            runCount++
+            next(null, value)
+          },
+          (_, value, next) => {
+            runCount++
+            next(null, value)
+          },
+        ])
+      )
+
+      return createMockAction(test)('foo').then(() => {
+        expect(runCount).toEqual(2)
+      })
+    })
     test('filter - truthy', () => {
       expect.assertions(1)
       const test: Operator<string, string> = pipe(