TL;DR

@rx-angular/stateを使うことで、安全に、より簡潔に、よりリアクティブに実装することができそう。

まえおき

Angularにおける、リアクティブプログラミングの話です。

下記の記事を読んで自分なりに解釈をまとめた記事になります。
BehaviorSubject vs RxState
Research on Reactive-Ephemeral-State in component-oriented frameworks

間違いや語弊などありましたら、優しくコメントいただけると嬉しいです。

なぜリアクティブに実装したいのか

こちらの記事で書きました。
なぜリアクティブプログラミングをしたいのかを考えてみる

リアクティブでない実装

まずは、リアクティブでない命令的な実装例です。

簡単なTodoアプリケーションを１つのコンポーネントで実装しています。
コンポーネントが保持している状態と、発生しうるイベントは次のとおりです。

状態

Todoリスト
フィルタ条件
フィルタされたTodoリスト
フィルタされたTodoリストの数

イベント

Todoリストの追加
Todoリストの削除
フィルタ条件の変更

この実装を後ほどリアクティブに置き換えていきます。

todo-page-component.ts

import { Component, OnInit, ChangeDetectionStrategy, Input, Output, EventEmitter } from '@angular/core';
import { v4 as uuidv4 } from 'uuid';

type FilterCondition = 'completed' | 'all' | 'todo';
type TodoItem = { id: string; title: string; completed: boolean; };

@Component({
  selector: 'app-todo-page',
  templateUrl: './todo-page.component.html',
  styleUrls: ['./todo-page.component.scss'],
  changeDetection: ChangeDetectionStrategy.OnPush
})
export class TodoPageComponent implements OnInit {
  @Input() title = '';
  @Output() changedTitle: EventEmitter<string> = new EventEmitter();

  // item一覧
  items: TodoItem[] = [
    {id: uuidv4(), title: 'todo0', completed: false},
    {id: uuidv4(), title: 'todo1', completed: true},
    {id: uuidv4(), title: 'todo2', completed: false},
    {id: uuidv4(), title: 'todo3', completed: true},
    {id: uuidv4(), title: 'todo4', completed: false},
    {id: uuidv4(), title: 'todo5', completed: true},
  ];

  // フィルタされたitem一覧
  filteredItems: TodoItem[] = [];

  // フィルタ条件
  filterCondition: FilterCondition = 'all';

  // フィルタされたitemの数
  filteredItemsCount = 0;

  constructor() {
    this.executeFilter();
  }

  ngOnInit(): void {}

  /**
   * filterConditionに基づいて、itemsをフィルタする
   */
  executeFilter() {
    switch (this.filterCondition) {
      case 'todo':
        this.filteredItems = this.items.filter(v => !v.completed);
        break;
      case 'completed':
        this.filteredItems = this.items.filter(v => v.completed);
        break;
      case 'all':
        this.filteredItems = this.items;
        break;
    }
    this.filteredItemsCount = this.filteredItems.length;
  }

  changeFilterCondition(condition: FilterCondition) {
    this.filterCondition = condition;
    this.executeFilter();
  }

  /**
   * itemsに新規itemを追加し、itemsのフィルタを実行する
   */
  add() {
    this.items = [...this.items, {
      id: uuidv4(),
      title: uuidv4().slice(0, 5),
      completed: false
    }];
    this.executeFilter();
  }

  /**
   * itemsからitemを削除し、itemsのフィルタを実行する
   */
  remove(id: string) {
    this.items = this.items.filter(v => v.id !== id);
    this.executeFilter();
  }

}

todo-page-component.html

<h2>{{ title }}</h2>
<button (click)="changedTitle.emit('Hoge')">change title to "Hoge"</button>
<p>filter condition: {{filterCondition}}, total: {{ filteredItemsCount }}</p>
<div>
  <button (click)="changeFilterCondition('all')">all</button>
  <button (click)="changeFilterCondition('todo')">todo</button>
  <button (click)="changeFilterCondition('completed')">completed</button>
</div>
<button (click)="add()">add</button>
<ul>
  <ng-container *ngFor="let item of filteredItems; trackBy: trackByFn">
    <li>
      <span>title: {{ item.title }}, completed: {{ item.completed }}</span>
      <button (click)="remove(item.id)">remove</button>
    </li>
  </ng-container>
