Enable cancelling listeners when unsubscribing

Author: markerikson

docs/api/createListenerMiddleware.mdx

@@ -144,7 +144,7 @@ const store = configureStore({
 Adds a new listener entry to the middleware. Typically used to "statically" add new listeners during application setup.
 
 ```ts no-transpile
-const startListening = (options: AddListenerOptions) => Unsubscribe
+const startListening = (options: AddListenerOptions) => UnsubscribeListener
 
 interface AddListenerOptions {
   // Four options for deciding when the listener will run:
@@ -170,6 +170,12 @@ type ListenerPredicate<Action extends AnyAction, State> = (
   currentState?: State,
   originalState?: State
 ) => boolean
+
+type UnsubscribeListener = (unsuscribeOptions?: UnsubscribeOptions) => void
+
+interface UnsubscribeOptions {
+  cancelActive?: true
+}
 ```
 
 **You must provide exactly _one_ of the four options for deciding when the listener will run: `type`, `actionCreator`, `matcher`, or `predicate`**. Every time an action is dispatched, each listener will be checked to see if it should run based on the current action vs the comparison option provided.
@@ -199,36 +205,54 @@ Note that the `predicate` option actually allows matching solely against state-r
 
 The ["matcher" utility functions included in RTK](./matching-utilities.mdx) are acceptable as either the `matcher` or `predicate` option.
 
-The return value is a standard `unsubscribe()` callback that will remove this listener. If you try to add a listener entry but another entry with this exact function reference already exists, no new entry will be added, and the existing `unsubscribe` method will be returned.
+The return value is an `unsubscribe()` callback that will remove this listener. By default, unsubscribing will _not_ cancel any active instances of the listener. However, you may also pass in `{cancelActive: true}` to cancel running instances.
+
+If you try to add a listener entry but another entry with this exact function reference already exists, no new entry will be added, and the existing `unsubscribe` method will be returned.
 
 The `effect` callback will receive the current action as its first argument, as well as a "listener API" object similar to the "thunk API" object in `createAsyncThunk`.
 
 All listener predicates and callbacks are checked _after_ the root reducer has already processed the action and updated the state. The `listenerApi.getOriginalState()` method can be used to get the state value that existed before the action that triggered this listener was processed.
 
 ### `stopListening`
 
-Removes a given listener. It accepts the same arguments as `startListening()`. It checks for an existing listener entry by comparing the function references of `listener` and the provided `actionCreator/matcher/predicate` function or `type` string.
+Removes a given listener entry.
+
+It accepts the same arguments as `startListening()`. It checks for an existing listener entry by comparing the function references of `listener` and the provided `actionCreator/matcher/predicate` function or `type` string.
+
+By default, this does _not_ cancel any active running instances. However, you may also pass in `{cancelActive: true}` to cancel running instances.
 
 ```ts no-transpile
-const stopListening = (options: AddListenerOptions) => boolean
+const stopListening = (options: AddListenerOptions & UnsubscribeOptions) =>
+  boolean
+
+interface UnsubscribeOptions {
+  cancelActive?: true
+}
 ```
 
 Returns `true` if the `options.effect` listener has been removed, or `false` if no subscription matching the input provided has been found.
 
 ```js
+// Examples:
 // 1) Action type string
-listenerMiddleware.stopListening({ type: 'todos/todoAdded', listener })
+listenerMiddleware.stopListening({
+  type: 'todos/todoAdded',
+  listener,
+  cancelActive: true,
+})
 // 2) RTK action creator
 listenerMiddleware.stopListening({ actionCreator: todoAdded, listener })
 // 3) RTK matcher function
-listenerMiddleware.stopListening({ matcher, listener })
+listenerMiddleware.stopListening({ matcher, listener, cancelActive: true })
 // 4) Listener predicate
 listenerMiddleware.stopListening({ predicate, listener })
 ```
 
 ### `clearListeners`
 
-Removes all current listener entries. This is most likely useful for test scenarios where a single middleware or store instance might be used in multiple tests, as well as some app cleanup situations.
+Removes all current listener entries. It also cancels all active running instances of those listeners as well.
+
+This is most likely useful for test scenarios where a single middleware or store instance might be used in multiple tests, as well as some app cleanup situations.
 
 ```ts no-transpile
 const clearListeners = () => void;
@@ -253,15 +277,19 @@ const unsubscribe = store.dispatch(addListener({ predicate, listener }))
 
 A standard RTK action creator, imported from the package. Dispatching this action tells the middleware to dynamically remove a listener at runtime. Accepts the same arguments as `stopListening()`.
 
+By default, this does _not_ cancel any active running instances. However, you may also pass in `{cancelActive: true}` to cancel running instances.
+
 Returns `true` if the `options.listener` listener has been removed, `false` if no subscription matching the input provided has been found.
 
 ```js
-store.dispatch(removeListener({ predicate, listener }))
+const wasRemoved = store.dispatch(
+  removeListener({ predicate, listener, cancelActive: true })
+)
 ```
 
 ### `clearAllListeners`
 
-A standard RTK action creator, imported from the package. Dispatching this action tells the middleware to dynamically remove all listeners at runtime.
+A standard RTK action creator, imported from the package. Dispatching this action tells the middleware to remove all current listener entries. It also cancels all active running instances of those listeners as well.
 
 ```js
 store.dispatch(clearAllListeners())
@@ -284,7 +312,7 @@ The `listenerApi` object is the second argument to each listener callback. It co
 
 ### Listener Subscription Management
 
-- `unsubscribe: () => void`: removes the listener entry from the middleware, and prevent future instances of the listener from running.
+- `unsubscribe: () => void`: removes the listener entry from the middleware, and prevent future instances of the listener from running. (This does _not_ cancel any active instances.)
 - `subscribe: () => void`: will re-subscribe the listener entry if it was previously removed, or no-op if currently subscribed
 - `cancelActiveListeners: () => void`: cancels all other running instances of this same listener _except_ for the one that made this call. (The cancellation will only have a meaningful effect if the other instances are paused using one of the cancellation-aware APIs like `take/cancel/pause/delay` - see "Cancelation and Task Management" in the "Usage" section for more details)
 - `signal: AbortSignal`: An [`AbortSignal`](https://developer.mozilla.org/en-US/docs/Web/API/AbortSignal) whose `aborted` property will be set to `true` if the listener execution is aborted or completed.

packages/toolkit/src/index.ts

@@ -162,7 +162,7 @@ export type {
   TypedAddListener,
   TypedStopListening,
   TypedRemoveListener,
-  Unsubscribe,
+  UnsubscribeListener,
   ForkedTaskExecutor,
   ForkedTask,
   ForkedTaskAPI,

packages/toolkit/src/listenerMiddleware/index.ts

@@ -14,14 +14,15 @@ import type {
   FallbackAddListenerOptions,
   ListenerEntry,
   ListenerErrorHandler,
-  Unsubscribe,
+  UnsubscribeListener,
   TakePattern,
   ListenerErrorInfo,
   ForkedTaskExecutor,
   ForkedTask,
   TypedRemoveListener,
   TaskResult,
   AbortSignalWithReason,
+  UnsubscribeOptions,
 } from './types'
 import {
   abortControllerWithReason,
@@ -55,7 +56,7 @@ export type {
   TypedAddListener,
   TypedStopListening,
   TypedRemoveListener,
-  Unsubscribe,
+  UnsubscribeListener,
   ForkedTaskExecutor,
   ForkedTask,
   ForkedTaskAPI,
@@ -113,7 +114,11 @@ const createFork = (parentAbortSignal: AbortSignalWithReason<unknown>) => {
 }
 
 const createTakePattern = <S>(
-  startListening: AddListenerOverloads<Unsubscribe, S, Dispatch<AnyAction>>,
+  startListening: AddListenerOverloads<
+    UnsubscribeListener,
+    S,
+    Dispatch<AnyAction>
+  >,
   signal: AbortSignal
 ): TakePattern<S> => {
   /**
@@ -130,7 +135,7 @@ const createTakePattern = <S>(
     validateActive(signal)
 
     // Placeholder unsubscribe function until the listener is added
-    let unsubscribe: Unsubscribe = () => {}
+    let unsubscribe: UnsubscribeListener = () => {}
 
     const tuplePromise = new Promise<[AnyAction, S, S]>((resolve) => {
       // Inside the Promise, we synchronously add the listener.
@@ -223,11 +228,7 @@ const createClearListenerMiddleware = (
   listenerMap: Map<string, ListenerEntry>
 ) => {
   return () => {
-    listenerMap.forEach((entry) => {
-      entry.pending.forEach((controller) => {
-        abortControllerWithReason(controller, listenerCancelled)
-      })
-    })
+    listenerMap.forEach(cancelActiveListeners)
 
     listenerMap.clear()
   }
@@ -257,19 +258,19 @@ const safelyNotifyError = (
 }
 
 /**
- * @alpha
+ * @public
  */
 export const addListener = createAction(
   `${alm}/add`
 ) as TypedAddListener<unknown>
 
 /**
- * @alpha
+ * @public
  */
 export const clearAllListeners = createAction(`${alm}/removeAll`)
 
 /**
- * @alpha
+ * @public
  */
 export const removeListener = createAction(
   `${alm}/remove`
@@ -279,8 +280,16 @@ const defaultErrorHandler: ListenerErrorHandler = (...args: unknown[]) => {
   console.error(`${alm}/error`, ...args)
 }
 
+function cancelActiveListeners(
+  entry: ListenerEntry<unknown, Dispatch<AnyAction>>
+) {
+  entry.pending.forEach((controller) => {
+    abortControllerWithReason(controller, listenerCancelled)
+  })
+}
+
 /**
- * @alpha
+ * @public
  */
 export function createListenerMiddleware<
   S = unknown,
@@ -296,7 +305,12 @@ export function createListenerMiddleware<
     entry.unsubscribe = () => listenerMap.delete(entry!.id)
 
     listenerMap.set(entry.id, entry)
-    return entry.unsubscribe
+    return (cancelOptions?: UnsubscribeOptions) => {
+      entry.unsubscribe()
+      if (cancelOptions?.cancelActive) {
+        cancelActiveListeners(entry)
+      }
+    }
   }
 
   const findListenerEntry = (
@@ -323,7 +337,9 @@ export function createListenerMiddleware<
     return insertEntry(entry)
   }
 
-  const stopListening = (options: FallbackAddListenerOptions): boolean => {
+  const stopListening = (
+    options: FallbackAddListenerOptions & UnsubscribeOptions
+  ): boolean => {
     const { type, effect, predicate } = getListenerEntryPropsFrom(options)
 
     const entry = findListenerEntry((entry) => {
@@ -335,7 +351,12 @@ export function createListenerMiddleware<
       return matchPredicateOrType && entry.effect === effect
     })
 
-    entry?.unsubscribe()
+    if (entry) {
+      entry.unsubscribe()
+      if (options.cancelActive) {
+        cancelActiveListeners(entry)
+      }
+    }
 
     return !!entry
   }

packages/toolkit/src/listenerMiddleware/tests/listenerMiddleware.test.ts

@@ -22,7 +22,7 @@ import type {
   ListenerEffectAPI,
   TypedAddListener,
   TypedStartListening,
-  Unsubscribe,
+  UnsubscribeListener,
   ListenerMiddleware,
 } from '../index'
 import type {
@@ -445,7 +445,7 @@ describe('createListenerMiddleware', () => {
         })
       )
 
-      expectType<Unsubscribe>(unsubscribe)
+      expectType<UnsubscribeListener>(unsubscribe)
 
       store.dispatch(testAction1('a'))
 
@@ -478,6 +478,80 @@ describe('createListenerMiddleware', () => {
       expect(effect.mock.calls).toEqual([[testAction1('a'), middlewareApi]])
     })
 
+    test('can cancel an active listener when unsubscribing directly', async () => {
+      let wasCancelled = false
+      const unsubscribe = startListening({
+        actionCreator: testAction1,
+        effect: async (action, listenerApi) => {
+          try {
+            await listenerApi.condition(testAction2.match)
+          } catch (err) {
+            if (err instanceof TaskAbortError) {
+              wasCancelled = true
+            }
+          }
+        },
+      })
+
+      store.dispatch(testAction1('a'))
+      unsubscribe({ cancelActive: true })
+      expect(wasCancelled).toBe(false)
+      await delay(10)
+      expect(wasCancelled).toBe(true)
+    })
+
+    test('can cancel an active listener when unsubscribing via stopListening', async () => {
+      let wasCancelled = false
+      const effect = async (action: any, listenerApi: any) => {
+        try {
+          await listenerApi.condition(testAction2.match)
+        } catch (err) {
+          if (err instanceof TaskAbortError) {
+            wasCancelled = true
+          }
+        }
+      }
+      startListening({
+        actionCreator: testAction1,
+        effect,
+      })
+
+      store.dispatch(testAction1('a'))
+      stopListening({ actionCreator: testAction1, effect, cancelActive: true })
+      expect(wasCancelled).toBe(false)
+      await delay(10)
+      expect(wasCancelled).toBe(true)
+    })
+
+    test('can cancel an active listener when unsubscribing via removeListener', async () => {
+      let wasCancelled = false
+      const effect = async (action: any, listenerApi: any) => {
+        try {
+          await listenerApi.condition(testAction2.match)
+        } catch (err) {
+          if (err instanceof TaskAbortError) {
+            wasCancelled = true
+          }
+        }
+      }
+      startListening({
+        actionCreator: testAction1,
+        effect,
+      })
+
+      store.dispatch(testAction1('a'))
+      store.dispatch(
+        removeListener({
+          actionCreator: testAction1,
+          effect,
+          cancelActive: true,
+        })
+      )
+      expect(wasCancelled).toBe(false)
+      await delay(10)
+      expect(wasCancelled).toBe(true)
+    })
+
     const addListenerOptions: [
       string,
       Omit<