javascript - Bagaimana menjalankan array permintaan dengan rxjs seperti forkJoin dan combineLatest tetapi tanpa harus menunggu SEMUA selesai sebelum melihat hasilnya?

Jadi bayangkan Anda memiliki array URL:

urls: string[]

Anda membuat kumpulan permintaan (dalam contoh ini saya menggunakan HTTPClient.get Angular yang mengembalikan sebuah Observable)

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

Sekarang saya ingin mengeksekusi permintaan ini secara bersamaan tetapi tidak menunggu semua respon untuk melihat semuanya. Dengan kata lain, jika saya memiliki sesuatu seperti films$: Observable<Film[]>, saya ingin films$ diperbarui secara bertahap setiap kali sebuah respon datang.

Sekarang untuk mensimulasikan ini, Anda dapat memperbarui requests di atas menjadi sesuatu seperti ini

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

Dengan array Observable di atas, Anda seharusnya mendapatkan data dari setiap permintaan satu per satu karena mereka tidak diminta pada saat yang sama. Perlu dicatat bahwa ini hanya meniru waktu kedatangan data yang berbeda dari masing-masing permintaan. Permintaan itu sendiri seharusnya dilakukan secara bersamaan.

Tujuannya adalah untuk memperbarui elemen di films$ setiap kali ada nilai yang dihasilkan oleh salah satu permintaan.

Jadi sebelumnya saya memiliki sesuatu seperti ini ketika saya salah paham tentang cara kerja combineLatest

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);

Saya mengharapkan kode di atas untuk memperbarui films$ secara bertahap, mengabaikan sebagian dari dokumentasi

Untuk memastikan array output selalu memiliki panjang yang sama, combineLatest sebenarnya akan menunggu semua Observable input untuk mengeluarkan setidaknya sekali, sebelum mulai mengeluarkan hasil.

Itu bukan yang saya cari.

Jika ada operator atau fungsi rxjs yang dapat mencapai apa yang saya cari, saya bisa memiliki template yang lebih bersih dengan hanya memanfaatkan async pipe dan tidak perlu menangani nilai null dan permintaan yang gagal.

Saya juga sudah mencoba

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

dan

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

yang mana tidak benar karena tipe nilai yang dikembalikan adalah Observable<Film> alih-alih Observable<Film[]> yang bisa saya gunakan di template dengan *ngFor="let film of films$ | async". Sebaliknya, jika saya berlangganan ke itu, rasanya seperti saya mendengarkan socket untuk satu rekaman, mendapatkan pembaruan secara realtime (respon individu yang masuk). Saya bisa berlangganan secara manual ke salah satu dari keduanya dan melakukan Array.push ke properti terpisah films: Film[] misalnya, tapi itu mengalahkan tujuan (menggunakan Observable di template dengan async pipe).

Jawaban #1

Operator scan akan bekerja dengan baik untuk Anda di sini:

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), [])
);

Aliran :

  • from mengeluarkan url satu per satu
  • mergeMap berlangganan ke "makeRequest" dan memancarkan hasil ke dalam aliran
  • scan mengakumulasi hasil ke dalam array dan memancarkan setiap kali emisi baru diterima

Untuk menjaga urutan, saya mungkin akan menggunakan combineLatest karena ia memancarkan array dalam urutan yang sama dengan observables input. Kita dapat memulai setiap observable dengan startWith(undefined), lalu menyaring item yang nilainya undefined:

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))
);

Artikel terkait