RxJava - Holen Sie sich eine Liste, bearbeiten Sie ihre Artikel, gruppieren Sie sie erneut in die Liste. toList() wartet auf onCompleted()

Question

Ich möchte alle nicht synchronisierten Elemente aus meiner Datenbank jedes Mal, wenn sie dort angezeigt werden, versuchen, sie mit dem Back-End zu synchronisieren und dann in die Datenbank schreiben, dass sie synchronisiert sind. Um dies durchzuführen, mache ich diese Kette von Operationen (mit RxJava Observablen):

1. Wenn eine Änderung in einer Datenbanktabelle auftritt, erhalten Sie die Liste der unsynchronisierten Elemente.
2. Für jedes Element in der Liste - senden Sie es an das Back-End.
3. Sammeln Sie alle Artikel, die erfolgreich an Backend gesendet wurden ...
4. ... und schreiben Sie sie in einer einzigen Transaktion in die Datenbank.

Ich habe versucht, so etwas wie diese:

newMsgsObservable //from realm database 
      .flatMapIterable(new Func1<List<Message>, Iterable<Message>>() { //1 
       @Override 
       public Iterable<Message> call(List<Message> messages) { 
        return messages; 
       } 
      }) 
      .flatMap(new Func1<Message, Observable<Message>>() { //2 
       @Override 
       public Observable<Message> call(Message message) { 
        return restService.sendMessage(message); 
       } 
      }) 
      .toList() //3 
      .flatMap(new Func1<List<Message>, Observable<?>>() { //4 
       @Override 
       public Observable<?> call(List<Message> messages) { 
        for(Message message : messages) { 
         message.setSynchronised(true); 
        } 
        persistence.saveMessages(messages); 
        return null; //yeah, I know, don't bother ;) 
       } 
      }) 
      .subscribe(); 

Das Problem

newMsgsObservable immer nach neuen Änderungen in der Datenbank warten (es wird nie abgeschlossen) , die ein gutes ist. Leider wartet toList() auf das Observable zu complete, bevor es die Liste der Elemente erneut zurückgibt. Ich möchte eine Lösung, die eine einzelne Liste von der Datenbank erhält, führt alle Aktionen für jedes Element auf der Liste durch, und erhält dann die gleiche Liste und gibt sie an die nächste beobachtbare zurück. Und tut es jedes Mal, wenn eine neue Liste von der ersten Observablen verfügbar ist.

Oder vielleicht mache ich es falsch und ich sollte dieses Problem auf andere Weise angehen?

Answer 1

Die Grundidee ist, Ihre toList() auf eine andere Ebene zu verschieben. So wird Ihr Code in etwa so aussehen: