import { OperatorAsyncFunction } from '../../interfaces' ; import { AsyncIterableX , interval , concat , of } from '../' ; import { map } from './map' ; import { merge } from '../merge' ; import { wrapWithAbort } from './withabort' ; const timerEvent = { } ; const ended = { } ; class BufferCountOrTime < TSource > extends AsyncIterableX < TSource [ ] > { constructor ( private readonly source , private readonly bufferSize , private readonly maxWaitTime ) { super ( ) ; } async * [ Symbol . asyncIterator ] ( signal ? ) { const buffer = [ ] ; const timer = interval ( this . maxWaitTime ) . pipe ( map ( ( ) => timerEvent ) ) ; const source = concat ( this . source , of ( ended ) ) ; const merged = merge ( source , timer ) ; for await ( const item of wrapWithAbort ( merged , signal ) ) { if ( item === ended ) { break ; } if ( item !== timerEvent ) { buffer . push ( item as TSource ) ; } if ( buffer . length >= this . bufferSize || ( buffer . length && item === timerEvent ) ) { yield buffer . slice ( ) ; buffer . length = 0 ; } } if ( buffer . length ) { yield buffer ; } } } template TSource param { number } count param { number } time returns { OperatorAsyncFunction < TSource , TSource [ ] > } export function bufferCountOrTime < TSource > ( count , time ) { return function bufferOperatorFunction ( source ) : AsyncIterableX < TSource [ ] > { return new BufferCountOrTime < TSource > ( source , count , time ) ; } ; }