Angular RxJS と Signals の使い分け｜toSignal判断フロー

Angular の RxJS と Signals の使い分けは、「時間の流れ（いつ・何回・どの順で起きるか）を扱うなら RxJS、画面に出す“今の値”を持つなら Signal」です。 両者は @angular/core/rxjs-interop の toSignal() / toObservable() でつなげるので、「イベントは RxJS で整えて、最後に signal にして表示する」という組み合わせが基本形になります。

この記事では、Angular 22.2.0 で実際にプロジェクトを作り、検索入力（debounce・switchMap）、HTTP 取得、ポーリング、イベントストリーム、フォーム値、グローバル状態の6ケースを コピペで動くコード で書きました。どれを選ぶかの 判断フロー と ケース別の対応表、toSignal を注入コンテキストの外で呼んだときの NG0203、requireSync で同期値がないときの NG0601 など 実際のエラー文、Vitest と Playwright（Chromium）での 実行結果 もそのまま載せています。

Angular RxJS と Signals の使い分け早見表

迷ったときの判断フロー

「値が1つあればよいか」「時間やキャンセルの制御が要るか」の2点で決まります。 次のフローで上から順に答えていくと、使う API が1つに絞れます。

flowchart TD
    A[扱いたいものは?] --> B{画面に出す<br>今の値だけ?}
    B -- はい --> S[signal / computed]
    B -- いいえ --> C{HTTPで取得する?}
    C -- はい --> D{Observableを返す<br>既存サービスがある?}
    D -- いいえ --> H[httpResource]
    D -- はい --> R[rxResource]
    C -- いいえ --> E{待つ・間引く・キャンセル<br>などの時間制御が要る?}
    E -- はい --> T[RxJSで整えてから toSignal]
    E -- いいえ --> F{イベントが次々に来る?<br>WebSocket・通知など}
    F -- はい --> G[scanで状態にして toSignal]
    F -- いいえ --> S

ポイントは「最後は signal にして表示する」ことです。 テンプレートでは async パイプの代わりに items() のように読むだけになり、computed で派生値も作れます。逆に「signal の変化を待って何かする（遅らせる・まとめる）」ときだけ toObservable で RxJS 側へ渡します。

検証環境とプロジェクトの作り方

この記事のコードはすべて Angular 22.2.0 / RxJS 7.8.2 / TypeScript 6.0.3 / Node.js 24.21.0 で動かし、テストは Vitest 5.0.2、画面確認は Playwright 1.63.0（Chromium）で行いました。 Angular 22.2 の CLI は Node.js ^22.22.3・^24.15.0・>=26.0.0 のいずれかが必要です。

npx -y -p @angular/cli@22 ng new rxdemo --skip-git --defaults --style=css --ssr=false --routing --ai-config=none
cd rxdemo

学習用のモックAPI（HTTP インターセプター）

サーバーを立てずに試せるよう、/api/* をブラウザ内で返すインターセプターを用意します。 以降のコードはこれを前提にしています。

// src/app/mock-backend.ts
import { HttpInterceptorFn, HttpResponse } from '@angular/common/http';
import { of } from 'rxjs';
import { delay } from 'rxjs/operators';

export interface Item {
  id: number;
  name: string;
  stock: number;
}

const ITEMS: Item[] = [
  { id: 1, name: 'ボールペン', stock: 120 },
  { id: 2, name: 'ペンケース', stock: 8 },
  { id: 3, name: 'ノート', stock: 45 },
  { id: 4, name: 'ホチキス', stock: 12 },
  { id: 5, name: 'ペン立て', stock: 3 },
];
let statsCount = 0;

/** /api/* をブラウザ内で返すモックAPI（学習用） */
export const mockBackend: HttpInterceptorFn = (req, next) => {
  const url = new URL(req.urlWithParams, 'http://localhost');
  if (!url.pathname.startsWith('/api/')) return next(req);

  let body: unknown;
  if (url.pathname === '/api/items') {
    const q = url.searchParams.get('q') ?? '';
    body = ITEMS.filter((i) => i.name.includes(q));
  } else if (url.pathname.startsWith('/api/items/')) {
    const id = Number(url.pathname.split('/').pop());
    body = ITEMS.find((i) => i.id === id) ?? null;
  } else if (url.pathname === '/api/stats') {
    body = { checkedAt: ++statsCount, lowStock: ITEMS.filter((i) => i.stock < 10).length };
  }
  console.log(`[mock] ${req.method} ${req.urlWithParams}`);
  return of(new HttpResponse({ status: 200, body })).pipe(delay(200));
};

