import { Stream , Sink , ScheduledTask , Time , Scheduler , Disposable } from '@most/types' ; import { MulticastSource } from '@most/core' ; import { asap , cancelTask } from '@most/scheduler' ; export const hold = < A > ( stream ) : Stream < A > => new Hold ( stream ) ; type HeldValue < A > = { value : A ; } ; class Hold < A > extends MulticastSource < A > implements Stream < A > , Disposable , Sink < A > { private pendingSinks : Sink < A > [ ] = [ ] ; private held ? : HeldValue < A > = undefined ; private task ? : ScheduledTask = undefined ; run ( sink , scheduler ) { if ( this . _shouldScheduleFlush ( ) ) { this . _scheduleFlush ( sink , scheduler ) ; } return super . run ( sink , scheduler ) ; } dispose ( ) { this . _cancelTask ( ) ; return super . dispose ( ) ; } event ( time , value ) { this . flushPending ( time ) ; this . held = { value } ; super . event ( time , value ) ; } end ( time ) { this . flushPending ( time ) ; super . end ( time ) ; } error ( time , err ) { this . flushPending ( time ) ; super . error ( time , err ) ; } flushPending ( time ) { if ( this . pendingSinks . length > 0 && this . held ) { const pendingSinks = this . pendingSinks ; this . pendingSinks = [ ] ; for ( let i = 0 ; i < pendingSinks . length ; ++ i ) { tryEvent ( time , this . held . value , pendingSinks [ i ] ) ; } } } private _hasValue ( ) { return this . held !== undefined ; } private _hasSinks ( ) { return this . sinks . length > 0 ; } private _shouldScheduleFlush ( ) { return this . _hasValue ( ) && this . _hasSinks ( ) ; } private _scheduleFlush ( sink , scheduler ) { this . pendingSinks . push ( sink ) ; if ( this . task ) { cancelTask ( this . task ) ; } this . task = asap ( new HoldTask ( this ) , scheduler ) ; } private _cancelTask ( ) { if ( this . task ) { cancelTask ( this . task ) ; this . task = undefined ; } } } class HoldTask < A > { hold : Hold < A > ; constructor ( hold ) { this . hold = hold ; } run ( t ) { this . hold . flushPending ( t ) ; } error ( t , e ) { this . hold . error ( t , e ) ; } dispose ( ) { } } function tryEvent < A > ( t , x , sink ) { try { sink . event ( t , x ) ; } catch ( e ) { sink . error ( t , e ) ; } }