</ul>

ポイントは、イベントハンドリングの処理内で、命令的にコンポーネントのメンバ変数の値を更新しているところです。

addボタンがクリックされた時には、this.itemsを更新した後にfiltereを実行する、removeボタンがクリックされた時には、this.itemsを更新した後にfiltereを実行する、という風に命令的に処理が実装されています。

リアクティブな実装

上記の実装をリアクティブな実装に段階的に置き換えていきます。

コンポーネント内で発生しうるイベントをSubjectで定義する

リアクティブな実装にする上で必要になってくるのは、全てのイベントや値の更新に対して反応できるようにすることです。
イベントの発火とイベント発火を検知できるようにするために、イベントをSubjectで定義していきます。

この時、コンポーネントのライフサイクルとコンポーネントのOutputもSubjectで定義しておきます。

  readonly onAdd$ = new Subject();
  readonly onRemove$ = new Subject<Pick<TodoItem, 'id'>>();
  readonly onChangedFilterCondition$ = new Subject<FilterCondition>();
  readonly onChangedTitle$: Subject<string> = new Subject();
  readonly onDestroy$ = new Subject();

template側でイベントを発火させるときの実装は下記のようになります。

<!-- itemの追加 -->
<button (click)="onAdd$.next()">add</button>

<!-- itemの削除 -->
<button (click)="onRemove$.next({id: item.id})">remove</button>

<!-- フィルタ条件の変更 -->
<button (click)="onChangedFilterCondition$.next('all')">all</button>

これで、イベントの発火をObservableとして扱うことができるので、rxjsでリアクティブに実装しやすくなります。

コンポーネントのメンバ変数をObservableで定義する

次に、コンポーネントのメンバ変数をObservableで定義していきます。

値を明示的に指定する必要があるメンバ変数は、まずBehaviorSubjectとして定義します。
今回の例だと、Todoリストとフィルタ条件がそれに当たります。
それぞれ別々のSubjectで定義してもいいんですが、ここではstateSubjectとして1つにまとめています。

フィルタされたTodoリストのように、他の値から値を導出できるようなメンバ変数は、nextを発行する必要がないので、単純なObservableとして定義します。

  type State = {
    items: TodoItem[];
    filterCondition: FilterCondition;
  };

  readonly stateSubject: BehaviorSubject<State> = new BehaviorSubject({
    items: [
      {id: uuidv4(), title: 'todo0', completed: false},
      {id: uuidv4(), title: 'todo1', completed: true},
      {id: uuidv4(), title: 'todo2', completed: false},
      {id: uuidv4(), title: 'todo3', completed: true},
      {id: uuidv4(), title: 'todo4', completed: false},
      {id: uuidv4(), title: 'todo5', completed: true},
    ],
    filterCondition: 'all',
  });

  readonly state$ = this.stateSubject.asObservable().pipe(
    distinctUntilChanged(),
    shareReplay({ bufferSize: 1, refCount: true }),
  );

  readonly items$ = this.state$.pipe(
    pluck('items'),
    distinctUntilChanged(),
    shareReplay({ bufferSize: 1, refCount: true }),
  );

  readonly filterCondition$ = this.state$.pipe(
    pluck('filterCondition'),
    distinctUntilChanged(),
    shareReplay({ bufferSize: 1, refCount: true }),
  );

  readonly filteredItems$ = combineLatest([
    this.items$,
    this.filterCondition$
  ]).pipe(
    map(([items, condition]) => {
      switch (condition) {
        case 'todo':
          return items.filter(v => !v.completed);
        case 'completed':
          return items.filter(v => v.completed);
        case 'all':
          return items;
      }
    }),
    distinctUntilChanged(),
    shareReplay({ bufferSize: 1, refCount: true }),
  );

  readonly filteredItemsCount$ = this.filteredItems$.pipe(
    map(items => items.length)
  );

それぞれのObservableに対して、pipeで必ず二つののオペレータを指定しています。

pip(
  distinctUntilChanged(),
  shareReplay({ bufferSize: 1, refCount: true }),
)

