Subject

Subject는 ReactiveX의 일부 구현에서 사용할 수 있는 일종의 다리 또는 프록시로, 관찰자(Observer) 및 옵저버블(Observable)의 역할을 동시에 수행하는 것입니다.

• Observable이자 Observer
• 3가지의 Event형태를 가짐
  • next
  • error
  • completed
• .next(value)를 받고, 수신할 때마다 subscriber에게 방출합니다.
• Sequence가 종료된 Subject를 subscribe할 경우, 마지막 Event(.error 혹은 .completed)를 수신합니다.

Subject 종류

PublishSubject

PublishSubject는 구독 이후에 소스 Observable(들)이 배출한 항목들만 옵저버에게 배출한다.

• 초기값을 가지지 않습니다.
• subscriber가 subscribe 하기 전에 방출된 이벤트에 대해서 알지못합니다.
• subscribe 후 이벤트가 방출되어야 처음으로 이벤트를 받습니다.

let subject = PublishSubject<String>() // 초기값 없이 생성

subject.onNext("a")
subject.onNext("b")

subject.subscribe { element in
	print("subscribe 1:", element)
} onError: { error in
	print("subscribe error 1:", error)
} onCompleted: {
	print("subscribe 1 onCompleted")
}.disposed(by: disposeBag)

subject.onNext("c")
    
subject.subscribe { element in
	print("subscribe 2:", element)
} onError: { error in
    print("subscribe error 2:", error)
} onCompleted: {
    print("subscribe 2 onCompleted")
}.disposed(by: disposeBag)

subject.onNext("d")
subject.onNext("e")
subject.onCompleted()

• output

subscribe 1: c
subscribe 1: d
subscribe 2: d
subscribe 1: e
subscribe 2: e
subscribe 1 onCompleted
subscribe 2 onCompleted

BehaviorSubject

• 초기값을 가집니다.
• subscriber가 subscribe 하기 전에 방출된 이벤트에 대해서 알지 못합니다.
• 단, subscribe 후 가장 최근에 방출된 이벤트을 받습니다.

let subject = BehaviorSubject(value: "0") // 생성 시 초기값

subject.onNext("a")
subject.onNext("b")
        
subject.subscribe { element in
    print("subscribe 1:", element)
} onError: { error in
    print("subscribe error 1:", error)
} onCompleted: {
    print("subscribe 1 onCompleted")
}.disposed(by: disposeBag)
		
subject.onNext("c")

subject.subscribe { element in
    print("subscribe 2:", element)
} onError: { error in
    print("subscribe error 2:", error)
} onCompleted: {
    print("subscribe 2 onCompleted")
}.disposed(by: disposeBag)

subject.onNext("d")
subject.onNext("e")
subject.onCompleted()

• output

subscribe 1: b
subscribe 1: c
subscribe 2: c
subscribe 1: d
subscribe 2: d
subscribe 1: e
subscribe 2: e
subscribe 1 onCompleted
subscribe 2 onCompleted

ReplaySubject

• 이전 이벤트를 저장하고 새로운 구독자에게 전달하는데 사용됩니다.
• 이전 이벤트의 재생 및 공유를 지원하는 더 유연한 Observable입니다.
• subscribe 후 bufferSize만큼의 최근에 방출된 이벤트를 받습니다.

/// static func create(bufferSize: Int) -> ReplaySubject<Element>
/// 메서드로 생성
let subject = ReplaySubject<String>.create(bufferSize: 2)
        
subject.onNext("a")
subject.onNext("b")

subject.subscribe { element in
    print("subscribe 1:", element)
} onError: { error in
    print("subscribe error 1:", error)
} onCompleted: {
    print("subscribe 1 onCompleted")
}.disposed(by: disposeBag)

subject.onNext("c")

subject.subscribe { element in
    print("subscribe 2:", element)
} onError: { error in
    print("subscribe error 2:", error)
} onCompleted: {
    print("subscribe 2 onCompleted")
}.disposed(by: disposeBag)

subject.onNext("d")
subject.onNext("e")
subject.onCompleted()

• output

subscribe 1: a
subscribe 1: b
subscribe 1: c
subscribe 2: b
subscribe 2: c
subscribe 1: d
subscribe 2: d
subscribe 1: e
subscribe 2: e
subscribe 1 onCompleted
subscribe 2 onCompleted

AsyncSubject

• completed 이벤트가 방출되기 전까진 어떤 이벤트도 방출하지 않습니다.
• completed 이벤트가 방출되면 마지막 방출되었던 이벤트를 subscriber에게 전달합니다.
• error 이벤트가 방출되면 아무 이벤트도 방출하지 않습니다.

enum AsyncSubjectError: Error {
	case asyncSubjectError
}
let subject1 = AsyncSubject<String>()

subject1.subscribe { element in
    print("subscribe 1:", element)
} onError: { error in
    print("subscribe error 1:", error)
} onCompleted: {
    print("subscribe 1 onCompleted")
}.disposed(by: disposeBag)
        
subject1.onNext("a")
subject1.onNext("b") // subscribe event가 전달되지 않습니다.
subject1.onCompleted() // subscribe event 마지막 이벤트 .next("b")가 전달됩니다.
        
let subject2 = AsyncSubject<String>()

subject2.subscribe { element in
    print("subscribe 2:", element)
} onError: { error in
    print("subscribe error 2:", error)
} onCompleted: {
    print("subscribe 2 onCompleted")
}.disposed(by: disposeBag)

subject2.onNext("a")
subject2.onNext("b")
subject2.onError(AsyncSubjectError.asyncSubjectError) //subscribe event 마지막 이벤트 .next("b")가 전달됩니다.

• output

subscribe 1: b
subscribe 1 onCompleted
subscribe error 2: asyncSubjectError