RxJava Muster für die Rückkehr kalten Ergebnisse, mehr Arbeit zu tun, dann wieder heiße Ergebnisse

Question

Ich lerne RxJava also bitte sanft sein. Ich habe mir die Tutorials angeschaut, die Lektüre gemacht, SO durchsucht, aber ich habe immer noch Probleme, meine AsyncTaskLoader zu transformieren. Aus irgendeinem Grund kann ich kein Muster von Operatoren finden, um meine Aufgabe zu erfüllen (obwohl ich denke, dass es eine übliche ist). Was ich versuche zu tun, ist folgendes: Gebe ein Observable zurück, das mein Fragment abonnieren könnte. Das Observable sollte folgende Schritte beim Abonnieren ausführen:

1) Daten aus der lokalen Datenbank abrufen, indem 2 Abfragen ausgeführt werden, eine Logik ausgeführt wird und Ergebnisse zurückgegeben werden;
2) Abrufen von Daten aus API;
3) Synchronisieren der neuen API-Daten mit der Datenbank;
4) Schritt eins wiederholen und Ergebnisse zurückgeben;

Bisher habe ich meine DB-Aufrufe und meine API-Aufrufe zu Observablen zurückgegeben. Ich versuche zu verstehen, wie ich die kalten Ergebnisse ausstrahlen und mit der Kette fortfahren kann. Ich könnte wahrscheinlich die beiden Operationen getrennt halten und den gleichen Abonnenten verwenden, um beide zu abonnieren? Aber ich bin nicht sicher, wie das funktionieren würde, wenn meine neue Loader-Ersetzungsklasse eine beobachtbare zurückgibt ... Auch muss ich die Ergebnisse von der zweiten Observablen nicht wirklich verarbeiten - ich brauche nur die erste zu wiederholen, wenn die der zweite ist fertig.

Bisher habe ich folgende:

public Observable<StuffFetchResult> getColdStuff() { 
    return Observable.zip(mDataSource.listStuff(), mDataSource.listOtherStuff(), 
      (stuff, moreStuff) -> { 
       List<Stuff> mergedList = new ArrayList<>(); 
       // do some merging stuff 
       return new StuffFetchResult(mergedList); 
      }).subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()); 
} 

Angenommen, ich habe getHotStuff() auch, dass der API-Aufruf und die Synchronisation mit der Datenbank tun wird, wenn das der richtige Ansatz ist, und gibt die gleichen Observable. Allerdings stecke ich im nächsten Schritt fest - wie kann ich das erste Observable neu starten, um es erneut abzuspielen, nachdem hotStuff abgeschlossen ist, ohne einen weiteren Abonnenten hinzuzufügen?

EDIT:

Ich habe einige Fortschritte gemacht und ich denke, alles, was ich jetzt brauchen, ist alles bis zu verbinden. Ich habe meine zwei Methoden:

1) getColdStuff() ziemlich viel ist, wie oben
2) getHotStuff() beschrieben wird an die API nennt, synchronisieren mit der Datenbank und kehrt ein beobachtbares. Die Idee war, getColdStuff() erneut aufzurufen, nachdem getHotStuff() beendet wurde, um die Benutzeroberfläche zu aktualisieren, sodass das tatsächliche Ergebnis, das von getHotStuff() zurückgegeben wird, ignoriert werden kann. Alles, was es tun muss, ist getColdStuff() einmal ausgelöst zu lösen.

Ich habe auf den Vorschlag in der Antwort versucht, und erstellt die folgenden:

BehaviorRelay<Observable<StuffFetchResult>> callSequence = BehaviorRelay.create(); 
Observable<StuffFetchResult> valueSequence = Observable.switchOnNextDelayError(callSequence.toSerialized()); 
valueSequence.subscribe(new Subscriber<StuffFetchResult>() { 
    @Override 
    public void onCompleted() {} 

    @Override 
    public void onError(Throwable e) {} 

    @Override 
    public void onNext(StuffFetchResult result) { 
     // UI stuff 
    } 
}); 
callSequence.call(loader.getColdStuff()); 

ich valueSequence hier und callSequence.call(loader.getColdStuff()); verwenden abonnieren können, die die erste Methode und produzieren Ergebnisse in onNext() meines Abonnement läuft , die ich für meine Benutzeroberfläche verwenden kann. Ich bin mir jedoch nicht sicher, wie ich getHotStuff() parallel ausführen und auch eine andere Aktion auf es ausführen, wenn es zurückgibt. Auch getHotStuff() gibt eine andere Art von Observable zurück, also kann ich nicht wirklich das gleiche callSequence verwenden?

EDIT 2

zwei Teilnehmer verwenden, kann ich das erforderliche Verhalten Ich denke erreichen. Nicht wirklich sicher, ob das der richtige Weg ist.

loader.getHotStuff() 
    .subscribeOn(Schedulers.io()) 
    .subscribe(new Subscriber<Object>() { 
     @Override 
     public void onCompleted() {} 

     @Override 
     public void onError(Throwable e) {} 

     @Override 
     public void onNext(Object stuffWeDontCareAbout) { 
      callSequence.call(loader.getColdStuff()); 
     } 
    }); 

Answer 1

, wenn ich Ihr Szenario richtig verstanden hat, kann man so etwas will -

BehaviorSubject<Observable<T> callSequence = BehaviorSubject.create(); 
Observable<T> valueSequence = Observable.swithOnNextDelayError(callSequence.toSerialized()); 

Ihre Teilnehmer werden den valueSequence zuhören, und wenn Sie auf „Neustart“ benötigen, werden Sie nennen dies -

callSequence.onNext(call.cache()); // *call* is Observable<T> 

(i die .subscribeOn/.observeOn Konfiguration Sie verlassen)