[RxSwift] Subjects

이번 포스팅에서 알아볼 Subject를 알아보기에 앞서, HOT Observable과 COLD Observable에 대해 알아보자! 

 

Cold Observable & Hot Observable 

Cold Observable이란 구독이 되기 전까지 동작하지 않다가, 

구독이 되면, 모든 데이터를 한번에 방출한다.

Cold Observable을 marble diagram으로 표현하면 아래와 같다.(점선은 구독을 의미한다)

지금까지 알아보았던 Observable이 Cold observable이다. 

또한, Unicast방식으로 동작하여,

Subscriber(Observer)마다 고유의 Stream이 생성된다.

 

 

Hot Observable은 구독 여부와 상관없이 방출할 데이터가 있으면 방출한다.

구독이 되면, 모든 데이터를 방출하는 것이 아닌! 

구독 시점 이후로 방출된 데이터만 받는다.

이번 포스팅에서 알아볼 Subject가 Hot Observable이다. 

이는 Multicast 방식으로 동작하여, 

하나의 stream을 통해 여러 subscriber에게 데이터를 방출한다.

 

또한, Subject는 Observable이자 Observer이다.

즉! 이벤트를 받을 수 있고, 이 이벤트를 Subscriber에게 전달 할 수 있다. 

이러한 특성 때문에 원하는 시점에 (수동적으로) 이벤트를 방출 할 수 있다.

 

 

PublishSubject

let subject = PublishSubject<String>()

이런식으로 타입제한을 해주어야 한다.

 

 

또한 Observable은 생성과 동시에 방출할 데이터를 명시해야 했다면, Subject에서는 그렇지 않다. 

let Observable = Observable.just("1")

이러한 이유 또한,

Observable은 Cold Observable이므로, 구독과 동시에 모든 데이터를 방출해야 했지만, 

Subject는 Hot Observable로, 원하는 시점에 데이터를 방출 할 수 있기 때문이다. 

 

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

subject
    .subscribe(
        onNext: { print($) },
        onCompleted: { print("completed") }
    )
    .disposed(by: DisposeBag)

subject.onNext("3")
subject.onNext("4")
subject.onComleted() //stream 종료

/* Prints:
 3
 4
 completed
 */

위와 같이 onNext, onCompleted, onError를 통해 publishSubject 이벤트를 전달 받고!!(Observer처럼) 

그리고 이를 subscriber에게 전달한다!

 

BehaviorSubject

BehaviorSubject는 PublishSubject와 유사하지만, 

한가지 다른점은 "직전에 방출 되었던 데이터" 도 같이 방출 한다는 점이다.

 

아무런 방출된 데이터가 없을때는 초기값을 방출하는데 

이러한 이유 때문에 추가적으로 value:를 통해 초기값을 설정 해주어야 한다.

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

subject.onNext("1")

    .subscribe(
        onNext: { print($) },
        onCompleted: { print("completed") }
    )
    .disposed(by: DisposeBag)

subject.onNext("3")
subject.onNext("4")
subject.onComleted() //stream 종료

/* Prints:
 1
 3
 4
 completed
 */

위의 예시에서 구독전에 방출된 데이터("1") 이 없다면, 

1대신 initial이 방출된다.

 

 

ReplaySubject

ReplaySubject는 BehaviorSubject와 유사한데, 

BehaviorSubject는 직전의 데이터가 방출되었다면, 

ReplaySubject는 직전의 데이터를 Buffer에 저장하여, 같이 방출한다.

 

이러한 이유 때문에 bufferSize를 정해 주어야 한다.

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

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

    .subscribe(
        onNext: { print($) },
        onCompleted: { print("completed") }
    )
    .disposed(by: DisposeBag)

subject.onNext("3")
subject.onNext("4")
subject.onComleted() //stream 종료

/* Prints:
 1 
 2
 3
 4
 completed
 */

하지만, ReplaySubject를 남발하는 것은 좋지 않은데, 

이유는 buffer는 메모리를 사용하는 행위이기 때문에, 

