javascript - RxJS-takaisinkutsu prioriteetilla

Minulla on vaikeuksia keksiä tämä virta.

Etsimäni on jotain sellaista kuin debounceTime, mutta prioriteetilla.

Joten jos minulla on tapahtumia muodossa { type: 'a', priority: 2 }. Nämä tapahtumat on viivästettävä muutamalla sekunnilla, mutta sen sijaan että viimeinen tapahtuma lähetettäisiin, lähetetään korkeamman prioriteetin tapahtuma.

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


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

Olen yrittänyt katsella muita operaattoreita, kuten window, ja suodattaa tuloksen viimeisen tapahtuman osalta, mutta se ei ole ihanteellista, koska window toimii kiinteällä ajoituksella, kun taas haluan ajastimen alkavan ensimmäisestä tapahtumasta kuten debouncing tekee.

Vastaus #1

Sinun täytyy tallentaa ja päivittää kohde, jolla on korkein prioriteetti, ja kartoittaa tämä highest-arvo, jonka sitten annat debounceTime:lle.

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

Voit luoda oman operaattorisi, joka tekee tämän defer-avulla. defer varmistaa, että jokaisella tilauksentekijällä on oma highest-muuttujansa, sillä jokainen tilauksentekijä saa luotua uuden Observablen kutsumalla annettua tehtaan funktiota.

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

Yllä oleva koodi on Typescriptiä. Jos haluat tavallista Javascriptiä, poista vain kaikki tyypit.

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

Vastaus #2

Tarjoan seuraavan ratkaisun, joka perustuu scan-operaattorin käyttöön, jotta tähän mennessä korkein annettu prioriteetti voidaan tarjota debounceTime()-funktion käsiteltäväksi. Huomaa, että scanin täytyy tarkastella uutta dataa jokaisen onnistuneen debounce-erän jälkeen, joten käytän operaattoria window() jakamaan lähetteet, aloittaen uuden observable-ikkunan jokaisen debounceTime()-lähetyksen jälkeen.

Tässä on CodeSandbox

Ja tässä on hieman yksinkertaistettua koodia CodeSandboxista, joka näyttää tärkeimmät kohdat:

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