fix(asynciterable): use more yield

Author: jeengbe

spec/asynciterable-operators/finalize-spec.ts

@@ -1,6 +1,6 @@
 import { hasNext, hasErr, noNext } from '../asynciterablehelpers.js';
 import { range, throwError } from 'ix/asynciterable/index.js';
-import { flatMap, finalize, tap } from 'ix/asynciterable/operators/index.js';
+import { finalize, tap, concatMap } from 'ix/asynciterable/operators/index.js';
 
 test('AsyncIterable#finalize defers behavior', async () => {
   let done = false;
@@ -79,7 +79,7 @@ test('AsyncIterable#finalize calls with downstream error from flattening', async
     finalize(async () => {
       done = true;
     }),
-    flatMap(async (x) => {
+    concatMap(async (x) => {
       // srcValues.push(x);
       if (x === 1) {
         return throwError(err);

src/add/asynciterable-operators/mergeall.ts

@@ -8,7 +8,7 @@ export function mergeAllProto<T>(
   this: AsyncIterableX<AsyncIterable<T>>,
   concurrent = Infinity
 ): AsyncIterableX<T> {
-  return mergeAll(concurrent)(this);
+  return mergeAll<T>(concurrent)(this);
 }
 
 AsyncIterableX.prototype.mergeAll = mergeAllProto;

src/asynciterable/_extremaby.ts

@@ -9,29 +9,31 @@ export async function extremaBy<TSource, TKey>(
 ): Promise<TSource[]> {
   throwIfAborted(signal);
 
-  let result = [];
-  const it = wrapWithAbort(source, signal)[Symbol.asyncIterator]();
-  const { value, done } = await it.next();
-  if (done) {
-    throw new Error('Sequence contains no elements');
-  }
-
-  let resKey = await selector(value, signal);
-  result.push(value);
+  let hasValue = false;
+  let key: TKey | undefined;
+  let result: TSource[] = [];
 
-  let next: IteratorResult<TSource>;
-  while (!(next = await it.next()).done) {
-    const current = next.value;
-    const key = await selector(current, signal);
-    const cmp = await comparer(key, resKey, signal);
+  for await (const item of wrapWithAbort(source, signal)) {
+    if (!hasValue) {
+      key = await selector(item, signal);
+      result.push(item);
+      hasValue = true;
+    } else {
+      const currentKey = await selector(item, signal);
+      const cmp = await comparer(currentKey, key as TKey, signal);
 
-    if (cmp === 0) {
-      result.push(current);
-    } else if (cmp > 0) {
-      result = [current];
-      resKey = key;
+      if (cmp === 0) {
+        result.push(item);
+      } else if (cmp > 0) {
+        result = [item];
+        key = currentKey;
+      }
     }
   }
 
+  if (!hasValue) {
+    throw new Error('Sequence contains no elements');
+  }
+
   return result;
 }

src/asynciterable/asynciterablex.ts

@@ -290,8 +290,7 @@ export class FromPromiseIterable<TSource, TResult = TSource> extends AsyncIterab
   }
 
   async *[Symbol.asyncIterator]() {
-    const item = await this._source;
-    yield await this._selector(item, 0);
+    yield await this._selector(await this._source, 0);
   }
 }
 

src/asynciterable/average.ts

@@ -42,9 +42,12 @@ export async function average(
     ['signal']: signal,
     ['thisArg']: thisArg,
   } = options || {};
+
   throwIfAborted(signal);
+
   let sum = 0;
   let count = 0;
+
   for await (const item of wrapWithAbort(source, signal)) {
     sum += await selector.call(thisArg, item, signal);
     count++;

src/asynciterable/catcherror.ts

@@ -1,5 +1,4 @@
 import { AsyncIterableX } from './asynciterablex.js';
-import { returnAsyncIterator } from '../util/returniterator.js';
 import { wrapWithAbort } from './operators/withabort.js';
 import { throwIfAborted } from '../aborterror.js';
 
@@ -19,29 +18,14 @@ export class CatchAllAsyncIterable<TSource> extends AsyncIterableX<TSource> {
     let hasError = false;
 
     for (const source of this._source) {
-      const it = wrapWithAbort(source, signal)[Symbol.asyncIterator]();
-
       error = null;
       hasError = false;
 
-      while (1) {
-        let c = <TSource>{};
-
-        try {
-          const { done, value } = await it.next();
-          if (done) {
-            await returnAsyncIterator(it);
-            break;
-          }
-          c = value;
-        } catch (e) {
-          error = e;
-          hasError = true;
-          await returnAsyncIterator(it);
-          break;
-        }
-
-        yield c;
+      try {
+        yield* wrapWithAbort(source, signal);
+      } catch (e) {
+        error = e;
+        hasError = true;
       }
 
       if (!hasError) {
@@ -64,7 +48,7 @@ export class CatchAllAsyncIterable<TSource> extends AsyncIterableX<TSource> {
  * sequences until a source sequence terminates successfully.
  */
 export function catchAll<T>(source: Iterable<AsyncIterable<T>>): AsyncIterableX<T> {
-  return new CatchAllAsyncIterable<T>(source);
+  return new CatchAllAsyncIterable(source);
 }
 
 /**
@@ -76,5 +60,5 @@ export function catchAll<T>(source: Iterable<AsyncIterable<T>>): AsyncIterableX<
  * sequences until a source sequence terminates successfully.
  */
 export function catchError<T>(...args: AsyncIterable<T>[]): AsyncIterableX<T> {
-  return new CatchAllAsyncIterable<T>(args);
+  return new CatchAllAsyncIterable(args);
 }

src/asynciterable/combinelatest.ts

@@ -3,9 +3,10 @@ import { identity } from '../util/identity.js';
 import { wrapWithAbort } from './operators/withabort.js';
 import { throwIfAborted } from '../aborterror.js';
 import { safeRace } from '../util/safeRace.js';
+import { returnAsyncIterators } from '../util/returniterator.js';
 
 // eslint-disable-next-line @typescript-eslint/no-empty-function
-const NEVER_PROMISE = new Promise(() => {});
+const NEVER_PROMISE = new Promise<never>(() => {});
 
 type MergeResult<T> = { value: T; index: number };
 
@@ -28,39 +29,42 @@ export class CombineLatestAsyncIterable<TSource> extends AsyncIterableX<TSource[
     const length = this._sources.length;
     const iterators = new Array<AsyncIterator<TSource>>(length);
     const nexts = new Array<Promise<MergeResult<IteratorResult<TSource>>>>(length);
-    let hasValueAll = false;
-    const values = new Array<TSource>(length);
-    const hasValues = new Array<boolean>(length);
-    let active = length;
 
-    hasValues.fill(false);
+    let active = length;
+    let allValuesAvailable = false;
+    const values = new Array<TSource>(length);
+    const hasValues = new Array<boolean>(length).fill(false);
 
     for (let i = 0; i < length; i++) {
       const iterator = wrapWithAbort(this._sources[i], signal)[Symbol.asyncIterator]();
       iterators[i] = iterator;
       nexts[i] = wrapPromiseWithIndex(iterator.next(), i);
     }
 
-    while (active > 0) {
-      const next = safeRace(nexts);
-      const {
-        value: { value: value$, done: done$ },
-        index,
-      } = await next;
-      if (done$) {
-        nexts[index] = <Promise<MergeResult<IteratorResult<TSource>>>>NEVER_PROMISE;
-        active--;
-      } else {
-        values[index] = value$;
-        hasValues[index] = true;
+    try {
+      while (active > 0) {
+        const {
+          value: { value, done },
+          index,
+        } = await safeRace(nexts);
+
+        if (done) {
+          nexts[index] = NEVER_PROMISE;
+          active--;
+        } else {
+          values[index] = value;
+          hasValues[index] = true;
+          allValuesAvailable = allValuesAvailable || hasValues.every(identity);
 
-        const iterator$ = iterators[index];
-        nexts[index] = wrapPromiseWithIndex(iterator$.next(), index);
+          nexts[index] = wrapPromiseWithIndex(iterators[index].next(), index);
 
-        if (hasValueAll || (hasValueAll = hasValues.every(identity))) {
-          yield values;
+          if (allValuesAvailable) {
+            yield values;
+          }
         }
       }
+    } finally {
+      await returnAsyncIterators(iterators);
     }
   }
 }
@@ -176,5 +180,5 @@ export function combineLatest<T, T2, T3, T4, T5, T6>(
  */
 export function combineLatest<T>(...sources: AsyncIterable<T>[]): AsyncIterableX<T[]>;
 export function combineLatest<T>(...sources: any[]): AsyncIterableX<T[]> {
-  return new CombineLatestAsyncIterable<T>(sources);
+  return new CombineLatestAsyncIterable(sources);
 }

src/asynciterable/concat.ts

@@ -13,18 +13,17 @@ export class ConcatAsyncIterable<TSource> extends AsyncIterableX<TSource> {
 
   async *[Symbol.asyncIterator](signal?: AbortSignal) {
     throwIfAborted(signal);
+
     for (const outer of this._source) {
-      for await (const item of wrapWithAbort(outer, signal)) {
-        yield item;
-      }
+      yield* wrapWithAbort(outer, signal);
     }
   }
 }
 
 export function _concatAll<TSource>(
   source: Iterable<AsyncIterable<TSource>>
 ): AsyncIterableX<TSource> {
-  return new ConcatAsyncIterable<TSource>(source);
+  return new ConcatAsyncIterable(source);
 }
 
 /**
@@ -136,5 +135,5 @@ export function concat<T, T2, T3, T4, T5, T6>(
  * @returns {AsyncIterableX<T>} An async-iterable sequence that contains the elements of each given sequence, in sequential order.
  */
 export function concat<T>(...args: AsyncIterable<T>[]): AsyncIterableX<T> {
-  return new ConcatAsyncIterable<T>(args);
+  return new ConcatAsyncIterable(args);
 }

src/asynciterable/count.ts

@@ -18,9 +18,10 @@ export async function count<T>(
 ): Promise<number> {
   const { ['signal']: signal, ['thisArg']: thisArg, ['predicate']: predicate = async () => true } =
     options || {};
+
   throwIfAborted(signal);
-  let i = 0;
 
+  let i = 0;
   for await (const item of wrapWithAbort(source, signal)) {
     if (await predicate.call(thisArg, item, i, signal)) {
       i++;

src/asynciterable/create.ts

@@ -11,11 +11,12 @@ class AnonymousAsyncIterable<T> extends AsyncIterableX<T> {
 
   async *[Symbol.asyncIterator](signal?: AbortSignal) {
     throwIfAborted(signal);
+
     const it = await this._fn(signal);
-    let next: IteratorResult<T> | undefined;
-    while (!(next = await it.next()).done) {
-      yield next.value;
-    }
+
+    yield* {
+      [Symbol.asyncIterator]: () => it,
+    };
   }
 }