このオペレータを指定することによって、下記の恩恵を受けることができ、パフォーマンス面の改善につながります。


distinctUntilChanged	前回発行された値と差分がない場合は、値を流さない。比較方法をこちらで指定することは可能。
shareReplay({ bufferSize: 1, refCount: true })	Hotかつ、multicastにする。refCountの恩恵。

メンバ変数がObservableになったので、template側ではasyncパイプを指定する必要があります。

<!-- フィルタ条件とフィルタされたTodoリストの数 -->
<p>filter condition: {{filterCondition$ | async}}, total: {{ filteredItemsCount$ | async }}</p>

<!-- フィルタされたTodoリストの表示 -->
<ng-container *ngFor="let item of filteredItems$ | async">
  <li>
    <span>title: {{ item.title }}, completed: {{ item.completed }}</span>
    <button (click)="onRemove$.next({id: item.id})">remove</button>
  </li>
</ng-container>

Inputの値をObservableに変換

Inputで受け取る値がObservableでない時、変更をトリガに反応できるようにObservableにします。

Inputの値に変更があったというイベントをSubjectで定義し、Inputのsetterで、そのSubjectのnextを発行する形になります。

  @Input()
  set title(title: string) {
    this.onChangedInputTitle$.next(title);
  }

  readonly onChangedInputTitle$ = new Subject<string>();

イベントハンドラを実装

全てのイベントとメンバ変数がObservableになったので、これをもとにイベントハンドラの実装をおこなっていきます。

Observableで定義したイベントに対して、そのイベントで発生する副作用をtap内で実装する流れになります。

詳しくは後述するんですが、Subscriptionの管理を簡潔にするために、イベントハンドラごとにsubscribeは行いません。

  // Todoリストに項目が追加されたときのイベントハンドラ
  private readonly addItemHandler$ = this.onAdd$.pipe(
    withLatestFrom(this.state$),
    tap(([, state]) => {
      const newItem = {
        id: uuidv4(),
        title: uuidv4().slice(0, 5),
        completed: false
      };
      this.stateSubject.next({
        ...state,
        items: [...state.items, newItem],
      });
    })
  );

  // フィルタ条件が更新されたときのイベントハンドラ
  private readonly changeFilterConditionHandler$ = this.onChangedFilterCondition$.pipe(
    withLatestFrom(this.state$),
    tap(([condition, state]) => {
      this.stateSubject.next({
        ...state,
        filterCondition: condition
      });
    })
  );

Subscriptionの管理

上記で実装したイベントハンドラのObservableをまとめてsubscribeし、イベントハンドリングが行われるようにします。
コンポーネント破棄後は、イベントハンドラが動いて欲しくないので、takeUntilにonDestroy$を指定します。
これで、コンポーネント破棄時にcompleteされるので、イベントハンドリング処理が残ることはありません。

  constructor() {
    merge(
      this.addItemHandler$,
      this.removeItemHandler$,
      this.changeFilterConditionHandler$,
      this.changeInputTitleHandler$,
    ).pipe(takeUntil(this.onDestroy$)).subscribe();
  }

最終的なソースコード

reactive-todo-page.component.ts

import { Component, OnInit, ChangeDetectionStrategy, OnDestroy, Input, Output } from '@angular/core';
import { v4 as uuidv4 } from 'uuid';
import { BehaviorSubject, combineLatest, Subject, merge, } from 'rxjs';
import { map, shareReplay, pluck, withLatestFrom, tap, takeUntil, distinctUntilChanged } from 'rxjs/operators';

type FilterCondition = 'completed' | 'all' | 'todo';
type TodoItem = { id: string; title: string; completed: boolean; };
type State = {
  items: TodoItem[];
  filterCondition: FilterCondition;
  title: string;
};

@Component({
  selector: 'app-reactive-todo-page',
  templateUrl: './reactive-todo-page.component.html',
  styleUrls: ['./reactive-todo-page.component.scss'],
  changeDetection: ChangeDetectionStrategy.OnPush
})
export class ReactiveTodoPageComponent implements OnInit, OnDestroy {
  @Input()
  set title(title: string) {
    this.onChangedInputTitle$.next(title);
  }

