RxJava Observable永不執行onCompleted

Question

我使用RxJava和Realm來查詢移動數據庫。當查詢完成時，我需要通知我的視圖來更新列表，但它看起來像flatMap和doOnNext有效，但是它永遠不會完成。我需要一個觸發器才能知道它已經結束。

Realm.getDefaultInstance().asObservable().flatMap((realm) -> 
     realm.where(FileDoc.class).isNull("parent").isNotNull("updatedDate").findAllSortedAsync("name").asObservable() 
       .flatMap((files) -> { 
        documents.get("root").clear(); 
        documents.get("root").addAll(realm.copyFromRealm(files)); 
        return realm.where(Folder.class).isNull("parent").findAllSortedAsync("name").asObservable().doOnCompleted(() -> { 
         Log.e(TAG, "Barr!"); 
        }); 
       }) 
       .doOnNext((folders) -> { 
        documents.get("root").addAll(realm.copyFromRealm(folders)); 
       })).doOnCompleted(() -> Log.e(TAG, "Fooo!")) 
     .doOnError(error -> handleErrorEvent(error)) 
     .observeOn(AndroidSchedulers.mainThread()) 
     .subscribeOn(AndroidSchedulers.from(backgroundLooper)) 
     .subscribe(); 

這裏Barr!，並且從不打印Fooo! ......任何有線索？

---

修訂

Realm.getDefaultInstance().asObservable().first().flatMap((realm) -> 
       realm.where(IncidentTemplate.class).equalTo("deleted", false).findAllAsync().asObservable() 
         .filter(results -> results.isLoaded()) 
         .first() 
         .doOnNext((files) -> { 
          if (!incidents.containsKey("INCIDENT")) { 
           incidents.put("INCIDENT", new ArrayList<>()); 
          } 

          incidents.get("INCIDENT").clear(); 
          incidents.get("INCIDENT").addAll(realm.copyFromRealm(files)); 

         }) 
         .doOnTerminate(() -> { 
          Log.e(TAG, "Closing realm"); 
          realm.close(); 
         })) 
       .doOnCompleted(() -> { 
        emitStoreChange(new CobaltStore.CobaltStoreChangeEvent()); 
        Log.e(TAG, "EMIT INCIDENT"); 
       }) 
       .doOnError(error -> handleErrorEvent(error)) 
       .observeOn(AndroidSchedulers.mainThread()) 
       .subscribeOn(AndroidSchedulers.from(backgroundLooper)) 
       .subscribe(); 

Answer 1

1）Realm.getDefaultInstance().asObservable()

您獲得本地領域實例，你將永遠無法關閉，因此，你正在泄漏內存。

2）這不是asObservable()設計的工作方式。

OnClickListener完成併發出終端事件是否有意義？

例如，您單擊按鈕一次，然後可觀察的「結束」並且永遠不會再次發出任何事件，即使用戶之後單擊該按鈕？

當然不是，所以你不應該期待它來自Realm的RealmChangeListener;這是綁定到Realm或RealmResults的內容，以便您可以聆聽將來對Realm或其特定表進行的任何寫入。

解決方案可能是隻使用類似

Observable.fromCallable(() => { 
    try(Realm realm = Realm.getDefaultInstance()) { 
     RealmResults<FileDoc> docs = realm.where(FileDoc.class) 
      .isNull("parent") 
      .isNotNull("updatedDate") 
      .findAllSorted("name"); 
     documents.get("root").clear(); // <-- shouldn't you modify a new copy and return this? 
     List<FileDoc> unmanagedDocs = realm.copyFromRealm(files); 
     documents.get("root").addAll(unmanagedDocs); 
     return unmanagedDocs; 
    } 
} 

或者正如你所指出，.filter((data) -> data.isLoaded()).first()也工作，雖然我沒有想到這一點。