javascript - Comment exécuter un tableau de requêtes avec rxjs comme forkJoin et combineLatest mais sans avoir à attendre que TOUT soit terminé avant de voir les résultats ?

Alors imaginez que vous avez un tableau d'URL :

urls: string[]

Vous faites une collection de requêtes (dans cet exemple, j'utilise HTTPClient.get d'Angular, qui renvoie un Observable)

const requests = urls.map((url, index) => this.http.get<Film>(url)

Maintenant, je veux exécuter ces requêtes simultanément mais ne pas attendre toutes les réponses pour voir tout. En d'autres termes, si j'ai quelque chose comme films$: Observable<Film[]>, je veux que films$ se mette à jour graduellement à chaque fois qu'une réponse arrive.

Maintenant, pour simuler cela, vous pouvez mettre à jour le requests ci-dessus en quelque chose comme ceci

const requests = urls.map((url, index) => this.http.get<Film>(url).pipe(delay((index + 1)* 1000))

Avec le tableau ci-dessus de Observables, vous devriez obtenir les données de chaque requête une par une puisque elles ne sont pas demandées en même temps. Notez que cela ne fait que simuler les différents temps d'arrivée des données des requêtes individuelles. Les requêtes elles-mêmes devraient être effectuées en parallèle.

Le but est de mettre à jour les éléments dans films$ chaque fois qu'une valeur est émise par l'une des requêtes.

Donc avant, j'avais quelque chose comme ça quand je ne comprenais pas comment combineLatest fonctionne

let films$: Observable<Film[]> = of([]);
const requests = urls.map(url => this.http.get<Film>(url)
.pipe(
  take(1),
 // Without this handling, the respective observable does not emit a value and you need ALL of the Observables to emit a value before combineLatest gives you results.
 // rxjs EMPTY short circuits the function as documented. handle the null elements on the template with *ngIf. 
  catchError(()=> of(null))
));
// Expect a value like [{...film1}, null, {...film2}] for when one of the URL's are invalid for example.
films$ = combineLatest(requests); 

Je m'attendais à ce que le code ci-dessus mette à jour films$ progressivement, en négligeant une partie de la documentation

Pour garantir que le tableau de sortie ait toujours la même longueur, combineLatest attendra en réalité que toutes les Observables d'entrée émettent au moins une fois, avant de commencer à émettre des résultats.

Ce n’est pas ce que je cherchais.

S'il existe un opérateur ou une fonction rxjs qui peut accomplir ce que je recherche, je peux avoir un modèle plus propre en utilisant simplement le async pipe et sans avoir à gérer les valeurs null et les requêtes échouées.

J'ai aussi essayé

this.films$ = from(urls).pipe(mergeMap(url => this.http.get<Film>(url)));

et

this.films$ = from(requests).pipe(mergeAll());

ce qui n'est pas correct parce que le type de valeur retournée est Observable<Film> au lieu de Observable<Film[]> que je peux utiliser dans le modèle avec *ngFor="let film of films$ | async". Au lieu de cela, si je m'y abonne, c'est comme si j'écoutais un socket pour un enregistrement, recevant des mises à jour en temps réel (les réponses individuelles qui arrivent). Je peux m'abonner manuellement à l'un des deux et faire un Array.push vers une propriété séparée films: Film[] par exemple, mais cela va à l'encontre de l'objectif (utiliser Observable dans le modèle avec le tube async).

Réponse #1

L'opérateur scan fonctionnera bien pour vous ici :

const makeRequest = url => this.http.get<Film>(url).pipe(
  catchError(() => EMPTY))
);

films$: Observable<Film[]> = from(urls).pipe(
  mergeMap(url => makeRequest(url)),
  scan((films, film) => films.concat(film), [])
); 

Flux :

• from émet des URL une à la fois
• mergeMap s'abonne à "makeRequest" et émet le résultat dans le flux
• scan accumule les résultats dans un tableau et émet à chaque fois qu'une nouvelle émission est reçue

Pour préserver l'ordre, j'utiliserais probablement combineLatest puisqu'il émet un tableau dans le même ordre que les observables en entrée. Nous pouvons commencer chaque observable avec startWith(undefined), puis filtrer les éléments indéfinis :

const requests = urls.map(url => this.http.get<Film>(url).pipe(startWith(undefined));

films$: Observable<Film[]> = combineLatest(requests).pipe(
  map(films => films.filter(f => !!f))
);