Observable.ts
184 lines
| 4.8 KiB
| video/mp2t
|
TypeScriptLexer
cin
|
r39 | import { IObservable, IDestroyable, ICancellation } from "./interfaces"; | ||
| import { Cancellation } from "./Cancellation"; | ||||
| import { argumentNotNull } from "./safe"; | ||||
|
cin
|
r23 | |||
|
cin
|
r39 | type Handler<T> = (x: T) => void; | ||
|
cin
|
r23 | |||
|
cin
|
r39 | type Initializer<T> = (notify: Handler<T>, error?: (e: any) => void, complete?: () => void) => void; | ||
|
cin
|
r23 | |||
| // TODO: think about to move this interfaces.ts and make it public | ||||
| interface IObserver<T> { | ||||
|
cin
|
r39 | next(event: T): void; | ||
|
cin
|
r23 | |||
|
cin
|
r39 | error(e: any): void; | ||
|
cin
|
r23 | |||
|
cin
|
r39 | complete(): void; | ||
|
cin
|
r23 | } | ||
|
cin
|
r39 | const noop = () => { }; | ||
|
cin
|
r23 | |||
| export class Observable<T> implements IObservable<T> { | ||||
| private _once = new Array<IObserver<T>>(); | ||||
| private _observers = new Array<IObserver<T>>(); | ||||
|
cin
|
r39 | private _complete: boolean; | ||
|
cin
|
r23 | |||
|
cin
|
r39 | private _error: any; | ||
|
cin
|
r23 | |||
| constructor(func?: Initializer<T>) { | ||||
| if (func) | ||||
| func( | ||||
| this._notifyNext.bind(this), | ||||
| this._notifyError.bind(this), | ||||
| this._notifyCompleted.bind(this) | ||||
| ); | ||||
| } | ||||
| /** | ||||
| * Registers handlers for the current observable object. | ||||
|
cin
|
r39 | * | ||
|
cin
|
r23 | * @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. | ||||
|
cin
|
r39 | * | ||
|
cin
|
r23 | */ | ||
| on(next: Handler<T>, error?: Handler<any>, complete?: () => void): IDestroyable { | ||||
| argumentNotNull(next, "next"); | ||||
|
cin
|
r39 | const me = this; | ||
|
cin
|
r23 | |||
|
cin
|
r39 | const observer: IObserver<T> & IDestroyable = { | ||
| next, | ||||
|
cin
|
r23 | error: error ? error.bind(null) : noop, | ||
| complete: complete ? complete.bind(null) : noop, | ||||
| destroy() { | ||||
| me._removeObserver(this); | ||||
| } | ||||
|
andrei
|
r25 | }; | ||
|
cin
|
r23 | |||
| 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); | ||||
| } | ||||
| } | ||||
| /** | ||||
| * 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. | ||||
|
cin
|
r39 | * | ||
|
cin
|
r23 | * @param ct a cancellation token | ||
| */ | ||||
| next(ct: ICancellation = Cancellation.none): Promise<T> { | ||||
| return new Promise<T>((resolve, reject) => { | ||||
|
cin
|
r39 | const observer: IObserver<T> = { | ||
|
cin
|
r23 | next: resolve, | ||
| error: reject, | ||||
| complete: () => reject("No more events are available") | ||||
| }; | ||||
| if (this._addOnce(observer) && ct.isSupported()) { | ||||
|
cin
|
r39 | ct.register(e => { | ||
|
cin
|
r23 | this._removeOnce(observer); | ||
| reject(e); | ||||
| }); | ||||
| } | ||||
| }); | ||||
| } | ||||
| 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; | ||||
| } | ||||
| protected onObserverException(e: any) { | ||||
| } | ||||
| private _removeOnce(d: IObserver<T>) { | ||||
|
cin
|
r39 | const i = this._once.indexOf(d); | ||
|
cin
|
r23 | if (i >= 0) | ||
| this._once.splice(i, 1); | ||||
| } | ||||
| private _removeObserver(d: IObserver<T>) { | ||||
|
cin
|
r39 | const i = this._observers.indexOf(d); | ||
|
cin
|
r23 | if (i >= 0) | ||
| this._observers.splice(i, 1); | ||||
| } | ||||
| private _notify(guard: (observer: IObserver<T>) => void) { | ||||
|
cin
|
r39 | this._once.forEach(guard); | ||
| this._once = []; | ||||
|
cin
|
r23 | |||
|
cin
|
r39 | this._observers.forEach(guard); | ||
|
cin
|
r23 | } | ||
| protected _notifyNext(evt: T) { | ||||
|
cin
|
r39 | const guard = (observer: IObserver<T>) => { | ||
|
cin
|
r23 | try { | ||
| observer.next(evt); | ||||
| } catch (e) { | ||||
| this.onObserverException(e); | ||||
| } | ||||
|
cin
|
r39 | }; | ||
|
cin
|
r23 | |||
| this._notify(guard); | ||||
| } | ||||
| protected _notifyError(e: any) { | ||||
|
cin
|
r39 | const guard = (observer: IObserver<T>) => { | ||
|
cin
|
r23 | try { | ||
| observer.error(e); | ||||
| } catch (e) { | ||||
| this.onObserverException(e); | ||||
| } | ||||
|
cin
|
r39 | }; | ||
|
cin
|
r23 | |||
| this._notify(guard); | ||||
| this._observers = []; | ||||
|
cin
|
r26 | this._complete = true; | ||
|
cin
|
r23 | } | ||
| protected _notifyCompleted() { | ||||
|
cin
|
r39 | const guard = (observer: IObserver<T>) => { | ||
|
cin
|
r23 | try { | ||
| observer.complete(); | ||||
| } catch (e) { | ||||
| this.onObserverException(e); | ||||
| } | ||||
|
cin
|
r39 | }; | ||
|
cin
|
r23 | |||
| this._notify(guard); | ||||
| this._observers = []; | ||||
|
cin
|
r26 | this._complete = true; | ||
|
cin
|
r23 | } | ||
|
cin
|
r39 | } | ||