src/app/app.config.ts の providers には provideHttpClient(withInterceptors([mockBackend])) を追加します（provideHttpClient と withInterceptors は @angular/common/http から読み込み）。

// src/app/item-api.ts
import { HttpClient } from '@angular/common/http';
import { Injectable, inject } from '@angular/core';
import { Item } from './mock-backend';

@Injectable({ providedIn: 'root' })
export class ItemApi {
  private readonly http = inject(HttpClient);

  search(q: string) {
    return this.http.get<Item[]>('/api/items', { params: { q } });
  }
  get(id: number) {
    return this.http.get<Item>(`/api/items/${id}`);
  }
  stats() {
    return this.http.get<{ checkedAt: number; lowStock: number }>('/api/stats');
  }
}

rxjs-interop の API 一覧

@angular/core/rxjs-interop が公開している API は7つです（22.2.0 の型定義で確認）。

HTTP 専用の httpResource は @angular/common/http から読み込みます（こちらも型定義で publicApi 22.0）。

検索入力：debounce と switchMap の結果を signal で表示する

入力値は signal、時間の制御は RxJS、表示は signal という分担にします。 toObservable で signal を Observable にし、debounceTime で待ち、switchMap で古いリクエストを取り消してから、toSignal で signal に戻します。

// src/app/demos/item-search.ts
import { Component, inject, signal } from '@angular/core';
import { toObservable, toSignal } from '@angular/core/rxjs-interop';
import { debounceTime, distinctUntilChanged, map, switchMap } from 'rxjs';
import { ItemApi } from '../item-api';

@Component({
  selector: 'app-item-search',
  template: `
    <input id="q" placeholder="備品を検索" [value]="query()" (input)="query.set($any($event.target).value)" />
    <p id="hit">{{ results().length }} 件</p>
    <ul>
      @for (item of results(); track item.id) {
        <li>{{ item.name }}（在庫 {{ item.stock }}）</li>
      }
    </ul>
  `,
})
export class ItemSearch {
  private readonly api = inject(ItemApi);

  /** 入力値は signal で持つ */
  readonly query = signal('');

  /** 時間の扱い（debounce・キャンセル）は RxJS、表示は signal */
  readonly results = toSignal(
    toObservable(this.query).pipe(
      map((q) => q.trim()),
      debounceTime(300),
      distinctUntilChanged(),
      switchMap((q) => this.api.search(q)),
    ),
    { initialValue: [] },
  );
}

initialValue: [] を渡すと型が Signal<Item[]> になり、テンプレートで undefined を気にせず書けます。 渡さない場合の型は Signal<Item[] | undefined> です。

Vitest で「API は1回だけ呼ばれる」を確かめる

Vitest のフェイクタイマーで時間を進め、3回素早く入力しても API 呼び出しが最後の1回だけになることを確認しました。

// src/app/demos/item-search.spec.ts（要点）
beforeEach(() => vi.useFakeTimers());
afterEach(() => vi.useRealTimers());

it('素早く3回入力しても API は最後の1回だけ呼ばれる', async () => {
  const search = vi.fn((q: string) => of([{ id: 1, name: `${q}の結果`, stock: 1 }]));
  TestBed.configureTestingModule({ providers: [{ provide: ItemApi, useValue: { search } }] });
  const fixture = TestBed.createComponent(ItemSearch);
  const cmp = fixture.componentInstance;
  TestBed.tick();

  for (const q of ['ぺ', 'ペ', 'ペン']) {
    cmp.query.set(q);
    TestBed.tick(); // toObservable の effect を動かす
    await vi.advanceTimersByTimeAsync(100);
  }
  await vi.advanceTimersByTimeAsync(300);

  expect(search.mock.calls.map((c) => c[0])).toEqual(['ペン']);
  expect(cmp.results()).toEqual([{ id: 1, name: 'ペンの結果', stock: 1 }]);
});

 ✓ |rxdemo| src/app/demos/item-search.spec.ts > ItemSearch（debounce + switchMap） > 素早く3回入力しても API は最後の1回だけ呼ばれる 18ms

フェイクタイマー中に await fixture.whenStable() を使うと、Error: Test timed out in 5000ms. で止まりました。 変更検知の予約もタイマーで動くためです。フェイクタイマーを使うテストでは TestBed.tick() で変更検知を進めると安定します。

HTTP 取得：httpResource・rxResource・toSignal の選び方

GET で値を読むだけなら httpResource、Observable を返す既存サービスを使うなら rxResource が短く書けます。 どちらも v22 で安定版になり、value()・isLoading()・error()・status() を signal で持ちます。

