Używanie RxJava + Retrofit do tworzenia żądań API dla każdego elementu na liście

Question

Próbuję utworzyć obserwowalne przez połączenie wielu aprowizowanych api apofili. Kroki są:

1. Uzyskaj listę obiektów json przy użyciu wywołania API
2. Dla każdego obiektu na liście dokonać innego wywołanie API, aby uzyskać dodatkowe informacje o produkcie
3. Zapis danych pobranych z tej nowej szczegółowy obiekt do pliku na dysku (ta występuje dla każdej pozycji na liście)
4. Wreszcie zwraca obserwowalne osobnego obiektu, który wymaga, aby plik został utworzony dla każdej z poprzednich obiektów

to co Do tej pory:

public static Observable<DownloadedFiles> downloadFiles() { 
    DownloadedFiles downloadedFiles = new DownloadedFiles(); 
    Observable.create(subscriber -> { 
     return getRestService().getObjectList() 
     .flatMapIterable(objects -> objects) 
     .flatMap(objectLimited -> getRestService().getObject(objectLimited.getPath())) 
     .doOnNext(objectFull -> { 

      try { 
       File file = new File(); 
       // Extract data from objectFull and write new file to disk 
       // ... 
       } catch (IOException e) { 
       subscriber.onError(e); 
      } 

      downloadedFiles.putFile(file); 
     }) 
     .toList() 
     .map(objects -> downloadedFiles) 
     .finallyDo(() -> { 
      subscriber.onNext(downloadedFiles); 
      subscriber.onCompleted(); 
     }); 
    }); 
} 

@GET("/api/...") 
Observable<List<Object>> getObjectList(); 

@GET("/api/.../{path}") 
Observable<Object> getObject(@Path("path") String path); 

Czy ktoś mógłby potwierdzić, że użyłem odpowiednich operatorów. Dzięki .

Answer 1

Edycja: Usunięto Observable.create, modernizacja już daje się zauważyć, wystarczy ją zmienić.

Edycja 2: nie powinieneś też nic robić z subscriber.onError, jeśli zostanie zgłoszony błąd, zadzwoni on sam do subskrybenta.onError.

Całkiem nieźle, Nie wiem, po co poszło z płaską obserwacją. Zamiast tego robiłbym flatmap do Observable :: od, również zbieranie jest warte dodania. Zasadniczo zamierzam zmapować 1 rzecz do wielu, a następnie wykonać jakąś akcję, zebrać wiele z powrotem do jednego, a następnie zasubskrybować to, gdy zbieram wszystkie emitowane przedmioty.

public static Observable<DownloadedFiles> downloadFiles() {   
     return getRestService().getObjectList() 
     .flatMap(Observable::from) 
     .flatMap(objectLimited -> getRestService().getObject(objectLimited.getPath())) 
     .doOnNext(objectFull -> { 
      try { 
       File file = new File(); 
       // Extract data from objectFull and write new file to disk 
       // ... 
      } catch (IOException e) { 
       new IOException(e); 
      }}) 
     .collect(() -> new DownloadFiles<>(), (files, object) -> { files.add(object}); 

Answer 2

Myślę, że takie rozwiązanie powinno zadziałać. Zamiast ciągnąc u osobnika emitować mu DownloadedFile (y), można po prostu użyć obserwowalnym z usługi REST do map do każdego z DownloadedFiles:

public static Observable<DownloadedFile> downloadFiles() { 
    final Observable<Observable<FullObject>> observable = getRestService().getObjectList() 
      .flatMapIterable(objects -> objects) 
      .map(objectLimited -> getRestService().getObject(objectLimited.getPath())); 

    return Observable.mergeDelayError(observable) 
      .map(fullObject -> { 
       try { 
        File file = new File("path"); 
        // Extract data from objectFull and write new file to disk 
        // ... 

        return new DownloadedFile(); 
       } catch (IOException e) { 
        throw OnErrorThrowable.from(OnErrorThrowable.addValueAsLastCause(e, fullObject)); 
       } 
      }); 
} 

Można rozważyć użycie mergeDelayError (mapa()) zamiast flatMap, jeśli chcesz wysyłać pomyślnie zapisane pliki przed propagowaniem jakichkolwiek błędów.