Perfecting Logs with Rx Kotlin

This article’s goal is simple : Introducing generic logs on every Rx result’s events.

We all faced that moment when your stream does not emit the wanted result after writing a new reactive feature. I share in this article a nice way to debug your reactive code.

A brutal way to perform data logging would be to add logs everywhere we call Rx subscribeBy{}, but of course we will try to do something more sexy. First, let’s begin with Completable object. Thanks to Kotlin, I rewrite the Rx subscribeBy method as custom extension function.

import io.reactivex.disposables.Disposable
import io.reactivex.Completable
import io.reactivex.rxkotlin.subscribeBy
fun Completable.subscribeBy(
        onError: (Throwable) -> Unit,
        onComplete: () -> Unit
): Disposable = subscribeBy(onError, onComplete)

Just need to replace the io.reactivex.rxkotlin.subscribeBy imports by yourpackage.yourfile.subscribeBy in all your project’s files to call your extension function. And we’re almost done, it is finally with doOnError{} that we log the error. It “Returns a Completable which calls the given onError callback if this Completable emits an error.” Moreover it takes a Consumer<? super Throwable> as parameter, therefore we operate on a Throwable (“it” in the code).

fun Completable.subscribeBy(
        onError: (Throwable) -> Unit,
        onComplete: () -> Unit
): Disposable = this
        .doOnError { Log.d(TAG, it.message, it.cause) }
        .subscribeBy(onError, onComplete)

Every time the subscription to a Completable emits an error, doOnError will be called and the error will be logged.

Now that we understood the trick, we are able to create extension functions of Single, Observable, Maybe, and all other Rx objects to log their error emission. But why would we be satisfied with so little if we could do the same for successful events ?
Indeed, let’s log the emission of a Single success event. This time, it’s doOnSuccess{} that we will use. It “calls the shared consumer with the success value sent via onSuccess”, and takes a Consumer<? super T> as parameter.

fun <T : Any> Single<T>.subscribeBy(
        onError: (Throwable) -> Unit,
        onSuccess: (T) -> Unit
): Disposable = this
        .doOnSuccess { Log.d(TAG, it.javaClass.simpleName) }
        .doOnError { Log.d(TAG, it.message, it.cause) }
        .subscribeBy(onError, onSuccess)

Finally, those extensions methods log every reactive events in my app without adding disturbing code, and it really helped my PO during debugging.

This solution could be reused for many different use cases. Thanks Kotlin !

And to go further see how I exclude Logs on production.