// src/app/demos/item-detail.ts
import { Component, inject, signal } from '@angular/core';
import { httpResource } from '@angular/common/http';
import { rxResource } from '@angular/core/rxjs-interop';
import { retry } from 'rxjs';
import { ItemApi } from '../item-api';
import { Item } from '../mock-backend';

@Component({
  selector: 'app-item-detail',
  template: `
    <button id="next" (click)="id.update((v) => (v % 5) + 1)">次の備品（id={{ id() }}）</button>
    <p id="rx">rxResource: {{ rx.isLoading() ? '読み込み中…' : rx.value()?.name }}</p>
    <p id="http">httpResource: {{ http.isLoading() ? '読み込み中…' : http.value()?.name }}</p>
  `,
})
export class ItemDetail {
  private readonly api = inject(ItemApi);
  readonly id = signal(1);

  /** Observable を返す既存サービスをそのまま使える */
  readonly rx = rxResource({
    params: () => ({ id: this.id() }),
    stream: ({ params }) => this.api.get(params.id).pipe(retry(2)),
  });

  /** HTTP の GET だけなら httpResource が最短 */
  readonly http = httpResource<Item>(() => `/api/items/${this.id()}`);
}

params から undefined を返すと、リソースは読み込みをせず idle になります。 「ID が決まるまで取得しない」という条件はこれで書けます。

Vitest で HttpTestingController を使って確かめると、id を1つ変えるたびに rxResource と httpResource がそれぞれ1本ずつリクエストを出し、応答後に両方の value() が新しい備品名になりました。

ポーリング：timer と switchMap を toSignal で表示する

「一定間隔で取得する」は RxJS の timer が得意です。 toSignal はコンポーネントが破棄されると購読を自動で解除するため、画面を離れればポーリングも止まります。

// src/app/demos/stats-poller.ts
import { Component, inject } from '@angular/core';
import { toSignal } from '@angular/core/rxjs-interop';
import { switchMap, timer } from 'rxjs';
import { ItemApi } from '../item-api';

@Component({
  selector: 'app-stats-poller',
  template: `<p id="stats">確認 {{ stats()?.checkedAt ?? '-' }} 回目：在庫が少ない備品 {{ stats()?.lowStock ?? '-' }} 件</p>`,
})
export class StatsPoller {
  private readonly api = inject(ItemApi);

  /** 0秒後に1回、その後2秒ごとに取得。画面を離れると自動で止まる */
  readonly stats = toSignal(timer(0, 2000).pipe(switchMap(() => this.api.stats())));
}

前の応答が終わる前に次の周期が来たら、switchMap が前のリクエストを取り消します（全部使うなら concatMap、応答中は飛ばすなら exhaustMap）。

WebSocket・イベントストリーム：scan で状態にまとめてから signal にする

次々に届くイベントは、scan で「最新 N 件の配列」などの状態にしてから toSignal にします。 signal は「今の値」しか持たないため、イベントをそのまま signal にすると、同じ変更検知の間に届いた途中の値は画面に出ません。

// src/app/demos/notice-feed.ts
import { Component, Injectable, inject } from '@angular/core';
import { toSignal } from '@angular/core/rxjs-interop';
import { Subject, scan } from 'rxjs';

/** WebSocket などのイベント源の代わり */
@Injectable({ providedIn: 'root' })
export class NoticeBus {
  private readonly subject = new Subject<string>();
  readonly notices$ = this.subject.asObservable();
  push(message: string) {
    this.subject.next(message);
  }
}

@Component({
  selector: 'app-notice-feed',
  template: `
    <button id="push" (click)="bus.push('入荷: ノート ' + (latest().length + 1))">通知を送る</button>
    <ol id="notices">
      @for (n of latest(); track $index) { <li>{{ n }}</li> }
    </ol>
  `,
})
export class NoticeFeed {
  protected readonly bus = inject(NoticeBus);

  /** 流れてくるイベントを「最新3件の配列」という状態にまとめてから signal にする */
  readonly latest = toSignal(
    this.bus.notices$.pipe(scan((acc, n) => [n, ...acc].slice(0, 3), [] as string[])),
    { initialValue: [] },
  );
}

実際の WebSocket では、rxjs/webSocket の webSocket() が返す Observable を notices$ の位置に置けば同じ形で使えます（検証では Subject で代用）。

フォーム値：valueChanges を toSignal にするときは initialValue を渡す

Reactive Forms の valueChanges は、購読した時点の値を流しません。 そのため initialValue に現在の値を渡すと、最初の表示から正しい値になります。

