Múltiples requestes de modificación2 usando Flowable en Kotlin

Para mejorar mis habilidades en kotlin, Rx, Retrofit2, he decidido hacer un proyecto de demostración. El proyecto de demostración consiste en mostrar publicaciones en una vista de reciclador y luego mostrar detalles de la publicación en una actividad detallada.
Me han surgido dificultades para mostrar datos provenientes de diferentes llamadas a la API: el nombre de usuario, el título, el cuerpo de la publicación y el número de comentarios de la publicación.

Mi problema es que me gustaría hacer varias requestes y luego tener todos los datos necesarios para mostrarlos en la actividad detallada. Lo que significa hacer una llamada que me da el nombre de usuario y luego una llamada que me da la cantidad de comentarios para la publicación. El título y el cuerpo de la publicación proceden de una request realizada en la actividad principal. Simplemente la transmito con el package a la actividad detallada.

Api llama:
// devuelve los comentarios para la publicación 1
http://jsonplaceholder.typicode.com/comments?postId=1

// devuelve la información del usuario 2
http://jsonplaceholder.typicode.com/users/2

// llamada utilizada para mostrar publicaciones en la actividad principal
http: /jsonplaceholder.typicode.com/posts

Todavía soy nuevo en Rx, estaba pensando en usar un flatMap pero no sé cómo usarlo con Flowable en 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) } } })

He puesto la segunda llamada en un método 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 }

Otro que creo que he notado es que a veces la transmisión no llega a "Completa", a veces permanece bloqueada en "Siguiente". No entiendo por qué?

¡Cualquier ayuda será muy apreciada! Muchas gracias 🙂

así es como lo resolvería:

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

Utilice el zip o zipWith funciones para lograr su objective si las llamadas de actualización no dependen el uno del otro.
Puedes encontrar mas aqui:
RxZip (): http://reactivex.io/documentation/operators/zip .

Puede mapear fácilmente los datos del server con los datos de actividad principal juntos así:

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

Kotlin tiene una syntax muy bella para las funciones lambda, por lo que te animo a utilizarlas con la function de suscripción específica:
suscribirse (): http://reactivex.io/RxJava/javadoc/io/reactivex/Flowable.html#subscribe(io.reactivex.functions.Consumer,%20io.reactivex.functions.Consumer,%20io.reactivex.functions.Action)

También es muy importante tener en count que no utilicé solo la lib de Rxjava2 sin procesar. utilicé las libs a continuación: RxAndroid
para observeOn(AndroidSchedulers.mainThread()) para get el mainThread. Esto se debe a que manipuló la interfaz de usuario sin especificar el hilo en el que se suscribió. Con esto puede lograr que su suscripción se maneje en el hilo principal.
RxLifecycle
para .bindToLifecycle(this) esto se asegurará de no dejar pérdidas de memory si la actividad está cerrada pero su llamada de actualización no finalizó

Acabo de adaptar la solución sugerida por Kioba a mis necesidades. Publico esto aquí en caso de que pueda ser útil para alguien. No obstante, no sé si es una manera elegante de get la cantidad de comentarios. Acabo de utilizar List <Comment> en lugar de Comment y luego hago algo como it.second.size.toString () para get la cantidad de comentarios.
Como solo necesito dos datos: usuario y comentario, decidí usar Pair en lugar de 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") })