let publisher filter method use generic predicate

Author: yaacovCR

src/execution/execute.ts

@@ -748,15 +748,15 @@ function executeField(
       return completed.then(undefined, (rawError) => {
         const error = locatedError(rawError, fieldNodes, pathToArray(path));
         const handledError = handleFieldError(error, returnType, errors);
-        exeContext.publisher.filterSubsequentPayloads(path, asyncPayloadRecord);
+        filterSubsequentPayloads(exeContext, path, asyncPayloadRecord);
         return handledError;
       });
     }
     return completed;
   } catch (rawError) {
     const error = locatedError(rawError, fieldNodes, pathToArray(path));
     const handledError = handleFieldError(error, returnType, errors);
-    exeContext.publisher.filterSubsequentPayloads(path, asyncPayloadRecord);
+    filterSubsequentPayloads(exeContext, path, asyncPayloadRecord);
     return handledError;
   }
 }
@@ -1188,10 +1188,7 @@ function completeListItemValue(
             pathToArray(itemPath),
           );
           const handledError = handleFieldError(error, itemType, errors);
-          exeContext.publisher.filterSubsequentPayloads(
-            itemPath,
-            asyncPayloadRecord,
-          );
+          filterSubsequentPayloads(exeContext, itemPath, asyncPayloadRecord);
           return handledError;
         }),
       );
@@ -1203,7 +1200,7 @@ function completeListItemValue(
   } catch (rawError) {
     const error = locatedError(rawError, fieldNodes, pathToArray(itemPath));
     const handledError = handleFieldError(error, itemType, errors);
-    exeContext.publisher.filterSubsequentPayloads(itemPath, asyncPayloadRecord);
+    filterSubsequentPayloads(exeContext, itemPath, asyncPayloadRecord);
     completedResults.push(handledError);
   }
 
@@ -1929,10 +1926,7 @@ function executeStreamField(
             itemType,
             asyncPayloadRecord.errors,
           );
-          exeContext.publisher.filterSubsequentPayloads(
-            itemPath,
-            asyncPayloadRecord,
-          );
+          filterSubsequentPayloads(exeContext, itemPath, asyncPayloadRecord);
           return handledError;
         });
       }
@@ -1943,14 +1937,11 @@ function executeStreamField(
         itemType,
         asyncPayloadRecord.errors,
       );
-      exeContext.publisher.filterSubsequentPayloads(
-        itemPath,
-        asyncPayloadRecord,
-      );
+      filterSubsequentPayloads(exeContext, itemPath, asyncPayloadRecord);
     }
   } catch (error) {
     asyncPayloadRecord.errors.push(error);
-    exeContext.publisher.filterSubsequentPayloads(path, asyncPayloadRecord);
+    filterSubsequentPayloads(exeContext, path, asyncPayloadRecord);
     asyncPayloadRecord.addItems(null);
     return asyncPayloadRecord;
   }
@@ -1960,7 +1951,7 @@ function executeStreamField(
       (resolvedItem) => asyncPayloadRecord.addItems([resolvedItem]),
       (error) => {
         asyncPayloadRecord.errors.push(error);
-        exeContext.publisher.filterSubsequentPayloads(path, asyncPayloadRecord);
+        filterSubsequentPayloads(exeContext, path, asyncPayloadRecord);
         asyncPayloadRecord.addItems(null);
       },
     );
@@ -2014,18 +2005,15 @@ async function executeStreamIteratorItem(
           itemType,
           asyncPayloadRecord.errors,
         );
-        exeContext.publisher.filterSubsequentPayloads(
-          itemPath,
-          asyncPayloadRecord,
-        );
+        filterSubsequentPayloads(exeContext, itemPath, asyncPayloadRecord);
         return handledError;
       });
     }
     return { done: false, value: completedItem };
   } catch (rawError) {
     const error = locatedError(rawError, fieldNodes, pathToArray(itemPath));
     const value = handleFieldError(error, itemType, asyncPayloadRecord.errors);
-    exeContext.publisher.filterSubsequentPayloads(itemPath, asyncPayloadRecord);
+    filterSubsequentPayloads(exeContext, itemPath, asyncPayloadRecord);
     return { done: false, value };
   }
 }
@@ -2068,7 +2056,7 @@ async function executeStreamIterator(
       );
     } catch (error) {
       asyncPayloadRecord.errors.push(error);
-      exeContext.publisher.filterSubsequentPayloads(path, asyncPayloadRecord);
+      filterSubsequentPayloads(exeContext, path, asyncPayloadRecord);
       asyncPayloadRecord.addItems(null);
       // entire stream has errored and bubbled upwards
       if (iterator?.return) {
@@ -2086,10 +2074,7 @@ async function executeStreamIterator(
         (resolvedItem) => asyncPayloadRecord.addItems([resolvedItem]),
         (error) => {
           asyncPayloadRecord.errors.push(error);
-          exeContext.publisher.filterSubsequentPayloads(
-            path,
-            asyncPayloadRecord,
-          );
+          filterSubsequentPayloads(exeContext, path, asyncPayloadRecord);
           asyncPayloadRecord.addItems(null);
         },
       );
@@ -2105,6 +2090,28 @@ async function executeStreamIterator(
   }
 }
 
+function filterSubsequentPayloads(
+  exeContext: ExecutionContext,
+  nullPath: Path,
+  currentAsyncRecord: AsyncPayloadRecord | undefined,
+): void {
+  const nullPathArray = pathToArray(nullPath);
+  exeContext.publisher.filter((asyncRecord) => {
+    if (asyncRecord === currentAsyncRecord) {
+      // don't remove payload from where error originates
+      return true;
+    }
+    for (let i = 0; i < nullPathArray.length; i++) {
+      if (asyncRecord.path[i] !== nullPathArray[i]) {
+        // asyncRecord points to a path unaffected by this payload
+        return true;
+      }
+    }
+
+    return false;
+  });
+}
+
 class DeferredFragmentRecord {
   type: 'defer';
   errors: Array<GraphQLError>;

src/execution/publisher.ts

@@ -1,5 +1,3 @@
-import { Path, pathToArray } from '../jsutils/Path.js';
-
 import type {
   AsyncPayloadRecord,
   StreamRecord,
@@ -23,22 +21,11 @@ export class Publisher {
     this.subsequentPayloads.add(payload);
   }
 
-  filterSubsequentPayloads(
-    nullPath: Path,
-    currentAsyncRecord: AsyncPayloadRecord | undefined,
-  ): void {
-    const nullPathArray = pathToArray(nullPath);
+  filter(predicate: (payload: AsyncPayloadRecord) => boolean): void {
     this.subsequentPayloads.forEach((asyncRecord) => {
-      if (asyncRecord === currentAsyncRecord) {
-        // don't remove payload from where error originates
+      if (predicate(asyncRecord)) {
         return;
       }
-      for (let i = 0; i < nullPathArray.length; i++) {
-        if (asyncRecord.path[i] !== nullPathArray[i]) {
-          // asyncRecord points to a path unaffected by this payload
-          return;
-        }
-      }
       // asyncRecord path points to nulled error field
       if (isStreamPayload(asyncRecord) && asyncRecord.iterator?.return) {
         asyncRecord.iterator.return().catch(() => {