Wie kann ich eine gleichzeitige Mergenap in RXJs mit einer variablen Chargengröße ausführen? ⇐ JavaScript

[Anonymous]

Ich habe einen Strom von Prädikaten, die ich an eine API senden möchte, um auszuführen. Ich muss nur die ersten vier (zum Beispiel) identifizieren, die true zurückgeben. class = "Lang-JS PrettyPrint-Override">

Code: Select all

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

KoncatMap führt jedoch nur eine einzelne API -Anforderung zum anderen aus und wartet darauf, dass jedes vor dem Starten eines anderen abgeschlossen ist. Ich könnte zwanzig API-Anrufe aus dem Fledermaus abfeuern, wenn die ersten vier Prädikate bereits true zurückkehren.

Code: Select all

predicates$.pipe(
mergeMap(callExpensiveAPI, 4), // Limit to four concurrent API calls
filter(isTruthy),
take(4)
).subscribe();
< /code>
Es gibt jedoch noch zwei Mängel: < /p>

  Die Reihenfolge der Prädikate (und Ergebnisse) ist nicht erhalten. Ich werde nur vier Wahrheitsprädikate bekommen, aber ihre Reihenfolge im Stream hängt von der Reaktionszeit der API -Anrufe ab. Es besteht die Möglichkeit, dass nach dem Initiieren des vierten API -Anrufs für nachfolgende Prädikate im Stream ein paar weitere Anrufe (bis zu drei!) Tätig werden. Im schlimmsten Fall wird  ein nachfolgender Aufruf wahre < /Code> vor dem tatsächlichen "vierten wahrheitsgemäßen" Prädikat < /strong> zurückgegeben, und das Endergebnis wird nicht die ersten vier Wahrheitspräentikate sein. in der nachgelagerten Pipeline. Wenn nur ein weiteres Prädikat benötigt wird, wäre der Parameter Parallelität 1 

, was dazu führt, dass MergeMap sich wie ConcatMap .
Wie könnte ich eine solche Pipeline implementieren?>