WIP Sketch out ideas for Job-based condition handling

Author: markerikson

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

@@ -66,13 +66,20 @@ const defaultWhen: MiddlewarePhase = 'afterReducer'
 const actualMiddlewarePhases = ['beforeReducer', 'afterReducer'] as const
 
 function createTakePattern<S>(
-  addListener: AddListenerOverloads<Unsubscribe, S, Dispatch<AnyAction>>
+  addListener: AddListenerOverloads<Unsubscribe, S, Dispatch<AnyAction>>,
+  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) => {
       unsubscribe = addListener({
@@ -86,20 +93,22 @@ function createTakePattern<S>(
             listenerApi.getOriginalState(),
           ])
         },
+        parentJob,
       })
     })
 
-    if (timeout === undefined) {
-      return tuplePromise
-    }
+    let promises: Promise<unknown>[] = [tuplePromise]
 
-    const timedOutPromise = new Promise<null>((resolve, reject) => {
-      setTimeout(() => {
-        resolve(null)
-      }, timeout)
-    })
+    if (timeout !== undefined) {
+      const timedOutPromise = new Promise<null>((resolve, reject) => {
+        setTimeout(() => {
+          resolve(null)
+        }, timeout)
+      })
+      promises = [tuplePromise, timedOutPromise]
+    }
 
-    const result = await Promise.race([tuplePromise, timedOutPromise])
+    const result = await Promise.race(promises)
 
     unsubscribe()
     return result
@@ -386,7 +395,10 @@ export function createActionListenerMiddleware<
               },
             })
           )
-          if (result.isError()) {
+          if (
+            result.isError() &&
+            !(result.error instanceof JobCancellationException)
+          ) {
             safelyNotifyError(onError, result.error, {
               raisedBy: 'listener',
               phase: currentPhase,

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

@@ -93,13 +93,16 @@ describe('createActionListenerMiddleware', () => {
       increment(state) {
         state.value += 1
       },
+      decrement(state) {
+        state.value -= 1
+      },
       // Use the PayloadAction type to declare the contents of `action.payload`
       incrementByAmount: (state, action: PayloadAction<number>) => {
         state.value += action.payload
       },
     },
   })
-  const { increment, incrementByAmount } = counterSlice.actions
+  const { increment, decrement, incrementByAmount } = counterSlice.actions
 
   function delay(ms: number) {
     return new Promise((resolve) => setTimeout(resolve, ms))
@@ -846,6 +849,30 @@ describe('createActionListenerMiddleware', () => {
     })
   })
 
+  describe('Job API', () => {
+    test.skip('Allows canceling previous jobs', () => {
+      let jobsStarted = 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')
+          } else {
+            listenerApi.cancelPrevious()
+          }
+        },
+      })
+
+      // TODO Write the rest of this test
+    })
+  })
+
   describe('Type tests', () => {
     const middleware = createActionListenerMiddleware()
     const store = configureStore({

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

@@ -8,7 +8,7 @@ import type {
   ThunkDispatch,
 } from '@reduxjs/toolkit'
 
-import type { JobHandle } from './job'
+import type { JobHandle, Job } from './job'
 
 /**
  * Types copied from RTK
@@ -100,6 +100,7 @@ export interface ActionListenerOptions {
    * Defaults to 'before'.
    */
   when?: When
+  parentJob?: JobHandle
 }
 
 export interface CreateListenerMiddlewareOptions<ExtraArgument = unknown> {