RxKotlin- Part 7. ConnectableObservable Class

ConnectableObservable 클래스는 Subject 클래스와 같이 콜드 Observable을 핫 Observable로 변환합니다.

Observable을 여러 구독자에게 동시에 전달이 가능하며, subscribe() 함수를 호출하더라도 아무런 동작이 일어나지 않는다는 특징이 있습니다. 

 

subscribe() 함수를 호출하더라도 아무런 동작이 일어나지 않는 이유는 connect() 함수를 호출한 시점부터 subscribe() 함수를 호출한 구독자에게 데이터를 발행하기 때문입니다. 

 

ConnectableObsevable 객체를 생성하려면 먼저 Observable에 publish() 함수를 호출하면 됩니다. publish() 함수는 여러 구독자에게 데이터를 발행학 위해 connect() 함수를 호출하기 전까지 데이터 발행을 미루는 역할을 합니다.

 

http://reactivex.io/RxJava/javadoc/io/reactivex/observables/ConnectableObservable.html

즉,

publish() 함수는 Observable을 ConnectableObservable로 변환해주고

subscribe() 함수를 호출해도 데이터는 발행되지 않습니다.

 

ConnectableObservable Class의 메소드인 connect() 함수를 호출해야 데이터가 발행됩니다.

 

기존의 콜드 Observable은 subscribe() 함수를 호출하면 데이터가 차례로 발행되었는데, 핫 Observable은 구독자가 subscribe()를 한 시점부터의 데이터를 발행된다는 것을 알 수 있습니다.

 

/** ConnectableObservable Class **/
fun main(){
    val observable = PublishSubject.create<String>()

    val connectableObservable = observable.publish()

    observable.onNext("1")

    connectableObservable.subscribe {data -> println("Subscriber 1 : $data")}
    connectableObservable.subscribe {data -> println("Subscriber 2 : $data")}
    connectableObservable.connect()

    observable.onNext("2")

    connectableObservable.subscribe {data -> println("Subscriber 3 : $data")}

    observable.onNext("3")
}

실행결과 : 

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