[RxSwift] Subject

1. Subject란 

Observable과 Observer 모두로 동작할 수 있다.

Subject는 ObservableType프로토콜을 채택하고 있는 Observable을 상속하고 있고

ObserverType프로토콜을 채택하고 있기 때문이다.

 

 

 

쉽게 말하면, 데이터를 넣어줄 수도 있고(emit시킬수도 있고),  subscribe할 수도 있다.

            let subject = PublishSubject<String>()

            let subcriptionOne = subject.subscribe(onNext: { (string) in
                print(string)
            })
            
            subject.on(.next("1"))
            subject.onNext("2")
            
            Observable.just(3).subscribe(subject).disposed(by: disposeBag)

 

2. Subject의 종류 

1) PublishSubject
Starts empty & only emits new elements to subscribers

빈 상태로 시작한다. 오직 새로운 elements만을 subscribers한테 emit한다

 

 

만약, Observable이 오류 때문에 종료되면 PublishSubject는 아무런 항목도 배출하지 않고 Observable에서 발생한 오류를 그대로 전달한다.

2) BehaviorSubject 

Starts with an initial value & replays it or the latest element to new subsribers 

초기값을 가진채로 시작한다. 그 초기값을 다시 방출하든지, 아니면 가장 최신 값을 새로운 subscriber에게 방출한다.

 

만약, Observable이 오류 때문에 종료되면 BehaviorSubject는 아무런 항목도 배출하지 않고 Observable에서 발생한 오류를 그대로 전달한다.

 

 

3) ReplaySubject 

Initialized with a buffer size and will maintain a buffer of elements up to that size and replay it to new subscribers.

buffer size로 시작한다. 그 사이즈까지 버퍼의 원소들을 유지하며 그것들을 새로운 subscriber에게 방출한다.

 

만약, Observable이 오류 때문에 종료되면 ReplaySubject는 아무런 항목도 배출하지 않고 Observable에서 발생한 오류를 그대로 전달한다. ==> 기존 구독자들에게만...!  (subject.onError 전에 subcribe한 친구들)  // Rx사이트에 사진이 없다

 

subject.onError후에 새로 구독한 구독자들에게는 버퍼에 있는 값들 + error를 전달한다 // 이 점이 다른 subject들과 다른 특이점이여서 사진이 없나...? 🤔

 

 

4) AsyncSubject

Subject가 completed 되면 가장 마지막 데이터를 그때서야 새로운 subscriber에게 넘겨준다

Observable이 오류로 인해 종료될 경우 AsyncSubject는 아무 항목도 배출하지 않고 발생한 오류를 그대로 전달한다. 

 

(책에서는 없지만, Rx사이트에는 Async Subject에 대한 설명도 있어서 추가했음--! )

 

5) Variable
wraps a behaviorsubject,preserves its current value as state & replays only the latest/initail value to new subsribers 

BehaviourSubject를 래핑했다. 현재 vaule을 상태(state)로 보존하고 최신/초기 값(latest/initial)만 새로운 subscriber에게 방출한다.

 

 

2.1 PublishSubject 

PublishSubject는 구독된 순간 새로운 이벤트를 구독자에게 알리고 싶을 때 유용하다. 이 작업은 구독을 취소하거나, .completed나 .error 이벤트로 subject가 종료될 때까지 일어난다.

 

마블 다이어그램을 보면,

첫번째 라인은 publish subject / 두번째와 세번째 라인은 subsribers이다

upward-pointing arrow는 subscription을 나타내고 downward-pointing arrow는 방출되는 이벤트를 나타낸다

첫번째 subscriber는 1 이후에 구독했다. 그래서 1 이벤트를 받지 못했고 2,3만 받았다

두번째 subscriber는 2 이후에 구독했다. 그래서 3만 받았다.

 

        example(of: "Publish subject") {
            
            let subject = PublishSubject<String>()
            // 아직 Observer가 없어서 아무것도 출력안됨
            subject.onNext("Is anyone listening?")

            // observer 하나 추가
            let subcriptionOne = subject.subscribe(onNext: { (string) in
                print("1)",string)
            })
            
            // subsribe한 이후부터 출력됨
            // on(.next) = onNext
            subject.on(.next("1"))
            subject.onNext("2")
            
            let subscriptionTwo = subject.subscribe({ event in
                print("2)", event.element)
            })
            
            subject.onNext("3")
            
            subcriptionOne.dispose()
            subject.onNext("4")
    }



--- Example of: Publish subject ---
1) 1
1) 2
1) 3
2) Optional("3")
2) Optional("4")

 

