오류 연산자

Observable을 다루다 보면 다른 코드와 마찬가지로 예외 사항이 발생하기 마련입니다. RxKotlin(RxJava)에서는 예외 사항을 알리고자 Emitter.onError()를 호출하여 오류 이벤트를 통지합니다. 그리고 Observer(구독자)는 이를 올바르게 처리해야 합니다.

예외가 발생하는 코드

Observable.just("1", "2", "a", "3")
    .map { value ->
        value.toInt()
    }
    .subscribe {
        println(it)
    }

/*
결과
1
2
NumberFormatException
 */

Observer에서 예외 사항을 처리하는 코드

Observable.just("1", "2", "a", "3")
    .map { value ->
        value.toInt()
    }
    .subscribeBy(
        onNext = { value -> println(value) },
        onError = { println("Error!!") }
    )

/*
결과
1
2
Error!!
 */

위의 예제에서는 구독 시에 오류를 처리하는 Consumer를 subscribeBy()의 인자로 추가하였습니다. RxKotlin(RxJava)에서는 subscribe에서 오류를 처리하는 방법 이외에도 몇 가지 연산자들을 제공하고 있습니다.

onErrorReturn 연산자

• onErrorReturn 연산자는 오류가 발생하면 아이템 발행을 종료하고, onError()를 호출하는 대신에 오류 처리를 위한 함수를 실행합니다.

• 마블 다이어그램을 보면 Observable에서 오류가 발생하자 onErrorReturn의 인자로 주어진 함수가 실행되어 빨간 마름모가 방출된 것을 확인할 수 있습니다.

Observable.just("1", "2", "a", "3")
    .map { value ->
        value.toInt()
    }
    .onErrorReturn {
        -1
    }
    .subscribe {
        println(it)
    }

/*
결과
1
2
-1 // onErrorReturn의 인자로 주어진 -1이 방출됨
 */

onErrorResumeNext 연산자

• onErrorResumeNext 연산자는 오류 발생 시 기존 스트림을 종료시키고, 다른 Observable 소스로 스트림을 대체합니다.

• 마블 다이어그램을 보면 Observable에서 오류가 발생하자 onErrorResumeNext의 인자로 주어진 Observable 소스로 스트림을 대체하여 파란색 원, 분홍색 원이 방출된 것을 확인할 수 있습니다.

Observable.just("1", "2", "a", "3")
    .map { value ->
        value.toInt()
    }
    .onErrorResumeNext {
        Observable.just(100, 200, 300)
    }
    .subscribe {
        println(it)
    }

/*
결과
1
2
100
200
300
 */

• 예제 코드를 보면 "a"는 Int 타입으로 형 변환이 불가능하여 오류가 발생하기에 onErrorResumeNext의 Observable 소스로 스트림을 대체해서 100, 200, 300 아이템이 방출된 것을 확인할 수 있습니다.

retry 연산자

• retry 연산자는 에러 발생 시 인자로 주어진 횟수 만큼 재시도하는 연산자입니다. 즉, Observable이 error를 발행했을 때, Observable을 재구독하도록 하는 것입니다.

• 보통 네트워크 요청 후 응답을 받아올 경우 실패시에 몇 번 더 요청할 지 결정하는 상황에서 사용할 수 있습니다.

• 마블 다이어그램을 보면 오류가 발생했을 시, 재시도하여 정상적으로 데이터를 모두 방출하는 모습을 확인할 수 있습니다.

Observable.just("1", "2", "a", "3")
    .map { value ->
        value.toInt()
    }
    .retry(2)
    .onErrorReturn {
        -1
    }
    .subscribe {
        println(it)
    }

/*
결과
1
2
1
2
1
2
-1
 */

• 예제 코드를 보면 "a"를 Int 타입으로 변형하려고 하기에 오류가 발생하였습니다. 이때, retry의 인자로 넘긴 2번 만큼 더 재시도하는 모습을 확인할 수 있습니다. 최종적으로 오류가 발생하여 onErrorReturn에 지정한 데이터가 방출되는 코드입니다.

디버깅을 돕는 연산자

직접 Emitter를 제어하는 일은 별로 없어서, Observable 스트림 내부를 확인하기 힘든 경우가 있습니다. 이럴 때를 대비해서 RxKotlin(RxJava)에서는 doOn-- 연산자를 제공해 디버깅을 돕습니다.

doOnEach 연산자

• doOnEach 연산자는 Observable이 아이템을 발행하기 전에 이를 콜백으로 확인할 수 있도록 해줍니다. 콜백은 Notification 형태로 들어옵니다.

