javascript - RXJS Rantai observables bergantung secara berurutan, tetapi mendapatkan setiap pengeluaran untuk bar kemajuan

Saya menghadapi masalah, dan saya telah cuba mencari penyelesaian menggunakan RxJs, tetapi nampaknya tidak dapat menjumpai yang sesuai...

• Saya mempunyai 3 permintaan REST yang berbeza, yang akan dipanggil secara berurutan, dan setiap satu daripada mereka memerlukan respons daripada yang sebelumnya sebagai hujah
• Saya ingin melaksanakan bar kemajuan, yang meningkat apabila permintaan berjaya

Ini yang saya fikirkan :

• Saya akan menggunakan paip dan concatMap() untuk mengelakkan langganan bersarang dan melanggan setiap permintaan apabila permintaan sebelumnya selesai.

Pertimbangkan versi yang sangat ringkas ini. Anggap bahawa setiap of mewakili satu permintaan REST yang berjaya sepenuhnya (akan menangani ralat kemudian), dan bahawa saya akan melakukan kerja yang tidak ditunjukkan dengan parameter n...

const request1 = of('success 1').pipe(
  delay(500),
  tap(n => console.log('received ' + n)),
);

const request2 = (n) => of('success 2').pipe(
  delay(1000),
  tap(n => console.log('received ' + n))
);

const request3 = (n) => of('success 3').pipe(
  delay(400),
  tap(n => console.log('received ' + n))
);

request1.pipe(
  concatMap(n => request2(n).pipe(
    concatMap(n => request3(n))
  ))
)

Walau bagaimanapun, apabila saya melanggan kepada kod terakhir, saya hanya akan mendapat respons daripada permintaan terakhir, yang dijangka kerana tiub itu menyelesaikan kepada itu.

Jadi dengan concatMap(), saya boleh mengaitkan panggilan REST yang bergantung dengan betul, tetapi tidak dapat mengikuti kemajuannya.

Walaupun saya boleh mengikuti kemajuan dengan mudah menggunakan langganan bersarang, tetapi saya sedang berusaha keras untuk mengelakkan ini dan menggunakan cara amalan terbaik.

Bagaimana saya boleh mengaitkan panggilan REST bergantung saya, tetapi masih dapat melakukan perkara setiap kali panggilan berjaya?

Jawapan #1

Ini adalah penyelesaian yang digeneralisasikan, walaupun tidak semudah itu. Tetapi ia membolehkan kemajuan diperhatikan sambil masih mengelakkan operator share, yang boleh memperkenalkan keadaan tidak dijangka jika digunakan dengan salah.

const chainRequests = (firstRequestFn, ...otherRequestFns) => (
  initialParams
) => {
  return otherRequestFns.reduce(
    (chain, nextRequestFn) =>
      chain.pipe(op.concatMap((response) => nextRequestFn(response))),
    firstRequestFn(initialParams)
  );
};

chainRequests mengambil sejumlah fungsi yang berubah-ubah dan mengembalikan sebuah fungsi yang menerima parameter awal dan mengembalikan sebuah observable yang concatMaps fungsi-fungsi tersebut bersama-sama seperti yang ditunjukkan secara manual dalam soalan. Ia melakukan ini dengan mengurangkan setiap fungsi kepada nilai akumulasi yang kebetulan adalah sebuah observable.

Ingat, RxJS membawa kita keluar dari neraka callback jika kita mengetahui jalannya.

const chainRequestsWithProgress = (...requestFns) => (initialParams) =>  {
  const progress$ = new Rx.BehaviorSubject(0);
  const wrappedFns = requestFns.map((fn, i) => (...args) =>
    fn(...args).pipe(op.tap(() => progress$.next((i + 1) / requestFns.length)))
  );
  const chain$ = Rx.defer(() => {
    progress$.next(0);
    return chainRequests(...wrappedFns)(initialParams);
  });
  return [chain$, progress$];
};

chainRequestsWithProgress mengembalikan dua observables - satu yang akhirnya mengeluarkan respons terakhir, dan satu lagi yang mengeluarkan nilai kemajuan apabila observable pertama dilanggan. Kita melakukan ini dengan membuat BehaviorSubject untuk berfungsi sebagai aliran nilai kemajuan kita, dan membungkus setiap fungsi permintaan kita untuk mengembalikan observable yang sama seperti biasanya, tetapi kita juga menyalurkannya ke tap supaya ia boleh menolak nilai kemajuan baru ke BehaviorSubject.

Kemajuan akan dikembalikan kepada sifar setiap kali melanggan kepada pemerhati pertama.

Jika anda ingin memulangkan satu observable yang menghasilkan keadaan kemajuan serta nilai hasil akhirnya, anda boleh membiarkan chainRequestsWithProgress sebaliknya memulangkan:

chain$.pipe(
  op.startWith(null),
  op.combineLatest(progress$, (result, progress) => ({ result, progress }))
)

dan anda akan mempunyai sebuah observable yang memancarkan objek yang mewakili kemajuan ke arah hasil akhirnya, kemudian hasil itu sendiri. Untuk difikirkan - adakah progress$ perlu memancarkan hanya nombor?

Amaran

Ini menganggap bahawa pemerhatian permintaan memancarkan tepat satu nilai.

Jawapan #2

Penyelesaian paling mudah ialah dengan mempunyai pemboleh ubah pengira kemajuan yang dikemas kini dari ketukan apabila setiap respons kembali.

    let progressCounter = 0;

    request1.pipe(
      tap(_ => progressCounter = 0.33),
      concatMap(n => request2(n).pipe(
        tap(_ => progressCounter = 0.66),
        concatMap(n => request3(n)
          .pipe(tap(_ => progressCounter = 1)))
      ))
    );

Jika anda mahu kemajuan itu sendiri dapat diperhatikan, maka anda mahu berkongsi permintaan yang boleh diperhatikan supaya tidak membuat permintaan berganda) dan kemudian menggabungkannya untuk mendapatkan kemajuan.

Contoh bagaimana anda mungkin ingin mendekati perkara itu boleh didapati di: https://www.learnrxjs.io/recipes/progressbar.html