Subject

同時に観察者として観察できる
Subjectに渡されたイベントをObserverの最も基本的なSubjectに渡します.つまり、イベントを渡す役割と、イベントを受信する役割を同時に果たすことができます.

Subjectのタイプ

PublishSubject

空(要素を含まない)に作成され、サブスクリプション・サーバはサブスクリプション・ポイント以降に発生したイベントのみを受信します.

let subject = PublishSubject<String>()
subject.onNext("hello?")

上記のコードを実行しても何も起こりません.購読者がいないからです.

let subscriber = subject
    .subscribe(onNext: { str in
        print(str)
    })

同様に、サブスクリプションのコードを挿入すると、印刷文も実行されません.
次のコードを追加

subject.onNext("Hi")

今回はprint文を実行します.

PublishSubjectは時間ストリームに関連するモデルに非常に適している.
例えば、あるストリーミングメディアサイトでは、12時に放送を終了し、9時59分に「1分後に放送を終了する」.プロンプトでは、10時1分にログインしたユーザーはこの情報を知る必要はありません.
過去の通知が必要な場合は、BehaviorSubjectを使用します.

BehaviorSubject

PublishSubjectに似ています.差異は初期値で作成する必要があります.subscribeが発生すると、発生時間前に発生した最新イベントが受信されます.

上記のPublish Subjectと非常に似ていますが、Subscriberが発生すると、最近発生したイベントがすぐに発行されます.

let disposeBag = DisposeBag()

enum MyError: Error {
   case error
}

let b = BehaviorSubject<Int>(value: 0)
b.subscribe { print("BehaviorSubject >>", $0) }
    .disposed(by: disposeBag)

b.onNext(1)
b.subscribe { print("BehaviorSubject2 >>", $0) }
    .disposed(by: disposeBag)
    
b.onError(MyError.error)
b.subscribe { print("BehaviorSubject3 >>", $0) }
    .disposed(by: disposeBag)

最初にBehavior Subjectを購読すると、デフォルト値0が解放されます.
1がリリースされ、新しいサブスクライバが表示されると、最近のリリース値1が同時に受信されます.
errorの後に購読すると、errorメッセージが同時に受信されます.

ReplaySubject

BufferSizeとともに作成します.BehavorSubjectと似ていますが、受信した最新イベントはBufferSizeと同じです.

let disposeBag = DisposeBag()

enum MyError: Error {
   case error
}

let rs = ReplaySubject<Int>.create(bufferSize: 3)
(1...10).forEach { rs.onNext($0) }
rs.subscribe{ print("Observer 1 >>", $0) }
    .disposed(by: disposeBag)
rs.subscribe{ print("Observer 2 >>", $0) }
    .disposed(by: disposeBag)
rs.onNext(11)

rs.subscribe{ print("Observer 3 >>", $0) }
    .disposed(by: disposeBag)
rs.onError(MyError.error)
rs.subscribe{ print("Observer 4 >>", $0) }
    .disposed(by: disposeBag)

コメントとソース

http://reactivex.io/documentation/subject.html
https://yeonduing.tistory.com/61
https://rhammer.tistory.com/289