[RxSwift] Subject

개요

Subject는 Reactive X의 구현에서 사용 가능한 일종의 브릿지나 프록시를 의미합니다. 옵저버와 옵저버벌의 역할을 모두 수행할 수 있으며, 옵저버이기 때문에 하나나 여러 개의 옵저버블을 구독할 수 있으며 옵저버블이기 때문에 이것이 관찰하고 있던 아이템들을 다시 방출하여 전달하거나 새로운 아이템을 방출할 수 있습니다.

Subject는 옵저버블을 구독하기 때문에 옵저버블이 아이템을 방출하기 시작하도록 만들 수 있습니다(옵저버블은 옵저버가 있어야지 방출을 시작할 수 있으며 이에 대한 자세한 설명은 이 포스트를 참고하길 바랍니다). 즉 cold 옵저버블을 hot 옵저버블과 같은 방식으로 동작하게 만들 수 있습니다.

종류

AsyncSubject

AsyncSubject는 옵저버블에서 마지막으로 방출된 값만을 방출하며, 옵저버블이 완료된 후에만 값을 방출합니다. 만약 같은 옵저버블을 구독하는 후속 옵저버가 있다면 동일한 값을 이 옵저버에게도 전달하게 됩니다. 그리고 옵저버블이 오류로 종료하게 되면 AsyncSubject는 항목을 방출하지 않고 옵저버들에게 오류 알림만 전달하게 됩니다.

예제

let disposeBag = DisposeBag()

let subject = AsyncSubject<String>()

subject.onNext("1")

subject.subscribe(onNext: { value in
    print("Subscriber 1 : \(value)")
}).disposed(by: disposeBag)

subject.onNext("2")

subject.subscribe(onNext: { value in
    print("Subscriber 2 : \(value)")
}).disposed(by: disposeBag)

subject.onNext("3")

subject.onCompleted()

Subscriber 1 : 3
Subscriber 2 : 3

AsyncSubject는 마지막 값만 방출하기 때문에 Subscriber 1, 2 모두 옵저버블이 onCompleted() 된 이후 마지막 값인 3만을 방출합니다.

BehaviorSubject

BehaviorSubject는 최신 값을 저장하고 있다가 구독하는 옵저버가 생기면 즉시 최신 값을 방출합니다. 그 이후 옵저버블이 배출하는 갑들을 옵저버에게 바로 방출합니다. 만약 옵저버블이 오류로 종료되게 된다면 BehaviorSubject는 후속 옵저버에게 항목을 방출하지 않고, 오류 알림만 전달합니다.

예제

let disposeBag = DisposeBag()

let subject = BehaviorSubject<String>(value: "Initial Value")

subject.onNext("1")

subject.subscribe(onNext: { value in
    print("Subscriber 1 : \(value)")
}).disposed(by: disposeBag)

subject.onNext("2")

subject.subscribe(onNext: { value in
    print("Subscriber 2 : \(value)")
}).disposed(by: disposeBag)

subject.onNext("3")

Subscriber 1 : 1
Subscriber 1 : 2
Subscriber 2 : 2
Subscriber 1 : 3
Subscriber 2 : 3

BehaviorSubject는 최신값을 저장하고 있다가 새로운 옵저버가 생기면 최신값을 방출하기 때문에 Subscriber 2의 구독 시점이 '2'이벤트가 발생한 후여도 Subscriber 2에게 2의 값을 전달합니다.

PublishSubject

PublishSubject는 구독 이후에 발생한 아이템만을 옵저버에게 방출합니다. 그러나 PublishSubject가 생성된 시점과 옵저버가 구독을 시작한 시점의 차이로 인해서 하나 또는 여러개의 아이템이 손실될 수 있습니다. 따라서 옵저버가 구독하기 전에 옵저버블이 모든 구독자에게 값을 방출하기 시작하는 것을 확인해야 합니다(Create()을 사용해서 Cold 옵저버블을 수동으로 다시 생성하는 방법이 있습니다). 만약 옵저버블이 오류로 종료되면 오류의 알림만을 전달합니다.

예제

코드로 publishSubject의 동작을 자세히 살펴보겠습니다.

import UIKit
import RxSwift

let disposeBag = DisposeBag()

let subject = PublishSubject<String>()

subject.onNext("1")

subject.subscribe(onNext: { value in
    print("Subscriber 1 : \(value)")
}).disposed(by: disposeBag)

subject.onNext("2")

subject.subscribe(onNext: { value in
    print("Subscriber 2 : \(value)")
}).disposed(by: disposeBag)

subject.onNext("3")

Subscriber 1 : 2
Subscriber 1 : 3
Subscriber 2 : 3

PublishSubject는 구독 이후에 발생하는 이벤트만을 전달하기 때문에 Subscriber 1, 2 모두 구독 이후에 발생하는 이벤트만을 방출하는 것을 확인할 수 있습니다.

ReplaySubject

ReplaySubject는 옵저버가 어느 시점에 구독을 시작하던지 옵저버블이 방출한 모든 항목을 방출합니다. 그러나 아이템들이 생성된 후 일정 시간이 지났거나, 저장 공간이 일정 크기를 넘어선다면 오래된 아이템을 지우게 됩니다. ReplaySubject를 사용할 때는 onNext메소드를 여러 쓰레드에서 호출하지 않도록 유의해야 합니다. 만약 이 메서드를 호출하게 되면 Subject에서 어떤 아이템이나 알림이 재생될지 모르는 모호성을 지니게 되는데, 이는 옵저버블이 순차적으로 항목을 방출해야 한다는 옵저버블의 계약을 위배하는 것이기 때문입니다.

예제

let disposeBag = DisposeBag()

let subject = ReplaySubject<String>.create(bufferSize: 2)

subject.onNext("1")
subject.onNext("2")
subject.onNext("3")

subject.subscribe(onNext: { value in
    print("Subscriber 1 : \(value)")
}).disposed(by: disposeBag)

subject.onNext("4")

subject.subscribe(onNext: { value in
    print("Subscriber 2 : \(value)")
}).disposed(by: disposeBag)

subject.onNext("5")

Subscriber 1 : 2
Subscriber 1 : 3
Subscriber 1 : 4
Subscriber 2 : 3
Subscriber 2 : 4
Subscriber 1 : 5
Subscriber 2 : 5

위 코드에서 ReplaySubject의 버퍼 사이즈는 2로 설정이 되었기 때문에 Subscriber 1이 구독하자마자 이전에 저장된 2, 3을 전달해줍니다. Subscriber 2이 구독한 시점에서는 3과 4를 전달해줍니다.