  readonly onChangedTitle$: Subject<string> = new Subject();
  @Output() changedTitle = this.onChangedTitle$;

  // Component State
  readonly stateSubject: BehaviorSubject<State> = new BehaviorSubject({
    items: [
      {id: uuidv4(), title: 'todo0', completed: false},
      {id: uuidv4(), title: 'todo1', completed: true},
      {id: uuidv4(), title: 'todo2', completed: false},
      {id: uuidv4(), title: 'todo3', completed: true},
      {id: uuidv4(), title: 'todo4', completed: false},
      {id: uuidv4(), title: 'todo5', completed: true},
    ],
    filterCondition: 'all',
    title: '',
  });
  readonly state$ = this.stateSubject.asObservable().pipe(
    distinctUntilChanged(),
    shareReplay({ bufferSize: 1, refCount: true }),
  );
  readonly title$ = this.state$.pipe(
    pluck('title'),
    distinctUntilChanged(),
    shareReplay({ bufferSize: 1, refCount: true }),
  );
  readonly items$ = this.state$.pipe(
    pluck('items'),
    distinctUntilChanged(),
    shareReplay({ bufferSize: 1, refCount: true }),
  );
  readonly filterCondition$ = this.state$.pipe(
    pluck('filterCondition'),
    distinctUntilChanged(),
    shareReplay({ bufferSize: 1, refCount: true }),
  );
  readonly filteredItems$ = combineLatest([
    this.items$,
    this.filterCondition$
  ]).pipe(
    map(([items, condition]) => {
      switch (condition) {
        case 'todo':
          return items.filter(v => !v.completed);
        case 'completed':
          return items.filter(v => v.completed);
        case 'all':
          return items;
      }
    }),
    distinctUntilChanged(),
    shareReplay({ bufferSize: 1, refCount: true }),
  );
  readonly filteredItemsCount$ = this.filteredItems$.pipe(
    map(items => items.length),
    distinctUntilChanged(),
    shareReplay({ bufferSize: 1, refCount: true }),
  );

  // Events
  readonly onAdd$ = new Subject();
  readonly onRemove$ = new Subject<Pick<TodoItem, 'id'>>();
  readonly onChangedFilterCondition$ = new Subject<FilterCondition>();
  readonly onChangedInputTitle$ = new Subject<string>();
  readonly onDestroy$ = new Subject();

  // Event Handler
  private readonly addItemHandler$ = this.onAdd$.pipe(
    withLatestFrom(this.state$),
    tap(([, state]) => {
      const newItem = {
        id: uuidv4(),
        title: uuidv4().slice(0, 5),
        completed: false
      };
      this.stateSubject.next({
        ...state,
        items: [...state.items, newItem],
      });
    })
  );
  private readonly removeItemHandler$ = this.onRemove$.pipe(
    withLatestFrom(this.state$),
    tap(([removeEvent, state]) => {
      const itemsAfterRemove = state.items.filter(v => v.id !== removeEvent.id);
      this.stateSubject.next({
        ...state,
        items: itemsAfterRemove
      });
    })
  );
  private readonly changeFilterConditionHandler$ = this.onChangedFilterCondition$.pipe(
    withLatestFrom(this.state$),
    tap(([condition, state]) => {
      this.stateSubject.next({
        ...state,
        filterCondition: condition
      });
    })
  );
  private readonly changeInputTitleHandler$ = this.onChangedInputTitle$.pipe(
    withLatestFrom(this.state$),
    tap(([title, state]) => {
      this.stateSubject.next({
        ...state,
        title
      });
    })
  );

  constructor() {
    merge(
      this.addItemHandler$,
      this.removeItemHandler$,
      this.changeFilterConditionHandler$,
      this.changeInputTitleHandler$,
    ).pipe(takeUntil(this.onDestroy$)).subscribe();
  }

  ngOnInit(): void {
  }

  ngOnDestroy(): void {
    this.onDestroy$.next();
  }

}

reactive-todo-page.component.html