사이즈가 큰 데이터(이미지, 비디오)의 경우 치명적일 수 있다

 

 

 

Relay

Relay란 RxCocoa에 속해있는 클래스이다. 

크게 2가지 클래스가 있는데, 

• PublishRelay : PublishSubject를 wrapper
• BehaviorRelay : BehaviorSubject를 wrapper

Subject에서는 next, completed, error 이벤트를 받아 subscriber에게 전달 할 수 있었지만, 

Relay에서는 next이벤트만 받아 subscriber에게 전달한다.

 

Relay는 UI이벤트를 위해 사용이 되는데, 

UI업데이트를 해야 하는데 해당 stream이 끊기면 안되기 때문에! 

next이벤트만 전달 받는다.

 

또한, next이벤트를 전달 받을때, onNext가 아닌 accept 메서드를 통해 이루어 진다.

let subjectRelay = SubjectRelay<String>()
let behaviorRelay = BehaviorRelay<String>()

 

사용 예시 

우선 UITextField에서 숫자를 입력하면, 이를 UILabel에 반영하는 앱을 Subject를 사용하여 구현해보자

 

 

이를 코드로 구현하면, 

//ViewModel
class ViewModel {
    let disposBag = DisposeBag()
    
    let inputSubject = PublishSubject<String>()
    let inputObserver: AnyObserver<String>

    
    let outputSubject = PublishSubject<String>()
    let outputObserver: AnyObserver<String>

    
    init() {
        inputObserver = inputSubject.asObserver()
        outputObserver = outputSubject.asObserver()
        
        inputSubject // inputSubject를 subscribe -> 여기사 subscriber되는거임
            .subscribe(
                onNext: { [weak self] someData in
                    guard let strongSelf = self, let intergerValue = Int(someData) else {
                        return
                    }
                    strongSelf.outputObserver.onNext("\(intergerValue * 2)") //textField로 부터 onNext로 들어온 이벤트를! 연산을 수행후
                                                                             // outputSubject에세 onNext를 통해 이벤트를 전달한다.
                }
            )
            .disposed(by: disposBag)
        
        
    }
}

//ViewController
class ViewController: UIViewController {
    ...
    
    private let viewModel = ViewModel()
    
    override func viewDidLoad() {
        super.viewDidLoad()
        
        textField.rx.controlEvent(.editingDidEnd) // textField의 editing이 끝나면
            .withLatestFrom(textField.rx.text)    // textField의 text중 가장 최근 값을
            .map{ $0! }                           // Optional Binding하여
            .subscribe(onNext: viewModel.inputObserver.onNext) // inputSubject의 onNext로 이번트를 전달한다.
            .disposed(by: disposBag)
        
        viewModel.outputSubject // outputSubject의 Subscriber
            .subscribe(onNext: { [weak self] resultValue in
                guard let strongSelf = self else {
                    return
                }
                strongSelf.resultLabel.text = resultValue // UI update
            })
            .disposed(by: disposBag)
    }
    
}

하지만, 이전에도 말했듯이! 

UI업데이트는 끊기면 안되기 때문에, Relay로 이를 바꾸어 보면,

 

class ViewModel {
    ...

    let outputRelay = PublishRelay<String>()

    
    init() {
        ...
        inputSubject
            .subscribe(
                onNext: {
                    ...
                    strongSelf.outputRelay.accept("\(intergerValue * 2)") //여기만 accept로 바꿔줌
                }
            )
            .disposed(by: disposBag)
    }
}

ViewController는 동일하다. 

 

하지만, 나중에 공부해볼 Operator를 사용하게 되면! 

class ViewController: UIViewController {
    ...
    
    override func viewDidLoad() {
        ...

        viewModel.outputRelay
            .bind(to: resultLabel.rx.text)
            .disposed(by: disposBag)
    }
}

outputRelay의 구현부가 이렇게 짧아진다.

 

 

Reference

https://github.com/fimuxd/RxSwift/blob/master/Lectures/03_Subjects/Ch3.%20Subjects.md

https://www.notion.so/Wallaby-RxSwift-72194669a39a4557baa69c672268af38