ReactでRxJSは流行するのか？

こんにちは、エビリーの新規プロダクト開発を担当しているイサミと申します。

私は、今年の６月までは Unity を用いたゲーム制作のお仕事をしておりました。
前職では、イベント通知やネットワーク処理などの非同期関連の処理において、頻繁に UniRx を使用していました。

フロントエンドでも、ゲームと似たような非同期処理はたくさんあります。
しかし、フロントエンドを勉強する過程で Rx についてほとんど聞いたことがないのが現状・・・。
本記事は、便利な機能であるはずの Rx が、なぜフロントエンドに浸透していないのかを調べてみた雑記です。

時間のない方向けに、結論を先に書いてしまいますと
「可能性はありそうだけど、まだ数年は掛かりそう。」
というものが所感でした。

便利なのは間違いないのですが、「必要性の低さ」と「学習コストの高さ」が問題だと感じました。
詳しく気になる方は、記事読んでね 😎。

---

目次：
・Rx(Reactive Extensions)ってなんなの？
・RxJS を導入したらやりたいこと
・まとめ

Rx(Reactive Extensions)ってなんなの？

この記事を読まれてる方の中には Rx を知らない方もいると思います。
そんな方たちのために、まずは簡単な説明を挟んでみます。

あ、知ってる人は読み飛ばしていただいて結構です。
→ 次の章 （RxJS を導入したらやりたいこと）

The Reactive Extensions (Rx) is a library for composing asynchronous and event-based programs using observable sequences and LINQ-style query operators.

Reactive Extensions (Rx) は、観測可能なシーケンスと LINQ スタイルのクエリ演算子を用いて、非同期およびイベントベースのプログラムを構成するためのライブラリである。
引用：「dotnet/reactive」

RxJS 派生元の、ご本家様から持ってきました。
はい、ナンノコッチャですね。もう少し簡単にします。

• 観測可能なシーケンス
  → 「川、回転寿司、ベルトコンベアー」のような何かが流れてくるもの
• LINQ スタイルのクエリ演算子
  → Array の map や filter のようなもの
• 非同期及びイベントベース
  → イベントを待ち受けるように処理を書く（ex:ボタンが押されるまで待つ）

うーむ、これでも今ひとつピンときませんね。

よくある「ボタンを押したらカウントが上がる」コードを通常版と Rx 版で比較してみましょう。

---

まずは通常版、

index.tsx

import { useState } from "react";

export default function Home() {
  const [count, setCount] = useState(0);
  const countUp = () => {
    setCount(count + 1);
  };

  return (
    <div>
      <p>Count: {count}</p>
      <button onClick={countUp}>Count +1</button>
    </div>
  );
}

次は Rx 版、

index.tsx

import { useState } from "react";
import { Subject } from "rxjs";

export default function Home() {
  const [count, setCount] = useState(0);

  const subject = new Subject<
    React.MouseEvent<HTMLButtonElement, MouseEvent>
  >();

  subject.subscribe((_) => {
    setCount(count + 1);
  });

  return (
    <div>
      <p>Count: {count}</p>
      <button onClick={(e) => subject.next(e)}>Count +1</button>
    </div>
  );
}

早速、見慣れないモノが出てきましたね。

const subject = new Subject<React.MouseEvent<HTMLButtonElement, MouseEvent>>();

Subjectは、「観測可能なシーケンス」そのものと思ってください。

Subjectには、２つの機能が存在します。

１． 観測可能な値を送出する
送出する値にも種類があり、値の種類によって以下のように分類されます。
next → 正常値
error → 異常値
complete → シーケンスの終了を知らせる

<button onClick={(e) => subject.next(e)}>Count +1</button>

ここの行では、subjectに対してクリックイベントが格納された変数eを送出していますね。
ボタンがクリックされる度に、subjectに対して毎回クリックイベントeが渡されます。
この動作をするものを Rx では、Observable(観測可能)と呼びます。

２．観測者を登録する
subjectのsubscribeメソッドを呼び出してあげることで、観測者を登録できます。

subject.subscribe((_) => {
  setCount(count + 1);
});

ここの行は、subjectの観測者を登録しています。
ここで登録した処理は、subjectに何かしらの値が送出される度に発火します。
この動作をするものを Rx では、Observer(観測者)と呼びます。

---

「あれ、なんか処理長くなってない？」

はい、Rx を使うと単純な処理の場合は複雑度が増してしまうことがあります。
しかし、複雑な処理を実装しようとする場合、Rx を使うとより楽に見やすく実装できるパターンがあります。
次の章では、RxJS を導入したときに得られるであろうメリットを考えてみます。

もっと詳しく RxJS を知りたい方は公式サイトがわかりやすいです。

https://rxjs.dev/

RxJS を導入したらやりたいこと

Rx では、オペレータによるイベントの多様な制御や非同期による待機が可能です。
今回は実践でも使いそうな例を２つ、実装してみます。

長押し判定

