RxJava,RxKotlin - Observable (5) - ConnectableObservable

[출처 - RxJava 프로그래밍 : 리액티브 프로그래밍 기초부터 안드로이드 까지 한번에]

본 글은 'RxJava 프로그래밍 : 리액티브 프로그래밍 기초부터 안드로이드 까지 한번에' 를 학습하면서 정리한 글입니다

 

마블다이어그램참고 - http://reactivex.io/RxJava/javadoc/io/reactivex/subjects/Subject.html

ConnectableObservble은 차가운 Observable을 뜨거운 Observable로 변환합니다. Subject 처럼 말이죠.  이 ConnectableObservble은 Observable을 여러 구독자에게 공유할 수 있으므로 원 데이터 하나를 여러 구독자에게 동시에 전달할 때 사용합니다.

마블다이어그램으로 설명해 드리겠습니다.

이 ConnectableObservable은 위의 마블다이어그램을 보면 알 수 있듯이 subscribe()에 동작하지 않고, connect()라는 새로운 함수를 호출한 시점 부터 subscribe()를 한 구독자들에게 데이터를 발행합니다.

 

Code로 알아 보겠습니다.

Java

    public static void main(String[] args){

        PublishSubject<String> oPublishSubject = PublishSubject.create();

        ConnectableObservable<String> pc_Observable = oPublishSubject.publish();

        oPublishSubject.onNext("1");
        pc_Observable.subscribe(data->System.out.println("Subscribe #1 ->"+data));
        pc_Observable.subscribe(data->System.out.println("Subscribe #2 ->"+data));
        pc_Observable.connect();

        oPublishSubject.onNext("2");
        oPublishSubject.onNext("3");

        pc_Observable.subscribe(data->System.out.println("Subscribe #3 ->"+data));

        oPublishSubject.onNext("4");

        oPublishSubject.onComplete();


    }

Kotlin

fun main(){
    val oPublishSubject = PublishSubject.create<String>()

    val pc_Observable = oPublishSubject.publish()

    oPublishSubject.onNext("1")
    pc_Observable.subscribe { data -> println("Subscribe #1 ->$data") }
    pc_Observable.subscribe { data -> println("Subscribe #2 ->$data") }
    pc_Observable.connect()

    oPublishSubject.onNext("2")
    oPublishSubject.onNext("3")

    pc_Observable.subscribe { data -> println("Subscribe #3 ->$data") }

    oPublishSubject.onNext("4")

    oPublishSubject.onComplete()
}

결과

Code를 분석해보면 이전에 공부했던 PublishSubject를 사용하였습니다. 그래서 발행한 데이터를 바로바로 구독자에게 발행하게 되었습니다.

 

하지만 결과를 보면 subscribe #1과 subscribe #2에 2와 3이 발행되고, subscribe #1,subscribe #2,subscribe #3에 4가 발행됩니다.

이는 ConnectableObservable의 특성 때문으로, 데이터 1은 connect 전에 들어온 값이라 전달 받지 못하였고, connect 이후로 구독자들에게 데이터를 전달하기 때문입니다.

 

즉 connect 전에 1을 발행했기 때문에 1은 전달되지 못하였고, connect 이 후에 발행된 2와 3은 구독자 #1,#2에게 전달 되었습니다. 그리고 나서 구독자 #3이 구독을 하였고 4가 발행되어 #1,#2,#3에게 4가 발행 되었습니다.