rename PromiseCanceller to AbortSignalListener

and extract the cancellable<*> methods into functions

Author: yaacovCR

src/execution/AbortSignalListener.ts

@@ -1,76 +1,91 @@
 import { promiseWithResolvers } from '../jsutils/promiseWithResolvers.js';
 
 /**
- * A PromiseCanceller object can be used to trigger multiple responses
+ * A AbortSignalListener object can be used to trigger multiple responses
  * in response to a single AbortSignal.
  *
  * @internal
  */
-export class PromiseCanceller {
+export class AbortSignalListener {
   abortSignal: AbortSignal;
   abort: () => void;
 
-  private _aborts: Set<() => void>;
+  private _onAborts: Set<() => void>;
 
   constructor(abortSignal: AbortSignal) {
     this.abortSignal = abortSignal;
-    this._aborts = new Set<() => void>();
+    this._onAborts = new Set<() => void>();
     this.abort = () => {
-      for (const abort of this._aborts) {
+      for (const abort of this._onAborts) {
         abort();
       }
     };
 
     abortSignal.addEventListener('abort', this.abort);
   }
 
-  disconnect(): void {
-    this.abortSignal.removeEventListener('abort', this.abort);
+  add(onAbort: () => void): void {
+    this._onAborts.add(onAbort);
   }
 
-  cancellablePromise<T>(originalPromise: Promise<T>): Promise<T> {
-    if (this.abortSignal.aborted) {
-      // eslint-disable-next-line @typescript-eslint/prefer-promise-reject-errors
-      return Promise.reject(this.abortSignal.reason);
-    }
+  delete(onAbort: () => void): void {
+    this._onAborts.delete(onAbort);
+  }
 
-    const { promise, resolve, reject } = promiseWithResolvers<T>();
-    const abort = () => reject(this.abortSignal.reason);
-    this._aborts.add(abort);
-    originalPromise.then(
-      (resolved) => {
-        this._aborts.delete(abort);
-        resolve(resolved);
-      },
-      (error: unknown) => {
-        this._aborts.delete(abort);
-        reject(error);
-      },
-    );
+  disconnect(): void {
+    this.abortSignal.removeEventListener('abort', this.abort);
+  }
+}
 
-    return promise;
+export function cancellablePromise<T>(
+  originalPromise: Promise<T>,
+  abortSignalListener: AbortSignalListener,
+): Promise<T> {
+  const abortSignal = abortSignalListener.abortSignal;
+  if (abortSignal.aborted) {
+    // eslint-disable-next-line @typescript-eslint/prefer-promise-reject-errors
+    return Promise.reject(abortSignal.reason);
   }
 
-  cancellableIterable<T>(iterable: AsyncIterable<T>): AsyncIterable<T> {
-    const iterator = iterable[Symbol.asyncIterator]();
+  const { promise, resolve, reject } = promiseWithResolvers<T>();
+  const onAbort = () => reject(abortSignal.reason);
+  abortSignalListener.add(onAbort);
+  originalPromise.then(
+    (resolved) => {
+      abortSignalListener.delete(onAbort);
+      resolve(resolved);
+    },
+    (error: unknown) => {
+      abortSignalListener.delete(onAbort);
+      reject(error);
+    },
+  );
 
-    const _next = iterator.next.bind(iterator);
+  return promise;
+}
 
-    if (iterator.return) {
-      const _return = iterator.return.bind(iterator);
+export function cancellableIterable<T>(
+  iterable: AsyncIterable<T>,
+  abortSignalListener: AbortSignalListener,
+): AsyncIterable<T> {
+  const iterator = iterable[Symbol.asyncIterator]();
 
-      return {
-        [Symbol.asyncIterator]: () => ({
-          next: () => this.cancellablePromise(_next()),
-          return: () => this.cancellablePromise(_return()),
-        }),
-      };
-    }
+  const _next = iterator.next.bind(iterator);
+
+  if (iterator.return) {
+    const _return = iterator.return.bind(iterator);
 
     return {
       [Symbol.asyncIterator]: () => ({
-        next: () => this.cancellablePromise(_next()),
+        next: () => cancellablePromise(_next(), abortSignalListener),
+        return: () => cancellablePromise(_return(), abortSignalListener),
       }),
     };
   }
+
+  return {
+    [Symbol.asyncIterator]: () => ({
+      next: () => cancellablePromise(_next(), abortSignalListener),
+    }),
+  };
 }

src/execution/IncrementalPublisher.ts

@@ -4,7 +4,7 @@ import { pathToArray } from '../jsutils/Path.js';
 
 import type { GraphQLError } from '../error/GraphQLError.js';
 
-import type { PromiseCanceller } from './AbortSignalListener.js';
+import type { AbortSignalListener } from './AbortSignalListener.js';
 import { IncrementalGraph } from './IncrementalGraph.js';
 import type {
   CancellableStreamRecord,
@@ -44,7 +44,7 @@ export function buildIncrementalResponse(
 }
 
 interface IncrementalPublisherContext {
-  promiseCanceller: PromiseCanceller | undefined;
+  abortSignalListener: AbortSignalListener | undefined;
   cancellableStreams: Set<CancellableStreamRecord> | undefined;
 }
 
@@ -127,7 +127,7 @@ class IncrementalPublisher {
       IteratorResult<SubsequentIncrementalExecutionResult, void>
     > => {
       if (isDone) {
-        this._context.promiseCanceller?.disconnect();
+        this._context.abortSignalListener?.disconnect();
         await this._returnAsyncIteratorsIgnoringErrors();
         return { value: undefined, done: true };
       }
@@ -176,7 +176,7 @@ class IncrementalPublisher {
 
       // TODO: add test for this case
       /* c8 ignore next */
-      this._context.promiseCanceller?.disconnect();
+      this._context.abortSignalListener?.disconnect();
       await this._returnAsyncIteratorsIgnoringErrors();
       return { value: undefined, done: true };
     };