javascript - Contare gli eventi in un periodo di tempo e restituire la somma una volta al secondo in RxJS

È possibile contare gli eventi su un periodo di tempo e ottenere la somma una volta ogni secondo in RxJS? Ho un flusso continuo e infinito di eventi. Ogni 1 secondo vorrei ottenere il numero totale di eventi nell'ultima finestra di 5 minuti. L'idea è di usare questo per popolare un grafico in tempo reale.

So come fare questo nel modo tradizionale ma mi piacerebbe davvero capire come si fa con la programmazione reattiva.

Risposta n. 1

Ecco come lo affronterei.

Crea un observable che conta semplicemente il numero di eventi ricevuti ed emette questo come un totale progressivo (tramite scan). Crea un secondo observable che è solo il totale progressivo ritardato di 5 minuti. Crea un terzo observable che sottrae semplicemente l'observable ritardato dal primo observable. Questo restituirà il totale progressivo degli eventi che hanno meno di 5 minuti. Crea un observable finale che campiona questo terzo observable una volta al secondo.

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}`));

Risposta n. 2

Ecco un altro modo di farlo che penso possa essere un po' più semplice. Puoi usare windowWithTime - vedi la documentazione di ReactiveX e RxJS. Puoi creare finestre che si sovrappongono, quindi puoi avere una finestra di 5 minuti di eventi, seguita da un'altra finestra di 5 minuti di eventi che è iniziata un secondo dopo, e un'altra un secondo dopo ancora, ecc.

Se prendi il conteggio di ciascuna di queste finestre, otterrai una serie di conteggi di eventi di 5 minuti, ciascuno a un secondo di distanza. Apparirebbe qualcosa del genere:

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

Ho usato questo per fare esattamente quello che dici - ottenere un tasso di eventi su una finestra di tempo in un flusso continuo di eventi. Facendo sovrapporre le finestre, si genera una media mobile dell'attività sul flusso.