Как я могу запустить одновременную Mergemap в RXJ с переменным размером партии? ⇐ Javascript

[Anonymous]

У меня есть поток предикатов, которые я хочу отправить в API для выполнения. Мне нужно только идентифицировать первые четыре (например), которые возвращают True.
Хотя я могу отправлять запросы API параллельно, Каждый вызов API является дорогостоящим : я хочу минимизировать количество ненужных вызовов API, в то время как все еще используя параллелизм, где это возможно. />

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

predicates$.pipe(
concatMap(callExpensiveAPI),
filter(isTruthy),
take(4)
).subscribe();

Тем не менее, concatmap запускает только один запрос API за раз, ожидая, что каждый завершит, прежде чем запустить другой. Я мог бы выстрелить двадцати API отключает летучую мышь, когда первые четыре предиката уже возвращают true.
Параметр параллелизма для Mergemap кажется мне очень близко к желаемому поведению:

predicates$.pipe(
mergeMap(callExpensiveAPI, 4), // Limit to four concurrent API calls
filter(isTruthy),
take(4)
).subscribe();
< /code>
Однако все еще есть два недостатка: < /p>

Порядок предикатов (и результатов) не сохранился. Я получу только четыре предиката правды, но их порядок в потоке зависит от времени отклика каждого вызова API. Есть шанс, что после начала четвертого вызова API, предназначенного для возвращения True, еще несколько вызовов (до трех!) Будут сделаны для последующих предикатов в потоке. < /P>
В лучшем случае эти звонки просто не нужны. В худшем случае Последующий вызов вернет True < /code> до фактического «четвертого правдивого» предиката < /strong>, и конечный результат на самом деле не будет первым четырьмя правдами. Нижний трубопровод. Когда требуется только один предикат, параметр параллелизма будет 1 , что приведет к тому, что Mergemap ведет себя как concatmap .
Как я могу реализовать такой трубопровод?

Подробнее здесь: https://stackoverflow.com/questions/797 ... batch-size