在 Kotlin 中使用 Flowable 的多个改造 2 请求

Question

为了提高我在 kotlin、Rx、Retrofit2 方面的技能，我决定做一个演示项目。 演示项目包括在回收站视图中显示帖子，然后在详细活动中显示帖子的详细信息。
我在显示来自不同 api 调用的数据时遇到了困难：用户名、标题、帖子正文和帖子的 cmets 数。

我的问题是我想做多个请求，然后拥有所有需要的数据，以便在详细活动中显示它们。这意味着打电话给我用户名，然后打电话给我帖子的 cmets 数量。帖子的标题和正文来自主要活动中完成的请求，我只是将它与捆绑包一起传输到详细活动。

API 调用：
// 返回帖子 1 的 cmets
http://jsonplaceholder.typicode.com/comments?postId=1

//返回用户2的信息
http://jsonplaceholder.typicode.com/users/2

// 用于在主活动中显示帖子的调用
http://jsonplaceholder.typicode.com/posts

我还是 Rx 的新手，我正在考虑使用 flatMap，但我不知道如何在 kotlin 中将它与 Flowable 一起使用..

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)
                }
            }
        })

我已将第二个调用放入方法 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

}

我注意到的其他想法是，有时流没有进入 onComplete，有时它在 onNext 上仍然被阻止。不明白为什么？

任何帮助将不胜感激！非常感谢:)

Answer 1

我刚刚根据我的需要调整了 Kioba 建议的解决方案。我在这里发布它以防它对某人有用。 我不知道这是否是获取 cmets 数量的一种优雅方式。我刚刚使用了 List 而不是 Comment 然后我做了类似 it.second.size.toString() 来获取cmets 的数量。
因为我只需要两个数据：用户和评论，所以我决定使用 Pair 而不是 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")
            }) 

Answer 2

这就是我将如何解决它：

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")
      })

如果retrofit2 调用不相互依赖，请使用zip 或zipWith 函数来实现您的目标。
您可以在这里找到更多信息：
RxZip()：http://reactivex.io/documentation/operators/zip。

您可以像这样轻松地将来自服务器的数据与 mainActivity 数据映射在一起：

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

Kotlin 为 lambda 函数提供了非常优美的语法，因此我鼓励您将它们与特定的 subscribe 函数一起使用：
订阅（）：http://reactivex.io/RxJava/javadoc/io/reactivex/Flowable.html#subscribe(io.reactivex.functions.Consumer,%20io.reactivex.functions.Consumer,%20io.reactivex.functions.Action)

同样需要注意，我并没有只使用原始的 Rxjava2 库。我使用了以下库： RxAndroid
为observeOn(AndroidSchedulers.mainThread()) 获取主线程。这是因为您在未指定订阅线程的情况下操作了 UI。有了这个，您可以实现您的订阅将在主线程上处理。
RxLifecycle
对于.bindToLifecycle(this)，这将确保在活动已关闭但您的改造2 调用未完成时不会留下内存泄漏