import { Source, Scheduler, Disposable, Sink } from './interfaces' import { defaultScheduler } from './scheduler/defaultScheduler' export function withDefaultScheduler (f: (x: T) => any, source: Source) { return withScheduler(f, source, defaultScheduler) } export function withScheduler (f: (x: T) => any, source: Source, scheduler: Scheduler): Promise { return new Promise((resolve, reject) => { runSource(f, source, scheduler, resolve, reject) }) } export function runSource (f: (x: T) => any, source: Source, scheduler: Scheduler, end: (x?: T) => any, error: (e: Error) => any) { const disposable = new SettableDisposable() const observer = new Drain(f, end, error, disposable) disposable.setDisposable(source.run(observer, scheduler)) } class SettableDisposable implements Disposable { private disposable: Disposable private disposed: boolean private _resolve: (x: Promise | void) => any private _result: any constructor () { this.disposable = void 0 this.disposed = false this._resolve = void 0 const self = this this._result = new Promise | void>((resolve) => { self._resolve = resolve }) } dispose () { if (this.disposed) return this._result this.disposed = true if (this.disposable) { this._result = this.disposable.dispose() } return this._result } setDisposable (disposable: Disposable) { if (this.disposable !== void 0) { throw new Error('Disposable can only be set one time') } this.disposable = disposable if (this.disposed) { this._resolve(disposable.dispose()) } } } class Drain implements Sink { private active: boolean = true constructor (private _event: (x: T) => any, private _end: (x?: T) => any, private _error: (e: Error) => any, private disposable: Disposable) { } event (time: number, value: T) { if (!this.active) return this._event(value) } error (time: number, err: Error) { if (!this.active) return this.active = false disposeThen(this._error, this._error, this.disposable, err) } end (time: number, value?: T) { if (!this.active) return this.active = false disposeThen(this._error, this._error, this.disposable, value) } } function disposeThen(end: (x: any) => any, error: (x: any) => any, disposable: Disposable, value: T) { Promise.resolve(disposable.dispose()).then(() => { end(value) }, error) }