Consolidate sync/async error handling and expose job handle

Author: markerikson

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

@@ -30,6 +30,15 @@ import type {
   ListenerErrorInfo,
 } from './types'
 
+import {
+  Job,
+  SupervisorJob,
+  JobHandle,
+  JobCancellationReason,
+  JobCancellationException,
+} from './job'
+import { Outcome } from './outcome'
+
 export type {
   ActionListener,
   ActionListenerMiddleware,
@@ -132,6 +141,7 @@ export const createListenerEntry: TypedCreateListenerEntry<unknown> = (
     unsubscribe: () => {
       throw new Error('Unsubscribe not initialized')
     },
+    parentJob: new SupervisorJob(),
   }
 
   return entry
@@ -342,7 +352,6 @@ export function createActionListenerMiddleware<
             runListener = false
 
             safelyNotifyError(onError, predicateError, {
-              async: false,
               raisedBy: 'predicate',
               phase: currentPhase,
             })
@@ -353,38 +362,39 @@ export function createActionListenerMiddleware<
           continue
         }
 
-        try {
-          let promiseLikeOrUndefined = entry.listener(action, {
-            ...api,
-            getOriginalState,
-            condition,
-            take,
-            currentPhase,
-            extra,
-            unsubscribe: entry.unsubscribe,
-            subscribe: () => {
-              listenerMap.set(entry.id, entry)
-            },
-          })
-
-          if (promiseLikeOrUndefined) {
-            Promise.resolve(promiseLikeOrUndefined).catch(
-              (asyncListenerError) => {
-                safelyNotifyError(onError, asyncListenerError, {
-                  async: true,
-                  raisedBy: 'listener',
-                  phase: currentPhase,
-                })
-              }
-            )
+        entry.parentJob.launchAndRun(async (jobHandle) => {
+          const result = await Outcome.try(async () =>
+            entry.listener(action, {
+              ...api,
+              getOriginalState,
+              condition,
+              take,
+              currentPhase,
+              extra,
+              unsubscribe: entry.unsubscribe,
+              subscribe: () => {
+                listenerMap.set(entry.id, entry)
+              },
+              job: jobHandle,
+              cancelPrevious: () => {
+                entry.parentJob.cancelChildren(
+                  new JobCancellationException(
+                    JobCancellationReason.JobCancelled
+                  ),
+                  [jobHandle]
+                )
+              },
+            })
+          )
+          if (result.isError()) {
+            safelyNotifyError(onError, result.error, {
+              raisedBy: 'listener',
+              phase: currentPhase,
+            })
           }
-        } catch (syncListenerError) {
-          safelyNotifyError(onError, syncListenerError, {
-            async: false,
-            raisedBy: 'listener',
-            phase: currentPhase,
-          })
-        }
+
+          return Outcome.ok(1)
+        })
       }
       if (currentPhase === 'beforeReducer') {
         result = next(action)

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

@@ -32,6 +32,8 @@ const middlewareApi = {
   currentPhase: expect.stringMatching(/beforeReducer|afterReducer/),
   unsubscribe: expect.any(Function),
   subscribe: expect.any(Function),
+  cancelPrevious: expect.any(Function),
+  job: expect.any(Object),
 }
 
 const noop = () => {}
@@ -626,7 +628,7 @@ describe('createActionListenerMiddleware', () => {
     ])
   })
 
-  test('Notifies sync listener errors to `onError`, if provided', () => {
+  test('Notifies sync listener errors to `onError`, if provided', async () => {
     const onError = jest.fn()
     middleware = createActionListenerMiddleware({
       onError,
@@ -649,8 +651,9 @@ describe('createActionListenerMiddleware', () => {
     })
 
     store.dispatch(testAction1('a'))
+    await delay(100)
+
     expect(onError).toBeCalledWith(listenerError, {
-      async: false,
       raisedBy: 'listener',
       phase: 'afterReducer',
     })
@@ -679,10 +682,9 @@ describe('createActionListenerMiddleware', () => {
 
     store.dispatch(testAction1('a'))
 
-    await Promise.resolve()
+    await delay(100)
 
     expect(onError).toBeCalledWith(listenerError, {
-      async: true,
       raisedBy: 'listener',
       phase: 'afterReducer',
     })

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

@@ -8,6 +8,8 @@ import type {
   ThunkDispatch,
 } from '@reduxjs/toolkit'
 
+import type { JobHandle } from './job'
+
 /**
  * Types copied from RTK
  */
@@ -71,6 +73,8 @@ export interface ActionListenerMiddlewareAPI<S, D extends Dispatch<AnyAction>>
   subscribe(): void
   condition: ConditionFunction<S>
   take: TakePattern<S>
+  cancelPrevious: () => void
+  job: JobHandle
   currentPhase: MiddlewarePhase
   // TODO Figure out how to pass this through the other types correctly
   extra: unknown
@@ -282,6 +286,7 @@ export type ListenerEntry<
   unsubscribe: () => void
   type?: string
   predicate: ListenerPredicate<AnyAction, S>
+  parentJob: JobHandle
 }
 
 const declaredMiddlewareType: unique symbol = undefined as any
@@ -322,7 +327,6 @@ export type ListenerPredicateGuardedActionType<T> = T extends ListenerPredicate<
  * Additional infos regarding the error raised.
  */
 export interface ListenerErrorInfo {
-  async: boolean
   /**
    * Which function has generated the exception.
    */