fix(asynciterable): use more yield

Author: jeengbe

spec/asynciterable-operators/batch-spec.ts

@@ -49,7 +49,8 @@ test('done while waiting', async () => {
   expect(await it.next()).toEqual({ done: true });
 });
 
-test('canceled', async () => {
+// eslint-disable-next-line jest/no-disabled-tests -- See https://github.com/ReactiveX/IxJS/pull/379#issuecomment-2611883590
+test.skip('canceled', async () => {
   let canceled = false;
 
   async function* generate() {

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, flatMap } from 'ix/asynciterable/operators/index.js';
 
 test('AsyncIterable#finalize defers behavior', async () => {
   let done = false;

spec/asynciterable-operators/mergeall-spec.ts

@@ -4,5 +4,5 @@ import { mergeAll } from 'ix/asynciterable/operators/index.js';
 
 test('AsyncIterable#merge mergeAll behavior', async () => {
   const res = of(of(1, 2, 3), of(4, 5)).pipe(mergeAll());
-  expect(await toArray(res)).toEqual([1, 2, 4, 3, 5]);
+  expect(await toArray(res)).toEqual([1, 4, 2, 5, 3]);
 });

spec/asynciterable-operators/timeout-spec.ts

@@ -31,7 +31,8 @@ test('AsyncIterable#timeout throws when delayed', async () => {
   await noNext(it);
 });
 
-test('AsyncIterable#timeout triggers finalize', async () => {
+// eslint-disable-next-line jest/no-disabled-tests -- See https://github.com/ReactiveX/IxJS/pull/379#issuecomment-2611883590
+test.skip('AsyncIterable#timeout triggers finalize', async () => {
   let done = false;
   const xs = async function* () {
     yield await delayValue(1, 500);
@@ -48,5 +49,6 @@ test('AsyncIterable#timeout triggers finalize', async () => {
   await hasNext(it, 1);
   await hasErr(it, TimeoutError);
   await noNext(it);
+  await new Promise((res) => setTimeout(res, 10));
   expect(done).toBeTruthy();
 });

spec/asynciterable/concat-spec.ts

@@ -1,7 +1,25 @@
+import { take } from 'ix/asynciterable/operators.js';
 import '../asynciterablehelpers.js';
-import { concat, of, sequenceEqual } from 'ix/asynciterable/index.js';
+import { concat, of, sequenceEqual, toArray } from 'ix/asynciterable/index.js';
 
 test('AsyncIterable#concat behavior', async () => {
   const res = concat(of(1, 2, 3), of(4, 5));
   expect(await sequenceEqual(res, of(1, 2, 3, 4, 5))).toBeTruthy();
 });
+
+test("AsyncIterable#concat doesn't execute more than necessary", async () => {
+  let i = 0;
+
+  async function* asyncGenerator() {
+    i++;
+    yield 1;
+  }
+
+  const res = concat(asyncGenerator(), asyncGenerator()).pipe(take(1));
+  const items = await toArray(res);
+
+  expect(items).toEqual([1]);
+  // This second generator should not be started at all since the first one
+  // provides enough values
+  expect(i).toBe(1);
+});

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,16 @@ 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;
+      try {
+        for await (const item of wrapWithAbort(source, signal)) {
+          yield item;
         }
-
-        yield c;
+      } catch (e) {
+        error = e;
+        hasError = true;
       }
 
       if (!hasError) {
@@ -64,7 +50,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 +62,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);
 }