Finally we are here to talk about Kotlin, RxJava & RxAndroid. This is an incredible combo and we are going to take advantages of both by providing a new layer to our App and decoupling our Main UI thread with our background tasks, in this case to request Reddit news from the server. We are going to leave everything prepare to connect with a real server (next Part) but for now the mock data will be OK to emulate a server request.

If this is the first time that you hear about RxJava, I encourage you to read about this incredible peace of software that will blow your mind.

RxJava is the implementation of Reactive Extensions, a library for composing asynchronous and event-based programs by using observable sequences.

And we are going to use the Observable class to avoid making server requests (or any long process execution) from our famous Main UI Thread.

News Manager

The NewsManager class will be in charge of providing news from Reddit API. It will be responsible for performing the server request and give us a list of news with our already created News UI Model.

The main idea behind of this is to make our API call to be executed outside of our Main UI Thread, in another thread. A big picture would be something like this:

Observable

The NewsManager class will provide a method that will return an Observable object which will allows you to run a piece of code (like an API call) in another context (in this case in a new thread).

fun getNews(): Observable<List<RedditNewsItem>> {
    ...
}

This Observable object will be created by us from the NewsManager. We will provide the implementation of this object:

fun getNews(): Observable<List<RedditNewsItem>> {
    return Observable.create {
        subscriber ->

        val news = mutableListOf<RedditNewsItem>()
        // api call...
        subscriber.onNext(news)
        subscriber.onCompleted()
    }
}

Thanks to Kotlin, we can use lambda expressions to provide the function required by the Observable.create(…) method, in this case we are providing the implementation of the code that will be executed in the context that we provide to the Observable (we will see more about this context later in this post but in short it means in which thread to be executed).

The create() method requires you to provide a function that receives a subscriber object, this subscriber object allows you to send events to the subscribers by executing 3 possible methods:

• onNext(item: T): We are going to call this method to provide to the subscribers with the news received from the API call. In this case, T will be our “List<RedditNewsItem>” that it’s already infer from the return type in the getNews() signature.
• onCompleted(): This method is invoked when we finished to provide all the data from this Observable, in this case, we will call it after we send the news from onNext() method and we are ready to finish this process.
• onError(e: Throwable): If we face any issue from the API call, we will be calling this method to tell the subscribers that there was an error. We are going to use it later to inform the user that there was a problem with the request by showing a SnackBar message.

Lazy NewsManager

Also we added our NewsManager with our lazy property that we reviewed in the 3rd Part of this tutorial:

private val newsManager by lazy { NewsManager() }

This newsManager will be initialized with the NewsManager() only the first time that we use newsManagers field.

Request News from NewsFragment

Now it’s time to use our NewsManager and request some news to show in our RecyclerView.

Subscribe > Subscription > Observer

In order to request the mocked Reddit news from the NewsManager, we will turn our NewsFragment into an Observer of the Observable that we receive from the NewsManager.getNews(). We are going to do this by calling the getNews() method and invoke the method “subscribe(…)” from the received Observable:

val subscription = newsManager.getNews().subscribe (
        { retrievedNews ->
            ...
        },
        { e ->
            ...
        }

The method subscribe has several overloads and we are going to use this one:

public final Subscription subscribe(
       final Action1<? super T> onNext, 
       final Action1<Throwable> onError) {
       ...
}

It receives two functions:

• onNext: A function to be invoked when the Observable calls the onNext() method that we saw previously. We will use it to set the NewsAdapter with the received news.
• onError: A function to be called when the onError() method from the Observable is invoked. We’ll use it to show a SnackBar with an error message.

And returns a Subscription object. This object will allow us to manage the subscription like checking if it still subscribed or to cancel the subscription (we are going to see more about this at the end of this tutorial).

Thanks to Kotlin again, we can provide these two functions using lambda expressions. For example in the first function, the retrievedNews variable is the name that I gave to the news received from the onNext() method:

{ 
    retrievedNews ->
     (news_list.adapter as NewsAdapter).addNews(retrievedNews)
}

Remember that the onNext returning type is a “List<RedditNewsItem>”, so what I did is to pass these news directly to our NewsAdapter.

And for the onError function, I’m just telling the user that there was an error with a SnackBar. I can use “e”, which is a Throwable type, to get more details about the received exception:

{ e ->
    Snackbar.make(...).show()
}

We are still in the Main Thread

If we run the App it will just work because we are mocking the data provided in this Observable. But if you really do a long running operation instead of the mocked data that we generate, the App will stop working as we are still in the Main UI Thread.

As we didn’t provide any specific details to our Observable, it will run with the default configuration which is to be executed in the same thread that it was invoked. So let’s configure our Observable to be executed in another thread but still notify any event in our current Main UI Thread.

SubscribeOn

val subscription = newsManager.getNews()
        .subscribeOn(Schedulers.io())
        .subscribe (...)

The method subscribeOn(…) allows you to move the execution of the Observable code into another thread and with an specific behaviour. To do this, RxJava use Schedulers which provides the behaviour that you need for your use case. RxJava also provides a list of Schedulers for common scenarios:

• io: intended for IO-bound work.
• computation: intended for computational work.
• newThread: creates a new Thread for each unit of work.
• test: useful for debugging.

In our case, we are going to use Schedulers.io() as we are going to be executing an API request.

ObserveOn

ObserveOn is another method from the Observable class and it allows you to define where do you want to execute the functions that you provided in the “subscribe(…)” method. In other words, in which thread is going to be executed the functions that we provided in the subscribe method with the onNext and onError code. As we need to execute this code in the Main UI Thread it could be a good idea to make our Observable to observeOn the Main UI Thread.

Here is where we need to add RxAndroid dependency to have a new Scheduler called:

AndroidSchedulers.mainThread()

And we need to update our code to set the observeOn Scheduler:

val subscription = newsManager.getNews()
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe ( ...
)

But we have to realize that our app can be closed at any time or perform an orientation change operation in the middle of a subscription. This could lead us to a problem if we don’t manage our subscriptions properly, so let’s do something with it.

Subscriptions

In order to avoid leaving our subscriptions in the middle of a process while we are leaving from a fragment or closing our App, we must manage it carefully and a good practice is to unsubscribe all the subscriptions that we create.

A good technique to do so is to add all the subscription that you create in a CompositeSubscription object, this class is provided by RxJava and allows you to unsubscribe all the subscription that it has with just one method call. We are going to initialize a CompositeSubscription object inside the onResume method and unsubscribe when the onPause method is invoked. The code will be something like this:

var subscriptions = CompositeSubscription()

override fun onResume() {
    super.onResume()
    subscriptions = CompositeSubscription()
}

override fun onPause() {
    super.onPause()
    subscriptions.clear()
}

The only thing that we have to do now is to add our subscription to this new CompositeSubscription object and we are safe to leave our App (or fragment in this case) at any moment:

private fun requestNews() {
    val subscription = newsManager.getNews()
            .subscribeOn(Schedulers.io())
            .subscribe (...)
    subscriptions.add(subscription) // add the subscription
}

And just to make it more readable I move all this logic to a new base fragment called “RxBaseFragment” which gives me all this logic by default. Checkout the code to see with more details.

Conclusion

As you can see Kotlin & RxJava works perfect together and in this tutorial we are just seeing a small part of all the things that you can do with RxJava and thanks to Kotlin the code is totally readable and easy to understand.

See you soon in the next article!