RxAndroid, Comment détecter si observable a fini l'émission

Question

J'écris l'extrait de code suivant pour aller chercher la liste des aliments sauvegardés à partir de la base de données Firebase, puis en utilisant cette liste, je récupère les détails de la base de données Firebase.

Le code suivant fonctionne bien, sauf que je suis incapable de comprendre comment laisser second flatMap savoir que l'émission du premier flatMap est terminée (toute la liste des aliments a été traitée). Donc, je suis incapable d'appeler la méthode onCompleted() donc incapable de détecter quand tout le processus se termine.

Jetez un oeil à des commentaires dans l'extrait suivant:

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

Merci.

Answer 1

Le Observable du flatMap sait « quand tous les éléments ont été fini » lorsque toutes les émis par elle observables ont appelé onCompleted(). Le deuxième flatMap dans votre code n'appelle jamais onCompleted() car aucun des observables qu'il crée n'appelle onCompleted(). Vous devez appeler onCompleted() dans votre méthode onDataChange(). Comme chacun des observables créés dans le flatMap n'émettent un élément, il peut être appelé directement après la méthode 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() 
    } 
}