import { Channel } from "./Channel" ; import { FlowConsumedError , ObserverError , HasCompletedError , FlowRemoveObserverError , } from "./Errors" ; import { ObserverFunction } from "./ObserverFunction" ; import { Scope } from "./Scope" ; import { CancelFunction , CoroutineFactory , Suspender , Observer , Collector , Coroutine , } from "./Types" ; /**
 * Abstract class for emitting multiple values. A cold flow is doesn't start producing values until
 * it is yield in a coroutine with .collect() or .collectLatest(). Sharing a cold flow with multiple
 * coroutines requires converting it into a SharedEventSubject or SharedStateSubject. Hot flows
 * are EventSubjects and StateSubjects.
 */ export abstract class Flow < T > { param { Observer < T > } observer abstract addObserver ( observer ) ; param { Observer < T > } observer abstract removeObserver ( observer ) ; param { ( value ) => R } mapper returns { Flow < R > } map < R > ( mapper ) { return new MapFlow < T , R > ( this , mapper ) ; } param predicate filter ( predicate ) { return new FilterFlow ( this , predicate ) ; } param binder mergeMap < R > ( binder ) { return new MergeMapFlow ( this , binder ) ; } param f onEach ( f ) { return new OnEachFlow ( this , f ) ; } returns { Flow < T > } sharedState ( ) { return new SharedStateFlow < T > ( this ) ; } returns { Flow < T > } sharedEvent ( ) { return new SharedEventFlow < T > ( this ) ; } param scope launchIn ( scope ) { const that = this ; scope . launch ( function * ( ) { yield that . collect ( ( ) => { } ) ; } ) ; } param { ( value : T ) => void } collector returns { Suspender < void > } collect ( collector ) { return ( resultCallback ) => { const observerFunction = new ObserverFunction ( collector , ( ) => { resultCallback ( { value : undefined } ) ; } , ( error ) => { resultCallback ( { tag : `error` , error } ) ; } ) ; this . addObserver ( observerFunction ) ; return ( ) => { this . removeObserver ( observerFunction ) ; } ; } ; } param { ( value : T ) => CoroutineFactory < void > } coroutineFactory collectLatest ( coroutineFactory ) { return ( resultCallback ) => { let cancelFunction ; const scope = new Scope ( { errorCallback : ( error ) => { resultCallback ( { tag : `error` , error } ) ; } } ) ; const observer = new ObserverFunction < T > ( ( value ) => { if ( cancelFunction !== undefined ) { cancelFunction ( ) ; } cancelFunction = scope . launch ( coroutineFactory ( value ) ) ; } , ( ) => { resultCallback ( { value : undefined } ) ; } , ( error ) => { resultCallback ( { tag : `error` , error } ) ; } ) ; this . addObserver ( observer ) ; return ( ) => { this . removeObserver ( observer ) ; if ( cancelFunction !== undefined ) { cancelFunction ( ) ; } } ; } ; } param transformer transform < R > ( transformer ) { return new TransformFlow ( this , transformer ) ; } param factory catch ( factory ) { return new TransformCatch ( this , factory ) ; } param transformer transformLatest < R > ( transformer ) { return new TransformLatestFlow ( this , transformer ) ; } } class MapFlow < T , R > extends Flow < R > implements Observer < T > { private _observer ? : Observer < R > ; private _hasCompleted = false ; constructor ( private _flow , private _mapper ) { super ( ) ; } addObserver ( observer ) { if ( this . _observer !== undefined ) { throw new FlowConsumedError ( ) ; } this . _observer = observer ; this . _flow . addObserver ( this ) ; } removeObserver ( observer ) { if ( this . _observer !== observer ) { throw new FlowRemoveObserverError ( ) ; } this . _flow . removeObserver ( this ) ; } emit ( value ) { if ( this . _observer === undefined ) { throw new ObserverError ( ) ; } if ( this . _hasCompleted ) { throw new HasCompletedError ( ) ; } this . _observer . emit ( this . _mapper ( value ) ) ; } complete ( ) { if ( this . _observer === undefined ) { throw new ObserverError ( ) ; } if ( this . _hasCompleted ) { throw new HasCompletedError ( ) ; } this . _hasCompleted = true ; this . _observer . complete ( ) ; } error ( error ) { if ( this . _observer === undefined ) { throw new ObserverError ( ) ; } if ( this . _hasCompleted ) { throw new HasCompletedError ( ) ; } this . _hasCompleted = true ; this . _observer . error ( error ) ; } } class FilterFlow < T > extends Flow < T > implements Observer < T > { private _observer ? : Observer < T > ; private _hasCompleted = false ; constructor ( private _flow , private _predicate ) { super ( ) ; } addObserver ( observer ) { if ( this . _observer !== undefined ) { throw new FlowConsumedError ( ) ; } this . _observer = observer ; this . _flow . addObserver ( this ) ; } removeObserver ( observer ) { if ( this . _observer !== observer ) { throw new FlowRemoveObserverError ( ) ; } this . _flow . removeObserver ( this ) ; } emit ( value ) { if ( this . _observer === undefined ) { throw new ObserverError ( ) ; } if ( this . _hasCompleted ) { throw new HasCompletedError ( ) ; } if ( this . _predicate ( value ) ) { this . _observer . emit ( value ) ; } } complete ( ) { if ( this . _observer === undefined ) { throw new ObserverError ( ) ; } if ( this . _hasCompleted ) { throw new HasCompletedError ( ) ; } this . _hasCompleted = true ; this . _observer . complete ( ) ; } error ( error ) { if ( this . _observer === undefined ) { throw new ObserverError ( ) ; } if ( this . _hasCompleted ) { throw new HasCompletedError ( ) ; } this . _hasCompleted = true ; this . _observer . error ( error ) ; } } class MergeMapFlow < T , R > extends Flow < R > implements Observer < T > { private _observer ? : Observer < R > ; private _hasCompleted = false ; constructor ( private _flow , private _binder ) { super ( ) ; } addObserver ( observer ) { if ( this . _observer !== undefined ) { throw new FlowConsumedError ( ) ; } this . _observer = observer ; this . _flow . addObserver ( this ) ; } removeObserver ( observer ) { if ( this . _observer !== observer ) { throw new FlowRemoveObserverError ( ) ; } this . _flow . removeObserver ( this ) ; } private _flows = new Map < ObserverFunction < R > , Flow < R > > ( ) ; emit ( value ) { if ( this . _observer === undefined ) { throw new ObserverError ( ) ; } if ( this . _hasCompleted ) { throw new HasCompletedError ( ) ; } const observerFunction = new ObserverFunction < R > ( ( value ) => { this . _observer ! . emit ( value ) ; } , ( ) => { this . _flows . delete ( observerFunction ) ; if ( this . _hasCompleted && this . _flows . size === 0 ) { this . _observer ! . complete ( ) ; } } , ( error ) => { this . _flows . delete ( observerFunction ) ; this . error ( error ) ; } ) ; const flow = this . _binder ( value ) ; this . _flows . set ( observerFunction , flow ) ; flow . addObserver ( observerFunction ) ; } complete ( ) { if ( this . _observer === undefined ) { throw new ObserverError ( ) ; } if ( this . _hasCompleted ) { throw new HasCompletedError ( ) ; } this . _hasCompleted = true ; } error ( error ) { if ( this . _observer === undefined ) { throw new ObserverError ( ) ; } if ( this . _hasCompleted ) { throw new HasCompletedError ( ) ; } this . _hasCompleted = true ; for ( const [ observerFunction , flow ] of this . _flows ) { flow . removeObserver ( observerFunction ) ; } this . _observer . error ( error ) ; } } class OnEachFlow < T > extends Flow < T > implements Observer < T > { private _observer ? : Observer < T > ; private _hasCompleted = false ; constructor ( private _flow , private _onEach ) { super ( ) ; } addObserver ( observer ) { if ( this . _observer !== undefined ) { throw new FlowConsumedError ( ) ; } this . _observer = observer ; this . _flow . addObserver ( this ) ; } removeObserver ( observer ) { if ( this . _observer !== observer ) { throw new FlowRemoveObserverError ( ) ; } this . _flow . removeObserver ( this ) ; } emit ( value ) { if ( this . _observer === undefined ) { throw new ObserverError ( ) ; } if ( this . _hasCompleted ) { throw new HasCompletedError ( ) ; } this . _onEach ( value ) ; this . _observer . emit ( value ) ; } complete ( ) { if ( this . _observer === undefined ) { throw new ObserverError ( ) ; } if ( this . _hasCompleted ) { throw new HasCompletedError ( ) ; } this . _hasCompleted = true ; this . _observer . complete ( ) ; } error ( error ) { if ( this . _observer === undefined ) { throw new ObserverError ( ) ; } if ( this . _hasCompleted ) { throw new HasCompletedError ( ) ; } this . _hasCompleted = true ; this . _observer . error ( error ) ; } } class TransformFlow < T , R > extends Flow < R > implements Observer < T > { private _scope ? : Scope ; private _observer ? : Observer < R > ; private _receiverChannel ? : Channel < T > ; private _observerFunction ? : ObserverFunction < R > ; private _hasCompleted = false ; constructor ( private _flow , private _transformerFactory ) { super ( ) ; } addObserver ( observer ) { if ( this . _observer !== undefined ) { throw new FlowConsumedError ( ) ; } this . _observer = observer ; this . _scope = new Scope ( { errorCallback : ( error ) => { observer . error ( error ) ; } } ) ; this . _receiverChannel = new Channel < T > ( { bufferSize : Infinity } ) ; this . _observerFunction = new ObserverFunction < R > ( ( value ) => { observer . emit ( value ) ; } , ( ) => { if ( this . _hasCompleted ) { observer . complete ( ) ; } } , ( error ) => { this . error ( error ) ; } ) ; const that = this ; this . _scope . launch ( function * ( ) { while ( ! that . _hasCompleted ) { yield * this . call ( that . _transformerFactory ( yield * this . suspend ( that . _receiverChannel ! . receive ) , that . _observerFunction ! ) ) ; } } ) ; this . _flow . addObserver ( this ) ; } removeObserver ( observer ) { if ( this . _observer !== observer ) { throw new FlowRemoveObserverError ( ) ; } this . _flow . removeObserver ( this ) ; this . _scope ! . cancel ( ) ; } emit ( value ) { if ( this . _receiverChannel === undefined ) { throw new ObserverError ( ) ; } if ( this . _hasCompleted ) { throw new HasCompletedError ( ) ; } this . _receiverChannel . trySend ( value ) ; } complete ( ) { if ( this . _observer === undefined ) { throw new ObserverError ( ) ; } if ( this . _hasCompleted ) { throw new HasCompletedError ( ) ; } this . _hasCompleted = true ; } error ( error ) { if ( this . _observer === undefined ) { throw new ObserverError ( ) ; } if ( this . _hasCompleted ) { throw new HasCompletedError ( ) ; } this . _hasCompleted = true ; this . _scope ! . cancel ( ) ; this . _observer . error ( error ) ; } } class TransformCatch < T > extends Flow < T > implements Observer < T > { private _observer ? : Observer < T > ; private _hasCompleted = false ; private _scope ? : Scope ; constructor ( private _flow , private _transformerFactory ) { super ( ) ; } addObserver ( observer ) { if ( this . _observer !== undefined ) { throw new FlowConsumedError ( ) ; } this . _observer = observer ; this . _flow . addObserver ( this ) ; } removeObserver ( observer ) { if ( this . _observer !== observer ) { throw new FlowRemoveObserverError ( ) ; } this . _flow . removeObserver ( this ) ; this . _scope ?. cancel ( ) ; } emit ( value ) { if ( this . _observer === undefined ) { throw new ObserverError ( ) ; } if ( this . _hasCompleted ) { throw new HasCompletedError ( ) ; } this . _observer . emit ( value ) ; } complete ( ) { if ( this . _observer === undefined ) { throw new ObserverError ( ) ; } if ( this . _hasCompleted ) { throw new HasCompletedError ( ) ; } this . _hasCompleted = true ; this . _observer . complete ( ) ; } error ( error ) { if ( this . _observer === undefined ) { throw new ObserverError ( ) ; } if ( this . _hasCompleted ) { throw new HasCompletedError ( ) ; } this . _hasCompleted = true ; this . _scope = new Scope ( { errorCallback : ( error ) => { this . _observer ?. error ( error ) ; } } ) ; this . _scope . launch ( this . _transformerFactory ( error , this . _observer ) ) ; } } class TransformLatestFlow < T , R > extends Flow < R > { private _observer ? : Observer < R > ; private _coroutine ? : Coroutine < void > ; private _hasCompleted = false ; private _scope = new Scope ( { errorCallback : ( error ) => { this . _observer ! . error ( error ) ; } } ) ; private _cancel ? : CancelFunction ; constructor ( private _flow , private _transformerFactory ) { super ( ) ; } addObserver ( observer ) { if ( this . _observer !== undefined ) { throw new FlowConsumedError ( ) ; } this . _observer = observer ; const downstreamObserver = new ObserverFunction < R > ( ( value ) => { observer . emit ( value ) ; } , ( ) => { if ( this . _hasCompleted ) { observer . complete ( ) ; } } , ( error ) => { this . _scope . _cancelWithError ( error ) ; } ) ; const upstreamObserver = new ObserverFunction < T > ( ( value ) => { if ( this . _coroutine !== undefined ) { this . _scope . _cancelCallbacks . get ( this . _coroutine ) ?. call ( undefined ) ; } this . _coroutine = this . _transformerFactory ( value , downstreamObserver ) . call ( this . _scope ) ; this . _scope . _resume ( this . _coroutine , { value : undefined } ) ; } , ( ) => { this . _hasCompleted = true ; } , ( error ) => { this . _scope . _cancelWithError ( error ) ; } ) ; this . _cancel = ( ) => { this . _flow . removeObserver ( upstreamObserver ) ; if ( this . _coroutine !== undefined ) { this . _scope . _cancelCallbacks . get ( this . _coroutine ) ?. call ( undefined ) ; } } ; this . _flow . addObserver ( upstreamObserver ) ; } removeObserver ( observer ) { if ( this . _observer === observer ) { throw new FlowRemoveObserverError ( ) ; } this . _scope . cancel ( ) ; } } class FlowOf < T > extends Flow < T > { private _observer ? : Observer < T > ; private _cancel ? : CancelFunction ; private _scope = new Scope ( { errorCallback : ( error ) => { this . _observer ! . error ( error ) ; } } ) ; constructor ( private _coroutineFactory ) { super ( ) ; } addObserver ( observer ) { if ( this . _observer !== undefined ) { throw new FlowConsumedError ( ) ; } this . _observer = observer ; const that = this ; this . _cancel = this . _scope . launch ( function * ( ) { yield * this . call ( that . _coroutineFactory ( observer ) ) ; observer . complete ( ) ; } ) ; } removeObserver ( observer ) { if ( this . _observer !== observer ) { throw new FlowRemoveObserverError ( ) ; } this . _cancel ! ( ) ; } } export const flowOf = < T > ( factory ) : Flow < T > => new FlowOf ( factory ) ; class FlowOfValues < T > extends Flow < T > { private readonly _values : Array < T > ; constructor ( ... args ) { super ( ) ; this . _values = args ; } addObserver ( observer ) { for ( const value of this . _values ) { observer . emit ( value ) ; } observer . complete ( ) ; } removeObserver ( ) { } } export const flowOfValues = < T > ( ... args ) : Flow < T > => new FlowOfValues ( ... args ) ; /**
 * Starts observing a upstream flow and shares received values to downstream observers.
 * SharedStateFlow replay the last emitted value to new observers.
 */ export class SharedStateFlow < T > extends Flow < T > implements Observer < T > { private _observers = new Set < Observer < T > > ( ) ; private _last ? : { value : T ; } ; private _hasCompleted = false ; constructor ( private _flow ) { super ( ) ; this . _flow . addObserver ( this ) ; } addObserver ( observer ) { this . _observers . add ( observer ) ; if ( this . _last !== undefined ) { observer . emit ( this . _last . value ) ; } if ( this . _hasCompleted ) { observer . complete ( ) ; } } removeObserver ( observer ) { if ( ! this . _observers . has ( observer ) ) { throw new FlowRemoveObserverError ( ) ; } this . _observers . delete ( observer ) ; } emit ( value ) { if ( this . _hasCompleted ) { throw new HasCompletedError ( ) ; } this . _last = { value } ; for ( const observer of this . _observers ) { observer . emit ( value ) ; } } complete ( ) { if ( this . _hasCompleted ) { throw new HasCompletedError ( ) ; } this . _hasCompleted = true ; for ( const observer of this . _observers ) { observer . complete ( ) ; } } error ( error ) { for ( const observer of this . _observers ) { observer ?. error ( error ) ; } } } /**
 * EventSubjects update their observers when there is a new event. Previously emitted values are not
 * replayed on new observers. To replay the last emitted value, use StateSubject. Subjects are hot
 * and can be shared with multipler observers. New flows that observe subjects start cold.
 */ export class EventSubject < T > extends Flow < T > implements Collector < T > { private _observers : Set < Observer < T > > = new Set ( ) ; addObserver ( observer ) { this . _observers . add ( observer ) ; } removeObserver ( observer ) { if ( ! this . _observers . has ( observer ) ) { throw new FlowRemoveObserverError ( ) ; } this . _observers . delete ( observer ) ; } param value emit ( value ) { for ( const observer of this . _observers ) { observer . emit ( value ) ; } } } /**
 * StateSubject always have a value. When new observers are added, the last emitted value is
 * replayed. This is generally used used for hot observables like the mouse position. Subjects are
 * hot and can be shared with multipler observers. New flows that observe subjects start cold.
 */ export class StateSubject < T > extends Flow < T > implements Collector < T > { private _observers : Set < Observer < T > > = new Set ( ) ; constructor ( public value ) { super ( ) ; } addObserver ( observer ) { this . _observers . add ( observer ) ; observer . emit ( this . value ) ; } removeObserver ( observer ) { if ( ! this . _observers . has ( observer ) ) { throw new FlowRemoveObserverError ( ) ; } this . _observers . delete ( observer ) ; } emit ( value ) { this . value = value ; for ( const observer of this . _observers ) { observer . emit ( value ) ; } } get ( ) { return this . value ; } } /**
 * Starts observing a upstream flow and shares received values to downstream observers.
 * SharedEventFlow doesn't replay any past emitted values.
 */ export class SharedEventFlow < T > extends Flow < T > implements Observer < T > { private _observers = new Set < Observer < T > > ( ) ; private _hasCompleted = false ; constructor ( private _flow ) { super ( ) ; this . _flow . addObserver ( this ) ; } addObserver ( observer ) { this . _observers . add ( observer ) ; } removeObserver ( observer ) { if ( ! this . _observers . has ( observer ) ) { throw new FlowRemoveObserverError ( ) ; } this . _observers . delete ( observer ) ; } emit ( value ) { if ( this . _hasCompleted ) { throw new HasCompletedError ( ) ; } for ( const observer of this . _observers ) { observer . emit ( value ) ; } } complete ( ) { if ( this . _hasCompleted ) { throw new HasCompletedError ( ) ; } this . _hasCompleted = true ; for ( const observer of this . _observers ) { observer . complete ( ) ; } } error ( error ) { for ( const observer of this . _observers ) { observer ?. error ( error ) ; } } }