[Angular] takeUntilから学ぶRxjsストリーム #TypeScript

★ 基本的にストリームはリレー形式のようなイメージです。
　リレー形式といっても、
　1つのバトンを変換・組合せ・加工しながらsubscribe(ゴール)するイメージです。

私の解釈

■ ストリーム
　ストリームはキャンセル、破棄という表現よりは
　ストリームは置き換わるという解釈をすることで腑に落ちました。

■ takeUntil
　引数のストリームをトリガーに、本ストリームを終了する。
obs$.interval(1000).pipe( takeUntile(this.destroy$), ・・・ )
　上記では、this.destroy$にストリームが流れたら(next())、
　obs＄の監視終了(observableストリームをキャンセル)します。
　要は意図的に手動でキャンセルできるということです。

本題

まずは説明用のためコードを記載します。

code

v.html


<div class="main-container">
    <div class="upbutton-container" (click)="changeSetting('up')">
        <img class="upbuttonImage" [src]="'assets/images/'+ upTap">
    </div>
    <div class="value-container">
        {{valueChange}}
    </div>
    <div class="downbutton-container" (click)="changeSetting('down')">
        <img class="downbuttonImage" [src]="'assets/images/'+ downTap">
    </div>
</div>

v.ts

  /**
   * 数値
   */
  public value:number = 16;
  /**
   * 退避数値
   */
  public evavalue:number = 16;

  public sub$ = new Subject();


  /**
   * ユーザ操作
   */
  public changeSetting(type:string){
    let temp = this.value;
    if(type == "up"){
      this.value = temp + 1;
    }else {
      this.value = temp - 1;
    }

    this.sub$.next(this.value);
    console.log("ユーザタップ後タイマー処理スタート");
    timer(3000).pipe(
      takeUntil(this.sub$),
      switchMap(()=>{
        console.log("3秒経過処理実行!!!")
        this.value = this.evavalue;
        return of("");
      })
    ).subscribe(()=>{
      console.log("処理完了");
    })
  }

describe

上記コードの処理実行順は以下の通りとなります。

1: this.sub＄.next(this.value);
2: takeUntile(this.sub$)
3: timer(3000)
4: switchMap以降～

[1:][2:]でthis.sub＄ストリーム（★）値を発行しますが、
[3:]のtimer(3000)オペレータによって[1:][2:]で発行されたストリーム（★）はtimer待機中を表すストリーム（■）に変換されます。
以降、timerオペレータで定義した3秒以内に再度this.sub$ストリーム（★）値が発行されると上記の[3:]で発行されたストリーム(■)からストリーム(★)に変換されます。

ストリーム(■)→ストリーム(★)に置き換わるということはつまり、
3秒待機中オペレータ（■）は破棄されると同じ意味を持つと考えると考えやすかったです。
ちなみに、上記処理ではストリーム（★）に置き換えられた直後に[3:]が即座に実行されるため
ストリーム（★）はtimer(3000)待機中を表すストリーム（■）に置き換えられます。

説明要約イメージ図

補足

v.ts

  /**
   * ユーザ操作
   */
  public changeSetting(type:string){
    let temp = this.value;
    if(type == "up"){
      this.value = temp + 1;
    }else {
      this.value = temp - 1;
    }

    let local$ = new Subject(); //★

    local$.next(this.value);
    console.log("ユーザタップ後タイマー処理スタート");
    timer(3000).pipe(
      takeUntil(local$),
      switchMap(()=>{
        console.log("3秒経過処理実行!!!")
        this.value = this.evavalue;
        return of("");
      })
    ).subscribe(()=>{
      console.log("処理完了");
    })
  }

ローカルスコープとグローバルスコープでの定義挙動の違い

method内(ローカルスコープ)でsubject()インスタンスを生成してtakeUntilの引数にセットした場合、
ユーザタップ回数分各ストリームが独立して並列にストリームが流れます。

イメージ：
　ユーザタップA　→　timerB→　　以降の処理～
　ユーザタップ1　→　timer2→　　以降の処理～
　ユーザタップあ　→　timerい→　以降の処理～

最初(上の)グローバルスコープでsubject()インスタンスを生成した場合では、
同じsubject()のストリームが生成されるためストリームは置き換えられるような挙動になってました。

一般的な使われ方とよく紹介されてるもの

code

T.ts

public destroy$:Subject<void> = new Subject();

  ngOnInit(){
    let obs$: Observable<unknown>;

    obs$ = this.service.startup().pipe(
      takeUntil(this.destroy$),
      switchMap((res:DeviceAction)=>{
        return of(res);
      })
    )
    obs$.subscribe((val:DeviceAction | unknown)=>{
      if (val && typeof val === 'object' && 'value' in val) {
        this.temp = Number(val.value);
        console.log(val.value);
      }
    })
  }

  ngOnDestroy(){
    console.log("ngOnDestroy実施");
    this.destroy$.next();
    this.destroy$.complete();
  }

describe

ngOnDestroy()はディレクティブ/コンポーネントを破棄する直前に呼び出されます。

takeUntil(this.destroy＄) を使用して、this.destroy＄が値を発行するまで obs＄を監視し、ngOnDestroy メソッドが呼び出されると this.destroy＄に値が発行され、obs＄の監視が解除されます。

observableの購読(監視)を確実に解除するために、
next()とcomplete()を実装しています。

こうすることでメモリリークを防ぎリソース解放されます。