[Angular] `AsyncPipe`はどうやってViewに値を反映させているのか？

この記事は Angular Advent Calendar 2021 の3日目の記事です。

昨日は @kasaharu さんの記事でした。「2021 年の Angular 振り返り」

私は19日目の枠を取っていたのですが、なんと3日目にして誰も記事を書いていないという自体に遭遇したため、急遽、記事を書くことにしました！クオリティは期待なさらず！

概要

AsyncPipeがobservableを受け取ったとき、どうやってViewに値を反映させているのだろうか？

その謎を解明するため、我々調査隊はAngularのソースへと向かった――

どうやってAsyncPipeはobservableに値が流れたときにViewに値を出力しているのか？

皆さんは「AsyncPipeを実装しろ！」と言われたら、どんなコードを書くでしょうか?

私は直感的に下記のようなコードを書きました。

@Pipe({name: 'async'})
export class AsyncPipe {
  transform<T>(observableOrValue: Observable<T> | T): T | null {
    if(!isObservable(observableOrValue)) return observableOrValue;
    observableOrValue.subscribe((value) => this.transform(value));
    return null;
  }
}

当然、動きません。

入力されたobservableを購読してAsyncPipe.transform()を実行すれば、Viewにその返り値が出力されるかと思いましたが、そんなことはありません。
Pipeの出力がViewに反映されるのは、Pipeに値が流れてきた1度きりです。
上記の例では永遠にnullしか出力されることはないでしょう。

では、どのようにしてAngularのAsyncPipeはobservableを購読して、そこに流れてきた値をViewに反映させているのでしょうか？

ソースコード

私がOSSを好きな理由はソースコードを見れば答えがわかるところです。
なので、まずはAsyncPipeのソースコードを読んで見ましょう！

ソースコードを見るとStrategyパターンを用いて入力がPromiseとobservableの場合で処理を切り分けていたり、ngOnDestroyの処理があったりで、少々、処理の流れを追うのが大変そうです...

今回の関心事のみを切り出して整形したコード

そこで今回はPromiseの場合の処理やunsubscribeの処理など今回の関心の外にあるコードを削除して、少し読みやすく整形しました。(※unsubscribeの処理が消えているのでこのまま動かすと予期せぬバグが起こりやすいので注意)

import {ChangeDetectorRef, EventEmitter, OnDestroy, Pipe, PipeTransform} from '@angular/core';
import {Observable, Subscribable} from 'rxjs';


@Pipe({name: 'async', pure: false})
export class AsyncPipe implements OnDestroy, PipeTransform {
  private _latestValue: any = null;
  private _obs: Subscribable<any>|EventEmitter<any>|null = null;

  constructor(private _ref: ChangeDetectorRef) {}

  transform<T>(obs: Observable<T>|Subscribable<T>|null|undefined): T|null {
    // CASE_1: 購読中のobservableがnullの場合
    if (!this._obs) {
      if (obs) {
        this._obs = obs;
        obs.subscribe((value: Object) => {
          if (obs === this._obs) {
            this._latestValue = value;
            // 1. ChangeDetectorRef.markForCheck()を呼ぶことで前回と同じ値がPipeに流れる
            // 2. AsyncPipはpureではないため、同じ値でもtransformは実行される
            // 3. 1と2からobservableに値が流れるたびに、Pipeに値が流れtransform()の返り値が出力されることになる
            // 4. このとき呼ばれるtransform()ではCASE_3が実行されるため、最新の値が出力される
            this._ref.markForCheck();
          }
        });
      }
      return this._latestValue;

    // CASE_2: 入力されたobservableが購読中のobservableと異なる場合
    } else if (obs !== this._obs) {
      this._latestValue = null;
      this._obs = null;
      // このtransformではCASE_1が呼ばれるため、入力されたobservableが、最新の値が出力される
      return this.transform(obs);
      
    // CASE_3: 入力されたobservableが購読中のobservableと同じ場合
    } else {
      // 最新の値が出力される
      return this._latestValue;
    }
  }
}

AyncPipe.tranform()の3つのケース

AyncPipe.tranform()に値が渡されたときの挙動は3つのケースに分けられます。
ここでは処理の流れを追うためにCASE_3->CASE_1->CASE_2の順で説明します。

CASE_3: 入力されたobservableが購読中のobservableと同じ場合

    // CASE_3: 入力されたobservableが購読中のobservableと同じ場合
    } else {
      // 最新の値が出力される
      return this._latestValue;
    }

AsyncPipeは@Pipe({pure: false})が指定されていため、Pipeに同じ値の参照が流れてきてもtransform()が実行され、その返り値が出力されます。
入力されたobservableが購読中のobservableと同じ場合、購読中のている最新の値が出力されます。

CASE_1: 購読中のobservableがnullの場合

    // CASE_1: 購読中のobservableがnullの場合
    if (!this._obs) {
      if (obs) {
        this._obs = obs;
        obs.subscribe((value: Object) => {
          if (obs === this._obs) {
            this._latestValue = value;
            // 1. ChangeDetectorRef.markForCheck()を呼ぶことで前回と同じ値がPipeに流れる
            // 2. AsyncPipはpureではないため、同じ値でもtransformは実行される
            // 3. 1と2からobservableに値が流れるたびに、Pipeに値が流れtransform()の返り値が出力されることになる
            // 4. このとき呼ばれるtransform()ではCASE_3が実行されるため、最新の値が出力される
            this._ref.markForCheck();
          }
        });
      }
      return this._latestValue;

    // CASE_2: 入力されたobservableが購読中のobservableと異なる場合
    } else if (obs !== this._obs) {

購読中のobservableがnullの場合は、AsyncPipeに入力されたobservableを購読し、値が流れるたびに最新の値を(this._latestValue)を更新して、this._ref.markForCheck()を呼びます。

このthis._ref.markForCheck()が今回のAsyncPipeの疑問への答えであり、
これによってViewに変更があったことを示す処理が走り、再度、AsyncPipeに同じ値が流れます。
その結果、CASE_3の処理が走り、直前に更新した最新の値(this._latestValue)が出力されます。

CASE_2: 入力されたobservableが購読中のobservableと異なる場合

    // CASE_2: 入力されたobservableが購読中のobservableと異なる場合
    } else if (obs !== this._obs) {
      this._latestValue = null;
      this._obs = null;
      // このtransformではCASE_1が呼ばれるため、入力されたobservableが、最新の値が出力される
      return this.transform(obs);
      
    // CASE_3: 入力されたobservableが購読中のobservableと同じ場合
    } else {

最後に、入力されたobservableが購読中のobservableと異なる場合ですが、このケースではこれまで購読していたobservableを捨てて、新しく入力されたobservableを購読し直します。

これらの処理によって私達の普段使うAsyncPipeの処理は成り立っています。

まとめ

純粋でないPipeの中でChangeDetectorRef.markForCheck()を呼ぶことでAsyncPipeが再評価され出力を変化させることができます。AsyncPipeは入力されたobservableを購読し、そこに値が流れるたびにChangeDetectorRef.markForCheck()を呼ぶことで、Viewに最新の値を出力しているのでした。