| | |
| | | F: 24.6, |
| | | }); |
| | | }); |
| | | |
| | | it('waits a full flush interval before timer flushes newly cached data', async () => { |
| | | const originalNow = Date.now; |
| | | let now = 1_000; |
| | | Date.now = () => now; |
| | | |
| | | try { |
| | | const flushed = []; |
| | | const aggregator = new MetricAggregator({ |
| | | mode: 'batch', |
| | | flushIntervalMs: 60_000, |
| | | onFlush: async (_device, payload) => { |
| | | flushed.push(payload); |
| | | return true; |
| | | }, |
| | | }); |
| | | |
| | | aggregator.ingest({ deviceId: 'JH-001' }, { F: 24.6 }); |
| | | |
| | | now = 30_000; |
| | | await aggregator.flush({ reason: 'timer' }); |
| | | assert.strictEqual(flushed.length, 0); |
| | | |
| | | now = 61_000; |
| | | await aggregator.flush({ reason: 'timer' }); |
| | | assert.strictEqual(flushed.length, 1); |
| | | assert.deepStrictEqual(flushed[0], { |
| | | n: 'JH-001', |
| | | F: 24.6, |
| | | }); |
| | | } finally { |
| | | Date.now = originalNow; |
| | | } |
| | | }); |
| | | |
| | | it('preserves immediate flush intent when a second flush is queued during an ongoing flush', async () => { |
| | | const flushed = []; |
| | | let releaseFirstFlush = null; |
| | | const firstFlushDone = new Promise((resolve) => { |
| | | releaseFirstFlush = resolve; |
| | | }); |
| | | |
| | | const aggregator = new MetricAggregator({ |
| | | mode: 'batch', |
| | | onFlush: async (_device, payload) => { |
| | | flushed.push(payload); |
| | | |
| | | if (flushed.length === 1) { |
| | | await firstFlushDone; |
| | | } |
| | | |
| | | return true; |
| | | }, |
| | | }); |
| | | |
| | | aggregator.ingest({ deviceId: 'JH-001' }, { F: 24.6 }); |
| | | const firstFlushPromise = aggregator.flush({ reason: 'manual' }); |
| | | |
| | | await new Promise((resolve) => setTimeout(resolve, 0)); |
| | | |
| | | aggregator.ingest({ deviceId: 'JH-001' }, { N: 120, O: 80, P: 89 }); |
| | | const queuedFlushResult = await aggregator.flush({ reason: 'blood-pressure', deviceId: 'JH-001' }); |
| | | |
| | | assert.strictEqual(queuedFlushResult, false); |
| | | |
| | | releaseFirstFlush(); |
| | | await firstFlushPromise; |
| | | |
| | | assert.strictEqual(flushed.length, 2); |
| | | assert.deepStrictEqual(flushed[0], { |
| | | n: 'JH-001', |
| | | F: 24.6, |
| | | }); |
| | | assert.deepStrictEqual(flushed[1], { |
| | | n: 'JH-001', |
| | | F: 24.6, |
| | | N: 120, |
| | | O: 80, |
| | | P: 89, |
| | | }); |
| | | }); |
| | | }); |