<h2>{{ title$ | async }}</h2>
<button (click)="onChangedTitle$.next('Hoge')">change title to "Hoge"</button>
<p>filter condition: {{filterCondition$ | async}}, total: {{ filteredItemsCount$ | async }}</p>
<div>
  <button (click)="onChangedFilterCondition$.next('all')">all</button>
  <button (click)="onChangedFilterCondition$.next('todo')">todo</button>
  <button (click)="onChangedFilterCondition$.next('completed')">completed</button>
</div>
<button (click)="onAdd$.next()">add</button>
<ul>
  <ng-container *ngFor="let item of filteredItems$ | async">
    <li>
      <span>title: {{ item.title }}, completed: {{ item.completed }}</span>
      <button (click)="onRemove$.next({id: item.id})">remove</button>
    </li>
  </ng-container>
</ul>

これでリアクティブな実装になりました！
とはいえ、記述量が結構多くなってしまっています。

そこで、@rx-angular/stateというライブラリが活用できます。
このライブラリを使うことで、上記のコードと同じようなことができる、かつより簡潔に記述することができます。

@rx-angular/stateを利用した実装

@rx-angular/stateを利用した最終的な実装コードです。
templateの実装はリアクティブにした実装例と同様なので割愛します。

rx-angular-todo-page.component.ts

import { Component, OnInit, Input } from '@angular/core';
import { RxState, stateful } from '@rx-angular/state';
import { Subject, combineLatest } from 'rxjs';
import { map } from 'rxjs/operators';
import { v4 as uuidv4 } from 'uuid';

type FilterCondition = 'completed' | 'all' | 'todo';
type TodoItem = { id: string; title: string; completed: boolean; };

type State = {
  items: TodoItem[];
  filterCondition: FilterCondition;
  title: string;
};

@Component({
  selector: 'app-rx-angular-todo-page',
  templateUrl: './rx-angular-todo-page.component.html',
  styleUrls: ['./rx-angular-todo-page.component.scss'],
  providers: [RxState]
})
export class RxAngularTodoPageComponent {
  @Input()
  set title(title: string) {
    this.onChangedInputTitle$.next(title);
  }

  // Component State
  readonly state$ = this.state.$;
  readonly title$ = this.state.select('title');
  readonly items$ = this.state.select('items');
  readonly filterCondition$ = this.state.select('filterCondition');
  readonly filteredItems$ = combineLatest([
    this.items$,
    this.filterCondition$
  ]).pipe(
    stateful(
      map(([items, condition]) => {
        switch (condition) {
          case 'todo':
            return items.filter(v => !v.completed);
          case 'completed':
            return items.filter(v => v.completed);
          case 'all':
            return items;
        }
      }),
    )
  );
  readonly filteredItemsCount$ = this.filteredItems$.pipe(
    stateful(map(items => items.length))
  );

  // EVENTS
  readonly onAdd$ = new Subject<Pick<TodoItem, 'title'>>();
  readonly onRemove$ = new Subject<Pick<TodoItem, 'id'>>();
  readonly onChangedFilterCondition$ = new Subject<FilterCondition>();
  readonly onChangedInputTitle$ = new Subject<string>();
  readonly onChangedTitle$ = new Subject<string>();

  constructor(
    private state: RxState<State>
  ) {
    this.state.connect('title', this.onChangedInputTitle$);
    this.state.connect('items', this.onAdd$, (s, { title }) => (
      [
        ...s.items,
        {
          id: uuidv4(),
          title: `${title}${uuidv4().slice(0, 5)}`,
          completed: false
        }
      ]
    ));
    this.state.connect('items', this.onRemove$, (s, { id }) => (
      s.items.filter(v => v.id !== id)
    ));
    this.state.connect('filterCondition', this.onChangedFilterCondition$);
    this.state.set({
      items: [
        {id: uuidv4(), title: 'todo0', completed: false},
        {id: uuidv4(), title: 'todo1', completed: true},
        {id: uuidv4(), title: 'todo2', completed: false},
        {id: uuidv4(), title: 'todo3', completed: true},
        {id: uuidv4(), title: 'todo4', completed: false},
        {id: uuidv4(), title: 'todo5', completed: true},
      ],
      filterCondition: 'all',
      title: '',
    });
  }

}

DIされているRxState<State> 型のstateが、コンポーネントの状態と、簡潔にリアクティブに実装するためのメソッドをもっています。

