Observable
Observable是一个类，是多个值的推送集合。
Observable类通过静态方法create创建Observable实例
用法
const observable = Observable.create(function (observer) {
observer.next(1);
observer.next(2);
observer.next(3);
setTimeout(() => {
observer.next(4);
observer.complete();
}, 1000);
});
源码
/**
* 创建一个Observable实例
* @static true
* @owner Observable
* @method create
* @param {Function} subscribe函数，函数参数为observer对象
* @return {Observable} 返回一个Observable实例
**/
static create: Function = <T>(subscribe?: (subscriber: Subscriber<T>) => TeardownLogic) => {
return new Observable<T>(subscribe);
}
Observable实例通过subscribe方法订阅事件，返回Subscriber实例
用法
observable.subscribe({
next: x => console.log('got value ' + x),
error: err => console.error('something wrong occurred: ' + err),
complete: () => console.log('done'),
});
observable.subscribe(function (x) {
console.log(x);
});
源码
/**
* 参数为observer对象或者next函数，返回Subscriber实例subscription（执行subscription.unsubscribe()订阅取消）
**/
subscribe(observerOrNext?: PartialObserver<T> | ((value: T) => void),
error?: (error: any) => void,
complete?: () => void): Subscription {
const { operator } = this;//operator操作对象，包含map，scan等方法
const sink = toSubscriber(observerOrNext, error, complete);//生成Subscriber实例
if (operator) {//如果有operator，执行operator，source指Observable实例
operator.call(sink, this.source);
} else {//如果没有，订阅Subscriber实例
sink.add(
this.source || (config.useDeprecatedSynchronousErrorHandling && !sink.syncErrorThrowable) ?
this._subscribe(sink) :
this._trySubscribe(sink)
);
}
if (config.useDeprecatedSynchronousErrorHandling) {
if (sink.syncErrorThrowable) {
sink.syncErrorThrowable = false;
if (sink.syncErrorThrown) {
throw sink.syncErrorValue;
}
}
}
return sink;
}
_subscribe(subscriber: Subscriber<any>): TeardownLogic {
const { source } = this;
return source && source.subscribe(subscriber);//订阅
}
Observable实例通过forEach方法也可订阅事件，返回一个Promise实例
还有另外一个函式可以达到跟subscribe一样的结果，forEach只接受一个函式，这个函式只负责处理next阶段的行为，且返回的是一个Promise实例，而不是 subscription。返回Promise实例的好处是方便使用await处理异步，例如
//输出1,2,3，finish，不加await会先输出finish
async function execute() {
await Observable.from([1, 2, 3]).delay(1000).forEach(v => console.log(v));
console.log('finish');
}
源码
/**
* @method forEach
* @param {Function} next a handler for each value emitted by the observable
* @param {PromiseConstructor} [promiseCtor] a constructor function used to instantiate the Promise
* @return {Promise} a promise that either resolves on observable completion or
* rejects with the handled error
*/
forEach(next: (value: T) => void, promiseCtor?: PromiseConstructorLike): Promise<void> {
promiseCtor = getPromiseCtor(promiseCtor);
//返回一个Promise实例
return new promiseCtor<void>((resolve, reject) => {
// Must be declared in a separate statement to avoid a RefernceError when
// accessing subscription below in the closure due to Temporal Dead Zone.
let subscription: Subscription;
subscription = this.subscribe((value) => {
try {
next(value);//执行next方法
} catch (err) {
reject(err);
if (subscription) {
subscription.unsubscribe();
}
}
}, reject, resolve);
}) as Promise<void>;
}
Observable实例通过toPromise将实例转为promise
用法
Observable.of('foo').toPromise().then(res => console.log(res));
源码
toPromise(promiseCtor?: PromiseConstructorLike): Promise<T> {
promiseCtor = getPromiseCtor(promiseCtor);
//执行订阅，并返回一个promise
return new promiseCtor((resolve, reject) => {
let value: any;
this.subscribe((x: T) => value = x, (err: any) => reject(err), () => resolve(value));
}) as Promise<T>;
}
Observable实例通过pipe来组合操作符，返回observable实例
用法
range(0, 10).pipe(
filter(x => x % 2 === 0),
map(x => x + x),
scan((acc, x) => acc + x, 0)
)
.subscribe(x => console.log(x))
源码
function pipeFromArray<T, R>(fns: Array<UnaryFunction<T, R>>): UnaryFunction<T, R> {
if (!fns) {
return noop as UnaryFunction<any, any>;
}
if (fns.length === 1) {
return fns[0];
}
return function piped(input: T): R {// 通过reduce来组合操作符
return fns.reduce((prev: any, fn: UnaryFunction<T, R>) => fn(prev), input as any);
};
}
pipe(...operations: OperatorFunction<any, any>[]): Observable<any> {
if (operations.length === 0) {
return this as any;
}
//返回observable实例
return pipeFromArray(operations)(this);
}