// src/app/demos/filter-form.ts
import { Component, computed } from '@angular/core';
import { FormControl, ReactiveFormsModule } from '@angular/forms';
import { toSignal } from '@angular/core/rxjs-interop';

@Component({
  selector: 'app-filter-form',
  imports: [ReactiveFormsModule],
  template: `
    <input id="min" type="number" [formControl]="minStock" />
    <p id="min-label">{{ label() }}</p>
  `,
})
export class FilterForm {
  readonly minStock = new FormControl(10, { nonNullable: true });

  /** valueChanges は最初の値を流さないので initialValue に現在値を渡す */
  readonly min = toSignal(this.minStock.valueChanges, { initialValue: this.minStock.value });
  readonly label = computed(() => `在庫 ${this.min()} 個以上を表示`);
}

新しく作るフォームなら、@angular/forms/signals の Signal Forms（v22 の型定義で publicApi 22.0）も選べます。 フォームの値そのものを signal として持てるので、toSignal の橋渡しが要りません。既存の Reactive Forms はこの toSignal の形で少しずつ signal に寄せていけます。

Observable を output にする：outputFromObservable

子コンポーネントのイベントを RxJS で間引いてから親へ渡したいときは outputFromObservable を使います。 購読と解除は Angular が行います。

// src/app/demos/tick-button.ts
import { Component, signal } from '@angular/core';
import { outputFromObservable, takeUntilDestroyed } from '@angular/core/rxjs-interop';
import { Subject, fromEvent, throttleTime, map } from 'rxjs';

@Component({
  selector: 'app-tick-button',
  template: `<button id="tick" (click)="clicks$.next()">記録する</button> <span id="width">幅 {{ width() }}px</span>`,
})
export class TickButton {
  protected readonly clicks$ = new Subject<void>();

  /** Observable をそのまま output にする（1秒に1回まで） */
  readonly recorded = outputFromObservable(
    this.clicks$.pipe(throttleTime(1000), map(() => new Date().toISOString())),
  );

  readonly width = signal(window.innerWidth);

  constructor() {
    // subscribe するときは takeUntilDestroyed で後片付けを任せる
    fromEvent(window, 'resize')
      .pipe(takeUntilDestroyed())
      .subscribe(() => this.width.set(window.innerWidth));
  }
}

親では <app-tick-button (recorded)="records.update((r) => [...r, $event])" /> のように普通の output と同じく受け取れます。

subscribe するなら takeUntilDestroyed

subscribe は「副作用（ログ送信、外部ライブラリの操作など）」を書くときに使い、takeUntilDestroyed() を必ず付けます。 引数なしで使えるのはコンストラクターやフィールド初期化の中だけです（前の TickButton の例）。ngOnInit などで使うときは inject(DestroyRef) で取得した DestroyRef を takeUntilDestroyed(this.destroyRef) のように渡します。

グローバル状態：サービスに signal を持たせる

アプリ全体で共有する状態は、サービスに signal を持たせ、外には読み取り専用で公開する形が基本です。 読み取りは同期でよく、computed で派生値も作れるため、状態そのものは Signal で持つのが扱いやすくなります。

@Injectable({ providedIn: 'root' })
export class CartState {
  private readonly _lines = signal<{ itemId: number; qty: number }[]>([]);
  readonly lines = this._lines.asReadonly(); // 外からは読み取り専用
  readonly totalQty = computed(() => this._lines().reduce((sum, l) => sum + l.qty, 0));
  add(itemId: number) { this._lines.update((lines) => [...lines, { itemId, qty: 1 }]); }
}

状態と操作が増えてきたら NgRx SignalStore に移すと、1か所で整理できます（非同期処理は rxMethod で RxJS の演算子をそのまま使えます）。

toSignal のオプション一覧

initialValue と requireSync: true は同時に指定しません。 型定義上も、どちらか一方を選ぶ形になっています。

// BehaviorSubject は購読した瞬間に値が来る → Signal<string>（undefined なし）
readonly status = toSignal(this.status$, { requireSync: true });

// ngOnInit などコンテキストの外では、フィールドで取得した Injector を渡す
this.later = toSignal(this.status$, { injector: this.injector, requireSync: true });

よくあるエラーと直し方（実際のエラー文）

Vitest で実際に発生させたエラー文です。 どれも原因がはっきりしているので、文言で見分けられます。

NG0203：注入コンテキストの外で toSignal を呼んだ

直す前（ngOnInit の中で呼んでいる）→ 直した後（フィールド初期化で作る、または injector を渡す）：