ドラッグ＆ドロップの操作を行うときなどには、クリックイベントと分類分けするために長押し判定を行うことがあります。
この長押しですが、単純に実装しようとすると結構面倒くさいです。
ですが、Rx を使えば驚くほど直感的なコードで実装が可能となります。

index.tsx

import { useEffect, useRef, useState } from "react";
import { delay, fromEvent, mergeMap, of, takeUntil } from "rxjs";

const longPressMilliSeconds = 500;

export default function Home() {
  const buttonRef = useRef(null);
  const [isLongPress, setIsLongPress] = useState(false);

  useEffect(() => {
    if (!buttonRef.current) return;
    const mouseDown$ = fromEvent(buttonRef.current, "mousedown");
    const mouseUp$ = fromEvent(buttonRef.current, "mouseup");

    // 長押し(0.5秒)されたらイベントを送出するストリームを作成
    const longPress$ = mouseDown$.pipe(
      mergeMap((e) => {
        return of(e).pipe(delay(longPressMilliSeconds), takeUntil(mouseUp$));
      })
    );

    // 長押しのストリームにイベントが流れてきたならば、isLongPress を true にする
    const longPressSubscription = longPress$.subscribe(() => {
      setIsLongPress(true);
    });

    // mouseup のストリームにイベントが流れてきたならば、isLongPress を false にする
    const mouseUpSubscription = mouseUp$.subscribe(() => {
      setIsLongPress(false);
    });

    return () => {
      longPressSubscription.unsubscribe();
      mouseUpSubscription.unsubscribe();
    };
  }, []);

  return (
    <div>
      <p>{isLongPress ? "長押ししてるよ" : "長押ししてないよ"}</p>
      <button ref={buttonRef}>長押しボタン</button>
    </div>
  );
}

上記のコードは、大きく分類すると以下の４段階となります。

1. mouseUp と mouseDown のストリーム^[1]を作成する
2. mouseUp と mouseDown を使用して、長押し判定のストリームを作成する
3. 長押しストリームにイベントが流れてきたならば、判定を True に更新する
4. mouseUp にイベントが流れてきたならば、判定を False に更新する

特に重要な箇所は３と４です。
長押し処理に必要な最もコアな機能を完結に実装することが出来ています。

複数の fetch イベントを待機する

通常、データを Fetch してくる場合は Promise を使ってデータの取得を待機します。
Rx では Promise の代わりに Observable を使用してデータの取得を待機します。
Observable で待機するメリットは、データの変化やタイムアウトなどの処理の差し込みを容易にすることです。

index.tsx

import { useFetchSample } from "../hooks/useFetchSample";

export default function Home() {
  const [data1, data2, dataAll] = useFetchSample();
  let dispData1 = "未取得";
  let dispData2 = "未取得";
  let dispDataAll = "未取得";

  if (data1) {
    dispData1 = data1.error ? data1.message : "取得完了！";
  }
  if (data2) {
    dispData2 = data2.error ? data2.message : "取得完了！";
  }
  if (dataAll) {
    dispDataAll = dataAll.error ? dataAll.message : "全件処理完了";
  }

  return (
    <div>
      <p>Data1 : {dispData1}</p>
      <p>Data2 : {dispData2}</p>
      <p>DataAll : {dispDataAll}</p>
    </div>
  );
}

useFetchSample.tsx

import { useEffect, useState } from "react";
import {
  catchError,
  delay,
  lastValueFrom,
  of,
  Subscription,
  switchMap,
  timeout,
} from "rxjs";
import { fromFetch } from "rxjs/fetch";

// データの取得サンプル１
const fetchSample1 = () => {
  return fromFetch("https://jsonplaceholder.typicode.com/posts/1").pipe(
    delay(1000), // 重い処理を再現（１秒待機）
    switchMap((response) => {
      if (response.ok) {
        return of({ error: false, body: response.json() });
      } else {
        return of({ error: true, message: `Error ${response.status}` });
      }
    }),
    catchError((err) => {
      // Network or other error, handle appropriately
      console.error(err);
      return of({ error: true, message: err.message });
    })
  );
};

// データの取得サンプル２
const fetchSample2 = () => {
  return fromFetch("https://jsonplaceholder.typicode.com/posts/2").pipe(
    delay(2000), // 重い処理を再現（２秒待機）
    switchMap((response) => {
      if (response.ok) {
        return of({ error: false, body: response.json() });
      } else {
        return of({ error: true, message: `Error ${response.status}` });
      }
    }),
    catchError((err) => {
      // Network or other error, handle appropriately
      console.error(err);
      return of({ error: true, message: err.message });
    })
  );
};

export const useFetchSample = () => {
  const [data1, setData1] = useState<any>();
  const [data2, setData2] = useState<any>();
  const [dataAll, setDataAll] = useState<any>();

  useEffect(() => {
    const data1$ = fetchSample1();
    const data2$ = fetchSample2();

    const subscriptions = new Subscription();

    // Fetchを単体で待ち受けることも・・・
    subscriptions.add(
      data1$.subscribe((d1) => {
        setData1(d1);
      })
    );

    subscriptions.add(
      data2$.subscribe((d2) => {
        setData2(d2);
      })
    );

    // Fetchをすべて完了するまで待機することも可能
    Promise.all([lastValueFrom(data1$), lastValueFrom(data2$)]).then(
      ([d1, d2]) => {
        setDataAll([d1, d2]);
      }
    );

    return () => {
      subscriptions.unsubscribe();
    };
  }, []);

  return [data1, data2, dataAll];
};

