RxSwift - Subject

Subject 란?

 

먼저 Subject 중 하나인 PublishSubject 코드를 살펴보자.

public final class PublishSubject<Element>
    : Observable<Element>
    , SubjectType
    , Cancelable
    , ObserverType
    , SynchronizeUnsubscribeType {
...

 

Subject는 ObservableType 프로토콜을 채택하고 있는 Observable 을 상속하고 있으며, ObserverType 프로토콜을 채택하고 있다. 즉, Subject 가 Observable 로서 이벤트를 발행할 수도 있고, Observer 로서 발행하는 이벤트를 subscribe 할 수도 있는 것이다. 

 

또한, Subject 는 multicast 방식이다. multicast가 방식이라는 것은 하나의 observable 실행이 여러 subscriber 에게 공유되는 것을 의미한다. 

 

예시처럼 ViewModel 에 Subject 중 하나인 BehaviorSubject 를 선언하였고, 이를 subscribe 하는 observer 변수인 countFruit, priceFruit 도 선언하였다. 

Observable 프로토콜을 준수하는 BehaviorSubject 인 fruitSubject 가 init 함수 내에서 fruitModel 이라는 데이터가 들어오는, observable 의 onNext 이벤트를 발행한다. 그렇다면 fruitSubject 를 subscribe 하는 두 개의 subscriber(Observer) 에도 이러한 변화가 공유되는 것이다.

class ViewModel {
    var fruitSubject = BehaviorSubject<[Fruit]>(value: [])
    
    lazy var countFruit = fruitSubject.map(...)
    
    lazy var priceFruit = fruitSubject.map(...)
    
    init() {
        let fruitModel = [
        	Fruit(name: "apple", count: 100, price: 120),
        	Fruit(name: "banana", count: 200, price: 240),
        ]
        
        fruitSubject.onNext(fruitModel)
    }
}

 

Subject 의 종류

 

① PublishSubject

subscribe 된 시점부터 발생한 이벤트를 전달하며, subscribe 된 순간 새로운 이벤트에 대한 수신을 알리고 싶을 때 유용하다.

이런 활동은 subscribe 를  멈추거나, completed/error 이벤트를 통해 subject 가 완전히 종료될때까지 계속된다.

 

let publishSubject = PublishSubject<String>()
 
// subject 를 만들고
publishSubject.onNext("1. Hello Everyone")

// subscribe 를 하나 생성 후 구독을 시작한다
let subscriber1 = publishSubject
    .subscribe(
        onNext: {
            print("첫 번째 구독자: \($0)")
        }
    )

// next 이벤트를 두 개 방출한 상황
publishSubject.onNext("2. Can you hear me?")
publishSubject.on(.next("3. Nothing?"))

subscriber1.dispose()

// 모든 이벤트를 방출한 후 구독 시작
let subscriber2 = publishSubject
    .subscribe(
        onNext: {
            print("두 번째 구독자: \($0)")
        })

publishSubject.onNext("4. 여보세요")
publishSubject.onCompleted()

publishSubject.onNext("5. 혹시 모르니..?")

subscriber2.dispose()
 

publishSubject
    .subscribe {
        print("세 번째 구독자:", $0.element ?? $0)
    }
    .disposed(by: disposeBag)

publishSubject.onNext("6. completed 후 읽힐까?")


-----Publish Subject-----
첫 번째 구독자: 2. Can you hear me?
첫 번째 구독자: 3. Nothing?
두 번째 구독자: 4. 여보세요
세 번째 구독자: completed

 

② BehaviorSubject

let behaviorSubject = BehaviorSubject<String>(value: "")

새로운 Subscriber 에게 초기값 또는 최신값을 방출한다.

 

 

enum SubjectError: Error {
    case error1
}

// behaviorSubject 는 반드시 초기값을 가진다
let behaviorSubject = BehaviorSubject<String>(value: "0. 초기 값")

behaviorSubject.onNext("1. 첫 번째 값")

behaviorSubject
    .subscribe {
        print("첫 번째 구독", $0.element ?? $0)
    }
    .disposed(by: disposeBag)

behaviorSubject.onError(SubjectError.error1)

behaviorSubject
    .subscribe {
        print("두 번째 구독", $0.element ?? $0)
    }
    .disposed(by: disposeBag)


-----Behavior Subject-----
첫 번째 구독 1. 첫 번째 값
첫 번째 구독 error(error1)
두 번째 구독 error(error1)

 

onNext 이벤트 이후에 subscribe 하여도, 직전에 발생한 onNext 이벤트에서 방출한 value 를 가진다.

두 번째 subscribe 에서도 직전의 error 이벤트에서 방출한 error 값을 가진다.

 

BehaviorSubject 의 특징 중 하나가 value 값을 가져올 수 있다.

let value = try? behaviorSubject.value()
print(value)

 

예시를 위해 onError 을 주석처리하고 실행한다면, 아직 종료되지 않았기 때문에 가장 최신의 onNext 값을 value 로 꺼내주는 것을 확인할 수 있다.

//behaviorSubject.onError(SubjectError.error1)

Optional("1. 첫 번째 값")

 

 

③ ReplaySubject

Subject 를 생성할 때, 선택한 특정 크기(bufferSize)까지 방출하는 최신 요소를 일시적으로 캐싱하거나 버퍼를 두고 초기화하며, 버퍼 사이즈 만큼의 값들을 유지하면서, 새로운 Subscriber 에게 값을 방출한다.

 

// bufferSize 를 2로 설정하였다
let replaySubject = ReplaySubject<String>.create(bufferSize: 2)

replaySubject.onNext("1. 안녕")
replaySubject.onNext("2. 클레오파트라")
replaySubject.onNext("3. 세상에서")

replaySubject
    .subscribe {
        print("첫 번째 구독:", $0.element ?? $0)
    }
    .disposed(by: disposeBag)

replaySubject
    .subscribe {
        print("두 번째 구독:", $0.element ?? $0)
    }
    .disposed(by: disposeBag)

replaySubject.onNext("4. 제일 가는")
replaySubject.onError(SubjectError.error1)
replaySubject.dispose()

replaySubject
    .subscribe {
        print("세 번째 구독:", $0.element ?? $0)
    }
    .disposed(by: disposeBag)
    
    
-----Replay Subject-----
첫 번째 구독: 2. 클레오파트라
첫 번째 구독: 3. 세상에서
두 번째 구독: 2. 클레오파트라
두 번째 구독: 3. 세상에서
첫 번째 구독: 4. 제일 가는
두 번째 구독: 4. 제일 가는
첫 번째 구독: error(error1)
두 번째 구독: error(error1)
세 번째 구독: error(Object `RxSwift.(unknown context at $10e985990).ReplayMany<Swift.String>` was already disposed.)

 

세 번째 구독의 error 는 왜...?

세 번째 구독자는 이미 dispose 되었는데, 다시 subscribe 한다고 하니 RxSwift 에서 발생시킨 에러를 내뱉는다.