RxJS: Subjectで複数のObserverにマルチキャスト

• Subject、BehaviorSubject、ReplaySubject

前提

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

ドキュメント

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

Subject

• 複数のObserverに通知(マルチキャスト)できるObservables
• Subjectのsubscribe()はObserverを登録するだけでまだ実現しない
  • next()で実現

• 自身はObserverでもあり、nextメソッドなどで実現する
• 他のObservablesのObserverにもなる

複数のObserverを登録できる

• subscribe()の段階ではなくnext()でマルチキャスト

const Subject = rxjs.Subject
const subject = new Subject()

// 一つ目のObserver登録
subject.subscribe(x => console.log('X', x))
// 二つ目のObserver登録
subject.subscribe(a => console.log('A', a))

console.log(0)

// nextメソッドでマルチキャスト
subject.next(1)

console.log(2)
subject.next(3)

実行結果

0
X 1
A 1
2
X 3
A 3

他のObservablesのObserverにもなる

• 他のObservablesのsubscribe()にSubjectを渡す

const Subject = rxjs.Subject
const interval = rxjs.interval

const subject = new Subject()
subject.subscribe(x => console.log('X', x))
subject.subscribe(a => console.log('A', a))

console.log(0)

// マルチキャスト
interval(500).subscribe(subject)

// 非同期なのでこれが先
console.log(1)

実行結果

0
1
X 0
A 0
X 1
A 1
X 2
A 2
(つづく)

BehaviorSubject

• 実現の後で追加されたObserverが、直前の値を受け取れる
  • これはsubscribe()で実現
• コンストラクタに初期値を渡す

const BehaviorSubject = rxjs.BehaviorSubject

const subject = new BehaviorSubject(1)                                                                                

// 一つ目のObserver
subject.subscribe(x => console.log('X', x)) // <= 初期値実現
console.log(1.5)
subject.next(2)

// 二つ目のObserver
// subscribe()で直近の値も実現
subject.subscribe(a => console.log('A', a)) // A 2

subject.next(3)

実行結果

X 1
1.5
X 2
A 2
X 3
A 3

ReplaySubject

• 実現の後で追加されたObserverが、指定の回数だけ過去の値を受け取れる
• コンストラクタに、過去何回分記憶するか渡す
  • 引数2に制限時間を渡すこともできる

const ReplaySubject = rxjs.ReplaySubject

const subject = new ReplaySubject(2) // 2回分

// 一つ目のObserver
subject.subscribe(x => console.log('X', x))
subject.next(1)
subject.next(2)
subject.next(3)

// 二つ目のObserver
// subscribe()で過去の値も実現
subject.subscribe(a => console.log('A', a))

実行結果

X 1
X 2
X 3
A 2
A 3