Angularでは、非同期処理やイベントのストリームを扱うためにRxJS（Reactive Extensions for JavaScript）が組み込まれています。
本章では、RxJSの基礎であるObservableの作成と購読、代表的な演算子（map、filter、switchMapなど）の使い方、そしてフォームやHTTP通信とのリアクティブな連携方法を解説します。
1. Observableの作成と購読（subscribe）
Observableは、非同期の値の流れ（ストリーム）を表現します。
イベントやHTTPレスポンス、フォーム入力などをリアクティブに扱えるのが特徴です。
■ Observableの基本的な作成方法
import { Observable } from 'rxjs';
const obs = new Observable(observer => {
observer.next('データ1');
observer.next('データ2');
observer.complete();
});
obs.subscribe(value => console.log(value));
出力：
データ1
データ2
■ of() や from() を使ったObservableの生成
import { of, from } from 'rxjs';
const obs1 = of(1, 2, 3);
const obs2 = from([10, 20, 30]);
2. 代表的なRxJS演算子
RxJSには、ストリームを操作するための多くの「演算子（Operators）」が存在します。
特に重要なものを以下に紹介します。
■ map
値を変換します。
obs.pipe(map(value => value * 2)).subscribe(console.log);
■ filter
条件に合う値だけを通過させます。
obs.pipe(filter(value => value > 10)).subscribe(console.log);
■ switchMap
内部で新たなObservableを返し、前のObservableはキャンセルされます。
主にHTTP通信やフォームイベントと組み合わせます。
input$.pipe(
debounceTime(300),
switchMap(keyword => this.http.get(api/search?q=${keyword}))
).subscribe(result => this.results = result);
3. フォームやHTTP通信との連携
■ Reactive FormとObservable
Angularのリアクティブフォームでは、フォームの変更値はvalueChangesでObservableとして受け取れます。
this.searchForm.get('keyword')?.valueChanges.pipe(
debounceTime(300),
distinctUntilChanged(),
switchMap(keyword => this.apiService.search(keyword))
).subscribe(results => this.results = results);
■ HTTP通信のレスポンスもObservable
AngularのHttpClientの各メソッド（get(), post()など）はObservableを返します。
これに対してpipe()で演算子をチェーンすることで、柔軟なデータ処理が可能になります。
this.apiService.getItems().pipe(
map(items => items.filter(item => item.active))
).subscribe(filteredItems => this.items = filteredItems);
まとめ
RxJSはAngularにおける非同期処理の中核であり、イベントやデータの流れを効率よく制御できます。Observableの概念を理解し、pipeと各種演算子を使いこなすことで、フォーム入力やAPI通信といった実務シーンでも力を発揮します。
