feat(action-listener-middleware): add fork API

- Changes

* forked task executor now receive an API, currently { pause, delay, signal }
* added lisnter.fork tests
* re-enabled 'Long polling loop' test
* removed outcome.ts it has been replaced by an ADT
* micro optimazations to keep the size at 4.0K
* added pause to listenerApi
* encapsulated abortController in fork() output
* listenerApi.delay now rejects as soon as the listner is cancelled

- Build log

Build "actionListenerMiddleware" to dist/esm:
       1763 B: index.modern.js.gz
       1575 B: index.modern.js.br
Build "actionListenerMiddleware" to dist/module:
       2.4 kB: index.js.gz
      2.15 kB: index.js.br
Build "actionListenerMiddleware" to dist/cjs:
       2.4 kB: index.js.gz
      2.15 kB: index.js.br

- Test log

 PASS  src/tests/listenerMiddleware.test.ts
 PASS  src/tests/effectScenarios.test.ts
 PASS  src/tests/fork.test.ts
 PASS  src/tests/useCases.test.ts

Test Suites: 4 passed, 4 total
Tests:       56 passed, 56 total

Author: FaberVitale

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

@@ -29,13 +29,18 @@ import type {
   WithMiddlewareType,
   TakePattern,
   ListenerErrorInfo,
-  TaskExecutor,
+  ForkedTaskExecutor,
   ForkedTask,
 } from './types'
-
+import { assertFunction } from './utils'
 import { TaskAbortError } from './exceptions'
