Observable.ts
197 lines
| 4.9 KiB
| video/mp2t
|
TypeScriptLexer
cin
|
r14 | import { IObservable, IDestroyable, ICancellation } from '../interfaces'; | |
|
cin
|
r13 | import { Cancellation } from '../Cancellation' | |
| import { argumentNotNull } from '../safe'; | |||
|
cin
|
r14 | interface Handler<T> { | |
|
cin
|
r15 | (x: T): void | |
|
cin
|
r14 | } | |
|
cin
|
r13 | ||
|
cin
|
r14 | interface Initializer<T> { | |
|
cin
|
r18 | (notify: Handler<T>, error?: (e: any) => void, complete?: () => void): void; | |
|
cin
|
r14 | } | |
|
cin
|
r15 | // TODO: think about to move this interfaces.ts and make it public | |
| interface IObserver<T> { | |||
| next(event: T): void | |||
| error(e: any): void | |||
| complete(): void | |||
| } | |||
|
cin
|
r13 | ||
|
cin
|
r18 | const noop = () => {}; | |
| class Observable<T> implements IObservable<T> { | |||
|
cin
|
r15 | private _once = new Array<IObserver<T>>(); | |
| private _observers = new Array<IObserver<T>>(); | |||
|
cin
|
r14 | ||
|
cin
|
r13 | ||
|
cin
|
r15 | private _complete: boolean | |
| private _error: any | |||
|
cin
|
r14 | ||
| constructor(func?: Initializer<T>) { | |||
|
cin
|
r18 | if (func) | |
| func( | |||
| this._notifyNext.bind(this), | |||
| this._notifyError.bind(this), | |||
| this._notifyCompleted.bind(this) | |||
| ); | |||
|
cin
|
r13 | } | |
|
cin
|
r15 | /** | |
| * Registers handlers for the current observable object. | |||
| * | |||
| * @param next the handler for events | |||
| * @param error the handler for a error | |||
| * @param complete the handler for a completion | |||
| * @returns {IDestroyable} the handler for the current subscription, this | |||
| * handler can be used to unsubscribe from events. | |||
| * | |||
| */ | |||
| on(next: Handler<T>, error?: Handler<any>, complete?: () => void): IDestroyable { | |||
| argumentNotNull(next, "next"); | |||
|
cin
|
r13 | ||
| let me = this; | |||
|
cin
|
r15 | ||
| let observer: IObserver<T> & IDestroyable = { | |||
| next: next, | |||
|
cin
|
r18 | error: error ? error.bind(null) : noop, | |
| complete: complete ? complete.bind(null) : noop, | |||
|
cin
|
r15 | ||
|
cin
|
r13 | destroy() { | |
|
cin
|
r15 | me._removeObserver(this); | |
|
cin
|
r13 | } | |
| } | |||
|
cin
|
r15 | ||
| this._addObserver(observer); | |||
| return observer; | |||
| } | |||
| private _addObserver(observer: IObserver<T>) { | |||
| if (this._complete) { | |||
| try { | |||
| if (this._error) | |||
| observer.error(this._error); | |||
| else | |||
| observer.complete(); | |||
| } catch (e) { | |||
| this.onObserverException(e); | |||
| } | |||
| } else { | |||
| this._observers.push(observer); | |||
| } | |||
|
cin
|
r13 | } | |
|
cin
|
r15 | /** | |
| * Waits for the next event. This method can't be used to read messages | |||
| * as a sequence since it can skip some messages between calls. | |||
| * | |||
| * @param ct a cancellation token | |||
| */ | |||
|
cin
|
r14 | next(ct: ICancellation = Cancellation.none): Promise<T> { | |
|
cin
|
r13 | return new Promise<T>((resolve, reject) => { | |
|
cin
|
r15 | let observer: IObserver<T> = { | |
| next: resolve, | |||
| error: reject, | |||
| complete: () => reject("No more events are available") | |||
| }; | |||
| if (this._addOnce(observer) && ct.isSupported()) { | |||
|
cin
|
r13 | ct.register((e) => { | |
|
cin
|
r15 | this._removeOnce(observer); | |
|
cin
|
r13 | reject(e); | |
| }); | |||
| } | |||
| }); | |||
| } | |||
|
cin
|
r15 | private _addOnce(observer: IObserver<T>) { | |
| if (this._complete) { | |||
| try { | |||
| if (this._error) | |||
| observer.error(this._error); | |||
| else | |||
| observer.complete(); | |||
| } catch (e) { | |||
| this.onObserverException(e); | |||
| } | |||
| return false; | |||
| } | |||
| this._once.push(observer); | |||
| return true; | |||
| } | |||
|
cin
|
r14 | protected onObserverException(e: any) { | |
| } | |||
|
cin
|
r15 | private _removeOnce(d: IObserver<T>) { | |
|
cin
|
r13 | let i = this._once.indexOf(d); | |
| if (i >= 0) | |||
|
cin
|
r18 | this._once.splice(i, 1); | |
|
cin
|
r13 | } | |
|
cin
|
r15 | private _removeObserver(d: IObserver<T>) { | |
|
cin
|
r13 | let i = this._observers.indexOf(d); | |
| if (i >= 0) | |||
|
cin
|
r18 | this._observers.splice(i, 1); | |
|
cin
|
r13 | } | |
|
cin
|
r15 | private _notify(guard: (observer: IObserver<T>) => void) { | |
|
cin
|
r13 | if (this._once.length) { | |
| for (let i = 0; i < this._once.length; i++) | |||
| guard(this._once[i]); | |||
| this._once = []; | |||
| } | |||
| for (let i = 0; i < this._observers.length; i++) | |||
| guard(this._observers[i]); | |||
| } | |||
|
cin
|
r15 | ||
| protected _notifyNext(evt: T) { | |||
| let guard = (observer: IObserver<T>) => { | |||
| try { | |||
| observer.next(evt); | |||
| } catch (e) { | |||
| this.onObserverException(e); | |||
| } | |||
| } | |||
| this._notify(guard); | |||
| } | |||
| protected _notifyError(e: any) { | |||
| let guard = (observer: IObserver<T>) => { | |||
| try { | |||
| observer.error(e); | |||
| } catch (e) { | |||
| this.onObserverException(e); | |||
| } | |||
| } | |||
| this._notify(guard); | |||
| this._observers = []; | |||
| } | |||
| protected _notifyCompleted() { | |||
| let guard = (observer: IObserver<T>) => { | |||
| try { | |||
| observer.complete(); | |||
| } catch (e) { | |||
| this.onObserverException(e); | |||
| } | |||
| } | |||
| this._notify(guard); | |||
| this._observers = []; | |||
| } | |||
|
cin
|
r13 | } | |
| namespace Observable { | |||
| } | |||
| export = Observable; |