RxJSをソースコードから理解したい

# はじめに
主に(?) Angular の裏側で使われるRxJSは`Promise`などと同様に非同期処理を解決するためにObservableパターンを採用したライブラリで、実際複雑な非同期処理を簡潔に書くことができる。

ただObservableパターンに不慣れだったり、oepratorの多さと複雑さに気圧されて、しばしば書いたコードが魔法にしか見えないことがある（結果は分かってもなぜそうなるのかさっぱり分からない・・・）。

いい加減裏側で何が行われているのかしっかりと理解したかったので、以下のようなシンプルなコードを元に、RxJSのソースコードを追いかけていく。

```typescript
of(1, 2, 3).pipe(
  map(value => value * 2)
).subscribe(value => console.log(value));

// output => 2, 4, 6
```

# バージョンと注意
この記事では 2019/04/07 時点の [RxJS 6.4.0](https://github.com/ReactiveX/rxjs/releases/tag/6.4.0) を参照している。

https://github.com/ReactiveX/rxjs/tree/6.4.0

また記事中では理解しやすいように、今回動かすコードとは関係の薄い部分はかなり省いて記載する。

# 登場人物
いきなり処理の流れを追う前に、上のコードで登場する型と代表的なinterfaceをまとめる。

- `Observable<T>` ([internal/Observable.ts](https://github.com/ReactiveX/rxjs/blob/6.4.0/src/internal/Observable.ts#L19))
    - subscribe()
    - pipe()
    - lift()
- `Observer<T>` ([internal/types.ts#L84](https://github.com/ReactiveX/rxjs/blob/6.4.0/src/internal/types.ts#L84))
    - next()
    - error()
    - complete()
- `Subscriber<T>` ([internal/Subscriber.ts](https://github.com/ReactiveX/rxjs/blob/6.4.0/src/internal/Subscriber.ts))
    - next()
    - error()
    - complete()
    - unsubscribe()
- Operator<T, R> ([internal/Operator.ts](https://github.com/ReactiveX/rxjs/blob/6.4.0/src/internal/Operator.ts))
    - `call(subscriber: Subscriber<R>, source: any): TeardownLogic;`

# of(1, 2, 3)
`of`の定義と実装は以下の通り、引数の違いはあれど最終的に`Observable`型を返す関数となっている。

[internal/observable/of.ts](https://github.com/ReactiveX/rxjs/blob/6.4.0/src/internal/observable/of.ts)

```typescript
export function of<T, T2, T3>(a: T, b: T2, c: T3, scheduler?: SchedulerLike): Observable<T | T2 | T3>;

export function of<T>(...args: Array<T | SchedulerLike>): Observable<T> {
  // ~~省略~~
  return fromArray(args as T[], scheduler);
}
```

今回(`of(1, 2, 3)`)は引数を複数渡しており、`scheduler`を渡していないためなんやかんやで`fromArray(args as T[], scheduler)`が呼ばれる。

`fromArray`では以下のように`Observable`が生成されている。

[internal/observable/fromArray.ts](https://github.com/ReactiveX/rxjs/blob/6.4.0/src/internal/observable/fromArray.ts)
[internal/util/subscribeToArray.ts](https://github.com/ReactiveX/rxjs/blob/6.4.0/src/internal/util/subscribeToArray.ts)

```typescript
export function fromArray<T>(input: ArrayLike<T>, scheduler?: SchedulerLike) {
  return new Observable<T>(subscribeToArray(input));
}

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]);
  }
  if (!subscriber.closed) {
    subscriber.complete();
  }
};
```

`Observable`は`subscribe()`に反応する`Subscribable<T>`を実装している。`Observable`のconstructorを見ると

```typescript
constructor(subscribe?: (this: Observable<T>, subscriber: Subscriber<T>) => TeardownLogic) {
  if (subscribe) {
    this._subscribe = subscribe;
  }
}
```
なので、`subscribeToArray`の結果が`this._subscribe`に渡される。この`Observable`の`subscribe()`が呼ばれると、中身の配列（今回は `[1, 2, 3]` )に従って `subscriber.next(array[i]);` が呼び出されそうである。

# pipe()
`pipe`の定義は以下の通り。

[internal/Observable.ts#L333](https://github.com/ReactiveX/rxjs/blob/6.4.0/src/internal/Observable.ts#L333)
[internal/util/pipe.ts#L23](https://github.com/ReactiveX/rxjs/blob/6.4.0/src/internal/util/pipe.ts#L23)

```typescript
pipe(...operations: OperatorFunction<any, any>[]): Observable<any> {
  if (operations.length === 0) {
    return this as any;
  }

  return pipeFromArray(operations)(this);
}

/** @internal */
export 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 {
    return fns.reduce((prev: any, fn: UnaryFunction<T, R>) => fn(prev), input as any);
  };
}
```

面白いのは`pipe`でreturnするときに`pipeFromArray`の戻り値に自分自身(`Observable`)を渡して実行していること。

`pipeFromArray`では、受け取ったoperator(今回は`map`)を`reduce`で**前の実行結果を入力にしながら**順番に実行する関数を返している。

`pipe`自身の戻り値も`Observable`なので、`pipe`周りの挙動をまとめると

- 自分自身(`Observable`)を入力として、受け取ったoperatorを`reduce`で順番に実行して結果(`Observable`)を返す
- operatorとは `Observable`を受け取って`Observable`を返す関数

と言うことが分かった。

# map(value => value * 2)
map operatorの実装は以下の通り。

[internal/operators/map.ts](https://github.com/ReactiveX/rxjs/blob/6.4.0/src/internal/operators/map.ts)

```typescript
export function map<T, R>(project: (value: T, index: number) => R, thisArg?: any): OperatorFunction<T, R> {
  return function mapOperation(source: Observable<T>): Observable<R> {
    return source.lift(new MapOperator(project, thisArg));
  };
}
```

一つ前で分かった「operatorとは `Observable`を受け取って`Observable`を返す関数」と言うのが`function mapOperation(source: Observable<T>): Observable<R>`に現れている。

ただそれだけでは不十分で、本質的(?)な実装は`source.lift()`の中に見える。

[internal/Observable.ts#L66](https://github.com/ReactiveX/rxjs/blob/6.4.0/src/internal/Observable.ts#L66)

```typescript
lift<R>(operator: Operator<T, R>): Observable<R> {
  const observable = new Observable<R>();
  observable.source = this;
  observable.operator = operator;
  return observable;
}
```

上記のように`source`と`operator`が設定された`Observable`を新しく生成して返している。

ここの`operator`がどんなものか`map`の実装から見ると

```typescript
export class MapOperator<T, R> implements Operator<T, R> {
  constructor(private project: (value: T, index: number) => R, private thisArg: any) {
  }

  call(subscriber: Subscriber<R>, source: any): any {
    return source.subscribe(new MapSubscriber(subscriber, this.project, this.thisArg));
  }
}

class MapSubscriber<T, R> extends Subscriber<T> {
  count: number = 0;
  private thisArg: any;

  constructor(destination: Subscriber<R>,
              private project: (value: T, index: number) => R,
              thisArg: any) {
    super(destination);
    this.thisArg = thisArg || this;
  }

  protected _next(value: T) {
    let result: any;
    try {
      result = this.project.call(this.thisArg, value, this.count++);
    } catch (err) {
      this.destination.error(err);
      return;
    }
    this.destination.next(result);
  }
}
```

少しややこしいが、`call()`の中で`subscribe()`を呼んでいること覚えておく。ただしこの時はまだ`Observable`が新規に生成されただけで、まだ`call()`は呼ばれてないことに注意する。

`MapSubscriber`についてはこの後見ていく。

# subscribe(value => console.log(value))
いよいよ`subscribe()`について見ていく。実装は以下のようになっている。
(本来は`sink`に`add`するコードがあるが、`unsubscribe()`の時に親をさかのぼるためのものなので、今は省略している。）

[internal/Observable.ts#L199](https://github.com/ReactiveX/rxjs/blob/6.4.0/src/internal/Observable.ts#L199)

```typescript
subscribe(observerOrNext?: PartialObserver<T> | ((value: T) => void),
            error?: (error: any) => void,
            complete?: () => void): Subscription {

  const { operator } = this;
  const sink = toSubscriber(observerOrNext, error, complete);

  if (operator) {
    operator.call(sink, this.source);
  } else {
    this._subscribe(sink);
  }

  return sink;
}
```

ここから今までのソースコードを順番にさかのぼっていくことができる。

最初に`subscribe()`される`Observable`はmap operatorの戻り値なので`operator`を持っている。なので`operator.call(sink, this.source);`が実行される。

上で書いた通り`call()`の中身は`source.subscribe()`なので、`source`である`of(1, 2, 3)`の戻り値の`Observable`が`subscribe()`される。
こちらの`Observable`は`operator`を持っていないので、今度は`this._subscribe()`が呼ばれる。

上で見た通りこれは

```typescript
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]);
  }
  if (!subscriber.closed) {
    subscriber.complete();
  }
};
```

である。さらにここでいう`subscriber: Subscriber<T>`とはmapで生成される`MapSubscriber<T, R>`である。

`MapSubscriber<T, R>`の`next()`を呼び出しているので、以下のコードが実行される。

```typescript
  protected _next(value: T) {
    let result: any;
    try {
      result = this.project.call(this.thisArg, value, this.count++);
    } catch (err) {
      this.destination.error(err);
      return;
    }
    this.destination.next(result);
  }
```

ここで言う`this.project`とは`value => value * 2`である。その結果(`2, 4, 6`)が`this.destination.next()`に渡される。
この`this.destination`とは`const sink = toSubscriber(observerOrNext, error, complete);`の`sink`である（ややこしい・・・）。
`sink`の`next()`とはつまり`value => console.log(value)`であるので、無事に`output => 2, 4, 6`が得られるということだった。

# 終わりに
単純な`Observable#of`と、基本的なoperatorである`map`を題材に結果が得られるまでの流れをソースコードで追ってみた。

魔法にしか見えなかったコードも、裏側で何が起きているか実感できたような気がする。

今回でObservableの生成からpipeline処理、subscribeまで基礎となる流れを理解できたため、次はAngularの`HttpClient`などでどのようにObservableが生成されているかや、各operatorがどのように実装されているか個別に追っていきたいと思う。

間違っている所等あれば是非ご指摘ください。