javascript - RxJS Penundaan dengan keutamaan

Saya menghadapi masalah untuk menghasilkan aliran ini.

Apa yang saya cari adalah sesuatu seperti debounceTime tetapi dengan keutamaan.

Jadi jika saya mempunyai acara dengan bentuk { type: 'a', priority: 2 }. Acara ini perlu ditangguhkan selama beberapa saat tetapi bukannya acara terakhir yang dihantar, acara dengan keutamaan tertinggi dihantar.

input stream:
------(a|1)--(b|3)---(c|2)-----------------------(a|1)-----------------


output stream:
-----------------------------------(b|3)---------------------(a|1)-----

Saya telah mencuba melihat kepada operator lain seperti window dan menapis melalui hasil untuk acara terakhir tetapi ia tidak ideal kerana window berfungsi pada kadar tetap di mana saya mahu pemasa bermula pada acara pertama seperti yang dilakukan oleh debouncing.

Jawapan #1

Anda perlu menyimpan dan mengemas kini item dengan keutamaan tertinggi dan memetakan kepada nilai highest ini yang kemudian anda teruskan ke debounceTime.

let highest = null;
source$.pipe(
  map(v => highest = highest && highest.priority > v.priority ? highest : v),
  debounceTime(2000),
  tap(() => highest = null)
);

Anda boleh membuat pengendali anda sendiri yang melakukan ini dengan bantuan defer. defer memastikan bahawa setiap pelanggan mendapat pembolehubah highest mereka sendiri, kerana setiap pelanggan akan mendapat Observable baru mereka sendiri yang dibuat dengan memanggil fungsi kilang yang diberikan.

function debounceTimeHighest<T>(dueTime: number, getHighest: (curr: T, high: T) => T): MonoTypeOperatorFunction<T> {
  return (source: Observable<T>) => defer(() => {
    let highest: T = null;
    return source.pipe(
      map(item => highest = highest ? getHighest(item, highest) : item),
      debounceTime(dueTime),
      tap(() => highest = null)
    );
  });
}

// Usage
source$.pipe(
  debounceTimeHighest(2000, (v1, v2) => v1.priority >= v2.priority ? v1 : v2)
)

Kod di atas adalah Typescript. Jika anda mahu Javascript biasa, cuma buang semua jenis.

https://stackblitz.com/edit/rxjs-hitqxk

Jawapan #2

Saya akan mencadangkan penyelesaian berikut, berasaskan penggunaan scan untuk memberikan emisi keutamaan tertinggi yang diberikan setakat ini untuk pertimbangan oleh debounceTime(). Perlu diingat bahawa scan perlu menilai semula data baru selepas setiap debounce yang berjaya, jadi saya menggunakan operator window() untuk membahagikan emisi, memulakan tetingkap observable baru selepas setiap emisi oleh debounceTime().

Berikut ialah CodeSandbox

Dan di sini adalah beberapa kod yang dipermudahkan dari CodeSandbox yang menunjukkan bahagian penting:

const resetScan$ = new Subject();

source$.pipe(
  window(resetScan$),
  mergeMap(win$ => win$.pipe(
    scan((acc, cur) => acc.priority >= cur.priority ? acc : cur )
  )),
  debounceTime(debounceDelay),
  tap(() => resetScan$.next())
);