サンプルコードで身につける: Observable と RxJS — 時間とともに届く値

解説で学んだ概念を、5つの具体例で確認します。素朴な例から実務寄りの例へと進みます。

1of と subscribe — 最小の Observable

of(1, 2, 3) は「1, 2, 3 を順に流す Observable」を作ります。Observable は作っただけでは何も起きず、subscribe(購読)して初めて値が流れ、コールバックが1つずつ受け取ります。「値の流れを定義してから、購読して受け取る」という基本の型をこの最小例で押さえましょう。

import { of } from 'rxjs';

// 「1, 2, 3 を順に流す」という値の流れを定義する
const numbers$ = of(1, 2, 3);

// subscribe して初めて値が流れ始める
numbers$.subscribe((value) => {
  console.log('届いた値:', value);
});

// 出力:
// 届いた値: 1
// 届いた値: 2
// 届いた値: 3

// 変数名の末尾に $ を付けるのは「Observableである」印の慣習

2interval — 時間とともに届く値

interval(1000) は1秒ごとに 0, 1, 2, ... と値を流し続ける Observable です。1回で終わる Promise と違い、Observable は時間をまたいで何度も値を届けられるのが特徴です。「あとから何度も届く値の流れ」という Observable のイメージをつかむのに最適な例です。

import { interval } from 'rxjs';

// 1000ミリ秒ごとに 0, 1, 2, ... を流し続ける Observable
const tick$ = interval(1000);

tick$.subscribe((count) => {
  console.log(count + ' 回目の通知');
});

// 出力(1秒ごとに1行ずつ増えていく):
// 0 回目の通知
// 1 回目の通知
// 2 回目の通知
// ...止めるまで永遠に続く

3map で流れる値を加工する

pipe() の中に map を書くと、流れてくる値を1つずつ変換した新しい流れが作れます。元の Observable はそのままに、「加工後の流れ」を派生させるのが RxJS の考え方です。APIのレスポンスを表示用の形に整えるなど、実務では subscribe の前に pipe で加工するのが定番の形です。

import { of, map } from 'rxjs';

const prices$ = of(100, 250, 980);

// pipe + map で「税込価格の流れ」に変換する
const taxed$ = prices$.pipe(
  map((price) => Math.floor(price * 1.1)),
);

taxed$.subscribe((price) => {
  console.log('税込:', price, '円');
});

// 出力:
// 税込: 110 円
// 税込: 275 円
// 税込: 1078 円

4HttpClient が返す Observable

HttpClient の get() は結果を Observable で返すため、subscribe して初めてリクエストが飛び、レスポンスが届くとコールバックが呼ばれます。of や interval で見た「subscribe して値を受け取る」型が、通信でもそのまま通用することを示す例です。1回きりの通信結果も「値の流れ」として同じ仕組みに乗っています。

import { Component, inject, signal } from '@angular/core';
import { HttpClient } from '@angular/common/http';

type User = { id: number; name: string };

@Component({
  selector: 'app-user-list',
  template: '@for (u of users(); track u.id) { <p>{{ u.name }}</p> }',
})
export class UserListComponent {
  private http = inject(HttpClient);
  users = signal<User[]>([]);

  load() {
    // get() は Observable を返す。subscribe して初めて通信が始まる
    this.http.get<User[]>('/api/users').subscribe((users) => {
      this.users.set(users); // レスポンスが届いたら状態に反映
    });
  }
}

5unsubscribe — 購読の後片付け

interval のように終わらない Observable を subscribe すると、止めない限り処理が動き続けます。subscribe の戻り値(Subscription)を取っておき、不要になったら unsubscribe() で止めるのが基本の後片付けです。コンポーネントが消えても購読が残るとメモリリークになるため、実務では解除とセットで考える習慣をつけます。

import { interval, Subscription } from 'rxjs';

// 終わらない流れは、止める手段とセットで持つ
const sub: Subscription = interval(1000).subscribe((count) => {
  console.log('経過:', count);
});

// 5秒後に購読を解除すると、それ以降コールバックは呼ばれない
setTimeout(() => {
  sub.unsubscribe();
  console.log('購読を解除しました');
}, 5000);

// 解除を忘れると処理が動き続け、メモリリークの原因になる。
// コンポーネントでは ngOnDestroy で解除するのが定番(次のレッスンで学ぶ)