• Notification에는 다음과 같은 메서드(코틀린은 프로퍼티)가 존재하여 현재 어떤 상태인지 확인할 수 있습니다.

  • value: onNext에 의해 획득된 데이터입니다.
  • isOnNext: onNext 신호인 경우 true를 반환합니다.
  • isOnComplete: onComplete 신호인 경우 true를 반환합니다.
  • isOnError: onError 신호인 경우 true를 반환합니다.
  • error: onError 신호인 경우라면 Throwable을 반환하고, 그렇지 않으면 null을 반환합니다.

• 마블 다이어그램을 확인하면 매번 아이템이 발행될 때마다 notification이 방출되는 것을 확인할 수 있습니다.

Observable.just(1, 2, 3)
    .doOnEach { notification ->
        val num = notification.value
        val isOnNext = notification.isOnNext
        val isOnComplete = notification.isOnComplete
        val isOnError = notification.isOnError
        val throwable = notification.error
        println("num : $num")
        println("isOnNext : $isOnNext")
        println("isOnComplete : $isOnComplete")
        println("isOnError : $isOnError")
        throwable?.let {
            it.printStackTrace()
        }
    }.subscribe { value ->
        println("Subscribed : $value")
    }

/*
결과
num : 1
isOnNext : true
isOnComplete : false
isOnError : false
Subscribed : 1
num : 2
isOnNext : true
isOnComplete : false
isOnError : false
Subscribed : 2
num : 3
isOnNext : true
isOnComplete : false
isOnError : false
Subscribed : 3
num : null
isOnNext : false
isOnComplete : true
isOnError : false
 */

doOnNext 연산자

• doOnNext 연산자는 doOnEach와 형태가 매우 비슷하지만 Notification 대신 간단히 발행된 아이템을 확인할 수 있는 Consumer를 파라미터로 넘깁니다.

• 마블 다이어그램을 보면 아이템이 발행될 때마다 onNext가 호출되므로 해당 아이템을 가진 Consumer가 방출되는 것을 확인할 수 있습니다.

Observable.just(1, 2, 3)
    .doOnNext { item ->
        if(item > 2) {
            throw IllegalArgumentException()
        }
    }.subscribeBy(
        onNext = { println(it) },
        onError = { println("Error!!") }
    )

/*
결과
1
2
Error!!
 */

• 예제 코드는 doOnNext로 아이템을 가진 Consumer가 넘어오고, 해당 Consumer를 통해 아이템 값이 2가 넘는 경우 에러를 발생시키도록 작성하였습니다.

doOnSubscribe 연산자

• doOnSubscribe 연산자는 구독 시마다 콜백을 받을 수 있도록 합니다. 매개 변수로 Disposable을 받을 수 있습니다. Disposable에 대한 개념이 궁금하시다면 RxKotlin(RxJava) #1 - RxKotlin의 개념과 Observable을 통해 확인하실 수 있습니다.

• 마블 다이어그램을 보면 Observable을 구독한 Observer가 생기자, doOnSubscribe의 인자로 주어진 별이 방출된 것을 확인할 수 있습니다.

val observable = Observable.just(1, 2, 3)
    .doOnSubscribe { disposable ->
        println("구독")
    }

observable.subscribe {
    println(it)
}

observable.subscribe {
    println(it)
}

/*
결과
구독
1
2
3
구독
1
2
3
 */

doOnComplete 연산자

• doOnComplete 연산자는 Emitter의 onComplete() 호출로 Observable이 정상적으로 종료될 때 호출되는 콜백입니다. 오류가 발생하거나 스트림을 폐기하는 경우에는 콜백이 호출되지 않습니다.

• 마블 다이어그램을 보면 Observable에서 정상적으로 onComplte가 호출되자 doOnComplete의 인자로 주어진 별이 방출된 것을 확인할 수 있습니다.

val observable = Observable.create<Int> { emitter ->
    emitter.onNext(1)
    Thread.sleep(1000)
    emitter.onNext(2)
    Thread.sleep(1000)
    emitter.onComplete()
}.doOnComplete {
    println("Complete!!")
}

val disposable1 = observable.subscribe {
    println("First : $it")
}

val disposable2 = observable.doOnSubscribe {
    it.dispose()
}.subscribe {
    println(it)
}

/*
결과
First : 1
First : 2
Complete!!
 */

• 예제 코드를 보면 첫 번째 Observer는 구독을 시작하고 모든 데이터를 정상적으로 받고 onComplete 신호까지 받았기에 Complete 메시지가 출력되었습니다. 그러나 두 번째 Observer의 경우 Observable에서 구독했을 때, dispose를 통해 스트림을 폐기하도록 설정하였기에 Complete 메시지가 출력되지 않는 것을 볼 수 있습니다.

