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

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.

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