Преобразование цикла с условием в поток 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>. Как я могу изменить код для использования RxJava, не выполняя что-то вроде blockingGet?

Answer 1

Я думаю, что это элегантный способ

int n = 10;
Observable.range(0,n)
        .flatMap(i -> downloadItems(i))
        .toList()
        .subscribe(itemsList -> itemsList.forEach(item -> System.out.println(item)));

Answer 2

Вы можете использовать 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) })