import { AsyncIterableX } from '../asynciterablex' ; import { OperatorAsyncFunction } from '../../interfaces' ; import { wrapWithAbort } from './withabort' ; import { throwIfAborted } from '../../aborterror' ; interface AsyncResolver < T > { resolve : ( value : T | PromiseLike < T > ) => void ; reject : ( reason ? : any ) => void ; } const WAITING_TYPE = 'waiting' ; const BATCHING_TYPE = 'batching' ; interface WaitingState < T > { type : 'waiting' ; resolver : AsyncResolver < IteratorResult < T [ ] > > ; } interface BatchingState < T > { type : 'batching' ; values : T [ ] ; } type State < T > = WaitingState < T > | BatchingState < T > ; function assertNever ( value ) { throw new Error ( `Unhandled discriminated union member ${ value }` ) ; } export class BatchAsyncIterable < TSource > extends AsyncIterableX < TSource [ ] > { private _source : AsyncIterable < TSource > ; constructor ( source ) { super ( ) ; this . _source = source ; } [ Symbol . asyncIterator ] ( signal ? ) { throwIfAborted ( signal ) ; const it = wrapWithAbort ( this . _source , signal ) [ Symbol . asyncIterator ] ( ) ; let state = { type : BATCHING_TYPE , values : [ ] } ; let ended = null ; let error = null ; function consumeNext ( ) { it . next ( ) . then ( ( res ) => { if ( res . done ) { ended = Promise . resolve ( { done : true } as IteratorResult < TSource [ ] > ) ; if ( state . type === WAITING_TYPE ) { state . resolver . resolve ( ended ) ; } } else { if ( state . type === WAITING_TYPE ) { const { resolve } = state . resolver ; state = { type : BATCHING_TYPE , values : [ ] } ; resolve ( { done : res . done , value : [ res . value ] } ) ; } else if ( state . type === BATCHING_TYPE ) { state . values . push ( res . value ) ; } else { assertNever ( state ) ; } consumeNext ( ) ; } } , ( err ) => { error = err ; if ( state . type === WAITING_TYPE ) { state . resolver . reject ( err ) ; } } ) ; } consumeNext ( ) ; return { next ( ) { if ( error ) { return Promise . reject ( error ) ; } if ( state . type === BATCHING_TYPE && state . values . length > 0 ) { const { values } = state ; state . values = [ ] ; return Promise . resolve ( { done : false , value : values } ) ; } if ( ended ) { return ended ; } if ( state . type === WAITING_TYPE ) { throw new Error ( 'Previous `next()` is still in progress' ) ; } return new Promise < IteratorResult < TSource [ ] > > ( ( resolve , reject ) => { state = { type : WAITING_TYPE , resolver : { resolve , reject } , } ; } ) ; } , return ( value ) { return it . return ? it . return ( value ) . then ( ( ) => ( { done : true } as IteratorResult < TSource [ ] > ) ) : Promise . resolve ( { done : true } as IteratorResult < TSource [ ] > ) ; } , } ; } } template TSource returns { OperatorAsyncFunction < TSource , TSource [ ] > } export function batch < TSource > ( ) { return function batchOperator ( source ) : AsyncIterableX < TSource [ ] > { return new BatchAsyncIterable ( source ) ; } ; }