DevUI是一支兼具设计视角和工程视角的团队，服务于华为云 DevCloud平台和华为内部数个中后台系统，服务于设计师和前端工程师。
官方网站： devui.design
Ng组件库： ng-devui（欢迎Star）
官方交流群：添加DevUI小助手（微信号：devui-official）进群
DevUIHelper插件：DevUIHelper-LSP（欢迎Star）
本篇是 RxJS 源码解析的第三篇文章，使用源码的版本是 6.6.0 。本篇文章的内容可能会比较多，请耐心阅读。为了方便阅读，文中的相关代码均通过裁剪和处理。若有不妥，还请指正。前端
在 RxJS 中，Creation Operator 主要分为如下两类：git
在 pipe 中使用的 operator ，我称之为 Pipe Operator ，它主要分为如下几类：github
限于篇幅本篇将先介绍 Normal Creation Operator ，它的主要做用是帮助开发者快速建立 Observable。typescript
of 、empty 、throwError ，首先讲这三个 operator 的重要缘由是，它提供了一系列基础的操做：next、complete、以及 error。json
const observableA = of(1);
const observableB = empty();
const observableC = throwError(Error('test'));
observableA.subscribe({
next: (v) => console.log('A: ' + v),
complete: () => console.log('A: complete');
error: (e) => console.log('A: error is ' + e);
});
observableB.subscribe({
next: (v) => console.log('B: ' + v),
complete: () => console.log('B: complete');
error: (e) => console.log('B: error is ' + e);
});
observableC.subscribe({
next: (v) => console.log('C: ' + v),
complete: () => console.log('C: complete'),
error: (e) => console.log(`C: error is (${e}).`),
});
// 打印结果
// A: 1
// A: complete
// B: complete
// C: error is Error: test
它的构建方式以下，其中，调度器是最后一个参数。segmentfault
export function of<T>(...args: Array<T | SchedulerLike>): Observable<T> {
let scheduler = args[args.length - 1] as SchedulerLike;
if (isScheduler(scheduler)) {
args.pop();
return scheduleArray(args as T[], scheduler);
} else {
return fromArray(args as T[]);
}
}
of 由两个函数 fromArray 和 scheduleArray。fromArray 是一个简单循环的函数，它将数据循环发送给 Observable 的订阅者。数组
export function fromArray<T>(input: ArrayLike<T>) {
return new Observable<T>((subscriber: Subscriber<T>) => {
// 循环获取数据
for (let i = 0, len = array.length; i < len && !subscriber.closed; i++) {
subscriber.next(array[i]);
}
subscriber.complete();
});
}
这部分的代码很简单，scheduler 部分能够忽略。实际上就是在 Observable 中调用 subscriber.complete()。promise
export function empty(scheduler?: SchedulerLike) {
if (scheduler) {
return new Observable<never>(subscriber => scheduler.schedule(
() => subscriber.complete()
));
} else {
return new Observable<never>(subscriber => subscriber.complete());
}
}
throwError 跟 empty 的实现是一致的，只不过 complete 换成了 error 。浏览器
export function throwError(error: any, scheduler?: SchedulerLike): Observable<never> {
if (!scheduler) {
return new Observable(subscriber => subscriber.error(error));
} else {
return new Observable(subscriber => scheduler.schedule(
dispatch,
0,
{ error, subscriber }
));
}
}
// 如下是 调度器中想要执行的状态。
interface DispatchArg {
error: any;
subscriber: Subscriber<any>;
}
// 最终执行的是 subcriber 的 error 方法。
function dispatch({ error, subscriber }: DispatchArg) {
subscriber.error(error);
}
iif 和 defer 的表现是一致的。微信
let test = false;
const observableA = iif(
() => test,
of('1'),
of('2'),
);
const observableB = defer(function() {
return test ? of('1') : of('2');
});
看到 iif 的源码的那一刻我震惊了，什么叫大道至简（战术后仰）。
export function iif<T = never, F = never>(
condition: () => boolean,
trueResult: SubscribableOrPromise<T> = EMPTY,
falseResult: SubscribableOrPromise<F> = EMPTY
): Observable<T|F> {
// 直接调用了 defer
return defer(() => condition() ? trueResult : falseResult);
}
defer 原理上比较简单：在构造 Observable 的时候，在传入的订阅函数中返回一个 Subscription。那么在这个传入的订阅函数中，defer 的过程分为如下三步：
export function defer<R extends ObservableInput<any> | void>(observableFactory: () => R): Observable<ObservedValueOf<R>> {
return new Observable<ObservedValueOf<R>>(subscriber => {
let input: R | void;
try {
// 调用工厂函数，获取输入的数据。
input = observableFactory();
} catch (err) {
subscriber.error(err);
return undefined;
}
// 经过 from 将 input 转换为 observable。
const source = input ? from(input as ObservableInput<ObservedValueOf<R>>) : empty();
// 返回一个订阅器到外部。
return source.subscribe(subscriber);
});
}
其中的 ObservedValueOf 是这样定义的，使用了 ts 的 infer 来推导出 ObservableInput<T> 中 T 的具体类型。
export type ObservedValueOf<OV> = OV extends ObservableInput<infer T> ? T : never;
from 提供了一种映射的功能，能够将传入的数据映射成 Observables 。它能够接受如下参数：
Iterable<T> Promise<T>Observable<T>稍微的修剪了一下，源码以下：
export function from<T>(input: ObservableInput<T>): Observable<T> {
return new Observable<T>(subscribeTo(input));
}
它直接建立一个新的 Observable，而且调用了 subscribeTo ，根据输入类型，对输入进行不一样的处理。
若是输入是原生数组或者是实现了数组功能的数据结构，那么直接调用 subscriber.next 把全部数据依次发送给订阅者。
export const subscribeToArray = <T>(array: ArrayLike<T>) => (subscriber: Subscriber<T>) => {
for (let i = 0, len = array.length; i < len && !subscriber.closed; i++) {
subscriber.next(array[i]);
}
subscriber.complete();
};
若是输入是 Obervable，那么要经过一个特定的 Symbol 取出 Observable，而后再订阅它。
（基于 Symbol 的特性，当前不少项目都会使用一个固定的 Symbol 对特定数据取值，来验证这个数据是否是符合类型）。
export const subscribeToObservable = <T>(obj: any) => (subscriber: Subscriber<T>) => {
const obs = obj[Symbol_observable]();
if (typeof obs.subscribe !== 'function') {
throw new TypeError('Provided object does not correctly implement Symbol.observable');
} else {
return obs.subscribe(subscriber);
}
};
若是输入是一个 Promise，那么经过 then 获取到 Promise 的内容，并将内容发送给订阅者。
export const subscribeToPromise = <T>(promise: PromiseLike<T>) => (subscriber: Subscriber<T>) => {
promise.then(
(value) => {
if (!subscriber.closed) {
subscriber.next(value);
subscriber.complete();
}
},
(err: any) => subscriber.error(err)
);
return subscriber;
};
生成器跟数组的方式相似，也是经过循环的方式将数据发送给订阅者。
export const subscribeToIterable = <T>(iterable: Iterable<T>) => (subscriber: Subscriber<T>) => {
const iterator = (iterable as any)[Symbol_iterator]();
do {
let item: IteratorResult<T>;
try {
item = iterator.next();
} catch (err) {
subscriber.error(err);
return subscriber;
}
if (item.done) {
subscriber.complete();
break;
}
subscriber.next(item.value);
if (subscriber.closed) {
break;
}
} while (true);
return subscriber;
};
generate 可让你用一种相似 for 循环的方式得到数据流。不过，我目前尚未遇到过很是须要这种方式生成流的方式，若是你遇到这种状况，欢迎交流。通常来讲，我习惯于这样调用它。
const observable = generate({
initialState: 1,
condition: x => x < 5,
iterate: x => x + 1,
});
observable.subscribe((value) => {
console.log(value);
});
// 打印结果
// 1
// 2
// 3
// 4
原来的源码包含了较多的参数判断，把内部逻辑梳理一下，实际上就是分为三个大步骤：
export function generate<S>(options: GenerateOptions<S>): Observable<S> {
const initialState = options.initialState;
const condition = options.condition;
const iterate = options.iterate;
// 返回 Observable
return new Observable<S>(subscriber => {
let state = initialState;
try {
while (true) {
// 判断结束条件
if (condition && !condition(state)) {
subscriber.complete();
break;
}
// 发送数据给订阅者
subscriber.next(state);
// 调用迭代，获取下一组数据
state = iterate(state);
if (subscriber.closed) {
break;
}
}
} catch (err) {
subscriber.error(err);
}
return undefined;
});
}
其中 GenerateOptions 包含了三个成员，initialState，condition 以及 iterate 。
export interface GenerateOptions<S> {
// 初始状态
initialState: S;
// 结束条件
condition?: (x: S) => boolean;
// 迭代方式
iterate: (x: S) => S;
}
range 能够建立一个给定范围的数字流。这个主要就是提供了一个简单的语义化函数，主要就是经过循环给订阅者喂数据。
export function range(start: number = 0, count?: number): Observable<number> {
return new Observable<number>(subscriber => {
if (count === undefined) {
count = start;
start = 0;
}
for (let index = 0; index < count; ++index) {
subscriber.next(start + index);
if (subscriber.closed) {
break;
}
}
return undefined;
});
}
fromEvent 是的 Observable 能够封装一系列的系统事件。既能够接受 NodeJS EventEmitter，也能够接受 DOM EventTarget， JQuery-like event target, NodeList 或者 HTMLCollection 等浏览器对象。
const clicksA = fromEvent(document, 'click');
const clicksB = fromEvent($(document), 'click');
clicksA.subscribe(x => console.log('A: ', x));
clicksB.subscribe(x => console.log('B: ', x));
// 每当点击一下页面，都会打印出 event 。
它的实现很简单，根据 target 的对象类型调用其对应的事件监听函数，而后经过 subscriber 调用 next 获取到订阅的输出。为了方便阅读，我稍微的改了一下，让 fromEvent 只支持 DOM EventTarget。
export interface HasEventTargetAddRemove<E> {
addEventListener(type: string, listener: ((evt: E) => void) | null, options?: boolean | AddEventListenerOptions): void;
removeEventListener(type: string, listener?: ((evt: E) => void) | null, options?: EventListenerOptions | boolean): void;
}
// 一个只支持 DOM EventTarget 的 fromEvent
export function fromEvent<T>(
target: HasEventTargetAddRemove<T>,
eventName: string,
options?: EventListenerOptions,
): Observable<T> {
return new Observable<T>(subscriber => {
// 处理结果
const handler = (...e: T[]) => subscriber.next(e.length === 1 ? e[0] : e);
// 调用 addEventListener，并让其在 handler 中处理。
target.addEventListener(eventName, handler, options);
// 取消订阅的时候，直接调用 removeEventListener 对 dom 取消订阅。
// 返回的是一个函数，这个函数负责了取消订阅的时，所作的内容。
return () => {
target.removeEventListener(eventName, handler, options);
};
});
}
上面的代码能够分解成这三个步骤：
对于其余的事件监听，再也不赘述，流程彻底是同样。
fromEventPattern 则是对 fromEvent 的泛化。
function addClickHandler(handler) {
document.addEventListener('click', handler);
}
function removeClickHandler(handler) {
document.removeEventListener('click', handler);
}
const clicks = fromEventPattern(
addClickHandler,
removeClickHandler
);
clicks.subscribe(x => console.log(x));
// 点击的时候，就会输出点击事件。
它的源码的与 fromEvent 相似。
export type NodeEventHandler = (...args: any[]) => void;
export function fromEventPattern<T>(
addHandler: (handler: NodeEventHandler) => any,
removeHandler?: (handler: NodeEventHandler, signal?: any) => void,
): Observable<T | T[]> {
return new Observable<T | T[]>(subscriber => {
const handler = (...e: T[]) => subscriber.next(e.length === 1 ? e[0] : e);
// 有一点不一样的地方在于，获取了返回值 addHandler 的返回值
let retValue: any;
try {
retValue = addHandler(handler);
} catch (err) {
subscriber.error(err);
return undefined;
}
if (!isFunction(removeHandler)) {
return undefined;
}
// 而后在这里传入 removeHandler 中
return () => removeHandler(handler, retValue);
});
}
它们都是一种特殊的 Operator ，思路应该是源于 Function.bind ，提供一种转换操做，将带有回调的函数转换成 Observable Factory。
function setTimeoutWithCallback(callback: () => void) {
setTimeout(() => {
callback();
}, 2000);
}
const obfactory = bindCallback(setTimeoutWithCallback);
const ob1 = obfactory();
const ob2 = obfactory();
const now = Date.now();
ob1.subscribe(() => {
console.log('ob1' + (Date.now() - now));
});
setTimeout(() => {
ob1.subscribe(() => {
console.log('ob1 later: ' + (Date.now() - now));
});
ob2.subscribe(() => {
console.log('ob2: ' + (Date.now() - now));
});
}, 3000);
// 打印结果：
// ob1: 2001
// ob1 later: 3004
// ob2: 5008
如下是 bindNodeCallback 的例子。
/*
file: ~/desktop/test.json
{ "name": "Hello World" }
*/
import * as fs from 'fs';
const readerFactory = bindNodeCallback(fs.readFile);
const reader$ = readerFactory('./src/person.json');
reader$.subscribe({
next: (value) => console.log(value.toString()),
error: (err) => console.log(err),
complete: () => console.log('complete')
});
// 若是没有错误，打印结果以下：
// { name: 'Hello World' }
// complete
// 若是有错误，打印结果以下：
// [Error: ENOENT: no such file or directory, open './src/person.json'] {
// errno: -2,
// code: 'ENOENT',
// syscall: 'open',
// path: './src/person'
// }
bindCallback 和 bindNodeCallback 的源码很是相似。
export function bindCallback<T>(
callbackFunc: Function
): (...args: any[]) => Observable<T> {
return function (this: any, ...args: any[]): Observable<T> {
const context = this;
//
let subject: AsyncSubject<T>;
return new Observable<T>(subscriber => {
if (!subject) {
subject = new AsyncSubject<T>();
const handler = (...innerArgs: any[]) => {
subject.next(innerArgs.length <= 1 ? innerArgs[0] : innerArgs);
subject.complete();
};
try {
callbackFunc.apply(context, [...args, handler]);
} catch (err) {
subject.error(err);
}
}
return subject.subscribe(subscriber);
});
};
}
bindCallback 和 bindNodeCallback 的源码惟一不一样的地方就是在于 handler 这个函数处理的内容不一样，bindNodeCallback 传入的函数的回调，第一个参数为是错误信息。
const handler = (...innerArgs: any[]) => {
const err = innerArgs.shift();
// 若是第一个参数存在，说明有问题。
if (err) {
subject.error(err);
return;
}
subject.next(innerArgs.length <= 1 ? innerArgs[0] : innerArgs);
subject.complete();
};
源码中比较有趣的地方在于，建立的时候，返回的工厂函数包含了一个 AsyncSubject。这个 AsyncSubject 保存了已经到来数据，能够看看例子中，ob1 被订阅了2次，第二次订阅后其实是马上就能拿到返回值；而 ob2 仍要执行一次 setTimeoutWithCallback。这种设计与这个 bind 的语义相吻合。
上面的 operators 中，我已经把 scheduler 相关的内容进行了裁剪，基本上与 scheduler 无关。而 interval 和 timer 都必须经过 scheduler 来相应的定时操做，因此这部分放到了最后。它们是用于建立定时数据源的 operators 。
const observableA = interval(1000).pipe(take(2));
const observableB = timer(500, 1000).pipe(take(3));
console.log('hello');
observableA.subscribe(value => {
console.log('A: ' + value);
});
observableB.subscribe(value => {
console.log('B: ' + value);
});
// 打印结果
// hello
// B: 0
// A: 0
// B: 1
// A: 1
// B: 2
interval 和 timer 都使用了一个默认的异步调度器，这个异步调度器主要是经过 setInterval 来实现相应的功能，实际上 Rx 把异步调度器经过 interval 和 timer 转化成 Observable 的形式提供到给用户。
timer 的实现以下图所示。它首先建立了一个 Observable ，而后在订阅函数中，返回调度器的订阅。在这里， scheduler 的 schedule 函数返回了一个 Subscription。
export function timer(
dueTime: number | Date = 0,
period: number,
scheduler: SchedulerLike = async
): Observable<number> {
return new Observable(subscriber => {
let due = 0;
// 判断是否是 Date 类型
if (dueTime instanceof Date) {
due = +dueTime - scheduler.now();
}
// 判断是否是 number 类型
if (isNumeric(dueTime)) {
due = dueTime as number;
}
// 此处调用跟 interval 相似。
return scheduler.schedule(dispatch, due, {
index: 0, period, subscriber
});
});
}
dispatch 其实是一个递归函数，这个函数绑定了 SchedulerAction ，经过传入订阅者，使得 Action 内部的 setInterval 可以一直调用 subscriber.next。
interface TimerState {
index: number;
period: number;
subscriber: Subscriber<number>;
}
function dispatch(this: SchedulerAction<TimerState>, state: TimerState) {
const { index, period, subscriber } = state;
subscriber.next(index);
if (subscriber.closed) {
return;
} else if (period === -1) {
return subscriber.complete();
}
state.index = index + 1;
this.schedule(state, period);
}
如下是 interval 的源码。
export function interval(period = 0): Observable<number> {
if (!isNumeric(period) || period < 0) {
period = 0;
}
const scheduler = async;
return new Observable<number>(subscriber => {
// 订阅器接收 scheduler 的订阅结果。
subscriber.add(
scheduler.schedule(dispatch, period, { subscriber, counter: 0, period })
);
return subscriber;
});
}
仔细的分析上面的代码，我发现 interval 的实现实际上就是 timer 的一个约束版本，它能够改写成这样。
export function interval(
period = 0,
scheduler: SchedulerLike = async,
): Observable<number> {
if (!isNumeric(period) || period < 0) {
period = 0;
}
return timer(period, period, sch);
}
总体而言，这部分的源码并无写得很绕，删去了 scheduler 相关的内容后，逻辑马上就变得清晰了起来。同时，从源码的风格上能够看到它们由不一样的人来编写。
最后，限于本人的水平，若是文章中有错误的地方，欢迎指正。
咱们是DevUI团队，欢迎来这里和咱们一块儿打造优雅高效的人机设计/研发体系。招聘邮箱：muyang2@huawei.com。
做者：zcx（公众号：Coder写字的地方）
原文连接：https://mp.weixin.qq.com/s/6fVoI_JtSXu6YfZur1TDNw
往期文章推荐