Subject가 한 번 completed되면 새로운 구독이 되더라도 completed이벤트를 다시 방출한다 

            subject.onCompleted()
            subject.onNext("5")
            
            subscriptionTwo.dispose()
            
            let disposeBag = DisposeBag()
            
            subject.subscribe{
                print("3)", $0.element ?? $0)
            }.disposed(by: disposeBag)
            
            subject.onNext("?")


// output 
2) completed
3) completed

time-sensitive data를 모델링하는 앱에 사용하기 좋다.

때때로 새로운 구독자에게 completed가 아니라 마지막 element가 뭐였는지 알려주고 싶을 수도 있다

 

비록 그 element가 subscription전에 방출되었더라도!! behavior subject로 렛츠고--! 

 

2.2 BehaviorSubject

Behavior subject는 publish subject와 비슷하게 동작한다. 하지만 새로운 구독자에게 가장 최근의 .next 이벤트를 재방출한다는 점이 다르다.

첫번째 라인은 subject

두번재 라인은 first subscriber - 1 이후에 구독해서 구독하자마자 1을 즉각적으로 받아온다

세번째 라인은 second subscriber - 2 이후에 구독해서 구독하자마자 2을 즉각적으로 받아온다

BehaviorSubject instance 를 만들때 그것의 initializer는 항상 initial vaule를 받는다 (만약 creation time에 default initial value를 줄 수 없다면 PublishSubject를 써라)

 

구독하기 전 subject에 추가된 값이 없으므로 초기값을 방출해주었다.

        example(of: "behaviour subject") {
        
            let subject = BehaviorSubject(value: "Initial value")
            
            subject.subscribe({ (event) in
                print("1)",event)
            }).disposed(by: disposeBag)
        }

--- Example of: behaviour subject ---
1) next(Initial value)

 

subscription 전에 onNext를 했다. latest element가 X니까 X를 방출해준다. 

 

        example(of: "behaviour subject") {
            
            let subject = BehaviorSubject(value: "Initial value")
            subject.onNext("X")
            subject.subscribe({ (event) in
                print("1)",event)
            }).disposed(by: disposeBag)
        }

--- Example of: behaviour subject ---
1) next(X)

 

enum MyError: Error {
    case anError
}

example(of: "behaviour subject") {
            
            let subject = BehaviorSubject(value: "Initial value")
            subject.onNext("X")
            subject.subscribe({ (event) in
                print("1)",event)
            }).disposed(by: disposeBag)
            
            subject.onError(MyError.anError)
            subject.subscribe({ (event) in
                 print("2)",event)
            }).disposed(by: disposeBag)
            
        }


--- Example of: behaviour subject ---
1) next(X)
1) error(anError)
2) error(anError)

 

가장 최신의 이벤트를 구독을 시작하자마자 방출해주기 때문에 가장 최신의 데이터로 뷰를 먼저 구성하고 싶을 때 사용하면 좋다.

앱이 새로운 data를 fetch 하는 동안 최신의 값들로 화면을 미리 보여준다

 

 

그러나 만약 최신의 값 하나가 아니라 더 많은 값을 보여주고 싶다면?! 예를 들어 만약 검색 화면에서 가장 최신의 값들을 5개까지 보여주고 싶으면 어떻게 해야 할까? replay subject로 가보자--! 

 

 

 

2.3 ReplaySubject 

ReplaySubject는 그동안 방출한 이벤트 중 최근의 것들을 특정 사이즈만큼 캐시(버퍼)시켜 놓는다. 이것들을 새로운 구독자에게 재방출한다.

마블 다이어그램을 보면

 

버퍼사이즈가 2인 replay subject

first subscriber (middle line)은 이미 replay subject를 구독했다. 그래서 방출된 element를 받고 있다.

second subscriber (bottom line)은 2 이후에 구독한다 그래서 1,2 를 받는다.

        example(of: "Replay Subject") {
            let subject = ReplaySubject<String>.create(bufferSize: 2)
            subject.onNext("1")
            subject.onNext("2")
            subject.onNext("3")
            
            subject.subscribe{
                print("1)",$0)
            }.disposed(by: disposeBag)
            
            subject.subscribe{
                print("2)",$0)
            }.disposed(by: disposeBag)
        }

--- Example of: Replay Subject ---
1) next(2)
1) next(3)
2) next(2)
2) next(3)

            subject.onNext("4")
            subject.subscribe{
                print("3)",$0)
            }.disposed(by: disposeBag)

// 
1) next(4)
2) next(4)
3) next(3)
3) next(4)

