جافاسكريبت - RxJS التثبيط مع الأولوية

أواجه صعوبة في ابتكار هذا البث.

ما أبحث عنه هو شيء مثل debounceTime ولكن مع أولوية.

لذلك إذا كان لدي أحداث بالشكل { type: 'a', priority: 2 }. يجب تأخير هذه الأحداث لبضع ثوانٍ، ولكن بدلاً من إصدار آخر حدث، يتم إصدار الحدث ذو الأولوية الأعلى.

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


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

لقد حاولت النظر في مشغلات أخرى مثل window وتصفية النتيجة للحصول على الحدث الأخير، لكن هذا ليس مثالياً لأن window يعمل على وتيرة ثابتة بينما أريد أن يبدأ المؤقت عند الحدث الأول كما يفعل التجميع الزمني (debouncing).

الإجابة رقم 1

يجب عليك تخزين وتحديث العنصر ذي الأولوية الأعلى وربطه بهذه القيمة highest التي تمررها بعد ذلك إلى debounceTime.

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

يمكنك إنشاء المشغّل الخاص بك الذي يقوم بذلك بمساعدة defer. يضمن defer أن يحصل كل مشترك على متغير highest الخاص به، حيث سيحصل كل مشترك على Observable جديد خاص به قد تم إنشاؤه عن طريق استدعاء دالة المصنع المعطاة.

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

الكود أعلاه هو تايبسكريبت. إذا كنت تريد جافاسكريبت عادية فقط قم بإزالة كل الأنواع.

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

الإجابة رقم 2

سأقدم الحل التالي، المبني على استخدام scan لعرض أعلى إصدار ذو أولوية حتى الآن للنظر فيه من قبل debounceTime(). لاحظ أن scan يحتاج إلى إعادة النظر في البيانات الجديدة بعد كل debounce ناجح، لذا أستخدم المعامل window() لتقسيم الإصدارات، بدء نافذة قابلة للمراقبة جديدة بعد كل إصدار بواسطة debounceTime().

إليك CodeSandbox

وإليك بعض الكود المبسط من CodeSandbox يوضح الأجزاء المهمة:

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