はじめに

この記事を書いたきっかけ

担当しているプロジェクトでAngularを利用したフロントエンドのWebアプリケーションを作ることになり、RxJSの概念が非常にわかりづらく、つまづきまくりました。

そこでRxJSとは何かを理解するために、RxJSについて整理することにしました。
私のように命令型プログラミングしかやってきておらず、リアクティブプログラミングのコードを見てなにこれ？となった方の助けになれば幸いです。

対象読者

AngularなどのフロントエンドWebアプリケーションフレームワークを使う方
いわゆる命令型プログラミングをずっとやってきて、リアクティブプログラミングでつまづいた方

RxJSの概要

RxJS (Reactive Extensions for JavaScript)とは、リアクティブプログラミングを行うためのJavaScriptライブラリ。

リアクティブプログラミングとは？

リアクティブプロぐらいミングとはなにか？については以下のサイトの説明が非常にわかりやすかったです。（説明丸投げ）

リアクティブプログラミングとは、データが流れるように来ること（ストリーム）に着目し、データを受け取るたびに関連したプログラムが反応（リアクション）して処理を行うようにするプログラミングの考え方です。

参考:[https://codezine.jp/article/detail/9570](https://codezine.jp/article/detail/9570)

何のために（何を解決するために）作られたのか？

バックエンドではWebAPIが呼び出されると処理を行い、レスポンスを返すという一連の流れをすればよいですが、フロントエンドだとユーザの操作やサーバ側との通信などのイベントが非同期で発生します。

非同期で発生するイベントに対して同時で並行処理をするというニーズが発生し、これを解決するのにリアクティブプログラミングのほうが相性が良いため様々なフロントエンドのフレームワークで採用していったのだろうと考えています。

(私が)わかりにくいと感じる理由

Javaのような命令型プログラミングでは、命令されると一連の処理をコードの記載順に処理をしていきます。

命令型プログラミングの場合は処理の流れを追いかける場合に、コードを上から読んでいけば処理の流れがわかります。

あるクラスのメソッドが呼び出されるとそのメソッド内でほかのメソッドを呼び出して・・・、というように追っていけば処理の流れがわかる、ということです。

しかし、リアクティブプログラミングの場合はどのようなデータがどのタイミングで発生するのか？データが発生した場合は誰がリアクションするのか？がコードの様々な箇所に分散するためコードから処理の流れを理解するのが難しいと感じていました。

これは例えば1つのデータが流れた場合に、複数のリアクションが発生して並行して処理が行われる、リアクションの処理はデータの流れとは別の箇所にコードが記載されているということが起きるためです。

RxJSの構成要素

ここからは、RxJSで重要となるObservableとObserverを説明します。

Observable

「データが流れるように来ること（ストリーム）」

のデータを流す役割を持つ。

Observer

「データを受け取るたびに関連したプログラムが反応（リアクション）して処理を行う」

のデータを受け取ってアクションをする役割を持つ。

ObservableとObserverのやり取り

基本的な流れは以下の通りです。

Observerはデータの発信元であるObservableに対してsubscribe（購読）を行う。
Observableは、データを流す際に購読しているObserver対してデータが来た旨の通知を行う。
Observerはデータを受け取ると、これに対して反応して処理を行う

Observerの通知パターンはどのようなデータが流れてくるかを気にして、その後のリアクションの処理だけに専念すればよいことになるため、データを流す側とデータを受け取って処理をする側を分離することができます。また、同じデータに対して複数の処理をする場合は1つのObservableに対して複数のObserverがsubscribeすれば、subscribeしたObserverすべてに通知がされます。

Observableの通知パターン

Observableの通知パターンは、データを流す際の状況に応じて以下の3つがあり、Observerはそれぞれの通知パターンに対するリアクションを用意しておく必要があります。

next
- Observableがデータを正常に流せた場合
- Observerはデータが正常に流れてきた際にどのようなリアクションをするかを定義
error
- Observableがデータを正常に流せなかった場合
- Observerはデータが正常に流れてこずにエラーが発生した際にどのようなリアクションをするかを定義
complete
- Observableがすべてのデータを正常に流し終わった後
- Observerはデータがすべて流れ終わった後にどのようなリアクションをするかを定義

errorとcompleteはどちらか一方しか発生しません。つまりnextが何回か（0回の場合もあり）通知された後に、errorまたはcompleteのどちらかが通知されることになります。
Observableがデータを流し、2つのobserverが受け取る際のサンプルコードはこちらです。
Rxの処理イメージとして以下のサイトがわかりやすかったので、川に桃やヤマメが流れていく、というイメージにしています。

https://qiita.com/MasanobuAkiba/items/a5026bd37603cc29e9e7#rx-のイメージ

		// '桃', 'ヤマメ'というデータが流れるObservableを作成しています。
    let dataRiver:Observable<string> = of('桃', 'ヤマメ');

    // observer1がデータを受け取った時の処理を定義します。
    let observer1:Observer<string> = {
      next: (item:string) => console.log(`observer1が${item}を受け取りました`),
      error: (error) => console.error(error),
      complete: () => console.log(`observer1は完了しました`)
    }
    // observer2がデータを受け取った時の処理を定義します。
    let observer2:Observer<string> = {
      next: (item:string) => console.log(`observer2が${item}を受け取りました`),
      error: (error) => console.error(error),
      complete: () => console.log(`observer2は完了しました`)
    }
    // observer1, observer2がdataRiverをsubscribeします
    dataRiver.subscribe(observer1);
    dataRiver.subscribe(observer2);

実行すると以下のようなログが出力されます。

'observer1が桃を受け取りました'
'observer1がヤマメを受け取りました'
'observer1は完了しました'
'observer2が桃を受け取りました'
'observer2がヤマメを受け取りました'
'observer2は完了しました'

observer1, observer2がdataRiver(Observable)をsubscribe(購読)。
dataRiverはデータ(’桃’, ‘ヤマメ’)を流すと、observer1, observer2が結果を取得して定義された処理を実行するという流れです。

さいごに

まずはRxJSの基本概念とObservable,Observerについて整理してみました。
これだけでも、最初は理解するのにかなり戸惑いました・・・。
Observableを扱う上では、pipeとoperatorの理解が必要となりますので、次はこのあたりについて整理していく予定です。

参考サイト