// 直す前
ngOnInit() {
  this.items = toSignal(this.items$); // NG0203
}

// 直した後
readonly items = toSignal(this.items$, { initialValue: [] }); // フィールド初期化ならOK
private readonly injector = inject(Injector);
ngOnInit() {
  const later = toSignal(this.items$, { injector: this.injector }); // injector を渡せばOK
}

Error: NG0203: toSignal() can only be used within an injection context such as a constructor, a factory function, a field initializer, or a function used with `runInInjectionContext`. Find more at https://v22.angular.dev/errors/NG0203

NG0601：requireSync なのに同期で値が来ない

toSignal(of(1).pipe(delay(10)), { requireSync: true }) を実行すると NG0601 になりました。非同期の Observable には initialValue を使い、requireSync は BehaviorSubject や startWith(...) 付きのように「購読した瞬間に値が出る」ものだけに使います。

toObservable は最初の値だけ同期で流れる

toObservable は内部で effect を使うため、2回目以降の値は非同期で流れ、途中の値はまとめられます（テストで確認）。

const count = signal(1);
toObservable(count).subscribe((v) => values.push(v)); // 注入コンテキスト内で購読
TestBed.tick();
count.set(2);
count.set(3);
// この時点で values は [1]
TestBed.tick();
// values は [1, 3]（2 は飛ばされ、最新の 3 だけ流れる）

「全部の変化を1つずつ処理したい」ときは、signal ではなく Subject などの Observable で持つ のが向いています。

この記事のテスト（検索・リソース・エラー7種）は、Vitest 5.0.2 で Tests 9 passed (9) でした。

Playwright（Chromium）で画面の動きを確認する

ng serve で起動した画面を Playwright で操作し、各ケースが期待どおり動くことを確かめました。 検索欄に「ペン」を1文字ずつ入力し、詳細の「次へ」、2秒以上の待機、通知ボタン2回、フォームへの入力、記録ボタン5回連打、ウィンドウ幅の変更を順に行い、モックAPIのログ（[mock] GET ...）から検索リクエストの回数を数えています。

初期: 5 件
「ペン」入力後: 3 件 | items 検索リクエスト: [ '[mock] GET /api/items?q=%E3%83%9A%E3%83%B3' ]
rxResource: ボールペン / httpResource: ボールペン
次へ後: rxResource: ペンケース / httpResource: ペンケース
確認 3 回目：在庫が少ない備品 2 件
通知: 入荷: ノート 2, 入荷: ノート 1
在庫 25 個以上を表示
output(throttle 1秒): 記録 1 件
幅 900px

「ペ」「ン」と2文字入力しても検索リクエストは1回だけで、5回連続クリックしても throttleTime(1000) により記録は1件でした。 検証では Playwright 1.63.0 から、既にインストール済みの Chromium を executablePath で指定して起動しています。

よくある質問

Q. RxJS はもう使わなくてよいのですか？ A. 状態を持つ部分は Signal に置き換えやすくなりましたが、debounce・キャンセル・リトライ・ポーリング・イベントの合成など「時間の扱い」は RxJS が得意です。状態は Signal、時間の流れは RxJS と分担し、toSignal でつなぐのが扱いやすい形です。

Q. async パイプと toSignal はどちらを使えばよいですか？ A. どちらも購読と解除を Angular に任せられます。toSignal にすると computed で派生値を作れて、テンプレートの外（クラスの中）でも同期で値を読めるため、新しく書くコードでは toSignal がまとめやすくなります。

Q. httpResource と rxResource はどう選びますか？ A. GET で JSON を読むだけなら httpResource、HttpClient を包んだ既存のサービスがある、retry などの演算子を足したい、HTTP 以外の Observable を読む、という場合は rxResource です。どちらも v22 で安定版です。

まとめ

• 状態（今の値）は Signal、時間の流れ（待つ・間引く・キャンセル）は RxJS と分担する
• RxJS で整えた結果は toSignal で signal にして表示し、initialValue で undefined を消す
• HTTP の GET は httpResource、既存の Observable サービスは rxResource（どちらも v22 で安定版）
• イベント列は scan で状態にしてから toSignal、フォーム値は initialValue に現在値を渡す
• subscribe するときは takeUntilDestroyed、ngOnInit では DestroyRef を渡す
• NG0203 は注入コンテキスト外、NG0601 は requireSync の対象違い が原因。フィールド初期化・injector・initialValue で直せる

株式会社RJCでは、RxJS と Signals のような新しい書き方も、実際に手を動かして確かめながらチームで取り入れています。こうした知識を共有しながら、日々技術を磨いています。

株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。