import { readFileSync } from 'node:fs'; import assert from 'node:assert/strict'; import { test } from 'node:test'; import ts from 'typescript'; const source = readFileSync(new URL('../src/lib/stream.ts', import.meta.url), 'utf8'); const compiled = ts.transpileModule(source, { compilerOptions: { module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2022 } }).outputText; const exports = {}; new Function('require', 'exports', compiled)(name => { assert.equal(name, './types'); return { FRAME_HEADER_LEN: 16 }; }, exports); const { StreamClient } = exports; globalThis.window = { clearTimeout, setTimeout }; const tick = () => new Promise(resolve => setImmediate(resolve)); const bitmap = () => ({ closed: false, close() { this.closed = true; } }); function frame(id, camera = 42) { const bytes = new ArrayBuffer(20); const header = new DataView(bytes); header.setUint32(0, camera, true); header.setUint32(4, id, true); header.setUint16(8, 640, true); header.setUint16(10, 480, true); header.setUint32(12, 4, true); return bytes; } function client(now, wallNow) { return new StreamClient({ onMessage() {}, onConnection() {} },now,wallNow); } function deferredDecoder() { const requests = []; globalThis.createImageBitmap = () => new Promise((resolve, reject) => requests.push({ resolve, reject })); return requests; } test('a slow decoder keeps only the latest pending frame for each camera', async () => { const requests = deferredDecoder(); const rx = client(); const finished = rx.onBinary(frame(1)); for (let id = 2; id <= 100; id++) await rx.onBinary(frame(id)); assert.equal(requests.length, 1); assert.equal(rx.pending.size, 1); const first = bitmap(); requests[0].resolve(first); await tick(); assert.equal(requests.length, 2); requests[1].resolve(bitmap()); await finished; assert.equal(rx.latest(42).frameId, 100); assert.equal(rx.counters.skippedDecodes, 98); assert.equal(first.closed, true); rx.close(); }); test('two cameras decode independently without queues growing behind either', async () => { const requests = deferredDecoder(); const rx = client(); const a = rx.onBinary(frame(1, 1)); const b = rx.onBinary(frame(2, 2)); assert.equal(requests.length, 2); requests[1].resolve(bitmap()); await b; assert.equal(rx.latest(2).frameId, 2); assert.equal(rx.latest(1), undefined); requests[0].resolve(bitmap()); await a; rx.close(); }); test('closing during decode releases the late bitmap and discards pending work', async () => { const requests = deferredDecoder(); const rx = client(); const work = rx.onBinary(frame(1)); await rx.onBinary(frame(2)); rx.close(); const late = bitmap(); requests[0].resolve(late); await work; assert.equal(late.closed, true); assert.equal(requests.length, 1); assert.equal(rx.latest(42), undefined); }); test('a corrupt JPEG does not stop the next frame from being decoded', async () => { const requests = deferredDecoder(); const rx = client(); const work = rx.onBinary(frame(1)); await rx.onBinary(frame(2)); requests[0].reject(new Error('bad JPEG')); await tick(); requests[1].resolve(bitmap()); await work; assert.equal(rx.counters.decodeFailures, 1); assert.equal(rx.latest(42).frameId, 2); rx.close(); }); test('late frames are rejected but a reboot after silence can restart the counter', () => { const rx = client(); rx.injectFrame(42, 100, 640, 480, bitmap()); const old = bitmap(); rx.injectFrame(42, 99, 640, 480, old); assert.equal(old.closed, true); assert.equal(rx.latest(42).frameId, 100); rx.latest(42).receivedAt -= 2000; rx.injectFrame(42, 1, 640, 480, bitmap()); assert.equal(rx.latest(42).frameId, 1); rx.close(); }); test('a WebView backlog is discarded before JPEG decoding using native host timing', async () => { const requests = deferredDecoder(); const rx = client(() => 0, () => 2000); rx.onTiming({camera:42,frameId:1,publishedUnixMs:1500}); await rx.onBinary(frame(1)); assert.equal(requests.length,0); assert.equal(rx.counters.staleFrames,1); rx.onTiming({camera:42,frameId:2,publishedUnixMs:2000}); const work = rx.onBinary(frame(2)); requests[0].resolve(bitmap());await work; assert.equal(rx.latest(42).frameId,2); rx.close(); }); test('a blocked decoder releases its old result and resumes with the fresh pending frame', async () => { let clock=0; const requests=deferredDecoder(),rx=client(()=>clock); const work=rx.onBinary(frame(1)); clock=110;await rx.onBinary(frame(2)); const old=bitmap();requests[0].resolve(old);await tick(); assert.equal(old.closed,true); assert.equal(rx.latest(42),undefined); requests[1].resolve(bitmap());await work; assert.equal(rx.latest(42).frameId,2); assert.equal(rx.counters.staleFrames,1);rx.close(); }); test('pending frames that expired during a decoder stall never reach the decoder', async () => { let clock=0; const requests=deferredDecoder(),rx=client(()=>clock); const work=rx.onBinary(frame(1));await rx.onBinary(frame(2)); clock=200;const old=bitmap();requests[0].resolve(old);await work; assert.equal(old.closed,true);assert.equal(requests.length,1); assert.equal(rx.counters.staleFrames,2);assert.equal(rx.latest(42),undefined);rx.close(); });