RxAndroid, cómo detectar si observable ha finalizado la emisión
Estoy escribiendo el siguiente fragment de código para recuperar la list de alimentos guardados de la database de firebase y luego, al usar esa list, vuelvo a search detalles de alimentos individuales de la database de firebase.
El siguiente código funciona bien, excepto que no puedo averiguar cómo hacer que el segundo flatMap sepa que la emisión del primer flatMap ha finalizado (toda la list de alimentos ha sido procesada). Por lo tanto, no puedo llamar al onCompleted() , por lo tanto, no onCompleted() detectar cuándo finaliza todo el process.
- ¿Cómo especificar la versión de RxJava al usar RxKotlin?
- ¿Cómo puedo agregar de manera condicional una operación asincrónica en medio de una transmisión de RxJava?
- Excepción causada por: java.lang.ClassNotFoundException: org.reactivestreams.Publisher
- ¿Cómo corotines de Kotlin son mejores que RxKotlin?
- Confusión de la syntax de Kotlin lambda
Echa un vistazo a los comentarios en el siguiente fragment:
Observable.create<List<PersonalizedFood>> { FirebaseDTDatabase.getSavedDietFoodQuery(user.uid).addListenerForSingleValueEvent(object : ValueEventListener { override fun onCancelled(p0: DatabaseError?) { } override fun onDataChange(p0: DataSnapshot?) { val list = ArrayList<PersonalizedFood>() p0?.let { for (dateObject in p0.children) { for (foodItem in dateObject.children) { val food = foodItem.getValue(FBPersonalizedFood::class.java) as FBPersonalizedFood list.add(PersonalizedFood(food)) } } } it.onNext(list) it.onCompleted() } }) }.subscribeOn(Schedulers.io()).flatMap { Observable.from(it) // returning a Observable that emits items of list ("it" is the list here) }.observeOn(Schedulers.io()).flatMap { // How does this flatMap know that emission of all item has been finished so that onCompleted() method could be called. personalizedFood -> Observable.create<Boolean>{ FirebaseDTDatabase.getFoodListReference(personalizedFood.foodId).addListenerForSingleValueEvent(object :ValueEventListener{ override fun onCancelled(p0: DatabaseError?) { it.onError(p0?.toException()) } override fun onDataChange(p0: DataSnapshot?) { if(p0 != null) { val food = p0.getValue(FBFood::class.java)!! val repo = LocalFoodRepository() doAsync { repo.insertFood(this@LoginActivity, Food(food.foodId, food.foodName, food.foodDesc)) repo.insertServingDetails(this@LoginActivity, food.servingList.map { it.component2() }) repo.saveFood(this@LoginActivity, personalizedFood) it.onNext(true) } }else { it.onNext(false) } } }) } }.observeOn(Schedulers.io()).doOnCompleted{ dismissProgressDialog() finish() }.doOnError{ it.printStackTrace() dismissProgressDialog() finish() }.subscribe()
Gracias.
- Inyectar constructor y object complementario
- Cómo notificar a Observable cuando finalice CountdownTimer
- Llamada asincrónica para cada elemento dentro de una colección
- ¿Expresar "súper" generics en los types funcionales de Kotlin?
- Usando RxJava para unir datos locales con datos remotos (o en caching)
- ¿Puedo crear un método de extensión Kotlin para agregar una suscripción rxJava a una suscripción compuesta?
- Repetir acciones en estado con RxJava
- Obtiene N últimos objects emitidos por observables en RxJava2
El Observable de flatMap sabe "cuándo se han terminado todos los elementos" cuando todos los observables emitidos por él han llamado a onCompleted() . El segundo flatMap en su código nunca llama onCompleted() porque ninguno de los observables crea call onCompleted() .
Debe llamar a onCompleted() en su método onDataChange() . Como cada uno de los observables creados en flatMap solo emite un elemento, se puede onNext() directamente después del método onNext() :
override fun onDataChange(p0: DataSnapshot?) { if(p0 != null) { val food = p0.getValue(FBFood::class.java)!! val repo = LocalFoodRepository() doAsync { repo.insertFood(this@LoginActivity, Food(food.foodId, food.foodName, food.foodDesc)) repo.insertServingDetails(this@LoginActivity, food.servingList.map { it.component2() }) repo.saveFood(this@LoginActivity, personalizedFood) it.onNext(true) it.onCompleted() } } else { it.onNext(false) it.onCompleted() } }