Объединить Flowable с отфильтрованными элементами, обновленными любой пакетной операцией ⇐ JAVA

[Anonymous]

У меня есть rxjava3 Flowable с объектами Item, и я хочу запускать асинхронную операцию в пакетном режиме только для элементов, соответствующих условию. Затем я хочу разрешить нижестоящим операторам перебирать исходный поток объектов Item в исходном порядке.

Код: Выделить всё

Flowable source ...
Flowable updated = source.filter(i -> i.isCondition())
.buffer(3)
.flatMap(batch -> {
return apiCall(batch).flatMapPublisher(apiRes -> {
for (Item item : batch) {
item.updateFrom(apiRes)
}
return Flowable.fromIterable(batch);
}
}
// return Flowable from source in the same order
// but only after updated completed for the batch
// or isCondition is false

Предположим, у меня есть список идентификаторов, и для идентификаторов больше 10 я хочу получить имя группы, используя удаленный сервис, принимающий список идентификаторов.

Код: Выделить всё

5
4
12
3
15
6

Он должен вызвать getGroupIds(12, 15) и вернуть:

Код: Выделить всё

5, null
4, null
12, sport
3, null
15, music
6, null

Как лучше всего добиться этого для большого набора данных? Ошибка в apiCall должна завершить исходный поток.
В приведенном выше примере существует вероятность большого количества элементов от 12 до 15 - это увеличит размер буфера, используемого для сохранения исходного порядка и таким образом, он должен вызвать apiCall, если i.isCondition() имеет значение true для одного элемента и false для следующих n элементов, чтобы позволить буферу сбрасываться на выходные данные после обработки apiCall. Это можно упростить, используя source.buffer(1000) перед применением фильтра.

Подробнее здесь: https://stackoverflow.com/questions/792 ... -operation