使用RxJava和Retrofit迭代列表并根据子查询扩充结果

Question

我正在使用改造,我觉得rxjava(with retrolambda)非常适合以下流程:

1. 获取小部件列表(http)

2. 为每个小部件

   a)获取给定窗口小部件类型的文章列表(http)
   b)将所有文章保存到db
   c)获取列表中的第一篇(最新)文章并使用本文中的适当值更新widget.articleName和widget.articleUrl

3. 转换回列表并完成

但是我不确定在步骤2a之后该怎么做.到目前为止,这是我的代码

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

我玩过一些操作员,但是当链接时,我似乎总是缩小范围(例如,获取单个文章的句柄),然后再也无法访问原始小部件进行修改.

@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

您可以在流的某些位置插入doOnNext以添加副作用:

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

这是可运行的例子.