import { AsyncIterableInput , AsyncIterableX } from './asynciterablex' ; import { identityAsync } from '../util/identity' ; import { bindCallback } from '../util/bindcallback' ; import { isIterable , isAsyncIterable , isArrayLike , isIterator , isPromise , isObservable } from '../util/isiterable' ; import { Observable } from '../observer' ; import { toLength } from '../util/tolength' ; import { AsyncSink } from './asyncsink' ; import { AbortError , throwIfAborted } from '../aborterror' ; export let from ; export let FromArrayIterable ; export let FromAsyncIterable ; export let FromPromiseIterable ; export let FromObservableAsyncIterable ; export function _initialize ( Ctor ) { nocollapse from = function < TSource , TResult = TSource > ( source , selector = identityAsync , thisArg ? ) : AsyncIterableX < TResult > { const fn = bindCallback ( selector , thisArg , 2 ) ; if ( isIterable ( source ) || isAsyncIterable ( source ) ) { return new FromAsyncIterable < TSource , TResult > ( source , fn ) ; } if ( isPromise ( source ) ) { return new FromPromiseIterable < TSource , TResult > ( source , fn ) ; } if ( isObservable ( source ) ) { return new FromObservableAsyncIterable < TSource , TResult > ( source , fn ) ; } if ( isArrayLike ( source ) ) { return new FromArrayIterable < TSource , TResult > ( source , fn ) ; } if ( isIterator ( source ) ) { return new FromAsyncIterable < TSource , TResult > ( { [ Symbol . asyncIterator ] : ( ) => source } , fn ) ; } throw new TypeError ( 'Input type not supported' ) ; } ; FromArrayIterable = class FromArrayIterable < TSource , TResult = TSource > extends Ctor < TResult > { private _source : ArrayLike < TSource > ; private _selector : ( value : TSource , index : number ) => TResult | Promise < TResult > ; constructor ( source , selector ) { super ( ) ; this . _source = source ; this . _selector = selector ; } async * [ Symbol . asyncIterator ] ( ) { let i = 0 ; const length = toLength ( ( < ArrayLike < TSource > > this . _source ) . length ) ; while ( i < length ) { yield await this . _selector ( this . _source [ i ] , i ++ ) ; } } } ; FromAsyncIterable = class FromAsyncIterable < TSource , TResult = TSource > extends Ctor < TResult > { private _source : Iterable < TSource | PromiseLike < TSource > > | AsyncIterable < TSource > ; private _selector : ( value : TSource , index : number ) => TResult | Promise < TResult > ; constructor ( source , selector ) { super ( ) ; this . _source = source ; this . _selector = selector ; } async * [ Symbol . asyncIterator ] ( ) { let i = 0 ; for await ( const item of < AsyncIterable < TSource > > this . _source ) { yield await this . _selector ( item , i ++ ) ; } } } ; FromPromiseIterable = class FromPromiseIterable < TSource , TResult = TSource > extends Ctor < TResult > { private _source : PromiseLike < TSource > ; private _selector : ( value : TSource , index : number ) => TResult | Promise < TResult > ; constructor ( source , selector ) { super ( ) ; this . _source = source ; this . _selector = selector ; } async * [ Symbol . asyncIterator ] ( ) { const item = await this . _source ; yield await this . _selector ( item , 0 ) ; } } ; FromObservableAsyncIterable = class FromObservableAsyncIterable < TSource , TResult = TSource > extends Ctor < TResult > { private _observable : Observable < TSource > ; private _selector : ( value : TSource , index : number ) => TResult | Promise < TResult > ; constructor ( observable , selector ) { super ( ) ; this . _observable = observable ; this . _selector = selector ; } async * [ Symbol . asyncIterator ] ( signal ? ) { throwIfAborted ( signal ) ; const sink = new AsyncSink < TSource > ( ) ; const subscription = this . _observable . subscribe ( { next ( value ) { sink . write ( value ) ; } , error ( err ) { sink . error ( err ) ; } , complete ( ) { sink . end ( ) ; } } ) ; function onAbort ( ) { sink . error ( new AbortError ( ) ) ; } if ( signal ) { signal . addEventListener ( 'abort' , onAbort ) ; } let i = 0 ; try { for ( let next ; ! ( next = await sink . next ( ) ) . done ; ) { throwIfAborted ( signal ) ; yield await this . _selector ( next . value ! , i ++ ) ; } } finally { if ( signal ) { signal . removeEventListener ( 'abort' , onAbort ) ; } subscription . unsubscribe ( ) ; } } } ; }