import { AsyncIterableX } from '../asynciterablex.js'; import { MonoTypeOperatorAsyncFunction } from '../../interfaces.js'; import { wrapWithAbort } from './withabort.js'; import { throwIfAborted } from '../../aborterror.js'; /** @ignore */ export class ExpandAsyncIterable extends AsyncIterableX { private _source: AsyncIterable; private _selector: ( value: TSource, signal?: AbortSignal ) => AsyncIterable | Promise>; constructor( source: AsyncIterable, selector: ( value: TSource, signal?: AbortSignal ) => AsyncIterable | Promise> ) { super(); this._source = source; this._selector = selector; } async *[Symbol.asyncIterator](signal?: AbortSignal) { throwIfAborted(signal); const q = [this._source]; while (q.length > 0) { const src = q.shift(); for await (const item of wrapWithAbort(src!, signal)) { const items = await this._selector(item, signal); q.push(items); yield item; } } } } /** * Expands (breadth first) the async-iterable sequence by recursively applying a selector function to generate more sequences at each recursion level. * * @template TSource Source sequence element type. * @param {(( * value: TSource, * signal?: AbortSignal * ) => AsyncIterable | Promise>)} selector Selector function to retrieve the next sequence to expand. * @returns {MonoTypeOperatorAsyncFunction} An operator which returns a sequence with results * from the recursive expansion of the source sequence. */ export function expand( selector: ( value: TSource, signal?: AbortSignal ) => AsyncIterable | Promise> ): MonoTypeOperatorAsyncFunction { return function expandOperatorFunction(source: AsyncIterable): AsyncIterableX { return new ExpandAsyncIterable(source, selector); }; }