Merge pull request #1799 from FaberVitale/feature/middleware-jobs--take-implemented-with-jobs

markerikson · web-flow · commit e2eac2af797e · 2021-12-03T00:16:11.000-05:00
diff --git a/packages/action-listener-middleware/src/index.ts b/packages/action-listener-middleware/src/index.ts
@@ -67,51 +67,75 @@ const actualMiddlewarePhases = ['beforeReducer', 'afterReducer'] as const
 
 function createTakePattern<S>(
   addListener: AddListenerOverloads<Unsubscribe, S, Dispatch<AnyAction>>,
-  parentJob?: Job<any>
+  parentJob: Job<any>
 ): TakePattern<S> {
+  /**
+   * A function that takes an ActionListenerPredicate and an optional timeout,
+   * and resolves when either the predicate returns `true` based on an action
+   * state combination or when the timeout expires.
+   * If the parent listener is canceled while waiting, this will throw a
+   * JobCancellationException.
+   */
   async function take<P extends AnyActionListenerPredicate<S>>(
     predicate: P,
     timeout: number | undefined
   ) {
+    // Placeholder unsubscribe function until the listener is added
     let unsubscribe: Unsubscribe = () => {}
-    let job: JobHandle
-
-    // TODO Need to figure out how to propagate cancelations of the job that
-    // is locked inside of the middleware back up into here, so that a canceled
-    // listener waiting on a condition has that condition throw instead.
-    // Maybe rewrite these around use of `Job.pause()` instead?
-
-    const tuplePromise = new Promise<[AnyAction, S, S]>((resolve) => {
-      unsubscribe = addListener({
-        predicate: predicate as any,
-        listener: (action, listenerApi): void => {
-          // One-shot listener that cleans up as soon as the predicate resolves
-          listenerApi.unsubscribe()
-          resolve([
-            action,
-            listenerApi.getState(),
-            listenerApi.getOriginalState(),
-          ])
-        },
-        parentJob,
-      })
-    })
-
-    let promises: Promise<unknown>[] = [tuplePromise]
-
-    if (timeout !== undefined) {
-      const timedOutPromise = new Promise<null>((resolve, reject) => {
-        setTimeout(() => {
-          resolve(null)
-        }, timeout)
-      })
-      promises = [tuplePromise, timedOutPromise]
-    }
 
-    const result = await Promise.race(promises)
+    // We'll add an additional nested Job representing this function.
+    // TODO This is really a duplicate of the other job inside the middleware.
+    // This behavior requires some additional nesting:
+    // We're going to create a `Promise` representing the result of the listener,
+    // but then wrap that in an `Outcome` for consistent error handling.
+    let job: Job<[AnyAction, S, S]> = parentJob.launch(async (job) =>
+      Outcome.wrap(
+        new Promise<[AnyAction, S, S]>((resolve) => {
+          // Inside the Promise, we synchronously add the listener.
+          unsubscribe = addListener({
+            predicate: predicate as any,
+            listener: (action, listenerApi): void => {
+              // One-shot listener that cleans up as soon as the predicate passes
+              listenerApi.unsubscribe()
+              // Resolve the promise with the same arguments the predicate saw
+              resolve([
+                action,
+                listenerApi.getState(),
+                listenerApi.getOriginalState(),
+              ])
+            },
+            parentJob,
+          })
+        })
+      )
+    )
 
-    unsubscribe()
-    return result
+    let result: Outcome<[AnyAction, S, S]>
+
+    try {
+      // Run the job and use the timeout if given
+      result = await (timeout !== undefined
+        ? job.runWithTimeout(timeout)
+        : job.run())
+
+      if (result.isOk()) {
+        // Resolve the actual `take` promise with the action+states
+        return result.value
+      } else {
+        if (
+          result.error instanceof JobCancellationException &&
+          result.error.reason === JobCancellationReason.JobCancelled
+        ) {
+          // The `take` job itself was canceled due to timeout.
+          return null
+        }
+        // The parent was canceled - reject this promise with that error
+        throw result.error
+      }
+    } finally {
+      // Always clean up the listener
+      unsubscribe()
+    }
   }
 
   return take as TakePattern<S>
@@ -310,11 +334,6 @@ export function createActionListenerMiddleware<
     return true
   }
 
