[RxKotlin] 2. Observable (콜드 옵저버블, 핫 옵저버블, Subject)

간결함을 위해 생략한 내용이 있으니 RxKotlin 을 처음 접한다면 지난 글을 먼저 읽는 것이 좋다.

• [RxKotlin] 기본 개념 (Observable 과 Observer)
• [RxKotlin] Observable (콜드 옵저버블, 핫 옵저버블)
• [RxKotlin] Single, Maybe (작성 예정)
• [RxKotlin] Flowable, Subscriber, 백프레셔
• [RxKotlin] 연산자
• [RxKotlin] 스레드
• [RxKotlin] 안드로이드에 적용

 

1. Cold observable

우리가 지난 글에서 다루었던 Observable 은 콜드 옵저버블이다. 핫 옵저버블은 ConnectableObservable 타입을 갖는다. 그렇다면 콜드 옵저버블과 핫 옵저버블의 차이점은 무엇일까? 그것은 이미 배출되었던 데이터를 저장하고 있는지, 저장하지 않고 있는지에 따라 결정된다. 다음은 콜드 옵저버블의 예이다.

fun main(args: Array<String>) {
    val coldObservable: Observable<Int> = Observable.just(1, 2, 3)

    coldObservable.subscribe {
        println("cold 1 : $it")
    }

    coldObservable.subscribe {
        println("cold 2 : $it")
    }
}

실행 결과

위 코드의 실행결과를 보면 알 수 있듯이, 언제 구독했는지에 상관 없이 모든 Observer 가 모든 데이터를 받을 수 있다.

 

2. Hot observable

핫 옵저버블을 이용하는 다음 코드를 보자.

fun main(args: Array<String>) {
    val hotObservable: ConnectableObservable<Int> = Observable.just(1, 2, 3).publish()

    hotObservable.subscribe {
        println("hot 1 : $it")
    }

    hotObservable.subscribe {
        println("hot 2 : $it")
    }

    // connect 시점에 데이터를 배출한다.
    hotObservable.connect()

    hotObservable.subscribe {
        println("hot 3 : $it")
    }
}

실행 결과

위 코드의 실행 결과 1과 2는 실행되었지만 3은 실행되지 않았다. 따라서 데이터를 배출한 connect 시점 이후 구독한 구독자들은 그 값을 받지 못 했다.

 

3. Subject

Subject 는 Observable 이 갖춰야할 모든 메서드를 갖고 있다. 또한 Observer 의 라이프사이클 메서드를 직접 호출할 수 있다.

fun main(args: Array<String>) {
    val subject: Subject<Int> = PublishSubject.create()

    subject.subscribe {
        println("subject $it")
    }

    subject.onNext(1)
    subject.onNext(2)
    subject.onNext(3)
    subject.onComplete()
}

위 코드에서는 Subject 인스턴스를 생성한 이후에 onNext 를 호출하여 데이터를 배출하고 있다. 같은 출력을 보이는 또 다른 코드가 있다.

fun main(args: Array<String>) {
    val subject: Subject<Int> = PublishSubject.create()
    val observable = listOf(1, 2, 3).toObservable()

    subject.subscribe {
        println("subject $it")
    }

    // observable 이 배출하는 값을 subject 에서도 배출한다.
    observable.subscribe(subject)
}

위 코드는 Subject 에서 구독자를 먼저 구현해 준 이후에 Observable 을 지정해주는 방법이다. 이전 코드에서는 Subject 가 Observable 의 역할을 했다면 이 코드에서는 Observer 의 역할을 한 셈이다.

RxKotlin 에서는 다른 종류의 Subject (BehaviorSubject, ReplaySubject) 도 지원하지만 여기서는 가장 기본적인 PublishSubject 만 설명했다.