doOnError 연산자

• doOnError 연산자는 Observable 내부에서 onError() 호출로 Observable이 정상적으로 종료되지 않을 때 호출되는 콜백입니다. 콜백 매개 변수로 Throwable이 들어옵니다.

• 마블 다이어그램을 보면 Observable에서 에러가 발생했을 때, doOnError 연산자의 인자로 주어진 별이 발행되는 것을 확인할 수 있습니다.

Observable.just(2, 1, 0)
    .map { num ->
        10 / num
    }
    .doOnError { throwable ->
        println("Error!!")
    } // Throwable이 매개변수로 들어옴
    .subscribeBy(
        onNext = { println(it) },
        onError = { throwable -> throwable.printStackTrace() }
    )

/*
결과
5
10
Error!!
ArithmeticException: / by zero
 */

• 예제 코드를 보면 0으로 정수를 나누려 하기에 오류가 발생합니다. 따라서 doOnError가 호출되면서 인자로 넘긴 println("Error") 메서드가 호출되는 것입니다.

doOnTerminate 연산자

• doOnTerminate 연산자는 onComplete 연산자와 유사하게 Observable이 종료될 때 호출되는 콜백입니다. onComplete 연산자와의 차이점은 오류가 발생했을 때도 콜백이 호출된다는 점입니다.

• 마블 다이어그램을 보면 Observable이 onComplete를 호출했을 때도 doOnTerminate의 인자로 주어진 초록색 별이 방출되었고, onError를 호출했을 때도 초록색 별이 방출되는 것을 볼 수 있습니다.

Observable.just(2, 1, 0)
    .map { value -> 10 / value }
    .doOnComplete { println("doOnComplete") }
    .doOnTerminate { println("doOnTerminate") }
    .subscribeBy(
        onNext = { println(it) },
        onError = { println("Error!!") },
    )

println()

Observable.just(2, 1)
    .map { value -> 10 / value }
    .doOnComplete { println("doOnComplete") }
    .doOnTerminate { println("doOnTerminate") }
    .subscribeBy(
        onNext = { println(it) },
        onError = { println("Error!!") },
    )

/*
결과
5
10
doOnTerminate
Error!!

5
10
doOnComplete
doOnTerminate
 */

• 예제 코드를 보면 첫 번째 Observable은 오류가 발생했기에 doOnTerminate만 호출되었습니다. 두 번째 Observable의 경우는 정상적으로 onComplete를 호출하였기에 doOnTerminate, doOnComple 모두 호출되었습니다.

doOnDispose 연산자

• doOnDispose는 구독 중인 스트림이 dispose() 메서드 호출로 인해 폐기되는 경우에 콜백이 호출됩니다.

• 마블 다이어그램을 보면 unsubscribe가 이루어지자 doOnDispose 함수의 인자로 넘긴 별이 방출되는 것을 볼 수 있습니다.

val observable = Observable.interval(500, TimeUnit.MILLISECONDS)
    .doOnDispose {
        println("doOnDispose")
    }

val disposable = observable.subscribe {
    println(it)
}

Thread.sleep(1100)
disposable.dispose()

/*
결과
0
1
doOnDispose
 */

doFinally 연산자

• Observable이 onError(), onComplete() 또는 스트림이 폐기될 때 doFinally 콜백이 호출됩니다. Observable을 구독한 뒤 종료되는 어떠한 상황에서도 후속 조치를 해야 하는 상황에서 이 연산자를 이용할 수 있습니다.

• 자바의 try ~ catch ~ finally의 finally라고 생각하면 될 것 같습니다.

• 마블 다이어그램을 보면 onComplete()가 호출되던, doOnError()가 호출되던, dispose를 통해 스트림이 폐기되던 doFinally 메서드의 인자로 넘긴 초록별이 방출되는 것을 볼 수 있습니다.

val observable = Observable.interval(500, TimeUnit.MILLISECONDS)
    .doOnComplete { println("doOnComplete") }
    .doOnTerminate { println("doOnTerminate") }
    .doFinally { println("doFinally") }

val disposable = observable.subscribe {
    println(it)
}

Thread.sleep(1100)
disposable.dispose()

/*
결과
0
1
doFinally
 */

참조 및 참고
틀린 부분은 댓글로 남겨주시면 바로 수정하겠습니다..!!
2022-09-23에 작성되었습니다.

아키텍처를 알아야 앱 개발이 보인다.
RxJava Docs