import { AsyncIterableX } from '../asynciterablex.js'; import { MonoTypeOperatorAsyncFunction } from '../../interfaces.js'; import { wrapWithAbort } from './withabort.js'; import { throwIfAborted } from '../../aborterror.js'; import { safeRace } from '../../util/safeRace.js'; const DONE_PROMISE_VALUE = undefined; /** @ignore */ export class TakeUntilAsyncIterable extends AsyncIterableX { private _source: AsyncIterable; private _other: (signal?: AbortSignal) => Promise; constructor(source: AsyncIterable, other: (signal?: AbortSignal) => Promise) { super(); this._source = source; this._other = other; } async *[Symbol.asyncIterator](signal?: AbortSignal) { throwIfAborted(signal); 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 safeRace([donePromise, itemPromise]); if (result === DONE_PROMISE_VALUE || result.done) { break; } yield result.value; } } } /** * Returns the elements from the source async-iterable sequence until the other function * that returns a promise produces an element. * * @template TSource The type of the elements in the source sequence. * @param {(signal?: AbortSignal) => Promise} other A function that terminates the propagation of * elements in the source sequence. * @returns {MonoTypeOperatorAsyncFunction} An async-iterable sequence containing the elements of the * source sequence up to the point the other function which returns a promise interrupted further propagation. */ export function takeUntil( other: (signal?: AbortSignal) => Promise ): MonoTypeOperatorAsyncFunction { return function takeUntilOperatorFunction( source: AsyncIterable ): AsyncIterableX { return new TakeUntilAsyncIterable(source, other); }; }