概要

• ObservableとObserverの作成
• subscribeとunsubscribe

前提

• Node.jsインストール済み
• RxJS 6

ドキュメント

• https://rxjs.dev/
• https://github.com/ReactiveX/RxJS
• API

インストール

npm install rxjs

# 確認
npm list rxjs

└── rxjs@6.6.0

Observableの思想

• 様々な値(オブジェクト)が実現して流れていくプログラミングイメージ
• ある処理を「値の実現(発行)」と「その値の観測(受取)」に分けて考える

プログラミングの流れ

1. Observable作成: 実現コード
2. Observer作成: 観測コード
3. subscribe()メソッド: 実際に、実現・観測となる

基本構文

observable.subscribe(observer)

// 日本語直訳: 観察可能(な値)が観察者によって購読される
観察可能.購読(観察者)

• Observableインスタンスのsubscribe()メソッドにObserverを渡す

構成要素

Observable

• 将来の処理やデータの流れ(集まり)を表したインスタンス
  • コンストラクタ(new)で作る
  • 作成関数で作る
• subscribe()メソッドで実行

Observer(Subscriber)

• 値が実現または失敗したときにどうするか、値の観測担当のオブジェクト
• next(値)メソッドを持つ: 受け取った値をどうするか記述

オペレータ(Operator)

• 作成オペレータ: 新規にObservableを作成する関数
• パイプオペレータ: 発行前の値の加工やフィルタリングなどをする関数

// オペレータを使う場合
const observable = 作成オペレータ(引数)
observable.pipe(
  パイプオペレータ1(),
  パイプオペレータ2()
).subscribe(observer)

Subscription

• subscribe()メソッドの戻り値オブジェクト
• unsubscribe()メソッド: 途中でキャンセルする

準備: インポート

// Observableが含まれる
const rxjs = require('rxjs')

// オペレータが含まれる
const operators = require('rxjs/operators')

• 以降、インポート省略して、このように使う

// 例: Observable作成関数
const of = rxjs.of
// まとめてインポートなら
const {of, from, Observable} = rxjs
const {filter, map} = operators

Observableを作る

コンストラクタの場合

const Observable = rxjs.Observable

const observable = new Observable(observer => {
  observer.next(2)
  observer.next(3)
  observer.next(4)
})

• コンストラクタにコールバックを渡す
• コールバックの仮引数: Observer
  • 実現させたいタイミングでnext(値)メソッド (何回でも)

作成関数の例

const of = rxjs.of

// コンストラクタの場合と同じ意味
const observable = of(2, 3, 4)

Observerを作る

const observer = {
  next: (値) => console.log(値),
  next: console.log // 短縮形
}

• Observerはただのオブジェクト
  • 後でsubscribe()に渡す
• 実現したときnext(値)メソッドが呼ばれる

subscribe()メソッドで、実現・観測の実行となる

• (注目) Observableインスタンスを作った段階では、まだ実現も観測も行われない

const log = [] // ログ用の配列

// Observableインスタンス作成
const observable = new Observable(observer => {
  observer.next(2)
  observer.next(3)
  observer.next(4)
})

log.push(1)

// Observerオブジェクト作成
const observer = {next: 値 => log.push(値)}

// ここで実行
observable.subscribe(observer)

log.push(5)

console.log(log) // [ 1, 2, 3, 4, 5 ]
// 5が最後であることに注目

• subscribe()しても全体は同期処理のまま (Promise.then()との違い)

Observerの詳細

• next(値)以外のメソッドも
• subscribeにコールバークの引数列を渡す方法も

三つの通知メソッド(省略可)

1. 成功: next(値)
2. 失敗: error(エラー)
3. 完了: complete() (引数なし)

• error()やcomplete()の場合、Observable内の残りのコードは実現せず終了

subscribeに一つのオブジェクトで渡す

observable.subscribe({
  next: 値 => console.log(値),                                                                         
  error: エラー => console.error(エラー),
  complete: () => console.log('完了')
})

subscribeに順番にコールバックで渡す

observable.subscribe(
  値 => console.log(値),
  エラー => console.error(エラー),
  () => console.log('完了')
)

• そのまま出力するだけなら

  observable.subscribe(console.log, console.error)

error、completeメソッドでObservable終了

const log = [] // ログ用の配列

new Observable(observer => {
  observer.next(1)
  observer.next(2)
  observer.complete()
  // completeの後は実行されない
  observer.next('x')
  observer.error('error')
}).subscribe({
  next: 値 => log.push(値),
  error: エラー => log.push(エラー),
  complete: () => log.push('完了')
})

log.push(3)

console.log(log) // [ 1, 2, '完了', 3 ]

• error()メソッドの代わりにthrowでエラー発生させても同様