RxJava 2. Retry mit Verzögerung, während andere Elemente fortfahren, die verschieden sind

Question

Ich habe eine beobachtbare, die Elemente aus vielen Quellen bekommt:

Source { List<Item> data } 

Beziehung zwischen Quellen und Gegenstände ist many-to-many und in verschiedenen Quellenelemente könnten sich duplizieren. Element ist eine Entität, die auf den Server hochgeladen werden soll, und der Server akzeptiert keine Duplikate. Um dies zu erreichen, füge ich Sources zusammen und unterscheide ihre Items anhand ihrer IDs und lade dann einzelne Items auf den Server hoch. Wie unten:

Observable.merge(source1(), source2(), source3()) 
      .flatMapIterable(sources -> sources) 
      .flatMapIterable(source::getItems) 
      .distinct(item -> item.getId()) 
      .flatMapCompletabale(item -> uploadItem(item)) 

Artikel-Upload kann mehrere Fehler emittiert und auf einige von ihnen soll ich versuchen Artikel zu laden, sobald es später noch einmal und weitere Gegenstände gehen, während ‚nicht bestanden‘ ein für sein retrying wartet.

Wie kann ich das erneute Hochladen des "fehlgeschlagenen" Elements verschieben und andere Elemente ausführen, während dieses für seinen Versuch wartet?

Vielen Dank im Voraus!

Answer 1

Ich setze diese Funktion in retryWhen Methode und es funktioniert.

public class RetryWithDelay implements Function<Observable<? extends Throwable>, Observable<?>> { 

private final int maxRetryCount; 
private final int retryDelay; 
private int retryCount; 
private TimeUnit timeUnit; 

public RetryWithDelay(final int maxRetryCount, final int retryDelay, final TimeUnit timeUnit) { 
    this.maxRetryCount = maxRetryCount; 
    this.retryDelay = retryDelay; 
    this.timeUnit = timeUnit; 
    this.retryCount = 0; 
} 

@Override 
public Observable<?> apply(final Observable<? extends Throwable> attempts) { 
    return attempts.flatMap((Function<Throwable, Observable<?>>) throwable -> { 

     if (++retryCount < maxRetryCount) { 
      return Observable.timer(retryDelay, timeUnit); 
     } 

     return Observable.error(throwable); 
    }); 
} 
} 

Answer 2

Um nur der Ausfall eines Upload Griff, Sie einen Operator im letzten Schritt hinzufügen:

.flatMapCompletable(item->uploadItem(item)) 

.flatMapCompletable(item->uploadItem(item) 
           .retryWhen(throwable -> 
            throwable.delay(5, TimeUnit.SECONDS))) 

bearbeiten werden soll: ich von this Dan Lew blog entry viel über retryWhen() Operator gelernt . Sie finden mehrere Beispiele, einschließlich der Verwendung des Operators zip() mit Observable.range(3), um die Anzahl der Wiederholungen zu begrenzen.