Zimmer Flowable gibt keine Daten beim Einfügen

Question

Ich habe Schwierigkeiten zu verstehen, wie Flowable im Raum funktioniert. Ich habe Dao Methoden wie diese

@Insert(onConflict = OnConflictStrategy.REPLACE) 
void upsert(List<Site> sites); 

@Query("SELECT * FROM site ORDER BY distance ASC") 
Flowable<List<Site>> getSites(); 

erwarte ich, dass, wenn ich rufe upsert Teilnehmer Flowable Objekt zurückgegeben durch getSites() wird immer aufgerufen werden. Ist meine Annahme wahr?

Hier ist, wie ich susbcribing bin auf diese fließfähige

private final Flowable<List<Site>> siteFlowable; 
ApiService apiService; 
FuelDatabase database; 

@Override 
public void getSites(boolean showOnlyKeySites) { 
    // add sites from cache first, then fetch network -> update cache -> update ui 
    disposable = siteFlowable.flatMap(Flowable::fromIterable) 
     .filter(site -> site.isValid()) 
     .buffer(100, TimeUnit.MILLISECONDS, 20) 
     .takeUntil(sites -> sites.size() == 0) 
     .observeOn(AndroidSchedulers.mainThread()) 
     .doOnNext(mapView::addPins) 
     .subscribe(sites -> { 
      Timber.d("Flowable emitted %d items", sites.size()); 
     }, Timber::e); 

    apiService.getSites() 
     .map(SiteListResponse::getData) 
     .flatMap(Observable::fromIterable) 
     .filter(Site::isValidSite) 
     .toList().toObservable() 
     .subscribe(sites -> { 
      Timber.i("Success Fetching %d sites", sites.size()); 
      database.siteDao().clear(); 
      database.siteDao().upsert(sites); 
     }, throwable -> Timber.e(throwable, "Error fetching sites")); 
} 

Dieses fließfähige nicht genannt bekommt nach upsert() genannt wird. Die API gibt gültige Daten zurück, und die Daten werden in die Datenbank eingegeben.

Answer 1

Try this:

apiService.getSites() 
    .map(SiteListResponse::getData) 
    .flatMap(result -> 
     Observable.fromIterable(result) 
     .filter(Site::isValidSite) 
     .toList() 
     .toFlowable() 
    ) 
    .subscribe(sites -> { 
     Timber.i("Success Fetching %d sites", sites.size()); 
     database.siteDao().clear(); 
     database.siteDao().upsert(sites); 
    }, throwable -> Timber.e(throwable, "Error fetching sites"));