import * as Rx from 'rxjs' ; export interface Update < Event , State > { event ? : Event ; state : State ; } export type EventInjector < Event > = ( event : Event ) => void ; export type Summarizer < Event , State > = ( prev : State , event : Event , injectEvent : EventInjector < Event > ) => State | undefined ; /**
 * Transform an event stream into a state update stream which emits
 * the events and individual states for each event.
 */ export const summarizeEventStream = < Event , State > ( event$ , initialState , summarize ) => { return new Rx . Observable < Update < Event , State > > ( ( subscriber ) => { const eventBuffer = [ ] ; let processingEventBuffer = false ; let eventStreamComplete = false ; let previousState = initialState ; const injectEvent = ( nextEvent ) => { eventBuffer . push ( nextEvent ) ; if ( processingEventBuffer ) { return ; } try { processingEventBuffer = true ; while ( eventBuffer . length && ! subscriber . closed ) { const event = eventBuffer . shift ( ) ! ; const nextState = summarize ( previousState , event , injectEvent ) ; if ( nextState === undefined ) { continue ; } previousState = nextState ; subscriber . next ( { event , state : nextState , } ) ; } if ( eventStreamComplete ) { subscriber . complete ( ) ; } } catch ( error ) { subscriber . error ( error ) ; } finally { processingEventBuffer = false ; } } ; subscriber . next ( { state : initialState , } ) ; subscriber . add ( event$ . subscribe ( injectEvent , ( error ) => { subscriber . error ( error ) ; } , ( ) => { eventStreamComplete = true ; if ( ! processingEventBuffer && eventBuffer . length === 0 ) { subscriber . complete ( ) ; } } ) ) ; } ) ; } ;