Fix memory memory management for flatMap

prevent the flatMap operator from accumulating in an internal array while concurrency is maxed out
ReactiveX · Oct 7, 2024 · 752157f · 752157f
1 parent b8890f1
commit 752157f
Showing 1 changed file with 5 additions and 1 deletion.
diff --git a/src/asynciterable/operators/_flatten.ts b/src/asynciterable/operators/_flatten.ts
@@ -93,10 +93,12 @@ export class FlattenConcurrentAsyncIterable<TSource, TResult> extends AsyncItera
               }
               if (active < concurrent) {
                 pullNextOuter(value as TSource);
+                results[0] = outer.next();
               } else {
+                // remove the outer iterator from the race, we're full
+                results[0] = NEVER_PROMISE;
                 outerValues.push(value as TSource);
               }
-              results[0] = outer.next();
               break;
             }
             case Type.INNER: {
@@ -119,6 +121,8 @@ export class FlattenConcurrentAsyncIterable<TSource, TResult> extends AsyncItera
             }
             case Type.INNER: {
               --active;
+              // add the outer iterator to the race
+              results[0] = outer.next();
               // return the current slot to the pool
               innerIndices.push(index);
               // synchronously drain the `outerValues` buffer