diff --git a/packages/core/src/lib/file_processor.ts b/packages/core/src/lib/file_processor.ts index f0b0347b..a1471518 100644 --- a/packages/core/src/lib/file_processor.ts +++ b/packages/core/src/lib/file_processor.ts @@ -513,7 +513,7 @@ class FileProcessor { let error_string = '' // TODO use the input timestamp in a progress meter - for await (const line of this.#readlines(proc.stderr)) { + for await (const line of FileProcessor.read_lines(proc.stderr)) { if (line.includes('Parsed_showinfo') && line.includes(' n: ')) { const pts_time_str = line.match(/pts_time:(?\d+([.]\d*)?)/)?.groups?.pts_time if (pts_time_str === undefined) { @@ -649,12 +649,14 @@ class FileProcessor { } } - async * #readlines(stream: ReadableStream) { + static async * read_lines(stream: ReadableStream) { const reader = stream.pipeThrough(new TextDecoderStream()).getReader() let buffer = '' - let read_result = await reader.read() - while (!read_result.done) { - read_result = await reader.read() + while (true) { + const read_result = await reader.read() + if (read_result.done) { + break + } buffer += read_result.value ?? '' const lines = buffer.split('\n') @@ -662,9 +664,7 @@ class FileProcessor { for (let i = 0; i < lines.length - 1; i++) { yield lines[i] } - if (lines.length > 0) { - buffer = lines.at(-1)! - } + buffer = lines.at(-1)! } if (buffer.length > 0) { diff --git a/packages/core/test/file_processor.test.ts b/packages/core/test/file_processor.test.ts new file mode 100644 index 00000000..b6ce296d --- /dev/null +++ b/packages/core/test/file_processor.test.ts @@ -0,0 +1,48 @@ +import { test } from 'forager-test' +import { FileProcessor } from '~/lib/file_processor.ts' + +function stream_of(chunks: string[]): ReadableStream { + const encoder = new TextEncoder() + return new ReadableStream({ + start(controller) { + for (const chunk of chunks) { + controller.enqueue(encoder.encode(chunk)) + } + controller.close() + } + }) +} + +test('FileProcessor.read_lines', async (ctx) => { + await ctx.subtest('yields every line, including lines in the first chunk', async () => { + const lines = await Array.fromAsync(FileProcessor.read_lines(stream_of([ + 'first\nsecond\n', + 'third\nfourth\n', + ]))) + ctx.assert.equals(lines, ['first', 'second', 'third', 'fourth']) + }) + + await ctx.subtest('reads a single chunk stream', async () => { + const lines = await Array.fromAsync(FileProcessor.read_lines(stream_of(['only line\n']))) + ctx.assert.equals(lines, ['only line']) + }) + + await ctx.subtest('joins lines split across chunk boundaries', async () => { + const lines = await Array.fromAsync(FileProcessor.read_lines(stream_of([ + 'frame ', + 'one\nframe ', + 'two', + ]))) + ctx.assert.equals(lines, ['frame one', 'frame two']) + }) + + await ctx.subtest('yields a trailing line without a newline', async () => { + const lines = await Array.fromAsync(FileProcessor.read_lines(stream_of(['first\nno trailing newline']))) + ctx.assert.equals(lines, ['first', 'no trailing newline']) + }) + + await ctx.subtest('yields nothing for an empty stream', async () => { + const lines = await Array.fromAsync(FileProcessor.read_lines(stream_of([]))) + ctx.assert.equals(lines, []) + }) +})