Wykonywanie N sekwencyjnych wywołań api za pomocą RxJava i Retrofit

Question

Mam listę plików, które chciałbym przesłać do backendu z urządzenia z systemem Android. Z powodu ograniczeń pamięci chciałbym wykonać drugie wywołanie API tylko po pierwszym zakończeniu, trzecim po drugim zakończeniu i tak dalej.

pisałem coś

private Observable<Integer> uploadFiles(List<File> files) { 
     return Observable.create(subscriber -> { 
      for (int i = 0, size = files.size(); i < size; i++) { 
       UploadModel uploadModel = new UploadModel(files.get(0)); 
       int uploadResult = retrofitApi.uploadSynchronously(uploadModel); 
       subscriber.onNext(uploadResult); 
      } 
      subscriber.onCompleted(); 
     }).subscribeOn(Schedulers.newThread()); 
    } 

Ale czuję się jak to może być wbrew duchowi Rx, a powiedzenie jest, jeśli używasz Observable.create, jesteś prawdopodobnie robi to źle. .. Czy to rozsądne podejście? Czy jest lepszy sposób na osiągnięcie tego dzięki integracji RxJava Retrofit?

Answer 1

naiwnie by to zrobić (nie działa, chociaż, patrz niżej):

return Observable.from(files).concatMap(file -> retrofitApi.upload(uploadModel)); 

Teraz problemem jest to, że nie ma sposobu, aby powiedzieć modernizację używać tylko jeden wątek dla tych połączeń.

reduce, jednak przekazuje wynik jednego wywołania funkcji do następnego, wraz z następną wysyłaną wartością z oryginalnego obserwowalnego. To zadziała, ale funkcja przekazana do reduce musi być synchroniczna. Niedobrze.

Innym podejściem byłoby zmodyfikować zaobserwować rekurencyjnie:

void getNextFile(int i) { 
    return retrofit.upload(i). 
     onNext(result -> getNextFile(i + 1)); 
} 

mniej. Ale nie jestem pewien, jak to wyczyścić, aby był bardziej czytelny.

Najczystsze Myślę byłoby coś takiego:

Observable.from(files).map(file -> retrofitApi.uploadSynchronously(new UploadModel(file))); 

Answer 2

tubylców RxJava będzie emitować wszystkie przedmioty na Observable.from(...) jakby równolegle. To najlepszy sposób, by myśleć o tym jako o emisji równoległej. Jednak niektóre przypadki wymagają rzeczywistej konsekwentnej realizacji całego łańcucha. Doszedłem do następującego rozwiązania, prawdopodobnie nie najlepszego, ale działającego.

import rx.Observable; 
import rx.Subscriber; 

import java.util.Iterator; 
import java.util.function.Function; 

public class Rx { 
    public static void ignore(Object arg) { 
    } 

    public static <E, R> Observable<Void> sequential(Iterator<E> iterator, Function<E, Observable<R>> action) { 
     return Observable.create(collectorSubscriber -> 
       Observable.<Void>create(producerSubscriber -> 
         producerSubscriber.setProducer(ignoredCount -> { 
          if (!iterator.hasNext()) { 
           producerSubscriber.onCompleted(); 
           return; 
          } 

          E model = iterator.next(); 
          action.apply(model) 
            .subscribe(
              Rx::ignore, 
              producerSubscriber::onError, 
              () -> producerSubscriber.onNext(null)); 
         })) 
         .subscribe(new Subscriber<Void>() { 
          @Override 
          public void onStart() { 
           request(1); 
          } 

          @Override 
          public void onCompleted() { 
           collectorSubscriber.onNext(null); 
           collectorSubscriber.onCompleted(); 
          } 

          @Override 
          public void onError(Throwable e) { 
           collectorSubscriber.onError(e); 
          } 

          @Override 
          public void onNext(Void aVoid) { 
           request(1); 
          } 
         })); 
    } 
} 

Przykład użycia może być:

Iterator<? extends Model> iterator = models.iterator(); 

    Rx.sequential(iterator, model -> someFunctionReturnsObservable(model)) 
      .subscribe(...); 

metoda ta gwarantuje łańcuchu Wykonania z

Observable<Dummy> someFunctionReturnsObservable(Model model)

Answer 3

Obecnie preferowanym sposobem tworzenia obserwable jest z fromAsync:

Observable.fromAsync(new Action1<AsyncEmitter<Object>>() 
    { 
     @Override 
     public void call(final AsyncEmitter<Object> emitter) 
     { 
      emitter.onNext(object); 
      emitter.onCompleted(); 

      emitter.setCancellation(new AsyncEmitter.Cancellable() 
      { 
       @Override 
       public void cancel() throws Exception 
       { 
        // on unSubscribe() callback 
       } 
      }); 
     } 
    }, AsyncEmitter.BackpressureMode.BUFFER);