Use RxJava and Retrofit to iterate through list and augment results based on subqueries

Question

I'm using retrofit and I feel like rxjava (with retrolambda) would be a good fit for the following flow:

1. get list of widgets (http)

2. for each widget

   a) get a list of articles (http) for the given widget type
   b) save all those to db
   c) take the first (latest) article in list and update widget.articleName and widget.articleUrl with appropriate values from this article

3. transform back to list and complete

However I'm unsure what to do after step 2a. Here's my code so far

apiService.getWidgets(token)
  .flatMapIterable(widgets -> widgets)
  .flatMap(widget -> apiService.getArticles(token, widget.type))
  ...
  .toList()
  .subscribe(
     modifiedWidgets -> saveWidgets(modifiedWidgets),
     throwable -> processWidgetError(throwable)
  );

I've played around with some operators but when chaining, I always seem to narrow down too far (e.g. get a handle on a single article) and then no longer have access to the original widget to make modifications.

@GET("/widgets")
Observable<List<Widget>> getWidgets(@Header("Authorization") String token);

@GET("/articles")
Observable<List<Article>> getArticles(@Header("Authorization") String token, @Query("type") String type);

Joletta

Answer 1

You could insert doOnNext at certain points of the stream to add side-effects:

apiService.getWidgets(token)
.flatMapIterable(v -> v)
.flatMap(w -> 
    apiService.getArticles(token, w.type)
    .flatMapIterable(a -> a)
    .doOnNext(a -> db.insert(a))
    .doOnNext(a -> {
         w.articleName = a.name;
         w.articleUrl = a.url;
    })
    .takeLast(1)
    .map(a -> w)
)
.toList()
.subscribe(
    modifiedWidgets -> saveWidgets(modifiedWidgets),
    throwable -> processWidgetError(throwable)
);

Here is runnable example of this.

Intern

Brilliant! Thank you for taking the time to provide such a comprehensive answer. My knowledge of RxJava has just increased another notch :) – Joletta

Answer 2

adding this here since I couldn't find an example of iterating a list that is returned in an object as variable.

getUserAccount(token)
    .subscribeOn(Schedulers.newThread())
    .observeOn(AndroidSchedulers.mainThread())
    .flatMap(userResponse -> Observable.just(userResponse.list))      //get list from response
    .flatMapIterable(baseDatas -> baseDatas)                          //make the list iterable
    .flatMap(baseData ->                                              //on each project, get the details
            getProjectDetails(baseData.name,token)
                    .subscribeOn(Schedulers.io())                     //get network call off the main thread
                    .observeOn(AndroidSchedulers.mainThread()))
    .subscribe(
            (dataResponse) -> {
                Timber.d( "Got Data Details:" + dataResponse);
            },
            (error) -> {
                Timber.e( "Got Error:" + error.getMessage());
            },
            () -> {
                Timber.d("Completed Data Details");
            }
    );

Expeditious

Answer 3

akarnokd's answer is quite helpful but that may cause NetworkOnMainThreadException. To solve that I have added

.observeOn(AndroidSchedulers.mainThread())
.subscribeOn(Schedulers.io())

on every requests

apiService.getWidgets(token)
.observeOn(AndroidSchedulers.mainThread())      //added this
.subscribeOn(Schedulers.io())                   //added this
.flatMapIterable(v -> v)
.flatMap(w -> 
    apiService.getArticles(token, w.type)
    .observeOn(AndroidSchedulers.mainThread())  //added this
    .subscribeOn(Schedulers.io())               //added this
    .flatMapIterable(a -> a)
    .doOnNext(a -> db.insert(a))
    .doOnNext(a -> {
         w.articleName = a.name;
         w.articleUrl = a.url;
    })
    .takeLast(1)
    .map(a -> w)
)
.toList()
.subscribe(
    modifiedWidgets -> saveWidgets(modifiedWidgets),
    throwable -> processWidgetError(throwable)
);

Civilize