filter items in the beginning

elastic · Nov 22, 2020 · e9103b7 · e9103b7
1 parent 4bd2a96
commit e9103b7
Showing 1 changed file with 20 additions and 11 deletions.
diff --git a/src/plugins/bfetch/public/batching/create_streaming_batched_function.ts b/src/plugins/bfetch/public/batching/create_streaming_batched_function.ts
@@ -88,16 +88,28 @@ export const createStreamingBatchedFunction = <Payload, Result extends object>(
     },
     onBatch: async (items) => {
       try {
+        // Filter out any items whose signal is already aborted
+        items = items.filter((item) => {
+          if (item.signal?.aborted) item.future.reject(new AbortError());
+          return !item.signal?.aborted;
+        });
+
         const donePromises: Array<Promise<any>> = items.map((item) => {
-          if (!item.signal) return item.future.promise.catch(() => {});
+          return new Promise<void>((resolve) => {
+            const { promise: abortPromise, cleanup } = item.signal
+              ? abortSignalToPromise(item.signal)
+              : {
+                  promise: undefined,
+                  cleanup: () => {},
+                };
 
-          // Reject promise if aborted
-          const { promise: abortPromise, cleanup } = abortSignalToPromise(item.signal);
-          abortPromise.catch(() => {
-            item.future.reject(new AbortError());
-            cleanup();
+            const onDone = () => {
+              resolve();
+              cleanup();
+            };
+            if (abortPromise) abortPromise.catch(onDone);
+            item.future.promise.then(onDone, onDone);
           });
-          return item.future.promise.then(cleanup, cleanup);
         });
 
         // abort when all items were either resolved, rejected or aborted
@@ -107,10 +119,7 @@ export const createStreamingBatchedFunction = <Payload, Result extends object>(
           isBatchDone = true;
           abortController.abort();
         });
-        const batch = items
-          // Filter out any items whose signal is already aborted
-          .filter((item) => !item.signal?.aborted)
-          .map((item) => item.payload);
+        const batch = items.map((item) => item.payload);
 
         const { stream } = fetchStreamingInjected({
           url,