diff --git a/src/main/emulator/mjpeg-frame-parser.test.ts b/src/main/emulator/mjpeg-frame-parser.test.ts index 20bab778736..0bddb11a025 100644 --- a/src/main/emulator/mjpeg-frame-parser.test.ts +++ b/src/main/emulator/mjpeg-frame-parser.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from 'vitest' -import { extractJpegFrames } from './mjpeg-frame-parser' +import { MjpegFrameAccumulator, extractJpegFrames } from './mjpeg-frame-parser' const JPEG_A = Buffer.from([0xff, 0xd8, 0x01, 0x02, 0xff, 0xd9]) const JPEG_B = Buffer.from([0xff, 0xd8, 0x03, 0x04, 0xff, 0xd9]) @@ -40,3 +40,49 @@ describe('extractJpegFrames', () => { expect(result.pending).toEqual(Buffer.from([0xff, 0xd8, 0x09])) }) }) + +describe('MjpegFrameAccumulator', () => { + const LONG_JPEG = Buffer.concat([ + Buffer.from([0xff, 0xd8]), + Buffer.from(Array.from({ length: 40 }, (_, index) => (index % 5 === 0 ? 0xff : index))), + Buffer.from([0xff, 0xd9]) + ]) + const stream = Buffer.concat([ + Buffer.from('--frame\r\nContent-Type: image/jpeg\r\n\r\n'), + LONG_JPEG, + Buffer.from('\r\n--frame\r\n\r\n'), + JPEG_B + ]) + + it.each([1, 2, 3, 7, 16, stream.length])( + 'returns the same frames as extractJpegFrames for %i-byte chunks', + (size) => { + const accumulator = new MjpegFrameAccumulator() + const frames: Buffer[] = [] + let pending: Buffer = Buffer.alloc(0) + const expected: Buffer[] = [] + for (let offset = 0; offset < stream.length; offset += size) { + const chunk = stream.subarray(offset, offset + size) + frames.push(...accumulator.push(chunk).map((frame) => Buffer.from(frame))) + const result = extractJpegFrames(pending, chunk) + pending = result.pending + expected.push(...result.frames.map((frame) => Buffer.from(frame))) + } + + expect(frames).toEqual([LONG_JPEG, JPEG_B]) + expect(frames).toEqual(expected) + } + ) + + it('does not retain caller chunk memory while a frame is buffered', () => { + const accumulator = new MjpegFrameAccumulator() + const first = Buffer.from(LONG_JPEG.subarray(0, 10)) + const middle = Buffer.from(LONG_JPEG.subarray(10, 30)) + expect(accumulator.push(first)).toEqual([]) + expect(accumulator.push(middle)).toEqual([]) + first.fill(0) + middle.fill(0) + + expect(accumulator.push(LONG_JPEG.subarray(30))).toEqual([LONG_JPEG]) + }) +}) diff --git a/src/main/emulator/mjpeg-frame-parser.ts b/src/main/emulator/mjpeg-frame-parser.ts index 0bc4ef80d74..cf3837e0700 100644 --- a/src/main/emulator/mjpeg-frame-parser.ts +++ b/src/main/emulator/mjpeg-frame-parser.ts @@ -48,3 +48,41 @@ export function extractJpegFrames( return { frames, pending: Buffer.alloc(0) } } + +// Why: a frame spans many socket chunks; re-joining the growing tail on every chunk is +// quadratic in chunks per frame, so hold chunks until one could complete the frame. +export class MjpegFrameAccumulator { + private parts: Buffer[] = [] + private bytes = 0 + + constructor(private readonly maxPendingBytes = DEFAULT_MAX_PENDING_BYTES) {} + + push(chunk: Buffer): Buffer[] { + if (chunk.length === 0) { + return [] + } + const head = this.parts[0] + if ( + head?.[0] === 0xff && + head[1] === 0xd8 && + !(this.parts.at(-1)?.at(-1) === 0xff && chunk[0] === 0xd9) && + !chunk.includes(JPEG_END) && + this.bytes + chunk.length <= this.maxPendingBytes + ) { + // Copied like extractJpegFrames' pending, so no caller chunk memory is retained. + this.parts.push(Buffer.from(chunk)) + this.bytes += chunk.length + return [] + } + const pending = this.parts.length === 1 ? this.parts[0] : Buffer.concat(this.parts, this.bytes) + const result = extractJpegFrames(pending, chunk, this.maxPendingBytes) + this.parts = result.pending.length > 0 ? [result.pending] : [] + this.bytes = result.pending.length + return result.frames + } + + reset(): void { + this.parts = [] + this.bytes = 0 + } +} diff --git a/src/main/emulator/mjpeg-frame-stream.ts b/src/main/emulator/mjpeg-frame-stream.ts index 315b13f303d..d8ca91cb720 100644 --- a/src/main/emulator/mjpeg-frame-stream.ts +++ b/src/main/emulator/mjpeg-frame-stream.ts @@ -1,6 +1,6 @@ import { request as httpRequest, type ClientRequest, type IncomingMessage } from 'node:http' import { request as httpsRequest } from 'node:https' -import { extractJpegFrames } from './mjpeg-frame-parser' +import { MjpegFrameAccumulator } from './mjpeg-frame-parser' const RECONNECT_DELAY_MS = 1_000 const REQUEST_TIMEOUT_MS = 10_000 @@ -41,7 +41,7 @@ function requestForUrl(url: URL, response: (res: IncomingMessage) => void): Clie } export class MjpegFrameStream { - private pending: Buffer = Buffer.alloc(0) + private readonly accumulator = new MjpegFrameAccumulator() private reconnectTimer: NodeJS.Timeout | null = null private request: ClientRequest | null = null private stopped = false @@ -67,7 +67,7 @@ export class MjpegFrameStream { } this.request?.destroy() this.request = null - this.pending = Buffer.alloc(0) + this.accumulator.reset() } private scheduleReconnect(): void { @@ -114,9 +114,7 @@ export class MjpegFrameStream { } private handleChunk(chunk: Buffer): void { - const result = extractJpegFrames(this.pending, chunk) - this.pending = result.pending - for (const frame of result.frames) { + for (const frame of this.accumulator.push(chunk)) { const now = Date.now() if (this.lastFrameAt > 0 && now - this.lastFrameAt < MIN_FRAME_INTERVAL_MS) { continue