Using RxJava + Retrofit to Create API Requests for Each Element in a List

Question

I am trying to create an observable by combining several modified api calls. Steps:

• Get list of json objects using api call
• For each object in the list, make another api call to get additional information about the element
• Write the data taken from this new detailed object to a file on disk (this happens for each item in the list)
• Finally, we return the observed separate object, which requires that the file be created for each of the previous objects.

This is what I have so far:

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); 

Can someone please confirm that I have used the correct operators. Thanks.

Answer 1

Edit: removed Observable.create, the modification already makes it visible to you, you just need to convert it.

Edit 2: you also do not need to do anything with the .onError subscriber; if an error is selected, it will itself call the subscriber.onError function.

Pretty good, Not sure why you went with the plane observed. Instead, I would make a flatmap for Observable :: from, and it's worth adding as well. Basically, I am going to display one thing for many, then take some action, put it back a lot, and then subscribe to it when I have collected all the emitted elements.

 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

I think some of them should work for you. Instead of pulling the Subject to emit it DownloadedFile (s), you can simply use the Observable from the REST service to map to each of the 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)); } }); } 

You may want to use mergeDelayError (map ()) instead of flatMap if you want to emit successfully saved files before propagating any errors.