feat(alm): add alm.clear() method

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

@@ -14,8 +14,6 @@ import type {
   BaseActionCreator,
   AnyActionListenerPredicate,
   CreateListenerMiddlewareOptions,
-  ConditionFunction,
-  ListenerPredicate,
   TypedActionCreator,
   TypedAddListener,
   TypedAddListenerAction,
@@ -200,6 +198,20 @@ export const createListenerEntry: TypedCreateListenerEntry<unknown> = (
   return entry
 }
 
+const createClearListenerMiddleware = (
+  listenerMap: Map<string, ListenerEntry>
+) => {
+  return () => {
+    listenerMap.forEach((entry) => {
+      entry.pending.forEach((controller) => {
+        controller.abort()
+      })
+    })
+
+    listenerMap.clear()
+  }
+}
+
 /**
  * Safely reports errors to the `errorHandler` provided.
  * Errors that occur inside `errorHandler` are notified in a new task.
@@ -240,6 +252,11 @@ export const addListenerAction = createAction(
   }
 ) as TypedAddListenerAction<unknown>
 
+/**
+ * @alpha
+ */
+export const clearListenerMiddlewareAction = createAction(`${alm}/clear`)
+
 /**
  * @alpha
  */
@@ -412,6 +429,8 @@ export function createActionListenerMiddleware<
     }
   }
 
+  const clearListenerMiddleware = createClearListenerMiddleware(listenerMap)
+
   const middleware: Middleware<
     {
       (action: Action<`${typeof alm}/add`>): Unsubscribe
@@ -430,6 +449,12 @@ export function createActionListenerMiddleware<
 
       return insertEntry(entry)
     }
+
+    if (clearListenerMiddlewareAction.match(action)) {
+      clearListenerMiddleware()
+      return
+    }
+
     if (removeListenerAction.match(action)) {
       removeListener(action.payload.type, action.payload.listener)
       return
@@ -491,6 +516,7 @@ export function createActionListenerMiddleware<
     {
       addListener,
       removeListener,
+      clear: clearListenerMiddleware,
       addListenerAction: addListenerAction as TypedAddListenerAction<S>,
     },
     {} as WithMiddlewareType<typeof middleware>

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

@@ -14,6 +14,7 @@ import {
   addListenerAction,
   removeListenerAction,
   TaskAbortError,
+  clearListenerMiddlewareAction,
 } from '../index'
 
 import type {
@@ -492,6 +493,109 @@ describe('createActionListenerMiddleware', () => {
     })
   })
 
+  describe('clear listeners', () => {
+    test('dispatch(clearListenerAction()) cancels running listeners and removes all subscriptions', async () => {
+      const listener1Test = deferred()
+      let listener1Calls = 0
+      let listener2Calls = 0
+      let listener3Calls = 0
+
+      middleware.addListener({
+        actionCreator: testAction1,
+        async listener(_, listenerApi) {
+          listener1Calls++
+          listenerApi.signal.addEventListener(
+            'abort',
+            () => listener1Test.resolve(listener1Calls),
+            { once: true }
+          )
+          await listenerApi.condition(() => true)
+          listener1Test.reject(new Error('unreachable: listener1Test'))
+        },
+      })
+
+      middleware.addListener({
+        actionCreator: clearListenerMiddlewareAction,
+        listener() {
+          listener2Calls++
+        },
+      })
+
+      middleware.addListener({
+        predicate: () => true,
+        listener() {
+          listener3Calls++
+        },
+      })
+
+      store.dispatch(testAction1('a'))
+      store.dispatch(clearListenerMiddlewareAction())
+      store.dispatch(testAction1('b'))
+      expect(await listener1Test).toBe(1)
+      expect(listener1Calls).toBe(1)
+      expect(listener3Calls).toBe(1)
+      expect(listener2Calls).toBe(0)
+    })
+
+    test('clear() cancels running listeners and removes all subscriptions', async () => {
+      const listener1Test = deferred()
+
+      let listener1Calls = 0
+      let listener2Calls = 0
+
+      middleware.addListener({
+        actionCreator: testAction1,
+        async listener(_, listenerApi) {
+          listener1Calls++
+          listenerApi.signal.addEventListener(
+            'abort',
+            () => listener1Test.resolve(listener1Calls),
+            { once: true }
+          )
+          await listenerApi.condition(() => true)
+          listener1Test.reject(new Error('unreachable: listener1Test'))
+        },
+      })
+
+      middleware.addListener({
+        actionCreator: testAction2,
+        listener() {
+          listener2Calls++
+        },
+      })
+
+      store.dispatch(testAction1('a'))
+
+      middleware.clear()
+      store.dispatch(testAction1('b'))
+      store.dispatch(testAction2('c'))
+
+      expect(listener2Calls).toBe(0)
+      expect(await listener1Test).toBe(1)
+    })
+
+    test('clear() cancels all running forked tasks', async () => {
+      const fork1Test = deferred()
+
+      middleware.addListener({
+        actionCreator: testAction1,
+        async listener(_, { fork }) {
+          const taskResult = await fork(() => {
+            return 3
+          }).result
+          fork1Test.resolve(taskResult)
+        },
+      })
+
+      store.dispatch(testAction1('a'))
+
+      middleware.clear()
+      store.dispatch(testAction1('b'))
+
+      expect(await fork1Test).toHaveProperty('status', 'cancelled')
+    })
+  })
+
   describe('Listener API', () => {
     test('Passes both getState and getOriginalState in the API', () => {
       const store = configureStore({

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

@@ -218,6 +218,10 @@ export type ActionListenerMiddleware<
   addListener: AddListenerOverloads<Unsubscribe, S, D>
   removeListener: RemoveListenerOverloads<S, D>
   addListenerAction: TypedAddListenerAction<S, D>
+  /**
+   * Unsubscribes all listeners, cancels running listeners and tasks.
+   */
+  clear: () => void
 }
 
 /**