javascript - Počítat události během časového období a vracet součet jednou za sekundu v RxJS

Je možné spočítat události během určitého časového období a vydat součet každou sekundu v RxJS? Mám nepřetržitý nikdy nekončící tok událostí. Každou sekundu bych chtěl získat celkový počet událostí za posledních 5 minut. Myšlenka je použít toto k naplnění grafu v reálném čase.

Vím, jak to udělat tradičním způsobem, ale opravdu bych chtěl pochopit, jak se to dělá s reaktivním programováním.

Odpověď č. 1

Takhle bych se do toho pustil.

Vytvořte jeden observable, který pouze počítá počet přijatých událostí a vysílá je jako průběžný součet (pomocí scan). Vytvořte druhý observable, který je pouze průběžným součtem zpožděným o 5 minut. Vytvořte třetí observable, který jednoduše odečte zpožděný observable od prvního observable. To dá průběžný součet událostí mladších než 5 minut. Nakonec vytvořte observable, který vzorkuje tento třetí observable jednou za sekundu.

const totalLast5Minutes = eventSource.publish(events => {
    const runningTotal = events
        .scan((e, total) => total + 1, 0)
        .startWith(0);
    const totalDelayed5Minutes = runningTotal
        .delay(5000 * 60)
        .startWith(0);
    return Rx.Observable
        .combineLatest(total, totalDelayed5Minutes, (t, td) => t - td);
});

// only sample the value once per second
Rx.Observable
    .interval(1000)
    .withLatestFrom(totalLast5Minutes, (interval, total) => total)
    .subscribe(total => console.log(`total=${total}`));

Odpověď č. 2

Zde je další způsob, jak to udělat, který si myslím, že může být o něco jednodušší. Můžete použít windowWithTime - viz dokumentace ReactiveX a RxJS. Můžete vytvářet okna, která se překrývají, takže můžete mít 5minutové okno událostí, následované dalším 5minutovým oknem událostí, které začalo o jednu sekundu později, a dalším o jednu sekundu poté, atd.

Pokud vezmete počet každého z těchto oken, dostanete řadu počtů událostí po 5 minutách, každou sekundu oddělené. Vypadalo by to nějak takto:

source.windowWithTime(5 * 60000, 1000)   // create 5 minute windows, one second apart
      .flatMap(window => window.count()) // take each window and get the count once it completes (after 5 minutes)
      .subscribe(count => console.log(count));

Použil jsem to přesně k tomu, co říkáte - získat rychlost událostí během časového okna v nepřetržitém proudu událostí. Tím, že okna překrýváte, vytváříte klouzavý průměr aktivity na proudu.