Observable と Async Generator / Async Iterator の比較

はじめに

過去記事で、Observable と async/await　を比較しましたが、どうやらAsync Generator / Async Iteratorなる便利なものがあるみたいなので以下にまとめます。

非同期処理を扱う方法は、JavaScript/TypeScriptの進化とともに多様化しています。この記事では、非同期ストリームを扱う2つの主要な方法である Observable と Async Generator / Async Iterator の違いについて詳しく説明します。どちらを選ぶべきか、どのようなユースケースに適しているのかを理解していきましょう。

1. Observable とは？

Observable は リアクティブプログラミングの概念を取り入れたもので、非同期データストリームを扱うための仕組みです。
rxjsライブラリを使用して実現され、イベントの購読（subscribe）やストリームの操作（フィルタリングやマッピングなど）を行えます。

基本例

import { Observable } from 'rxjs';

const observable = new Observable<number>((subscriber) => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
  subscriber.complete();
});

observable.subscribe({
  next: (value) => console.log(`Received: ${value}`),
  complete: () => console.log('Stream complete'),
});

特徴

プッシュ型: データはストリーム側から一方的にプッシュされます。
遅延評価: 実際に購読（subscribe）するまでストリームは動作しません。
高機能なストリーム操作: RxJSの演算子（map, filter, merge, etc.）を利用可能。

2. Async Generator / Async Iterator とは？

Async Generator は非同期処理を連続的に生成する仕組みです。内部ではawaitを使えるため、非同期処理を自然な形で表現できます。データは イテレーション（反復）によって1つずつ取り出されます。

基本例

async function* asyncGenerator() {
  yield 1;
  yield await new Promise<number>((resolve) => setTimeout(() => resolve(2), 1000));
  yield 3;
}

const iterator = asyncGenerator();

(async () => {
  for await (const value of iterator) {
    console.log(`Received: ${value}`);
  }
})();

特徴

プル型: データは使用者（for await...ofなど）からリクエストされて初めて生成されます。
シンプルな構文: for await...ofで非同期ストリームを簡潔に扱える。
柔軟性: Observableほどの演算子はありませんが、生成処理の自由度が高い。

3. Observable と Async Generator の比較

特徴	Observable	Async Generator
データの流れ	プッシュ型（ストリームが主導）	プル型（利用者が主導）
遅延評価	購読時に開始	初回のイテレーションで開始
エラー処理	内部で処理可能（`catchError`）	try-catchで処理可能
ストリーム操作の機能	RxJS演算子が豊富	標準の構文のみ
ユースケース	高度なストリーム操作	シンプルな反復処理

4. 適切なユースケースの選択

Observable を選ぶべき場合

データストリームを高度に操作する必要がある場合（例: フィルタリング、結合、スロットリングなど）。
ユーザーインターフェースやリアルタイムデータ更新を扱う場合（例: ボタンのクリックイベント、WebSocketデータ）。
RxJSライブラリをプロジェクトで使用している場合。

Async Generator / Async Iterator を選ぶべき場合

シンプルな非同期ストリームが必要な場合（例: データの逐次読み込み）。
ライブラリを使わずにネイティブな機能で非同期処理を実装したい場合。
プル型のモデルが適している場合（例: API呼び出し結果を順次処理）。

5. 両者を組み合わせる

ObservableとAsync Generatorは完全に排他的な関係ではありません。必要に応じて、以下のように互いに変換することも可能です。

ObservableからAsync Generatorへの変換

import { Observable } from 'rxjs';

async function* observableToAsyncGenerator<T>(observable: Observable<T>): AsyncGenerator<T> {
  return async function* () {
    const values: T[] = [];
    const complete = false;
    const sub = observable.subscribe({
      next: (value) => values.push(value),
      complete: () => {
        complete = true,
      }
    });

    try {
      while (!complete || values.length > 0) {
        if (values.length > 0) {
          yield values.shift();
        } else {
          await new Promise((resolve) => setTimeout(resolve, 1000)
        }
      }
    } finally {
      sub.unsubscribe();
    }
  };
}

Async GeneratorからObservableへの変換

import { Observable } from 'rxjs';

function asyncGeneratorToObservable<T>(asyncGenerator: AsyncGenerator<T>): Observable<T> {
  return new Observable<T>(async (subscriber) => {
    try {
      for await (const value of asyncGenerator) {
        subscriber.next(value);
      }
      subscriber.complete();
    } catch (error) {
      subscriber.error(error);
    }
  });
}

おわりに

ObservableとAsync Generatorは、それぞれ得意な分野を持つ非同期ストリームのアプローチです。プロジェクトの要件に応じて、どちらを採用するかを選択しましょう。また、両者を適切に変換して利用することで、より柔軟な非同期処理を実現できます。