前提
結論
pipe() と concatMap() は、Observableの通知に対して処理を組み立てるための仕組み。
- pipe()：Observableに対して、複数の処理（Operator）を順番に適用する
- concatMap()：通知された値を別のObservableに変換し、そのObservableの処理を順番に実行する
- subscribe()：最終的なObservableを購読し、通知を受け取る
詳細
使用例
値を2倍にしてから文字列に変換する
import { of } from 'rxjs';
import { map } from 'rxjs/operators';
// ① 値を通知するObservableを作る
const observable = of(1, 2, 3);
// ② 通知される値に処理を適用する
const result = observable.pipe(
map(value => value * 2),
map(value => `結果：${value}`)
);
// ③ 購読して通知を受け取る
result.subscribe(value => {
console.log(value);
});
出力結果：
結果：2
結果：4
結果：6
→ pipe() の中に記述した処理が、上から順番に適用される。
→ この例では、最初の map() で値を2倍にし、次の map() で文字列に変換している。
pipe()
pipe()とは？
Observableに対して、Operator（演算子）を順番に適用するためのメソッド
const result = observable.pipe(
map(value => value * 2),
map(value => value + 1)
);
-
observable：元のObservable -
pipe()：Operatorを順番に適用する -
map()：通知された値を別の値に変換するOperator -
result：処理を適用した後のObservable
pipe()の処理イメージ
元のObservableが 1、2、3 を通知するとする。
元のObservable
│
├── 1
├── 2
└── 3
│
▼
pipe()
│
▼
map(value => value * 2)
│
├── 2
├── 4
└── 6
│
▼
map(value => value + 1)
│
├── 3
├── 5
└── 7
│
▼
subscribe()
出力結果：
3
5
7
このように、前のOperatorが出力した値を、次のOperatorが受け取る。
pipe()とsubscribe()の違い
| メソッド | 役割 |
|---|---|
pipe() |
Observableに処理を組み合わせる |
subscribe() |
Observableを購読して通知を受け取る |
pipe() は処理の流れを組み立てるためのもので、subscribe() はその処理の流れを購読するためのもの。
concatMap()
concatMap()とは？
通知された値を、それぞれ別のObservableに変換し、変換先のObservableを1つずつ順番に処理するOperator。
少し複雑なので、まずは普通の値をObservableに変換する例から考える。
① concatMap()を使わない場合
import { of } from 'rxjs';
const observable = of(1, 2, 3);
observable.subscribe(value => {
console.log(value);
});
出力結果：
1
2
3
この例では、Observableが数値を通知し、Observerがその値を受け取って表示している。
② concatMap()を使う場合
import { of } from 'rxjs';
import { concatMap } from 'rxjs/operators';
const observable = of(1, 2, 3);
const result = observable.pipe(
concatMap(value => of(value * 10))
);
result.subscribe(value => {
console.log(value);
});
出力結果：
10
20
30
-
value：元のObservableから通知された値 -
value * 10：通知された値を10倍にする -
of(value * 10)：計算結果を通知するObservableを作る -
concatMap()：そのObservableを順番に処理する
例えば、元のObservableが 1 を通知した場合、
value => of(value * 10)
に value = 1 が渡される。
すると、
of(1 * 10)
つまり、
of(10)
というObservableが返される。
concatMap() は、そのObservableが通知する 10 を最終的なObservableから通知できるようにする。
同様に、2 からは 20、3 からは 30 が通知される。
③ concatMap()の「順番に処理する」とは？
ここがconcatMap()の重要なポイント。
of() は値を同期的に通知し、すぐに完了するObservableなので、先ほどの例だけでは順番に処理する意味が少しわかりにくい。
そこで、処理に時間がかかるケースを考える。
import { of } from 'rxjs';
import { concatMap, delay } from 'rxjs/operators';
const observable = of(1, 2, 3);
const result = observable.pipe(
concatMap(value =>
of(value).pipe(
//delay() は、Observableが通知するタイミングを遅らせるOperator
delay(1000)
)
)
);
result.subscribe(value => {
console.log(value);
});
出力結果：
subscribe() を呼んでから約1秒後：1
subscribe() を呼んでから約2秒後：2
subscribe() を呼んでから約3秒後：3
※ 時間は概算。
- このコードでは、各値に対して約1秒後に値を通知するObservableを作っている
-
concatMap()は、先に変換したObservableが完了してから、次のObservableの処理を開始する - Observableの完了=「そのObservableが値を通知し終えて、complete 通知を送ったら」
元のObservable
│
├── 1
├── 2
└── 3
│
▼
concatMap()
│
▼
① 1を通知するObservable
│
└── 約1秒後に1を通知・完了
│
▼
② 2を通知するObservable
│
└── 約1秒後に2を通知・完了
│
▼
③ 3を通知するObservable
│
└── 約1秒後に3を通知・完了
元のObservableが 2 や 3 を通知しても、前の処理が完了していなければ、concatMap() はそれらの値を待ち行列に保持する。
そして、前のObservableが完了すると、次のObservableを処理する。
つまり、concatMap()は、複数のObservableの処理を同時進行させず、順番に処理するためのOperator。
pipe()とconcatMap()を組み合わせる
import { of } from 'rxjs';
import { concatMap, delay } from 'rxjs/operators';
const observable = of('A', 'B', 'C');
const result = observable.pipe(
concatMap(value => {
console.log(`${value}の処理を開始`);
return of(`${value}の処理が完了`).pipe(
delay(1000)
);
})
);
result.subscribe(value => {
console.log(value);
});
出力結果：
Aの処理を開始
約1秒後：Aの処理が完了
Bの処理を開始
約1秒後：Bの処理が完了
Cの処理を開始
約1秒後：Cの処理が完了
処理の流れ
-
of('A', 'B', 'C')が元のObservableを作る。 -
pipe()がconcatMap()を適用する。 -
concatMap()に'A'が渡される。 -
'A'に対応するObservableが作られ、約1秒後に通知して完了する。 - その後、
'B'に対応するObservableが処理される。 - 同様に
'C'が処理される。 -
subscribe()に登録した処理が、それぞれの通知を受け取って表示する。
Angularでの利用例
例えば、Todoを3件順番に保存したい場合を考える。
import { from } from 'rxjs';
import { concatMap } from 'rxjs/operators';
const todos = [
{ title: '買い物' },
{ title: '掃除' },
{ title: '勉強' }
];
from(todos).pipe(
concatMap(todo => this.todoService.create(todo))
).subscribe({
next: result => {
console.log('保存結果:', result);
},
error: error => {
console.error('保存に失敗しました:', error);
},
complete: () => {
console.log('すべての保存処理が完了しました');
}
});
※ todoService.create(todo) は、TodoをAPIに保存して結果を返すObservableを返すメソッドとする。
何が起きているのか？
-
from(todos)：配列の要素を1件ずつ通知するObservableを作る。 -
pipe()：通知されたTodoに対して処理を適用する。 -
concatMap(todo => this.todoService.create(todo))：Todoごとに保存処理を開始し、前の保存Observableが完了してから次の保存処理を開始する。 -
subscribe()：保存結果やエラー、すべての処理が完了したことを受け取る。
この場合、保存処理は次の順番になる。
買い物を保存
↓ 完了
掃除を保存
↓ 完了
勉強を保存
↓ 完了
completeが呼ばれる
ただし、これはAPIへのリクエストを順番に開始するという意味であり、DBのトランザクション全体を保証するものではない。また、エラーが発生すると、エラー処理を別途設計していない限り、後続の保存処理は実行されない。
まとめ
| 用語 | 役割 |
|---|---|
| Observable | 値・エラー・完了などを通知する仕組み |
| Observer | 通知を受け取ったときの対応方法 |
pipe() |
ObservableにOperatorを順番に適用する |
map() |
通知された値を別の値に変換する |
concatMap() |
値をObservableに変換し、変換先を順番に処理する |
subscribe() |
Observableを購読して通知を受け取る |
覚え方
-
pipe()：処理をつなぐ -
map()：値を変換する -
concatMap()：Observableに変換して順番に処理する -
subscribe()：通知を受け取る