rfactor: implement take using jobs

Author: FaberVitale

packages/action-listener-middleware/src/index.ts

@@ -67,51 +67,47 @@ const actualMiddlewarePhases = ['beforeReducer', 'afterReducer'] as const
 
 function createTakePattern<S>(
   addListener: AddListenerOverloads<Unsubscribe, S, Dispatch<AnyAction>>,
-  parentJob?: Job<any>
+  parentJob: Job<any>
 ): TakePattern<S> {
   async function take<P extends AnyActionListenerPredicate<S>>(
     predicate: P,
     timeout: number | undefined
   ) {
     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) => {
+    let job: Job<[AnyAction, S, S]> = parentJob.launch(async (job) => Outcome.wrap(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([
+          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]
-    }
+    let result: Outcome<[AnyAction, S, S]>;
+
+    try {
+      result = await (timeout !== undefined ? job.runWithTimeout(timeout) : job.run());
+
+      if(result.isOk()) {
+        return result.value;
+      } else if(result.error instanceof JobCancellationException) {
+        return false;
+      } 
 
-    const result = await Promise.race(promises)
+      throw result.error;
+
+    } finally {
+      unsubscribe()
+    }
 
-    unsubscribe()
-    return result
   }
 
   return take as TakePattern<S>
@@ -310,11 +306,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 +363,11 @@ export function createActionListenerMiddleware<
         }
 
         entry.parentJob.launchAndRun(async (jobHandle) => {
+          const take = createTakePattern(addListener, entry.parentJob 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,

packages/action-listener-middleware/src/tests/listenerMiddleware.test.ts

@@ -850,7 +850,7 @@ describe('createActionListenerMiddleware', () => {
   })
 
   describe('Job API', () => {
-    test.skip('Allows canceling previous jobs', () => {
+    test('Allows canceling previous jobs', () => {
       let jobsStarted = 0
 
       middleware.addListener({