java - 反复调用API，直到全部数据下载完毕

Question

我有一个用例，我想为每个用户重复调用 Web API，直到下载完整个数据。

我拥有的 Web API 允许每个 API 请求为用户提取最多 100 条记录。我可以指定要为该用户下载的记录的 startTime 和结束时间戳:

void downloadRecord(String userId, recordStartTime, recordEndTime, int countOfRecord, ResultCallBack)

如果recordStartTime和recordEndTime之间的记录数超过100条，那么API响应只返回100条记录。然后我必须循环使用新的开始时间(刚下载的第 100 条记录的时间)调用此 api，直到所有记录都下载完毕。

final long recordStartTime = timeStampFrom2DaysAgo//;
Observable.from(arrayListOfUserIds).flatMap(new Func1<String, Observable<?>>() {
    @Override
    public Observable<?> call(String userId) {
        long recordEndTime = getCurrentTimeMS();
        //keep downloading records 
        downloadRecord(userId, recordStartTime, recordEndTime, 100, new ResultCallBack() {
            //if records are < 100 then stop 
            //else keep downloading 
        });
    }
}).subscribe();

请建议是否有 RxJava 示例代码可以用来解决我的问题。

谢谢

Answer 1

这是一个循环依赖的 Observable 的例子。假设你有一个用于请求的 Observable 和一个用于响应的 Observable。 requests Observable 发出一些 DownloadParams对象，包含 userId、recordStartTime、recordEndTime 和 countOfRecord，因此 Observable<DownloadParams> requests . responses Observable 发出记录列表，即 Observable<List<Record>> responses .

responses显然取决于 requests , 但不那么明显的部分是我们需要 requests还要依赖responses ，因为接下来从服务器下载，带有一定的DownloadParams ，取决于我们从之前的下载中得到的响应。为了完整起见，requests实际上还依赖于一些初始化 Observable，它发出 userId 来执行第一次下载。你可以用 .startWith(firstDownloadParams) 替换这个初始化 Observable .

无论如何，困难的部分是表达循环依赖。好消息是这在 Rx 中是可能的，并且已经 the focus of Cycle.js framework based on RxJS .坏消息是我们无法在没有 Subject 的情况下解决这个问题，这可能是不受欢迎的。

最好让 RxJava 代码尽可能地保持功能性，但是尝试这样做会遇到问题。如果我们尝试将 Observable 声明为其他对象的函数，我们会得到:

Observable<DownloadParams> requests = responses.flatMap( /* ... */ )
    .startWith(firstDownloadParams);
Observable<List<Record>> responses = requests.flatMap( /* ... */ );

这不会编译，因为第一个声明需要第二个声明，反之亦然。这就是主题可以提供帮助的方式。我们将这些 Observable 中的任何一个声明为 Subject:

PublishSubject<List<Record>> responsesProxy = PublishSubject.create();
Observable<DownloadParams> requests = responsesProxy.flatMap( /* ... */ )
    .startWith(firstDownloadParams);
Observable<List<Record>> responses = requests.flatMap( /* ... */ );

主题 responsesProxy将作为 responses 的代理, 这样我们就可以声明 requests取决于 responsesProxy .但是现在我们需要 responsesProxy模仿responses .我们通过添加这个来做到这一点:

responses.subscribe(responsesProxy);

这将关闭循环依赖中的循环，但请记住在我们上面进行订阅后也要正确处理 Subject。

现在我们只需要在那些 flatMap 中填写转换.在 requests.flatMap( )您应该执行下载记录列表的网络调用。您可能希望将其作为 Observable 而不是回调来处理，以简化与其余 RxJava 代码的相互作用。

在responsesProxy.flatMap( )您应该检查记录列表是否为 100 条或更多并创建下一个 DownloadParams使用新的 recordStartTime，包装为 Observable.just(newDownloadParams) .如果小于 100，则返回 Observable.empty() .