gx

chenyc

5 天以前 747a86bb94006aaca721cfc0c0ce7061643a9ea6

 metric-aggregator.js

@@ -12,7 +12,7 @@
    this.timer = null;
    this.started = false;
    this.flushing = false;
    this.flushQueued = false;
    this.queuedFlushOptions = null;
  }

  start() {
@@ -47,8 +47,15 @@

    const entry = this.getOrCreateEntry(device);
    entry.device = device;
    const now = Date.now();

    // 首次进入脏状态时记录时间，便于控制“缓存满一轮再发送”。
    if (!entry.dirty) {
      entry.dirtySinceAt = now;
    }

    entry.dirty = true;
    entry.lastUpdateAt = Date.now();
    entry.lastUpdateAt = now;

    if (this.includeDeviceIdField) {
      entry.payload[this.deviceIdField] = device.deviceId;
@@ -69,6 +76,7 @@
        device,
        payload: {},
        dirty: false,
        dirtySinceAt: 0,
        firstSeenAt: Date.now(),
        lastUpdateAt: 0,
        lastFlushAt: 0,
@@ -98,7 +106,7 @@

    this.timer = setTimeout(() => {
      this.flush({ reason: 'timer' }).catch((error) => {
        this.logger.error(`[APP] Aggregator flush failed: ${error.message}`);
        this.logger.error(`[APP] 聚合器刷新失败: ${error.message}`);
      });
    }, this.computeDelayMs());

@@ -107,12 +115,36 @@
    }
  }

  mergeQueuedFlushOptions(options = {}) {
    const nextOptions = {
      reason: options.reason || 'manual',
      deviceId: options.deviceId || '',
    };

    if (!this.queuedFlushOptions) {
      this.queuedFlushOptions = nextOptions;
      return;
    }

    const currentOptions = this.queuedFlushOptions;
    const currentIsTimer = currentOptions.reason === 'timer' || currentOptions.reason === 'queued';
    const nextIsTimer = nextOptions.reason === 'timer' || nextOptions.reason === 'queued';

    if (currentIsTimer && !nextIsTimer) {
      currentOptions.reason = nextOptions.reason;
    }

    if (!currentOptions.deviceId || !nextOptions.deviceId || currentOptions.deviceId !== nextOptions.deviceId) {
      currentOptions.deviceId = '';
    }
  }

  async flush(options = {}) {
    const reason = options.reason || 'manual';
    const deviceId = options.deviceId || '';

    if (this.flushing) {
      this.flushQueued = true;
      this.mergeQueuedFlushOptions({ reason, deviceId });
      return false;
    }

@@ -133,10 +165,20 @@
          return false;
        }

        // 定时触发时要求数据至少缓存满一个周期，避免刚收到就被单独发出。
        if ((reason === 'timer' || reason === 'queued') && entry.dirtySinceAt > 0) {
          const dirtyAgeMs = Date.now() - entry.dirtySinceAt;

          if (dirtyAgeMs < this.flushIntervalMs) {
            return false;
          }
        }

        return true;
      });

      for (const entry of entries) {
        const flushUpdateAt = entry.lastUpdateAt;
        const payload = {
          ...entry.payload,
        };
@@ -157,10 +199,14 @@
            continue;
          }

          entry.dirty = false;
          if (entry.lastUpdateAt === flushUpdateAt) {
            entry.dirty = false;
            entry.dirtySinceAt = 0;
          }

          entry.lastFlushAt = Date.now();
        } catch (error) {
          this.logger.error(`[APP] Aggregator device flush failed deviceId=${entry.device.deviceId}: ${error.message}`);
          this.logger.error(`[APP] 聚合器设备刷新失败 deviceId=${entry.device.deviceId}: ${error.message}`);
        }
      }

@@ -172,9 +218,10 @@
        this.scheduleNextFlush();
      }

      if (this.flushQueued) {
        this.flushQueued = false;
        await this.flush({ reason: 'queued' });
      if (this.queuedFlushOptions) {
        const queuedOptions = this.queuedFlushOptions;
        this.queuedFlushOptions = null;
        await this.flush(queuedOptions);
      }
    }
  }