javascript - Räkna händelser över en tidsperiod och ge summan en gång varje sekund i RxJS

Är det möjligt att räkna händelser över en tidsperiod och ge summan varje sekund i RxJS? Jag har en kontinuerlig aldrig sinande ström av händelser. Varje 1 sekund skulle jag vilja få det totala antalet händelser under de senaste 5 minuterna. Idén är att använda detta för att fylla en realtidsgraf.

Jag vet hur man gör det på det traditionella sättet men skulle verkligen vilja förstå hur det görs med reaktiv programmering.

Svar #1

Så här skulle jag ta itu med det.

Skapa en observable som bara räknar antalet mottagna händelser och skickar detta som ett löpande totaltal (via scan). Skapa en andra observable som bara är det löpande totaltalet fördröjt med 5 minuter. Skapa en tredje observable som bara subtraherar den fördröjda observablen från den första observablen. Detta ger det löpande totaltalet av händelser som är yngre än 5 minuter. Skapa en sista observable som samplar denna tredje observable en gång per sekund.

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

Svar #2

Här är ett annat sätt att göra det som jag tror kan vara lite enklare. Du kan använda windowWithTime - se ReactiveX och RxJS-dokumentation. Du kan skapa fönster som överlappar, så att du kan ha ett 5 minuters fönster med händelser, följt av ett annat 5 minuters fönster med händelser som startade en sekund senare, och ett till en sekund efter det, etc.

Om du tar antalet för varje av dessa fönster får du en serie av räkningar av 5 minuters händelser, en sekund ifrån varandra. Det skulle se ut ungefär så här:

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

Jag har använt detta för att göra exakt det du säger - få en hastighet av händelser över ett tidsfönster i en oavbruten ström av händelser. Genom att låta fönstren överlappa genererar du ett glidande medelvärde av aktiviteten på strömmen.