feat: reuse task scheduler to clean up parser stats

Author: evgmel

src/NetworkScoresCalculator.ts

@@ -5,14 +5,24 @@ import {
   INetworkScoresCalculator,
   WebRTCStatsParsed,
 } from './types';
+import { scheduleTask } from './utils/tasks';
+import { CLEANUP_PREV_STATS_TTL_MS } from './utils/constants';
 
 class NetworkScoresCalculator implements INetworkScoresCalculator {
   #lastProcessedStats: { [connectionId: string]: WebRTCStatsParsed } = {};
 
   calculate(data: WebRTCStatsParsed): NetworkScores {
+    const { connection: { id: connectionId } } = data;
     const outbound = this.calculateOutboundScore(data);
     const inbound = this.calculateInboundScore(data);
-    this.#lastProcessedStats[data.connection.id] = data;
+    this.#lastProcessedStats[connectionId] = data;
+
+    scheduleTask({
+      taskId: connectionId,
+      delayMs: CLEANUP_PREV_STATS_TTL_MS,
+      callback: () => (delete this.#lastProcessedStats[connectionId]),
+    });
+
     return { outbound, inbound };
   }
 

src/parser/RTCStatsParser.ts

@@ -15,6 +15,8 @@ import {
   Logger,
 } from '../types';
 import { checkIsConnectionClosed, calcBitrate } from './utils';
+import { scheduleTask } from '../utils/tasks';
+import { CLEANUP_PREV_STATS_TTL_MS } from '../utils/constants';
 
 interface PrevStatsItem {
   stats: WebRTCStatsParsed;
@@ -23,14 +25,11 @@ interface PrevStatsItem {
 
 interface WebRTCStatsParserParams {
   logger: Logger;
-  prevConnectionStatsTtlMs?: number;
 }
 
 class RTCStatsParser implements StatsParser {
   private readonly prevStats = new Map<string, PrevStatsItem | undefined>();
 
-  private readonly prevStatsCleanupTimers = new Map<string, NodeJS.Timer>();
-
   private readonly allowedReportTypes: Set<RTCStatsType> = new Set<RTCStatsType>([
     'candidate-pair',
     'inbound-rtp',
@@ -43,11 +42,8 @@ class RTCStatsParser implements StatsParser {
 
   private readonly logger: Logger;
 
-  private readonly prevConnectionStatsTtlMs: number;
-
   constructor(params: WebRTCStatsParserParams) {
     this.logger = params.logger;
-    this.prevConnectionStatsTtlMs = params.prevConnectionStatsTtlMs ?? 55_000;
   }
 
   get previouslyParsedStatsConnectionsIds(): string[] {
@@ -128,7 +124,11 @@ class RTCStatsParser implements StatsParser {
       ts: Date.now(),
     });
 
-    this.resetStatsCleanupTimer(connectionId);
+    scheduleTask({
+      taskId: connectionId,
+      delayMs: CLEANUP_PREV_STATS_TTL_MS,
+      callback: () => (this.prevStats.delete(connectionId)),
+    });
 
     return mappedStats;
   }
@@ -287,21 +287,6 @@ class RTCStatsParser implements StatsParser {
 
     return connectionStats as ParsedConnectionStats;
   }
-
-  private resetStatsCleanupTimer(connectionId: string): void {
-    const existingTimer = this.prevStatsCleanupTimers.get(connectionId);
-
-    if (existingTimer) {
-      clearTimeout(existingTimer);
-    }
-
-    const timer = setTimeout(() => {
-      this.prevStats.delete(connectionId);
-      this.prevStatsCleanupTimers.delete(connectionId);
-    }, this.prevConnectionStatsTtlMs);
-
-    this.prevStatsCleanupTimers.set(connectionId, timer);
-  }
 }
 
 export default RTCStatsParser;

test/parser/RTCStatsParser.spec.ts

@@ -103,7 +103,7 @@ describe('wid/lib/parser/RTCStatsParser', () => {
     const clock = sandbox.useFakeTimers();
 
     await parser.parse(payload);
-    await clock.tickAsync(54_999);
+    await clock.tickAsync(34_999);
 
     expect(parser.previouslyParsedStatsConnectionsIds).to.deep.eq([payload.id]);
   });
@@ -114,7 +114,7 @@ describe('wid/lib/parser/RTCStatsParser', () => {
     const clock = sandbox.useFakeTimers();
 
     await parser.parse(payload);
-    await clock.tickAsync(55_000);
+    await clock.tickAsync(35_000);
 
     expect(parser.previouslyParsedStatsConnectionsIds).to.deep.eq([]);
   });