-  const take = createTakePattern(addListener)
-  const condition: ConditionFunction<S> = (predicate, timeout) => {
-    return take(predicate, timeout).then(Boolean)
-  }
-
   const middleware: Middleware<
     {
       (action: Action<'actionListenerMiddleware/add'>): Unsubscribe
@@ -372,6 +391,11 @@ export function createActionListenerMiddleware<
         }
 
         entry.parentJob.launchAndRun(async (jobHandle) => {
+          const take = createTakePattern(addListener, jobHandle as Job<any>)
+          const condition: ConditionFunction<S> = (predicate, timeout) => {
+            return take(predicate, timeout).then(Boolean)
+          }
+
           const result = await Outcome.try(async () =>
             entry.listener(action, {
               ...api,
diff --git a/packages/action-listener-middleware/src/job.ts b/packages/action-listener-middleware/src/job.ts
@@ -53,10 +53,10 @@ export class JobCancellationException implements Error {
  * [JobCompleted]: The current job was already completed. This only happens if the same job is run more than once.
  */
 export enum JobCancellationReason {
-  ParentJobCancelled,
-  ParentJobCompleted,
-  JobCancelled,
-  JobCompleted,
+  ParentJobCancelled = 'ParentJobCancelled',
+  ParentJobCompleted = 'ParentJobCompleted',
+  JobCancelled = 'JobCancelled',
+  JobCompleted = 'JobCompleted',
 }
 
 /**
diff --git a/packages/action-listener-middleware/src/tests/listenerMiddleware.test.ts b/packages/action-listener-middleware/src/tests/listenerMiddleware.test.ts
@@ -21,6 +21,8 @@ import type {
   TypedAddListener,
   Unsubscribe,
 } from '../index'
+import { JobCancellationException } from '../job'
+import { createNonNullChain } from 'typescript'
 
 const middlewareApi = {
   getState: expect.any(Function),
@@ -732,40 +734,42 @@ describe('createActionListenerMiddleware', () => {
         middleware: (gDM) => gDM().prepend(middleware),
       })
 
+      let takeResult: any = undefined
+
       middleware.addListener({
         predicate: incrementByAmount.match,
         listener: async (_, listenerApi) => {
-          const result = await listenerApi.take(increment.match, 50)
-
-          expect(result).toBe(null)
+          takeResult = await listenerApi.take(increment.match, 15)
         },
       })
       store.dispatch(incrementByAmount(1))
-      await delay(200)
+      await delay(25)
+
+      expect(takeResult).toBe(null)
     })
 
-    test("take resolves to [A, CurrentState, PreviousState] if the timeout is provided but doesn't expires", (done) => {
+    test("take resolves to [A, CurrentState, PreviousState] if the timeout is provided but doesn't expires", async () => {
       const store = configureStore({
         reducer: counterSlice.reducer,
         middleware: (gDM) => gDM().prepend(middleware),
       })
+      let takeResult: any = undefined
+      let stateBefore: any = undefined
+      let stateCurrent: any = undefined
 
       middleware.addListener({
         predicate: incrementByAmount.match,
         listener: async (_, listenerApi) => {
-          const stateBefore = listenerApi.getState()
-          const result = await listenerApi.take(increment.match, 50)
-
-          expect(result).toEqual([
-            increment(),
-            listenerApi.getState(),
-            stateBefore,
-          ])
-          done()
+          stateBefore = listenerApi.getState()
+          takeResult = await listenerApi.take(increment.match, 50)
+          stateCurrent = listenerApi.getState()
         },
       })
       store.dispatch(incrementByAmount(1))
       store.dispatch(increment())
+
+      await delay(25)
+      expect(takeResult).toEqual([increment(), stateCurrent, stateBefore])
     })
 
     test('condition method resolves promise when the predicate succeeds', async () => {
@@ -799,11 +803,11 @@ describe('createActionListenerMiddleware', () => {
 
       store.dispatch(increment())
       expect(listenerStarted).toBe(true)
-      await delay(50)
+      await delay(25)
       store.dispatch(increment())
       store.dispatch(increment())
 
-      await delay(50)
+      await delay(25)
 
       expect(finalCount).toBe(3)
     })
@@ -828,7 +832,7 @@ describe('createActionListenerMiddleware', () => {
           listenerStarted = true
           const result = await listenerApi.condition((action, currentState) => {
             return (currentState as CounterState).value === 3
-          }, 50)
+          }, 25)
 
           expect(result).toBe(false)
           const latestState = listenerApi.getState() as CounterState
@@ -842,34 +846,49 @@ describe('createActionListenerMiddleware', () => {
 
       store.dispatch(increment())
 
-      await delay(150)
+      await delay(50)
       store.dispatch(increment())
 
       expect(finalCount).toBe(2)
     })
   })
 
   describe('Job API', () => {
-    test.skip('Allows canceling previous jobs', () => {
+    test('Allows canceling previous jobs', async () => {
       let jobsStarted = 0
+      let jobsContinued = 0
+      let jobsCanceled = 0
 
       middleware.addListener({
         actionCreator: increment,
         listener: async (action, listenerApi) => {
           jobsStarted++
 
           if (jobsStarted < 3) {
-            await listenerApi.condition(decrement.match)
-            // Cancelation _should_ cause `condition()` to throw so we never
-            // end up hitting this next line
-            console.log('Continuing after decrement')
+            try {
+              await listenerApi.condition(decrement.match)
+              // Cancelation _should_ cause `condition()` to throw so we never
+              // end up hitting this next line
+              jobsContinued++
+            } catch (err) {
+              if (err instanceof JobCancellationException) {
+                jobsCanceled++
+              }
+            }
           } else {
             listenerApi.cancelPrevious()
           }
         },
       })
 
-      // TODO Write the rest of this test
+      store.dispatch(increment())
+      store.dispatch(increment())
+      store.dispatch(increment())
+
+      await delay(10)
+      expect(jobsStarted).toBe(3)
+      expect(jobsContinued).toBe(0)
+      expect(jobsCanceled).toBe(2)
     })
   })
 
