import { AsyncIterableX } from '../asynciterablex' ; import { OperatorAsyncFunction } from '../../interfaces' ; import { wrapWithAbort } from './withabort' ; import { throwIfAborted } from '../../aborterror' ; export class BufferAsyncIterable < TSource > extends AsyncIterableX < TSource [ ] > { private _source : AsyncIterable < TSource > ; private _count : number ; private _skip : number ; constructor ( source , count , skip ) { super ( ) ; this . _source = source ; this . _count = count ; this . _skip = skip ; } async * [ Symbol . asyncIterator ] ( signal ? ) { throwIfAborted ( signal ) ; const buffers = [ ] ; let i = 0 ; for await ( const item of wrapWithAbort ( this . _source , signal ) ) { if ( i % this . _skip === 0 ) { buffers . push ( [ ] ) ; } for ( const buff of buffers ) { buff . push ( item ) ; } if ( buffers . length > 0 && buffers [ 0 ] . length === this . _count ) { yield buffers . shift ( ) ! ; } i ++ ; } while ( buffers . length > 0 ) { yield buffers . shift ( ) ! ; } } } template TSource param { number } count param { number } skip returns { OperatorAsyncFunction < TSource , TSource [ ] > } export function buffer < TSource > ( count , skip ? ) { let s = skip ; if ( s == null ) { s = count ; } return function bufferOperatorFunction ( source ) : AsyncIterableX < TSource [ ] > { return new BufferAsyncIterable < TSource > ( source , count , s ! ) ; } ; } param count param skip