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

Question

我正在使用改造，我觉得 rxjava（带有 retrolambda）非常适合以下流程：

获取小部件列表 (http)
每个小部件

a) 获取给定小部件类型的文章列表 (http)
b) 将所有这些保存到 db
c) 获取列表中的第一篇（最新）文章，并使用本文中的适当值更新 widget.articleName 和 widget.articleUrl
转换回列表并完成

但是我不确定在第 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)
);

Here is 这个可运行的例子。

Answer 2

之所以在此处添加它，是因为我找不到迭代对象中作为变量返回的列表的示例。

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

Answer 3

akarnokd 的回答很有帮助，但这可能会导致 NetworkOnMainThreadException。为了解决这个问题，我添加了

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

每个请求

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

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

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

android

rx-java

retrofit

retrolambda