RxAndroid，如何检测可观测是否已经完成发射

Question

我正在编写以下代码片段来从Firebase数据库中获取保存的食物列表，然后使用该列表，我再次从Firebase数据库中获取单个食物详细信息。

以下代码工作正常，除了我无法弄清楚如何让第二个flatMap知道第一个flatMap的发射已经完成(所有食物清单都已处理(。所以我无法调用onCompleted()方法，因此无法检测整个过程何时完成。

请查看以下代码片段中的评论：

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

谢谢。

Answer 1

当flatMap发出的所有可观察量都调用onCompleted()时，来自的Observable知道"何时完成所有项目"。代码中的第二个flatMap从不调用onCompleted()，因为它创建的任何可观察量都不会调用onCompleted()。

应在onDataChange()方法中调用onCompleted()。由于在flatMap中创建的每个可观察量只发出一个项目，因此可以直接在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()
}
}