RxJava + 改造，获取列表并为每个项目添加额外信息

Question

我正在使用 RXJava，在 Android 中进行改造。我正在尝试完成以下任务：

我需要定期轮询给我一个 Observable 的电话>（从这里我可以做到）

一旦我得到这个列表，我想在每个 Delivery 中迭代并调用另一个方法，这会给我 ETA（所以只是更多信息）我想将此新信息附加到交付中，并返回完整列表和额外的附在每个项目上的信息。

一旦我得到列表，我知道如何在没有 rxjava 的情况下做到这一点，但我想练习一下。

这是我目前的代码：

pollDeliveries = Observable.interval(POLLING_INTERVAL, TimeUnit.SECONDS, Schedulers.from(AsyncTask.THREAD_POOL_EXECUTOR))
            .map(tick -> RestClient.getInstance().getApiService().getDeliveries())
                    .doOnError(err -> Log.e("MPB", "Error retrieving messages" + err))
                    .retry()
                    .subscribe(deliveries -> {
                        MainApp.getEventBus().postSticky(deliveries);
                    });

这是给我一份交货清单。现在我想完成第二部分。

希望我已经足够清楚了。 谢谢

Answer 1

我终于找到了一个不错的方法。

private void startPolling() {
    pollDeliveries = Observable.interval(POLLING_INTERVAL, TimeUnit.SECONDS, Schedulers.from(AsyncTask.THREAD_POOL_EXECUTOR))
            .flatMap(tick -> getDeliveriesObs())
            .doOnError(err -> Log.e("MPB", "Error retrieving messages" + err))
            .retry()
            .subscribe(this::parseDeliveries, Throwable::printStackTrace);
}

private Observable<List<Delivery>> getDeliveriesObs() {
    return RestClient.getInstance().getApiService().getDeliveries()
            .flatMap(Observable::from)
            .flatMap(this::getETAForDelivery)
            .toSortedList((d1, d2) -> {
                if (d1.getEta() == null) {
                    return -1;
                }
                if (d2.getEta() == null) {
                    return 1;
                }
                return d1.getEta().getDuration().getValue() > d2.getEta().getDuration().getValue() ? 1 : -1;
            });
}

让我们一步一步来。

1. 首先，我们创建一个可在每个 POLLING_INTERVAL 时间触发 getDeliveriesObs() 方法的 Observable，该方法将返回最终列表
2. 我们使用改造来获取调用的 Observable
3. 我们使用 flatMap 来展平结果列表，并在下一个 flatmap 中一个接一个地获取 Delivery 项目。
4. 然后我们获取 Delivery 对象中设置的预计到达时间并返回它
5. 我们按预计到达时间对列表进行排序。
6. 如果出现错误，我们会打印并重试，这样间隔就不会停止
7. 我们最终订阅以对列表进行排序并在其中包含 ETA，然后我们只需返回它或您需要对其进行任何处理。

它工作正常，非常好，我开始喜欢 rxjava :)

Answer 2

我没有花很多时间使用 Java 8 lambdas，但这里有一个示例，将每个对象映射到不同的对象，然后在普通的 Java 7 的另一端得到一个 List<...>：

List<Delivery> deliveries = ...;
Observable.from(deliveries).flatMap(new Func1<Delivery, Observable<ETA>>() {
    @Override
    public Observable<ETA> call(Delivery delivery) {
        // Convert delivery to ETA...
        return someEta;
    }
})
.toList().subscribe(new Action1<List<ETA>>() {
    @Override
    public void call(List<ETA> etas) {

    }
});

当然，最好接受 Retrofit 响应（大概是 Observable<List<Delivery>>？）并观察其中的每一个。为此，我们最好使用flatten()，doesn't appear to be coming to RxJava anytime soon。

为此，您可以改为执行类似这样的操作（使用 lambda 会更好）。您可以将上面示例中的 Observable.from(deliveries) 替换为以下内容：

apiService.getDeliveries().flatMap(new Func1<List<Delivery>, Observable<Delivery>>() {
    @Override
    public Observable<Delivery> call(List<Delivery> deliveries) {
        return Observable.from(deliveries);
    }
}).flatMap(...)