この例の this.state.$が、@rx-angular/stateを使う前のthis.stateSubject.asObservable()に相当します。

メンバ変数の定義

ここでは、stateオブジェクトに含まれてる値をObservableな値として取得する時に、this.state.selectを利用しています。
RxStateのselectの内部で、distinctUntilChangedやshareReplayが指定されているので、開発者が毎回これらのオペレータを指定する必要がなくなっています。

また、filteredItems$はstateオブジェクトのに含まれていないので、selectできません。
なので、自分でdistincutUntilChangedやshareReplayを指定する必要があります。

ただ、@rx-angular/stateがstatefulという独自オペレータも提供しており、このオペレータを使うことで、distinctUntilChangedやshareReplayの恩恵を受けることができます。

  readonly filteredItems$ = combineLatest([
    this.items$,
    this.filterCondition$
  ]).pipe(
    stateful(
      map(([items, condition]) => {
        switch (condition) {
          case 'todo':
            return items.filter(v => !v.completed);
          case 'completed':
            return items.filter(v => v.completed);
          case 'all':
            return items;
        }
      }),
    )
  );

selectの内部でもこのstatefulオペレータは利用されています。

ちなみにstatefulオペレータの実装は下記のようになっています。
https://github.com/BioPhoton/rx-angular/blob/master/libs/state/src/lib/rxjs/operators/stateful.ts#L110

selectやstatefulオペレータを利用することで、メンバ変数の定義がだいぶスッキリしました。

イベントハンドラの実装

イベントハンドラの実装は下記のようになっています。

    this.state.connect('title', this.onChangedInputTitle$);
    this.state.connect('items', this.onAdd$, (s, { title }) => (
      [
        ...s.items,
        {
          id: uuidv4(),
          title: `${title}${uuidv4().slice(0, 5)}`,
          completed: false
        }
      ]
    ));
    this.state.connect('items', this.onRemove$, (s, { id }) => (
      s.items.filter(v => v.id !== id)
    ));
    this.state.connect('filterCondition', this.onChangedFilterCondition$);

connectは、イベントと、コンポーネントの状態の関連づけをおこなっています。

@rx-angular/stateを利用する前の実装と比較すると下記の通りです。

  // 利用前
   withLatestFrom(this.state$),
    tap(([removeEvent, state]) => {
      const itemsAfterRemove = state.items.filter(v => v.id !== removeEvent.id);
      this.stateSubject.next({
        ...state,
        items: itemsAfterRemove
      });
    })
  );

  // 利用後
  this.state.connect('items', this.onRemove$, (oldItems, { id }) => (
    oldItems.items.filter(v => v.id !== id)
  ));

この実装例だと、connectの第二引数で指定したObservableに値が流れた時、第三引数で指定した関数を、stateの第一引数で指定したプロパティに適用するという動きになります。

connectで記述することで簡潔になる要因としては下記のようなものがあるかなと思います。

第一引数にstateオブジェクトのプロパティを指定することで、第三引数の関数で気にしなきゃいけない値は、指定されたプロパティだけになる。
第三引数の関数では、引数で状態の現在値を受け取ることができるようになっている。
明示的なnextが不要

connectの使い方はいくつかあるので、詳細はドキュメントをお読みください。
https://github.com/BioPhoton/rx-angular/blob/master/libs/state/docs/api/rx-state.md#connect

ちなみに、コンポーネントの状態に直接関連しないイベントハンドリングを行う場合は、holdを利用します。
例えば、コンポーネントの状態を更新するのではなく、ngrxのdispatchを実行するようなイベント処理などです。
https://github.com/BioPhoton/rx-angular/blob/master/libs/state/docs/api/rx-state.md#hold

Subscriptionの管理

@rx-angular/stateが内部でSubscriptionの管理を行うので、もう自分で管理する必要はありません！
https://github.com/BioPhoton/rx-angular/blob/master/libs/state/src/lib/rx-state.service.ts#L78

まとめ

リアクティブにかきたい。
安全にリアクティブにかくと記述量が多くなる。
@rx-angular/stateを使うことで、安全に、より簡潔に実装することができそう。