fix(takeUntil): complete iterable immediately (#315) (#316)

Don't wait for a value to be emitted before performing iterable completeness check

Author: KonanMentor

spec/asynciterable-operators/takeuntil-spec.ts

@@ -1,9 +1,10 @@
 import { hasNext, noNext, delayValue } from '../asynciterablehelpers';
 import { takeUntil } from 'ix/asynciterable/operators';
 import { as } from 'ix/asynciterable';
+import { AsyncSink } from 'ix/asynciterable';
 
 test('AsyncIterable#takeUntil hits', async () => {
-  const xs = async function*() {
+  const xs = async function* () {
     yield await delayValue(1, 100);
     yield await delayValue(2, 300);
     yield await delayValue(3, 1200);
@@ -17,7 +18,7 @@ test('AsyncIterable#takeUntil hits', async () => {
 });
 
 test('AsyncIterable#takeUntil misses', async () => {
-  const xs = async function*() {
+  const xs = async function* () {
     yield await delayValue(1, 100);
     yield await delayValue(2, 300);
     yield await delayValue(3, 600);
@@ -30,3 +31,11 @@ test('AsyncIterable#takeUntil misses', async () => {
   await hasNext(it, 3);
   await noNext(it);
 });
+
+test('AsyncIterable#takeUntil completes immediately', async () => {
+  const sink = new AsyncSink<void>();
+  const ys = as(sink).pipe(takeUntil(() => Promise.resolve()));
+
+  const it = ys[Symbol.asyncIterator]();
+  await noNext(it);
+});

src/asynciterable/operators/takeuntil.ts

@@ -3,6 +3,8 @@ import { MonoTypeOperatorAsyncFunction } from '../../interfaces';
 import { wrapWithAbort } from './withabort';
 import { throwIfAborted } from '../../aborterror';
 
+const DONE_PROMISE_VALUE = undefined;
+
 export class TakeUntilAsyncIterable<TSource> extends AsyncIterableX<TSource> {
   private _source: AsyncIterable<TSource>;
   private _other: (signal?: AbortSignal) => Promise<any>;
@@ -15,13 +17,15 @@ export class TakeUntilAsyncIterable<TSource> extends AsyncIterableX<TSource> {
 
   async *[Symbol.asyncIterator](signal?: AbortSignal) {
     throwIfAborted(signal);
-    let otherDone = false;
-    this._other(signal).then(() => (otherDone = true));
-    for await (const item of wrapWithAbort(this._source, signal)) {
-      if (otherDone) {
+    const donePromise = this._other(signal).then(() => DONE_PROMISE_VALUE);
+    const itemsAsyncIterator = wrapWithAbort(this._source, signal)[Symbol.asyncIterator]();
+    for (;;) {
+      const itemPromise = itemsAsyncIterator.next();
+      const result = await Promise.race([donePromise, itemPromise]);
+      if (result === DONE_PROMISE_VALUE || result.done) {
         break;
       }
-      yield item;
+      yield result.value;
     }
   }
 }