diff --git a/src/components/home/enhanced-canvas-traffic-graph.tsx b/src/components/home/enhanced-canvas-traffic-graph.tsx index 3562d3726..9390b8457 100644 --- a/src/components/home/enhanced-canvas-traffic-graph.tsx +++ b/src/components/home/enhanced-canvas-traffic-graph.tsx @@ -110,7 +110,7 @@ export const EnhancedCanvasTrafficGraph = memo( const { t } = useTranslation(); // 使用增强版全局流量数据管理 - const { dataPoints, getDataForTimeRange, samplerStats } = + const { dataPoints, requestRange, samplerStats } = useTrafficGraphDataEnhanced(); // 基础状态 @@ -183,8 +183,7 @@ export const EnhancedCanvasTrafficGraph = memo( // 监听数据变化 useEffect(() => { - const timeRangeData = getDataForTimeRange(timeRange); - updateDisplayData(timeRangeData); + updateDisplayData(dataPoints); return () => { if (debounceTimeoutRef.current !== null) { @@ -192,7 +191,11 @@ export const EnhancedCanvasTrafficGraph = memo( debounceTimeoutRef.current = null; } }; - }, [dataPoints, timeRange, getDataForTimeRange, updateDisplayData]); + }, [dataPoints, updateDisplayData]); + + useEffect(() => { + requestRange(timeRange); + }, [requestRange, timeRange]); useEffect(() => { if (displayData.length === 0) { diff --git a/src/hooks/use-traffic-data.ts b/src/hooks/use-traffic-data.ts index 1e386da79..eb8ca4111 100644 --- a/src/hooks/use-traffic-data.ts +++ b/src/hooks/use-traffic-data.ts @@ -15,7 +15,7 @@ export const useTrafficData = () => { const trafficRef = useRef(null); const { graphData: { appendData }, - } = useTrafficMonitorEnhanced(); + } = useTrafficMonitorEnhanced({ subscribe: false }); const ws = useRef(null); const wsFirstConnection = useRef(true); const timeoutRef = useRef>(null); diff --git a/src/hooks/use-traffic-monitor.ts b/src/hooks/use-traffic-monitor.ts index a92f3d409..0f1291be6 100644 --- a/src/hooks/use-traffic-monitor.ts +++ b/src/hooks/use-traffic-monitor.ts @@ -1,61 +1,51 @@ -import { useEffect, useRef, useCallback, useReducer } from "react"; +import { + useCallback, + useEffect, + useMemo, + useReducer, + useRef, + useState, +} from "react"; import { Traffic } from "tauri-plugin-mihomo-api"; +import type { + ISamplerStats, + ITrafficDataPoint, + ITrafficWorkerSnapshotMessage, + TrafficWorkerRequestMessage, + TrafficWorkerResponseMessage, +} from "@/types/traffic-monitor"; import { debugLog } from "@/utils/debug"; +import { TrafficDataSampler, formatTrafficName } from "@/utils/traffic-sampler"; -// 增强的流量数据点接口 -export interface ITrafficDataPoint { - up: number; - down: number; - timestamp: number; - name: string; -} - -// 压缩的数据点(用于长期存储) -interface ICompressedDataPoint { - up: number; - down: number; - timestamp: number; - samples: number; // 压缩了多少个原始数据点 -} - -// 数据采样器配置 -interface ISamplingConfig { - // 原始数据保持时间(分钟) - rawDataMinutes: number; - // 压缩数据保持时间(分钟) - compressedDataMinutes: number; - // 压缩比例(多少个原始点压缩成1个) - compressionRatio: number; -} +export type { ITrafficDataPoint } from "@/types/traffic-monitor"; // 引用计数管理器 class ReferenceCounter { private count = 0; - private callbacks: (() => void)[] = []; + private callbacks = new Set<() => void>(); + + private notify() { + this.callbacks.forEach((cb) => cb()); + } increment(): () => void { this.count++; debugLog(`[ReferenceCounter] 引用计数增加: ${this.count}`); - if (this.count === 1) { - // 从0到1,开始数据收集 - this.callbacks.forEach((cb) => cb()); - } + this.notify(); return () => { this.count--; debugLog(`[ReferenceCounter] 引用计数减少: ${this.count}`); - if (this.count === 0) { - // 从1到0,停止数据收集 - this.callbacks.forEach((cb) => cb()); - } + this.notify(); }; } onCountChange(callback: () => void) { - this.callbacks.push(callback); + this.callbacks.add(callback); + return () => this.callbacks.delete(callback); } getCount(): number { @@ -63,225 +53,405 @@ class ReferenceCounter { } } -// 智能数据采样器 -class TrafficDataSampler { - private rawBuffer: ITrafficDataPoint[] = []; - private compressedBuffer: ICompressedDataPoint[] = []; - private config: ISamplingConfig; - private compressionQueue: ITrafficDataPoint[] = []; +const WORKER_CONFIG = { + rawDataMinutes: 10, + compressedDataMinutes: 60, + compressionRatio: 5, + snapshotIntervalMs: 250, + defaultRangeMinutes: 10, +}; - constructor(config: ISamplingConfig) { - this.config = config; - } +class InlineTrafficMonitor { + private config = { ...WORKER_CONFIG }; + private sampler = new TrafficDataSampler(this.config); + private throttleTimer: ReturnType | null = null; + private currentRange = this.config.defaultRangeMinutes; + private lastTimestamp: number | undefined; - addDataPoint(point: ITrafficDataPoint): void { - // 添加到原始缓冲区 - this.rawBuffer.push(point); + constructor( + private emit: (snapshot: ITrafficWorkerSnapshotMessage) => void, + ) {} - // 清理过期的原始数据 - const rawCutoff = Date.now() - this.config.rawDataMinutes * 60 * 1000; - this.rawBuffer = this.rawBuffer.filter((p) => p.timestamp > rawCutoff); - - // 添加到压缩队列 - this.compressionQueue.push(point); - - // 当压缩队列达到压缩比例时,执行压缩 - if (this.compressionQueue.length >= this.config.compressionRatio) { - this.compressData(); - } - - // 清理过期的压缩数据 - const compressedCutoff = - Date.now() - this.config.compressedDataMinutes * 60 * 1000; - this.compressedBuffer = this.compressedBuffer.filter( - (p) => p.timestamp > compressedCutoff, - ); - } - - private compressData(): void { - if (this.compressionQueue.length === 0) return; - - // 计算平均值进行压缩 - const totalUp = this.compressionQueue.reduce((sum, p) => sum + p.up, 0); - const totalDown = this.compressionQueue.reduce((sum, p) => sum + p.down, 0); - const avgTimestamp = - this.compressionQueue.reduce((sum, p) => sum + p.timestamp, 0) / - this.compressionQueue.length; - - const compressedPoint: ICompressedDataPoint = { - up: totalUp / this.compressionQueue.length, - down: totalDown / this.compressionQueue.length, - timestamp: avgTimestamp, - samples: this.compressionQueue.length, - }; - - this.compressedBuffer.push(compressedPoint); - this.compressionQueue = []; - - debugLog(`[DataSampler] 压缩了 ${compressedPoint.samples} 个数据点`); - } - - getDataForTimeRange(minutes: number): ITrafficDataPoint[] { - const cutoff = Date.now() - minutes * 60 * 1000; - - // 如果请求的时间范围在原始数据范围内,直接返回原始数据 - if (minutes <= this.config.rawDataMinutes) { - return this.rawBuffer.filter((p) => p.timestamp > cutoff); - } - - // 否则组合原始数据和压缩数据 - const rawData = this.rawBuffer.filter((p) => p.timestamp > cutoff); - const compressedData = this.compressedBuffer - .filter( - (p) => - p.timestamp > cutoff && - p.timestamp <= Date.now() - this.config.rawDataMinutes * 60 * 1000, - ) - .map((p) => ({ - up: p.up, - down: p.down, - timestamp: p.timestamp, - name: new Date(p.timestamp).toLocaleTimeString("en-US", { - hour12: false, - hour: "2-digit", - minute: "2-digit", - second: "2-digit", - }), - })); - - return [...compressedData, ...rawData].sort( - (a, b) => a.timestamp - b.timestamp, - ); - } - - getStats() { - return { - rawBufferSize: this.rawBuffer.length, - compressedBufferSize: this.compressedBuffer.length, - compressionQueueSize: this.compressionQueue.length, - totalMemoryPoints: this.rawBuffer.length + this.compressedBuffer.length, - }; - } - - clear(): void { - this.rawBuffer = []; - this.compressedBuffer = []; - this.compressionQueue = []; - } -} - -// 全局单例 -const refCounter = new ReferenceCounter(); -let globalSampler: TrafficDataSampler | null = null; - -/** - * 增强的流量监控Hook - 支持数据压缩、采样和引用计数 - */ -export const useTrafficMonitorEnhanced = () => { - // 初始化采样器 - if (!globalSampler) { - globalSampler = new TrafficDataSampler({ - rawDataMinutes: 10, // 原始数据保持10分钟 - compressedDataMinutes: 60, // 压缩数据保持1小时 - compressionRatio: 5, // 每5个原始点压缩成1个 + start(rangeMinutes?: number) { + this.currentRange = rangeMinutes ?? this.currentRange; + this.handle({ + type: "init", + config: { + ...this.config, + defaultRangeMinutes: this.currentRange, + }, }); } - const [, forceRender] = useReducer((version: number) => version + 1, 0); - const cleanupRef = useRef<(() => void) | null>(null); + stop() { + if (this.throttleTimer !== null) { + clearTimeout(this.throttleTimer); + this.throttleTimer = null; + } + this.sampler.clear(); + this.lastTimestamp = undefined; + } - const bumpRenderVersion = useCallback(() => { - forceRender(); - }, []); + handle(message: TrafficWorkerRequestMessage) { + switch (message.type) { + case "init": { + this.config = { ...message.config }; + this.sampler = new TrafficDataSampler(this.config); + this.currentRange = message.config.defaultRangeMinutes; + this.emitSnapshot("init"); + break; + } + case "append": { + const timestamp = message.payload.timestamp ?? Date.now(); + const dataPoint: ITrafficDataPoint = { + up: message.payload.up || 0, + down: message.payload.down || 0, + timestamp, + name: formatTrafficName(timestamp), + }; - // 注册引用计数 + this.lastTimestamp = timestamp; + this.sampler.addDataPoint(dataPoint); + this.scheduleSnapshot("append-throttle"); + break; + } + case "clear": { + this.sampler.clear(); + this.lastTimestamp = undefined; + this.emitSnapshot("clear"); + break; + } + case "setRange": { + if (this.currentRange !== message.minutes) { + this.currentRange = message.minutes; + this.emitSnapshot("range-change"); + } + break; + } + case "requestSnapshot": { + this.emitSnapshot("request"); + break; + } + default: + break; + } + } + + private emitSnapshot(reason: ITrafficWorkerSnapshotMessage["reason"]) { + const dataPoints = this.sampler.getDataForTimeRange(this.currentRange); + const availableDataPoints = this.sampler.getDataForTimeRange( + this.config.compressedDataMinutes, + ); + + this.emit({ + type: "snapshot", + dataPoints, + availableDataPoints, + samplerStats: this.sampler.getStats(), + rangeMinutes: this.currentRange, + lastTimestamp: this.lastTimestamp, + reason, + }); + } + + private scheduleSnapshot(reason: ITrafficWorkerSnapshotMessage["reason"]) { + if (this.throttleTimer !== null) return; + this.throttleTimer = setTimeout(() => { + this.throttleTimer = null; + this.emitSnapshot(reason); + }, this.config.snapshotIntervalMs); + } +} + +class TrafficWorkerClient { + private worker: Worker | null = null; + private inlineMonitor: InlineTrafficMonitor | null = null; + private mode: "worker" | "inline" | null = null; + private listeners = new Set< + (snapshot: ITrafficWorkerSnapshotMessage) => void + >(); + private pendingMessages: TrafficWorkerRequestMessage[] = []; + private ready = false; + private currentRange = WORKER_CONFIG.defaultRangeMinutes; + + start(rangeMinutes?: number) { + if (typeof window === "undefined") { + debugLog("[TrafficWorkerClient] Window not available, skip start"); + return; + } + + this.currentRange = rangeMinutes ?? this.currentRange; + + if (this.ready) return; + + const initMessage: TrafficWorkerRequestMessage = { + type: "init", + config: { + rawDataMinutes: WORKER_CONFIG.rawDataMinutes, + compressedDataMinutes: WORKER_CONFIG.compressedDataMinutes, + compressionRatio: WORKER_CONFIG.compressionRatio, + snapshotIntervalMs: WORKER_CONFIG.snapshotIntervalMs, + defaultRangeMinutes: this.currentRange, + }, + }; + + if (typeof Worker !== "undefined") { + try { + this.worker = new Worker( + new URL("../services/traffic-monitor-worker.ts", import.meta.url), + { type: "module" }, + ); + this.mode = "worker"; + + this.worker.onmessage = ( + event: MessageEvent, + ) => { + const message = event.data; + if (message.type === "snapshot") { + this.listeners.forEach((listener) => listener(message)); + } + }; + + this.worker.onerror = (error) => { + debugLog(`[TrafficWorkerClient] Worker error: ${String(error)}`); + }; + + this.ready = true; + this.post(initMessage); + this.flushQueue(); + return; + } catch (error) { + debugLog( + `[TrafficWorkerClient] Worker initialization failed, falling back to inline sampler: ${String(error)}`, + ); + this.worker = null; + this.mode = null; + } + } else { + debugLog( + "[TrafficWorkerClient] Worker not supported, using inline sampler", + ); + } + + this.startInline(initMessage); + } + + private startInline(initMessage: TrafficWorkerRequestMessage) { + this.inlineMonitor = new InlineTrafficMonitor((snapshot) => + this.listeners.forEach((listener) => listener(snapshot)), + ); + this.mode = "inline"; + this.ready = true; + this.post(initMessage); + this.flushQueue(); + } + + stop() { + if (this.worker) { + this.worker.terminate(); + } + if (this.inlineMonitor) { + this.inlineMonitor.stop(); + } + this.worker = null; + this.inlineMonitor = null; + this.mode = null; + this.ready = false; + this.pendingMessages = []; + } + + onSnapshot(listener: (snapshot: ITrafficWorkerSnapshotMessage) => void) { + this.listeners.add(listener); + return () => { + this.listeners.delete(listener); + }; + } + + private post(message: TrafficWorkerRequestMessage) { + if (!this.ready) { + this.pendingMessages.push(message); + return; + } + + if (this.mode === "worker" && this.worker) { + this.worker.postMessage(message); + return; + } + + if (this.mode === "inline" && this.inlineMonitor) { + this.inlineMonitor.handle(message); + return; + } + + this.pendingMessages.push(message); + } + + private flushQueue() { + if (!this.ready || this.pendingMessages.length === 0) { + return; + } + const queued = [...this.pendingMessages]; + this.pendingMessages = []; + queued.forEach((message) => { + this.post(message); + }); + } + + private ensureStarted() { + if (!this.ready) { + this.start(this.currentRange); + } + } + + appendData(traffic: Traffic) { + this.ensureStarted(); + this.post({ + type: "append", + payload: { + up: traffic?.up ?? 0, + down: traffic?.down ?? 0, + timestamp: Date.now(), + }, + }); + } + + clearData() { + this.ensureStarted(); + this.post({ type: "clear" }); + } + + setRange(minutes: number) { + this.currentRange = minutes; + this.post({ type: "setRange", minutes }); + } + + requestSnapshot() { + this.ensureStarted(); + this.post({ type: "requestSnapshot" }); + } +} + +const refCounter = new ReferenceCounter(); +let workerClient: TrafficWorkerClient | null = null; +const getWorkerClient = () => { + if (!workerClient) { + workerClient = new TrafficWorkerClient(); + } + return workerClient; +}; + +const EMPTY_STATS: ISamplerStats = { + rawBufferSize: 0, + compressedBufferSize: 0, + compressionQueueSize: 0, + totalMemoryPoints: 0, +}; + +/** + * 增强的流量监控Hook - Web Worker驱动的数据采样与压缩 + */ +export const useTrafficMonitorEnhanced = (options?: { + subscribe?: boolean; +}) => { + const subscribeToSnapshots = options?.subscribe ?? true; + const [latestSnapshot, setLatestSnapshot] = useState<{ + availableDataPoints: ITrafficDataPoint[]; + samplerStats: ISamplerStats; + lastTimestamp?: number; + }>({ + availableDataPoints: [], + samplerStats: EMPTY_STATS, + lastTimestamp: undefined, + }); + const [rangeMinutes, setRangeMinutes] = useState( + WORKER_CONFIG.defaultRangeMinutes, + ); + const [now, setNow] = useState(() => Date.now()); + const [, forceRefCountRender] = useReducer((value) => value + 1, 0); + + const clientRef = useRef(getWorkerClient()); + const currentRangeRef = useRef(WORKER_CONFIG.defaultRangeMinutes); + + // 注册引用计数与Worker生命周期 useEffect(() => { - debugLog("[TrafficMonitorEnhanced] 组件挂载,注册引用计数"); + const client = getWorkerClient(); + clientRef.current = client; + + const stopWatchRefCount = refCounter.onCountChange(() => + forceRefCountRender(), + ); const cleanup = refCounter.increment(); - cleanupRef.current = cleanup; + client.start(currentRangeRef.current); + + let unsubscribe: (() => void) | undefined; + if (subscribeToSnapshots) { + unsubscribe = client.onSnapshot((message) => { + setLatestSnapshot({ + availableDataPoints: + message.availableDataPoints ?? message.dataPoints, + samplerStats: message.samplerStats, + lastTimestamp: message.lastTimestamp, + }); + }); + + client.requestSnapshot(); + } return () => { - debugLog("[TrafficMonitorEnhanced] 组件卸载,清理引用计数"); + unsubscribe?.(); + stopWatchRefCount(); cleanup(); - cleanupRef.current = null; - }; - }, []); - - // 设置引用计数变化回调 - useEffect(() => { - const handleCountChange = () => { - debugLog( - `[TrafficMonitorEnhanced] 引用计数变化: ${refCounter.getCount()}`, - ); if (refCounter.getCount() === 0) { - debugLog("[TrafficMonitorEnhanced] 所有组件已卸载,暂停数据收集"); - } else { - debugLog("[TrafficMonitorEnhanced] 开始数据收集"); + client.stop(); } }; + }, [subscribeToSnapshots]); - refCounter.onCountChange(handleCountChange); - }, []); + // Periodically refresh "now" so idle streams age out of the selected window when subscribed + useEffect(() => { + if (!subscribeToSnapshots) return; - // 获取指定时间范围的数据 - const getDataForTimeRange = useCallback( - (minutes: number): ITrafficDataPoint[] => { - if (!globalSampler) return []; - return globalSampler.getDataForTimeRange(minutes); - }, - [], - ); + const timer = window.setInterval(() => { + setNow(Date.now()); + }, 1000); + + return () => window.clearInterval(timer); + }, [subscribeToSnapshots]); // 添加流量数据 const appendData = useCallback((traffic: Traffic) => { - if (globalSampler) { - // 添加到采样器 - const timestamp = Date.now(); - const dataPoint: ITrafficDataPoint = { - up: traffic?.up || 0, - down: traffic?.down || 0, - timestamp, - name: new Date(timestamp).toLocaleTimeString("en-US", { - hour12: false, - hour: "2-digit", - minute: "2-digit", - second: "2-digit", - }), - }; - globalSampler.addDataPoint(dataPoint); - } + clientRef.current?.appendData(traffic); + }, []); + + // 请求不同时间范围的数据 + const requestRange = useCallback((minutes: number) => { + currentRangeRef.current = minutes; + setRangeMinutes(minutes); + clientRef.current?.setRange(minutes); }, []); // 清空数据 const clearData = useCallback(() => { - if (globalSampler) { - globalSampler.clear(); - bumpRenderVersion(); - } - }, [bumpRenderVersion]); - - // 获取采样器统计信息 - const getSamplerStats = useCallback(() => { - return ( - globalSampler?.getStats() || { - rawBufferSize: 0, - compressedBufferSize: 0, - compressionQueueSize: 0, - totalMemoryPoints: 0, - } - ); + clientRef.current?.clearData(); }, []); + const filteredDataPoints = useMemo(() => { + const sourceData = latestSnapshot.availableDataPoints; + if (sourceData.length === 0) return []; + + const cutoff = now - rangeMinutes * 60 * 1000; + return sourceData.filter((point) => point.timestamp > cutoff); + }, [latestSnapshot.availableDataPoints, rangeMinutes, now]); + return { - // 图表数据管理 graphData: { - dataPoints: globalSampler?.getDataForTimeRange(60) || [], // 默认获取1小时数据 - getDataForTimeRange, + dataPoints: filteredDataPoints, + currentRangeMinutes: rangeMinutes, + requestRange, appendData, clearData, }, - // 性能统计 - samplerStats: getSamplerStats(), + samplerStats: latestSnapshot.samplerStats, referenceCount: refCounter.getCount(), }; }; diff --git a/src/services/traffic-monitor-worker.ts b/src/services/traffic-monitor-worker.ts new file mode 100644 index 000000000..f88462f29 --- /dev/null +++ b/src/services/traffic-monitor-worker.ts @@ -0,0 +1,113 @@ +import type { + ISamplingConfig, + ITrafficDataPoint, + ITrafficWorkerSnapshotMessage, + TrafficWorkerRequestMessage, +} from "../types/traffic-monitor"; +import { + TrafficDataSampler, + formatTrafficName, +} from "../utils/traffic-sampler"; + +const DEFAULT_CONFIG: ISamplingConfig & { + snapshotIntervalMs: number; + defaultRangeMinutes: number; +} = { + rawDataMinutes: 10, + compressedDataMinutes: 60, + compressionRatio: 5, + snapshotIntervalMs: 250, + defaultRangeMinutes: 10, +}; + +interface WorkerScope { + postMessage: (message: unknown) => void; + onmessage: + | ((event: MessageEvent) => void) + | null; + setTimeout: typeof setTimeout; + clearTimeout: typeof clearTimeout; +} + +const ctx: WorkerScope = self as unknown as WorkerScope; + +let config = { ...DEFAULT_CONFIG }; +let sampler = new TrafficDataSampler(config); +let currentRangeMinutes = config.defaultRangeMinutes; +let throttleTimer: ReturnType | null = null; +let lastTimestamp: number | undefined; + +const broadcastSnapshot = (reason: ITrafficWorkerSnapshotMessage["reason"]) => { + const dataPoints = sampler.getDataForTimeRange(currentRangeMinutes); + const availableDataPoints = sampler.getDataForTimeRange( + config.compressedDataMinutes, + ); + const samplerStats = sampler.getStats(); + + const message: ITrafficWorkerSnapshotMessage = { + type: "snapshot", + dataPoints, + availableDataPoints, + samplerStats, + rangeMinutes: currentRangeMinutes, + lastTimestamp, + reason, + }; + + ctx.postMessage(message); +}; + +const scheduleSnapshot = (reason: ITrafficWorkerSnapshotMessage["reason"]) => { + if (throttleTimer !== null) return; + throttleTimer = ctx.setTimeout(() => { + throttleTimer = null; + broadcastSnapshot(reason); + }, config.snapshotIntervalMs); +}; + +ctx.onmessage = (event: MessageEvent) => { + const message = event.data; + + switch (message.type) { + case "init": { + config = { ...message.config }; + sampler = new TrafficDataSampler(config); + currentRangeMinutes = message.config.defaultRangeMinutes; + broadcastSnapshot("init"); + break; + } + case "append": { + const timestamp = message.payload.timestamp ?? Date.now(); + const dataPoint: ITrafficDataPoint = { + up: message.payload.up || 0, + down: message.payload.down || 0, + timestamp, + name: formatTrafficName(timestamp), + }; + + lastTimestamp = timestamp; + sampler.addDataPoint(dataPoint); + scheduleSnapshot("append-throttle"); + break; + } + case "clear": { + sampler.clear(); + lastTimestamp = undefined; + broadcastSnapshot("clear"); + break; + } + case "setRange": { + if (currentRangeMinutes !== message.minutes) { + currentRangeMinutes = message.minutes; + broadcastSnapshot("range-change"); + } + break; + } + case "requestSnapshot": { + broadcastSnapshot("request"); + break; + } + default: + break; + } +}; diff --git a/src/types/traffic-monitor.ts b/src/types/traffic-monitor.ts new file mode 100644 index 000000000..d0b104517 --- /dev/null +++ b/src/types/traffic-monitor.ts @@ -0,0 +1,81 @@ +export interface ITrafficDataPoint { + up: number; + down: number; + timestamp: number; + name: string; +} + +export interface ISamplingConfig { + rawDataMinutes: number; + compressedDataMinutes: number; + compressionRatio: number; +} + +export interface ISamplerStats { + rawBufferSize: number; + compressedBufferSize: number; + compressionQueueSize: number; + totalMemoryPoints: number; +} + +export interface ITrafficWorkerInitMessage { + type: "init"; + config: ISamplingConfig & { + snapshotIntervalMs: number; + defaultRangeMinutes: number; + }; +} + +export interface ITrafficWorkerAppendMessage { + type: "append"; + payload: { + up: number; + down: number; + timestamp?: number; + }; +} + +export interface ITrafficWorkerClearMessage { + type: "clear"; +} + +export interface ITrafficWorkerSetRangeMessage { + type: "setRange"; + minutes: number; +} + +export interface ITrafficWorkerRequestSnapshotMessage { + type: "requestSnapshot"; +} + +export type TrafficWorkerRequestMessage = + | ITrafficWorkerInitMessage + | ITrafficWorkerAppendMessage + | ITrafficWorkerClearMessage + | ITrafficWorkerSetRangeMessage + | ITrafficWorkerRequestSnapshotMessage; + +export interface ITrafficWorkerSnapshotMessage { + type: "snapshot"; + dataPoints: ITrafficDataPoint[]; + availableDataPoints: ITrafficDataPoint[]; + samplerStats: ISamplerStats; + rangeMinutes: number; + lastTimestamp?: number; + reason: + | "init" + | "interval" + | "range-change" + | "request" + | "append-throttle" + | "clear"; +} + +export interface ITrafficWorkerLogMessage { + type: "log"; + message: string; +} + +export type TrafficWorkerResponseMessage = + | ITrafficWorkerSnapshotMessage + | ITrafficWorkerLogMessage; diff --git a/src/utils/traffic-sampler.ts b/src/utils/traffic-sampler.ts new file mode 100644 index 000000000..a6eaba967 --- /dev/null +++ b/src/utils/traffic-sampler.ts @@ -0,0 +1,107 @@ +import type { + ISamplerStats, + ISamplingConfig, + ITrafficDataPoint, +} from "@/types/traffic-monitor"; + +interface ICompressedDataPoint { + up: number; + down: number; + timestamp: number; + samples: number; +} + +export const formatTrafficName = (timestamp: number) => + new Date(timestamp).toLocaleTimeString("en-US", { + hour12: false, + hour: "2-digit", + minute: "2-digit", + second: "2-digit", + }); + +export class TrafficDataSampler { + private rawBuffer: ITrafficDataPoint[] = []; + private compressedBuffer: ICompressedDataPoint[] = []; + private compressionQueue: ITrafficDataPoint[] = []; + + constructor(private config: ISamplingConfig) {} + + addDataPoint(point: ITrafficDataPoint) { + this.rawBuffer.push(point); + + const rawCutoff = Date.now() - this.config.rawDataMinutes * 60 * 1000; + this.rawBuffer = this.rawBuffer.filter((p) => p.timestamp > rawCutoff); + + this.compressionQueue.push(point); + if (this.compressionQueue.length >= this.config.compressionRatio) { + this.compressData(); + } + + const compressedCutoff = + Date.now() - this.config.compressedDataMinutes * 60 * 1000; + this.compressedBuffer = this.compressedBuffer.filter( + (p) => p.timestamp > compressedCutoff, + ); + } + + private compressData() { + if (this.compressionQueue.length === 0) return; + + const totalUp = this.compressionQueue.reduce((sum, p) => sum + p.up, 0); + const totalDown = this.compressionQueue.reduce((sum, p) => sum + p.down, 0); + const avgTimestamp = + this.compressionQueue.reduce((sum, p) => sum + p.timestamp, 0) / + this.compressionQueue.length; + + const compressedPoint: ICompressedDataPoint = { + up: totalUp / this.compressionQueue.length, + down: totalDown / this.compressionQueue.length, + timestamp: avgTimestamp, + samples: this.compressionQueue.length, + }; + + this.compressedBuffer.push(compressedPoint); + this.compressionQueue = []; + } + + getDataForTimeRange(minutes: number): ITrafficDataPoint[] { + const cutoff = Date.now() - minutes * 60 * 1000; + const rawData = this.rawBuffer.filter((p) => p.timestamp > cutoff); + + if (minutes <= this.config.rawDataMinutes) { + return rawData; + } + + const compressedData = this.compressedBuffer + .filter( + (p) => + p.timestamp > cutoff && + p.timestamp <= Date.now() - this.config.rawDataMinutes * 60 * 1000, + ) + .map((p) => ({ + up: p.up, + down: p.down, + timestamp: p.timestamp, + name: formatTrafficName(p.timestamp), + })); + + return [...compressedData, ...rawData].sort( + (a, b) => a.timestamp - b.timestamp, + ); + } + + getStats(): ISamplerStats { + return { + rawBufferSize: this.rawBuffer.length, + compressedBufferSize: this.compressedBuffer.length, + compressionQueueSize: this.compressionQueue.length, + totalMemoryPoints: this.rawBuffer.length + this.compressedBuffer.length, + }; + } + + clear() { + this.rawBuffer = []; + this.compressedBuffer = []; + this.compressionQueue = []; + } +}