Multiple retrofit2 requests using Flowable in Kotlin

Question

In order to improve my skills in kotlin, Rx, Retrofit2 I've decided to do a demo project. The demo project consist to display posts in a recycler view then display details of the post in a detail activity.
I've encountered difficulties displaying data coming from different api call: the user name, the title, the body of the post and the number of comments of the post.

My problem is that I would like to do multiple request and then have all the data needed in order to display them in the detail activity. Which mean doing a call that give me the user name and then a call that give me the number of comments for the post. The title and the body of the post are coming from a request done in the main activity I just transmit it with the bundle to the detail activity.

Api calls:
// return the comments for the post 1
http://jsonplaceholder.typicode.com/comments?postId=1

// return the information of the user 2
http://jsonplaceholder.typicode.com/users/2

// call used to display posts in the main activity
http:/jsonplaceholder.typicode.com/posts

I'm still new on Rx, I was thinking to use a flatMap but I don't know how to use it with Flowable in kotlin..

var post = viewModel.getPost()
var userStream: Flowable<User> = postService.getUser(post.userId)
var commentsByPostIdCall: Flowable<List<Comment>> = postService.getCommentsByPostId(post.id)

userStream.subscribeOn(Schedulers.io())
        .subscribe(object : Subscriber<User> {
            override fun onError(t: Throwable?) {
                Log.d(this.toString(), " Read of users failed with the following message: " + t?.message);
            }

            override fun onNext(user: User) {
                userTextView.text = user.name
                title.text = post.title
                body.text = post.body
            }

            override fun onComplete() {
            }

            override fun onSubscribe(s: Subscription?) {
                if (s != null) {
                    s.request(1)
                }
            }
        })

I have put the second call in a method getNumberComments:

    private fun getNumberComments(commentsByPostIdCall: Flowable<List<Comment>>): Int {
    var listComments = listOf<Comment>()
    var listCommentSize = 0

     commentsByPostIdCall
             .subscribeOn(Schedulers.io())
             .subscribe(object : Subscriber<List<Comment>> {
                override fun onError(t: Throwable?) {
                    Log.d(this.toString(), " Read of comments failed with the following message: " + t?.message);
                }

                override fun onNext(comment: List<Comment>) {
                    listComments = comment
                }

                override fun onComplete() {
                    print("onComplete!")
                    listCommentSize = listComments.size
                }

                override fun onSubscribe(s: Subscription?) {
                    if (s != null) {
                        s.request(1)
                    }
                }
            })
    return listCommentSize

}

Other think that I've noticed is that sometimes the stream didn't go to onComplete, sometimes it remains blocked on onNext. Don't understand why?

Any help will be much appreciate! Thanks a lot :)

Possible duplicate of https://stackoverflow.com/questions/41786439/retrofit-and-rxjava-how-to-combine-two-requests-and-get-access-to-both-results Also, make sure you are using RxJava's flatmap operator, not Kotlins! — Eric Bachhuber, Aug 01 '17 at 15:32
Thanks for your answer Eric :) I've already seen this question & the answer. The answer was around Observable not Flowable and the language used was java. I've tried to be inspired by the answer to do the same in Kotlin but I didn't success..arf — Hawa Zen, Aug 01 '17 at 16:20
For all intents and purposes, Flowable is the same as Observable except it [supports backpressure](https://github.com/ReactiveX/RxJava/wiki/Backpressure). So you can still use the same operators as you would on an Observable, i.e. .flatMap(). RxKotlin is just syntax extensions on top of RxJava so you are still using RxJava in Kotlin. If you are hung up on syntax, you can use the auto-conversion tool on https://try.kotlinlang.org (since it does get kinda messy when you throw Functions into the mix). — Eric Bachhuber, Aug 01 '17 at 16:25
Why don't you use Kotlin's async features? See https://stackoverflow.com/a/43151714/882912 — KTCO, Aug 02 '17 at 04:57

score 4 · Accepted Answer · answered Aug 04 '17 at 08:41

this is how i would solve it:

Flowable.zip<User, Comments, Pair<User, Comments>>(
      postService.getUser(postId),
      postService.getCommentsByPostId(postId),
      BiFunction { user, comments -> Pair(user, comments) })
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())
      .bindToLifecycle(this)
      .map { (first, second) -> Triple(first, second, ExtraDatasFromSomewhere) }
      .subscribe({
        Log.d("MainActivity", "OnNext")
      }, {
        Log.d("MainActivity", "OnError")
      }, {
        Log.d("MainActivity", "OnComplete")
      })

Use the zip or zipWith functions to achieve your goal if the retrofit2 calls dont depent on each other.
You can find out more here:
RxZip() : http://reactivex.io/documentation/operators/zip .

You can easily map the data from the server with the mainActivity data together like this:

.map { (first, second) -> Triple(first, second, ExtraDatasFromSomewhere) }

Kotlin has a very beautiful syntax for lambda functions so i would encourage you to use them with the specific subscribe function:
subscribe() : http://reactivex.io/RxJava/javadoc/io/reactivex/Flowable.html#subscribe(io.reactivex.functions.Consumer,%20io.reactivex.functions.Consumer,%20io.reactivex.functions.Action)

Also very important to note that i did not use only the raw Rxjava2 lib. i used the libs below: RxAndroid
for observeOn(AndroidSchedulers.mainThread()) to get the mainThread. This is because you manipulated the UI without specifying the thread you subscribed on. With this you can achieve that your subscription will be handled on the mainThread.
RxLifecycle
for .bindToLifecycle(this) this will make sure you don't leave memory leak if the activity is closed but your retrofit2 call did not finished

Thank you very much, I've tried your solution and it worked! — Hawa Zen, Aug 04 '17 at 15:04

score 1 · Answer 2 · answered Aug 04 '17 at 15:06

I've just adapted the solution suggested by Kioba with my needs. I post this here in case it can be useful to someone. I don't know if it's an elegant way to obtain the number of comments though. I've just used List < Comment > instead of Comment and then I do something like it.second.size.toString() for obtaining the number of comments.
Since I only need two data: user and comment I decided to use Pair instead of Triple.

Flowable.zip<User, List<Comment>, Pair<User, List<Comment>>>(
            postService.getUser(post.id),
            postService.getCommentsByPostId(post.id),
            BiFunction { user, comments -> Pair(user, comments) })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .map { (first, second) -> Pair(first, second) }
            .subscribe({
                Log.d("MainActivity", "OnNext")
                userTextView.text = it.first.name
                title.text = post.title
                body.text = post.body
                number_comments.text = it.second.size.toString()

            }, {
                Log.d("MainActivity", "OnError")
            }, {
                Log.d("MainActivity", "OnComplete")
            })

Multiple retrofit2 requests using Flowable in Kotlin

2 Answers2

Linked