import { AsyncIterableX } from '../asynciterablex' ; import { MonoTypeOperatorAsyncFunction } from '../../interfaces' ; import { wrapWithAbort } from './withabort' ; import { throwIfAborted } from '../../aborterror' ; import { safeRace } from '../../util/safeRace' ; const DONE_PROMISE_VALUE = undefined ; export class TakeUntilAsyncIterable < TSource > extends AsyncIterableX < TSource > { private _source : AsyncIterable < TSource > ; private _other : ( signal ? : AbortSignal ) => Promise < any > ; constructor ( source , other ) { super ( ) ; this . _source = source ; this . _other = other ; } async * [ Symbol . asyncIterator ] ( signal ? ) { 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 ; } } } template TSource param { ( signal ? : AbortSignal ) => Promise < any > } other returns { MonoTypeOperatorAsyncFunction < TSource > } export function takeUntil < TSource > ( other ) { return function takeUntilOperatorFunction ( source ) : AsyncIterableX < TSource > { return new TakeUntilAsyncIterable < TSource > ( source , other ) ; } ; }