RxJava Datenfluss mit SqlBrite und Retrofit

Question

In meiner Android-App verwende ich Domain-Repository-Schnittstelle, die mit lokalen DB unterstützt mit SqlBrite und Netzwerk-API mit Retrofit Observables unterstützt wird. Also habe ich Methode getDomains(): Observable<List<Domain>> im Repository und zwei entsprechende Methoden in meinem Retrofit und SqlBrite. Ich möchte diese beiden Observablen nicht verketten oder zusammenführen. Ich möchte, dass mein Repository Daten nur von SqlBrite nimmt und da SqlBrite QueryObservable zurückgibt, was onNext() auslöst, wenn die zugrunde liegenden Daten geändert werden, kann ich meine Netzwerkanforderung unabhängig ausführen und Ergebnisse in SqlBrite speichern und meine Observable mit aus dem Netzwerk abgerufen und in DB gespeichert aktualisieren Daten. Also versuchte ich mein Repository des getDomains() Methode wie folgt umzusetzen:

fun getDomains(): Observable<List<Domain>> { 
    return db.getDomains() 
        .doOnSubscribe { 
         networkClient.getDomains() 
            .doOnNext { db.putDomains(it) } 
            .onErrorReturn{ emptyList() } 
            .subscribe() 
        } 
} 

aber in diesem Fall jedes Mal, sollte der Kunde abonniert, würde jedes Mal, wenn Netzwerkanforderungen machen, dass nicht so gut ist. Ich dachte über andere do... Betreiber, um Anfragen dorthin zu verschieben, aber doOnCompleted() im Falle von QueryObservable würde nie aufgerufen werden, bis ich toBlocking() irgendwo anrufen, was ich nicht, doOnEach() auch nicht gut, da es Anfragen jedes Mal Element aus Db extrahiert. Ich habe auch versucht, replay() Operator zu verwenden, aber obwohl das Observable in diesem Fall zwischengespeichert wird, geschieht das Abonnement und führt zu Netzwerkanforderungen. Also, wie kann diese beiden Observables in der gewünschten Weise kombinieren?

Answer 1

Ok, es hängt vom konkreten Anwendungsfall ab, den Sie haben: Wenn Sie beispielsweise die neuesten Daten aus Ihrer lokalen Datenbank anzeigen und die Datenbank von Zeit zu Zeit aktualisieren möchten, indem Sie eine Netzwerkanforderung im Hintergrund ausführen.

Vielleicht gibt es einen besseren Weg, aber vielleicht könnten Sie so etwas wie diese

fun <T> createDataAwareObservable(databaseQuery: Observable<T>): Observable<T> = 
     stateDeterminer.getState().flatMap { 
     when (it) { 
      State.UP_TO_DATE -> databaseQuery // Nothing to do, data is up to date so observable can be returned directly 

      State.NO_DATA -> 
      networkClient.getDomains() // no data so first do the network call 
       .flatMap { db.save(it) } // save network call result in database 
       .flatMap { databaseQuery } // continue with original observable 

      State.SYNC_IN_BACKGROUND -> { 
      // Execute sync in background 
      networkClient.getDomains() 
       .flatMap { db.save(it) } 
       .observeOn(backgroundSyncScheduler) 
       .subscribeOn(backgroundSyncScheduler) 
       .subscribe({}, { Timber.e(it, "Error when starting background sync") }, {}) 

      // Continue with original observable in parallel, network call will then update database and thanks to sqlbrite databaseQuery will be update automatically 
      databaseQuery 
      } 
     } 
     } 

So am Ende erstellen Sie Ihre SQLBrite beobachtbare (QueryObservable) und in die createDataAwareObservable() Funktion übergeben. Dann wird sichergestellt, dass die Daten aus dem Netzwerk geladen werden, wenn keine Daten vorhanden sind. Andernfalls wird überprüft, ob die Daten im Hintergrund aktualisiert werden sollen (wird automatisch in der Datenbank gespeichert, wodurch SQLBrite QueryObservable automatisch aktualisiert wird) oder wenn Daten vorhanden sind auf dem Laufenden.

Grundsätzlich können Sie es wie folgt verwendet werden:

createDataAwareObservable(db.getAllDomains()).subscribe(...) 

für Sie also als Benutzer dieses createDataAwareObservable() Sie immer die gleiche Art erhalten Observable<T> zurück, wie Sie als Parameter übergeben. So im Wesentlichen scheint es, dass Sie immer db.getAllDomains() ...

Answer 2

, wenn Ihr Problem ist, dass Sie Ihren Betrachter jedes Mal abonnieren, die Sie möchten Daten erhalten Sie Relais verwenden können, die nie die Beobachter austragen, weil nicht onComplete implementieren

/** 
* Relay is just an observable which subscribe an observer, but it wont unsubscribe once emit the items. So the pipeline keep open 
* It should return 1,2,3,4,5 for first observer and just 3, 4, 5 fot the second observer since default relay emit last emitted item, 
* and all the next items passed to the pipeline. 
*/ 
@Test 
public void testRelay() throws InterruptedException { 
    BehaviorRelay<String> relay = BehaviorRelay.create("default"); 
    relay.subscribe(result -> System.out.println("Observer1:" + result)); 
    relay.call("1"); 
    relay.call("2"); 
    relay.call("3"); 
    relay.subscribe(result -> System.out.println("Observer2:" + result)); 
    relay.call("4"); 
    relay.call("5"); 
} 

Weitere Beispiele hier