Я пишу следующий фрагмент кода, чтобы получить список сохраненных продуктов из базы данных 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()
Спасибо.
