[RxJava] Observable 디버깅하기

🔔 앞으로의 Reactive X 시리즈는 RxJava, RxKotlin 기준으로 작성됩니다
공식 문서를 참고하여 작성된 포스팅입니다.

이전 포스팅과 이어집니다.

Observable 디버깅 연산자

데이터 스트림에서는 다양한 이벤트가 발생한다. 이러한 흐름이 어떻게 흘러가고 있는 지 손쉽게 파악할 수 있도록 다양한 연산자들을 제공해준다.

doOnEach() 메소드

이 연산자는, 데이터가 발행되기 직전에 Notification<T> 라는 녀석을 통해 발행되는 데이터를 콜백 형태로 확인할 수 있도록 해준다. 아래와 같은 프로퍼티를 지원한다.

fun main() {
    Observable.just(1, 2)
        .doOnEach { notification ->
            println("data? " + notification.value)
            println("isOnNext? " + notification.isOnNext)
            println("isOnComplete? " + notification.isOnComplete)
            println("isOnError? " + notification.isOnError)
        }
        .subscribe(System.out::println)
}

data? 1
isOnNext? true
isOnComplete? false
isOnError? false
1
data? 2
isOnNext? true
isOnComplete? false
isOnError? false
2
data? null
isOnNext? false
isOnComplete? true
isOnError? false

실행을 하게 되면, 발행받은 데이터를 출력하는 구독자가 데이터를 출력하기 이전에 doOnEach() 문 안에서 작성한 출력문이 먼저 실행되는 것을 확인할 수 있고, onNext() 를 통해 발행된 데이터들의 경우 isOnNext 프로퍼티가 true 로 되어있는 것을 확인할 수 있다.

또한, 데이터가 onNext() 이벤트를 통해 모두 발행이 되면 onComplete() 이벤트가 발생하여 마지막 출력문에 isOnComplete 프로퍼티가 true 로 되어있는 것을 확인할 수 있다.

doOnNext() 메소드

이 녀석도 데이터가 발행되기 전에 디버깅을 도와주는데, doOnEach() 메소드와 달리 Notification<T> 형식이 아니라 구독자 마냥 Consumer 형태를 사용하게 된다.

fun main() {
    Observable.just(1, 2, 3, 4)
        .doOnNext {
            println("doOnNext : $it")
        }
        .subscribe(System.out::println)
}

doOnNext : 1
1
doOnNext : 2
2
doOnNext : 3
3
doOnNext : 4
4

doOnSubscribe() 메소드

얘는 어떤 구독 주체가 해당 Observable (데이터 스트림) 을 구독하기 시작할 때 콜백을 받을 수 있게 해주는 연산자이다. Disposable 을 전달하기 때문에, dispose 관련 동작을 수행할 수 있다.

fun main() {
    Observable.just(1, 2, 3)
        .doOnSubscribe { it: Disposable ->
            if (it.isDisposed) println("Disposed!") else println("Subscribe Start")
        }
        .subscribe(System.out::println)
}

Subscribe Start
1
2
3

doOnComplete() 메소드

정말 메소드 이름들이 직관적이라 그저 편ㅡ안하다. 이 연산자도 이름에서 유추할 수 있듯 Emitter 인터페이스가 onComplete() 를 호출했을 때를 감지하여 콜백을 받을 수 있게 해주는 녀석이다.

fun main() {
    Observable.just(1, 2, 3)
        .doOnComplete { println("Completed!") }
        .subscribe(System.out::println)
}

1
2
3
Completed!

doOnError() 메소드

얘도 뭐, 말 그대로 에러가 발생했을 때 콜백을 받을 수 있게 해준다. 즉, Emitter 인터페이스가 onError() 이벤트를 발생했을 때 호출된다. Observable 이 오류가 발생하여 종료되기 직전에 실행된다.

파라미터로는 Throwable 이 제공된다. 아래는, map 을 통해 데이터 스트림의 데이터들을 Int 자료형으로 변환 시키는 동작을 하다가 정수형으로 변환 불가능한 데이터를 만났을 때의 오류 처리를 보여준다.

import io.reactivex.Observable

fun main() {
    Observable.just("1", "2", "H43RO", "3", "4")
        .map { it.toInt() }
        .doOnError { e: Throwable ->
            println(e.toString())
        }
        .subscribe(System.out::println)
}

1
2
java.lang.NumberFormatException: For input string: "H43RO"

다만 에러 디버깅 용으로만 사용되는 것이므로, 에러 처리는 별도로 해줘야 한다.