first랑 second subscription은 이미 구독중이기 때문에 새로운 element를 하나받고

 

새로운 third subscriber은 마지막 두개의 buffered elements를 받는다

 

            subject.onNext("4")
            subject.onError(MyError.anError)
            subject.subscribe{
                print("3)",$0)
            }.disposed(by: disposeBag)

// 
1) next(4)
2) next(4)
1) error(anError)
2) error(anError)
3) next(3)
3) next(4)
3) error(anError)

replay subject는 error와 함께 terminated되었지만, 새로운 subscriber에게 re-emit한다!

그러나 버퍼가 여전히 남아있어서 stop event가 re-emit되기 전에 새로운 subsriber에게 재생된다

📌 버퍼사이즈 2인데  3)이 next(3)을 받는 것을 보니 .... 에러는 안차고 element 만 차는 듯--!! 하다 

 

하지만 error후에 바로 dispose하면 새로운 subscriber는 오직 error event만 받는다. 이 에러 이벤트는 해당 subject가 이미 disposed되었음을 의미한다. 

            subject.onError(MyError.anError)
            subject.dispose()
            subject.subscribe{
                print("3)",$0)
            }.disposed(by: disposeBag)

//
1) error(anError)
2) error(anError)
3) error(Object `RxSwift.(unknown context at 0x101914d68).ReplayMany<Swift.String>` was already disposed.)

이렇게 dispose()를 명백히 불러주는 것은 너가 해야할 일이 아니다. subscriptions를 dispose bag에 추가해두면(strong reference cycle를 피하면서) the owner(예를들어 viewcontroller나 viewmodel) 이 deallocated될때 everything이 disposed되고 deallocated된다.

 

// from 준상

ReplaySubject를 사용할 때 염두해야 할 점이 두 가지가 있다.

• 버퍼가 메모리에 유지되기 때문에 너무 큰 버퍼 사이즈를 유지할 경우 시스템적으로 좋지 않다.

• 배열의 ReplaySubject를 만드는 경우 사이즈가 크게 늘어나 시스템에게 부하를 준다.

또한, ReplaySubject가 .error 이벤트로 끝나도 disposed되지 않았다면 새로운 구독자가 생길 때 버퍼에 저장된 이벤트들이 재방출되고 이후 .error 이벤트를 방출한다.

 

2.4 AsyncSubject

2.5 Variable

 

Variable은 BehaviorSubject를 래핑하고 그것의 현재 값을 상태로 저장한다. 현재 값을 value 프로퍼티로 접근이 가능하며 다른 subject나 observable과 달리 value 프로퍼티에 직접 새로운 값을 설정할 수 있다. 대신 onNext(_:)를 사용할 수 없다.

BehaviorSubject를 매핑했기 때문에 초기값을 통해 생성되고 초기값이나 최신값을 replay한다.

 

variable의 underlying subject에 접근하고 싶다면,asObservable() 메서드를 통해 접근이 가능하다.

다른 subject들과 또 다른 특징은 .error 이벤트를 방출하지 않음을 보장한다는 점이다. 그래서 .error 이벤트를 Variable에 추가할 수 없다.

variable은 deallocated 되려고 할때 자동적으로 complete된다. 그래서 .completed 이벤트를 추가할 수 없다

 

 

  example(of: "Variable") {
            let variable = Variable("Initial value")
            variable.value = "new initial value"
            
            variable.asObservable().subscribe{
                print("1)",$0)
            }.disposed(by: disposeBag)
            
            variable.value = "1"
            variable.asObservable().subscribe{
                print("2)",$0)
            }.disposed(by: disposeBag)
            
            variable.value = "2"
            
        }

--- Example of: Variable ---
ℹ️ [DEPRECATED] `Variable` is planned for future deprecation. Please consider `BehaviorRelay` as a replacement. Read more at: https://git.io/vNqvx
1) next(new initial value)
1) next(1)
2) next(1)
1) next(2)
2) next(2)
1) completed
2) completed

 

variable 자체에 .error 또는 .completed 이벤트를 추가할 방법이 없다

variable. 찍으면 자동완성으로 value만 나온다

이런식으로 다 해봐도 에러가 뜨고 안된다

            variable.value.onError(MyError.anError)
            variable.asObservable().onError(MyError.anError)
            variable.value = MyError.anError
            variable.value.onCompleted()
            variable.asObservable().onCompleted()

 

Variable은 current value를 단순히 체크해보고 싶을때 ( update를 받는 subscribing 없이 ) 사용한다.

 

 

 

Reference

RxSwift by raywenderlich.com