import { Observable , Subject , EMPTY , Subscription } from 'rxjs' ; import { catchError , concatMap , map } from 'rxjs/operators' ; import { BlocObserver , EventStreamClosedError , Transition } from '../bloc' ; export type NextFunction < Event , State > = ( value : Event ) => Observable < Transition < Event , State > > ; export abstract class extends Observable < State > template Event template State export abstract class Bloc < Event , State > extends Observable < State > { constructor ( private _state ) { super ( ) ; this . stateSubject = new Subject ( ) ; this . bindStateSubject ( ) ; } private emitted : boolean = false ; private eventSubject = new Subject < Event > ( ) ; private stateSubject : Subject < State > ; private transitionSubscription : Subscription = Subscription . EMPTY ; static type { BlocObserver } memberof static observer : BlocObserver = new BlocObserver ( ) ; readonly type { State } memberof get state ( ) : State { return this . _state ; } param { ( value : State ) => void } onData param { ( ( ( onError : any ) => any ) | undefined ) } onError param { ( ( ( ) => any ) | undefined ) } onDone return { * } memberof listen ( onData , onError ? , onDone ? ) { return this . stateSubject . subscribe ( onData , onError , onDone ) ; } param { Event } event memberof add ( event ) { try { if ( this . eventSubject . isStopped ) { throw new EventStreamClosedError ( ) ; } this . onEvent ( event ) ; this . eventSubject . next ( event ) ; } catch ( error ) { this . onError ( error ) ; } } param { Event } event memberof onEvent ( event ) { Bloc . observer . onEvent ( this , event ) ; } param { Observable < Event > } events param { NextFunction < Event , State > } next return { * } memberof transformEvents ( events , next ) { return events . pipe ( concatMap ( next ) ) ; } abstract param { Event } event return { * } memberof abstract mapEventToState ( event ) ; param { Observable < Transition < Event , State > > } transitions return { * } memberof transformTransitions ( transitions ) { return transitions ; } param { Transition < Event , State > } transition memberof onTransition ( transition ) { Bloc . observer . onTransition ( this , transition ) ; } param { * } error memberof onError ( error ) { Bloc . observer . onError ( this , error ) ; } memberof close ( ) { this . stateSubject . complete ( ) ; this . eventSubject . complete ( ) ; this . transitionSubscription . unsubscribe ( ) ; } private bindStateSubject ( ) { this . transitionSubscription = this . transformTransitions ( this . transformEvents ( this . eventSubject , ( event ) => { return asyncToObservable ( this . mapEventToState ( event ) ) . pipe ( map ( ( nextState , _ ) => { return new Transition ( this . state , event , nextState ) ; } ) , catchError ( error => { this . onError ( error ) ; return EMPTY ; } ) ) ; } ) ) . subscribe ( ( transition ) => { if ( transition . nextState === this . state && this . emitted ) return ; try { this . onTransition ( transition ) ; this . _state = transition . nextState ; this . stateSubject . next ( transition . nextState ) ; } catch ( error ) { this . onError ( error ) ; } this . emitted = true ; } ) ; } } function asyncToObservable < T > ( iterable ) { return new Observable < T > ( observer => void ( async ( ) => { try { for await ( const item of iterable ) { if ( observer . closed ) return ; observer . next ( item ) ; } observer . complete ( ) ; } catch ( e ) { observer . error ( e ) ; } } ) ( ) ) ; }