RxAndroid，如何檢測是否可觀察已完成發射

我在寫下面的代碼片段，從firebase數據庫中獲取保存的食物列表，然後使用該列表，我再次從firebase數據庫獲取單個食物詳細信息。RxAndroid，如何檢測是否可觀察已完成發射

以下代碼正常工作，除非我無法弄清楚如何讓第二個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([email protected], Food(food.foodId, food.foodName, food.foodDesc)) 
           repo.insertServingDetails([email protected], food.servingList.map { it.component2() }) 
           repo.saveFood([email protected], personalizedFood) 
           it.onNext(true) 
          } 

         }else { 
          it.onNext(false) 
         } 
        } 

       }) 
      } 
     }.observeOn(Schedulers.io()).doOnCompleted{ 
      dismissProgressDialog() 
      finish() 
     }.doOnError{ 
      it.printStackTrace() 
      dismissProgressDialog() 
      finish() 
     }.subscribe()

感謝。

來源

2017-07-19 chandil03

是否使用fireabse：由於每個在flatMap創建僅發射一個項目的觀測，就可以在onNext()方法後直接叫什麼名字？有一些第三方RxFirebase可以爲您提供Firebase數據庫的很好包裝。 –

@Phoenix王這是另一回事。其實我正在學習如此想知道如何實現，而不使用任何第三方包裝。我不喜歡使用第三方lib，如果這很容易做到這一點.. – chandil03

嗯，其實你不需要。在第一個可觀察到的情況下，您獲取所有PersonalizedFood項目並將它們作爲列表發出，然後完成流。然後將其轉換爲一組項目，每個項目都在第二個flatMap中處理。關鍵在於每個onCompleted都傳遞給流，所以第二個flatMap「知道」不會有更多的元素並完成自己。我建議複習代碼安全性和一致性，因爲有些觀點對我來說看起來不對。 – MightySeal

flatMap的Observable知道「什麼時候所有物品已經完成」，當它發出的所有可觀察物都稱爲onCompleted()。代碼中的第二個flatMap從未調用onCompleted()，因爲它創建的所有可觀察對象都不會調用onCompleted()。

您應該在onDataChange()方法中調用onCompleted()。

override fun onDataChange(p0: DataSnapshot?) { 
    if(p0 != null) { 
     val food = p0.getValue(FBFood::class.java)!! 
     val repo = LocalFoodRepository() 
     doAsync { 
      repo.insertFood([email protected], Food(food.foodId, food.foodName, food.foodDesc)) 
      repo.insertServingDetails([email protected], food.servingList.map { it.component2() }) 
      repo.saveFood([email protected], personalizedFood) 
      it.onNext(true) 
      it.onCompleted() 
     } 
    } else { 
     it.onNext(false) 
     it.onCompleted() 
    } 
}

來源

2017-07-19 20:08:50 Bryan

這不符合我的要求。當所有數據都被提取時，我想完成活動。這就是爲什麼我不能在'onDataChange（）'上調用'onComplete'。 – chandil03

@ chandil03這將是'doOnCompleted（）'方法，不是嗎？你可以在該方法中調用'finish（）'，因此添加'onComplete（）'應該觸發完成該活動。 – Bryan

我知道，如果你仔細檢查我的代碼，我已經在'doOnCompleted（）'中添加了'finish（）'方法調用。如果我在你提示我的地方調用'onComplete（）'，'doOnComplete方法將被調用的次數儘可能多地被observable'發出。因此'finish（）'將被多次調用，因此會導致應用程序崩潰，因爲活動將爲空。 – chandil03

RxAndroid，如何檢測是否可觀察已完成發射

回答

相關問題