android - 将带条件的循环转换为 RxJava 流

Question

我有代码在 while 循环中执行阻塞操作(从服务器下载一些数据)。客户不知道每一步要返回多少项目。下载 N 个项目时循环中断。

val n = 10
val list = ArrayList<T>()

while (list.size < n) {
    val lastItemId = list.last()?.id ?: 0
    val items = downloadItems(lastItemId)
    list.addAll(items)
}

downloadItems执行阻塞 HTTP 调用并返回列表。现在让我们假设 downloadItems更改和新的返回类型是 Observable<Item> .如果不执行 blockingGet 之类的操作，我该如何更改代码以使用 RxJava？ ？

Answer 1

您可以使用 repeatUntil 来实现这一点:

var totalItems = 0    
var id = 0
Observable.fromCallable {
            downloadItems(id)
        }
        .flatMap {
            list ->
                totalItems += list.size
                id = list.last()?.id ?: 0
                Observable.just(list)
        }
        .repeatUntil({totalItems > n})
        .subscribe({result -> System.out.println(result) })