fromFetchの処理に注目してみます。
fromFetchに続くpipeメソッドで、データ取得の流れに処理を追加しています。

delay
データの Fetch が完了して Observable に取得データが流れてきたあとに、１秒間待機を行っています。

    delay(1000), // 重い処理を再現（１秒待機）

switchMap
取得したデータのステータスを判断し、新しいデータを作成しています。

    switchMap((response) => {
      if (response.ok) {
        return of({ error: false, body: response.json() });
      } else {
        return of({ error: true, message: `Error ${response.status}` });
      }
    }),

catchError
ネットワークエラーなど、例外が投げられた場合の処理を実装しています。

catchError((err) => {
  // Network or other error, handle appropriately
  console.error(err);
  return of({ error: true, message: err.message });
});

switchMapとcatchErrorは、Promise のthenとcatchに該当するので Rx にするメリットはあまり感じられません。

しかし、delayは話が違います。
Promise の場合は、setTimeoutの使用したsleep関数のようなものを自前で用意し、await で待機するなどの実装が必要です。

似たように、setTimeoutの使用が必要な処理では、タイムアウト実装も Promise だと面倒な部類に入ります。

Rx ではタイムアウト処理も１行で実装が出来てしまいます。
先程の処理に追加してみましょう。

const fetchSample2 = () => {
  return fromFetch("https://jsonplaceholder.typicode.com/posts/2").pipe(
    delay(2000), // 重い処理を再現（２秒待機）
+   timeout(1500), // １．５秒後にタイムアウトエラーを投げる
    switchMap((response) => {
      if (response.ok) {
        return of({ error: false, body: response.json() });
      } else {
        return of({ error: true, message: `Error ${response.status}` });
      }
    }),
    catchError((err) => {
      // Network or other error, handle appropriately
      console.error(err);
      return of({ error: true, message: err.message });
    })
  );
};

え？タイムアウトは axios 使えば簡単に実装できるって？？？

では、タイムアウトエラーなどが発生した場合に２回までリトライする機能を実装してみます。
Rx ではリトライ処理も１行で実装が可能です。

const fetchSample2 = () => {
  return fromFetch("https://jsonplaceholder.typicode.com/posts/2").pipe(
    delay(2000), // 重い処理を再現（２秒待機）
    timeout(1500), // １．５秒後にタイムアウトエラーを投げる
+   retry(2), // エラーが発生した場合２回までリトライする
    switchMap((response) => {
      if (response.ok) {
        return of({ error: false, body: response.json() });
      } else {
        return of({ error: true, message: `Error ${response.status}` });
      }
    }),
    catchError((err) => {
      // Network or other error, handle appropriately
      console.error(err);
      return of({ error: true, message: err.message });
    })
  );
};

まとめ

Rx を使用することで、イベントの通知や非同期処理を Promise ではなく Observable として扱うことができるようになります。

ボタンのクリックイベントや、fetch イベントなどはファクトリメソッド(fromEvent, fromFetch)から Observable に変換することが容易にできることがわかりました。

そして、Observable なものに対しては pipe を通して多様な処理を追加することが可能でした。

これらの特徴を活用することで、今回実装したサンプルのように
Promise 「面倒で複雑な実装」→ Observable 「完結で直感的な実装」
へ置き換えることに成功しました。

■ この結果からも、「Rx には十分なポテンシャルがある」と考えています。

---

しかしながら、「fetch 処理のタイムアウトは axios でカバーできる」といったように、ライブラリを導入すれば解決できる問題も多くあります。

■ Rx の導入によって開発効率が大幅に上がるというケースは、ほとんどないと感じています。

---

また、今回は簡単に Rx を紹介しましたが、バグを引き起こさないように正しく使用するためにはかなりの知識が必要です。

今回のサンプルで言えば、何気なく書いていたunsubscribeや、pipeに渡していたメソッドの順番など、ここを間違えると解決困難なバグを発生させます。

紹介していない内容では、ストリームの Cold → Hot 変換や Schedule による実行タイミングの制御など、難易度が高い知識が裏に隠れています。

数多くの Operator の挙動を覚えることも必要になるでしょう。

■ Rx を正しく使うためには、全く新しい概念を１から勉強しなくてはなりません。

---

さて、冒頭にも書いた内容で改めてまとめると、

「可能性はありそうだけど、まだ数年は掛かりそう。」

便利なのは間違いないのですが、「必要性の低さ」 と 「学習コストの高さ」 が問題だと感じました。

その他

rxjs-hooksというものが便利そうなので、次回はこちらの内容を書くかもしれません。

脚注

1. Subject の Observable な機能