-import { Outcome } from './outcome'
-export type { Outcome, Ok, Error } from './outcome'
+import {
+  runTask,
+  promisifyAbortSignal,
+  validateActive,
+  createPause,
+  createDelay,
+} from './task'
 export { TaskAbortError } from './exceptions'
 export type {
   ActionListener,
@@ -49,86 +54,68 @@ export type {
   TypedAddListener,
   TypedAddListenerAction,
   Unsubscribe,
-  TaskExecutor,
+  ForkedTaskExecutor,
   ForkedTask,
+  ForkedTaskAPI,
   AsyncTaskExecutor,
   SyncTaskExecutor,
+  TaskCancelled,
+  TaskRejected,
+  TaskResolved,
+  TaskResult,
 } from './types'
 
-function assertFunction(
-  func: unknown,
-  expected: string
-): asserts func is (...args: unknown[]) => unknown {
-  if (typeof func !== 'function') {
-    throw new TypeError(`${expected} is not a function`)
-  }
-}
-
 const defaultWhen: MiddlewarePhase = 'afterReducer'
 const actualMiddlewarePhases = ['beforeReducer', 'afterReducer'] as const
 
-function assertActive(signal: AbortSignal, reason?: string) {
-  if (signal.aborted) {
-    throw new TaskAbortError(reason)
-  }
-}
-
-function createDelay(signal: AbortSignal) {
-  return async function delay(timeoutMs: number): Promise<void> {
-    assertActive(signal)
-    await new Promise((resolve) => setTimeout(resolve, timeoutMs))
-    assertActive(signal)
-  }
-}
-
-function createFork(parentAbortSignal: AbortSignal) {
-  return function fork<T>(childJobExecutor: TaskExecutor<T>): ForkedTask<T> {
+const createFork = (parentAbortSignal: AbortSignal) => {
+  return <T>(taskExecutor: ForkedTaskExecutor<T>): ForkedTask<T> => {
+    assertFunction(taskExecutor, 'taskExecutor')
     const childAbortController = new AbortController()
-    const promise = Outcome.try(async () => {
-      assertActive(parentAbortSignal)
-      const result = await Promise.resolve(childJobExecutor())
-      assertActive(parentAbortSignal)
-      assertActive(childAbortController.signal)
+    const cancel = () => {
+      childAbortController.abort()
+    }
 
+    const result = runTask<T>(async (): Promise<T> => {
+      validateActive(parentAbortSignal)
+      validateActive(childAbortController.signal)
+      const result = (await taskExecutor({
+        pause: createPause(childAbortController.signal),
+        delay: createDelay(childAbortController.signal),
+        signal: childAbortController.signal,
+      })) as T
+      validateActive(parentAbortSignal)
+      validateActive(childAbortController.signal)
       return result
-    })
+    }, cancel)
+
     return {
-      promise,
-      controller: childAbortController,
+      result,
+      cancel,
     }
   }
 }
 
-function createTakePattern<S>(
+const createTakePattern = <S>(
   addListener: AddListenerOverloads<Unsubscribe, S, Dispatch<AnyAction>>,
   signal: AbortSignal
-): TakePattern<S> {
+): 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
    * TaskAbortError.
    */
-  async function take<P extends AnyActionListenerPredicate<S>>(
+  const take = async <P extends AnyActionListenerPredicate<S>>(
     predicate: P,
     timeout: number | undefined
-  ) {
-    assertActive(signal)
+  ) => {
+    validateActive(signal)
 
     // Placeholder unsubscribe function until the listener is added
     let unsubscribe: Unsubscribe = () => {}
 
-    const signalPromise = new Promise<null>((_, reject) => {
-      signal.addEventListener(
-        'abort',
-        () => {
-          reject(new TaskAbortError())
-        },
-        { once: true }
-      )
-    })
-
     const tuplePromise = new Promise<[AnyAction, S, S]>((resolve) => {
       // Inside the Promise, we synchronously add the listener.
       unsubscribe = addListener({
@@ -147,7 +134,7 @@ function createTakePattern<S>(
     })
 
     const promises: (Promise<null> | Promise<[AnyAction, S, S]>)[] = [
-      signalPromise,
+      promisifyAbortSignal(signal),
       tuplePromise,
     ]
 
@@ -160,7 +147,7 @@ function createTakePattern<S>(
     try {
       const output = await Promise.race(promises)
 
-      assertActive(signal)
+      validateActive(signal)
       return output
     } finally {
       // Always clean up the listener
@@ -201,7 +188,7 @@ export const createListenerEntry: TypedCreateListenerEntry<unknown> = (
     listener: options.listener,
     type,
     predicate,
-    taskAbortControllerSet: new Set<AbortController>(),
+    pendingSet: new Set<AbortController>(),
     unsubscribe: () => {
       throw new Error('Unsubscribe not initialized')
     },
@@ -311,9 +298,9 @@ export function createActionListenerMiddleware<
     return entry.unsubscribe
   }
 
-  function findListenerEntry(
+  const findListenerEntry = (
     comparator: (entry: ListenerEntry) => boolean
-  ): ListenerEntry | undefined {
+  ): ListenerEntry | undefined => {
     for (const entry of listenerMap.values()) {
       if (comparator(entry)) {
         return entry
@@ -364,30 +351,33 @@ export function createActionListenerMiddleware<
     return true
   }
 
-  async function notifyListener(
+  const notifyListener = async (
     entry: ListenerEntry<unknown, Dispatch<AnyAction>>,
     action: AnyAction,
     api: MiddlewareAPI,
     getOriginalState: () => S,
     currentPhase: MiddlewarePhase
-  ) {
+  ) => {
     const internalTaskController = new AbortController()
     const take = createTakePattern(addListener, internalTaskController.signal)
     const condition: ConditionFunction<S> = (predicate, timeout) => {
       return take(predicate, timeout).then(Boolean)
     }
     const delay = createDelay(internalTaskController.signal)
     const fork = createFork(internalTaskController.signal)
-
+    const pause: (val: Promise<any>) => Promise<any> = createPause(
+      internalTaskController.signal
+    )
     try {
-      entry.taskAbortControllerSet.add(internalTaskController)
+      entry.pendingSet.add(internalTaskController)
       await Promise.resolve(
         entry.listener(action, {
           ...api,
           getOriginalState,
           condition,
           take,
           delay,
+          pause,
           currentPhase,
           extra,
           signal: internalTaskController.signal,
@@ -397,12 +387,10 @@ export function createActionListenerMiddleware<
             listenerMap.set(entry.id, entry)
           },
           cancelPrevious: () => {
-            entry.taskAbortControllerSet.forEach((controller, _, set) => {
-              if (
-                controller !== internalTaskController &&
-                !controller.signal.aborted
-              ) {
+            entry.pendingSet.forEach((controller, _, set) => {
+              if (controller !== internalTaskController) {
                 controller.abort()
+                set.delete(controller)
               }
             })
           },
@@ -417,7 +405,7 @@ export function createActionListenerMiddleware<
       }
     } finally {
       internalTaskController.abort() // Notify that the task has completed
-      entry.taskAbortControllerSet.delete(internalTaskController)
+      entry.pendingSet.delete(internalTaskController)
     }
   }
 

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

@@ -0,0 +1,96 @@
+import { TaskAbortError } from './exceptions'
+import type { TaskResult } from './types'
+import { noop } from './utils'
+
+/**
+ * Synchronously raises {@link TaskAbortError} if the task tied to the input `signal` has been cancelled.
+ * @param signal
+ * @param reason
+ * @see {TaskAbortError}
+ */
+export const validateActive = (signal: AbortSignal, reason?: string): void => {
+  if (signal.aborted) {
+    throw new TaskAbortError(reason)
+  }
+}
+
+/**
+ * Returns a promise that will reject {@link TaskAbortError} if the task is cancelled.
+ * @param signal
+ * @returns
+ */
+export const promisifyAbortSignal = (
+  signal: AbortSignal,
+  reason?: string
+): Promise<never> => {
+  const promise = new Promise<never>((_, reject) => {
+    const notifyRejection = () => reject(new TaskAbortError(reason))
+
+    if (signal.aborted) {
+      notifyRejection()
+    } else {
+      signal.addEventListener('abort', notifyRejection, { once: true })
+    }
+  })
+
+  // We do not want 'unhandledRejection' warnings or crashes caused by cancelled tasks
+  promise.catch(noop)
+
+  return promise
+}
+
+/**
+ * Runs a task and returns promise that resolves to {@link TaskResult}.
+ *
+ * Second argument is an optional `cleanUp` function that always runs after task.
+ * @returns
+ */
+export const runTask = async <T>(
+  task: () => Promise<T>,
+  cleanUp?: () => void
+): Promise<TaskResult<T>> => {
+  try {
+    await Promise.resolve()
+    const value = await task()
+    return {
+      status: 'ok',
+      value,
+    }
+  } catch (error: any) {
+    return {
+      status: error instanceof TaskAbortError ? 'cancelled' : 'rejected',
+      error,
+    }
+  } finally {
+    cleanUp?.()
+  }
+}
+
+/**
+ * Given an input `AbortSignal` and a promise returns another promise that resolves
+ * as soon the input promise is provided or rejects as soon as
+ * `AbortSignal.abort` is `true`.
+ * @param signal
+ * @returns
+ */
+export const createPause = <T>(signal: AbortSignal) => {
+  return async (promise: Promise<T>): Promise<T> => {
+    validateActive(signal)
+    const result = await Promise.race([promisifyAbortSignal(signal), promise])
+    validateActive(signal)
+    return result
+  }
+}
+
+/**
+ * Given an input `AbortSignal` and `timeoutMs` returns a promise that resolves
+ * after `timeoutMs` or rejects as soon as `AbortSignal.abort` is `true`.
+ * @param signal
+ * @returns
+ */
+export const createDelay = (signal: AbortSignal) => {
+  const pause = createPause<void>(signal)
+  return (timeoutMs: number): Promise<void> => {
+    return pause(new Promise<void>((resolve) => setTimeout(resolve, timeoutMs)))
+  }
+}