Move TextLineStream to foxts, adopt skipEmptyLines
Some checks failed
Build / Build (push) Has been cancelled
Build / Diff output (push) Has been cancelled
Build / Deploy to Cloudflare Pages (push) Has been cancelled
Build / Deploy to GitHub and GitLab (push) Has been cancelled

This commit is contained in:
SukkaW 2025-03-11 01:05:14 +08:00
parent ac200f67c5
commit c9ebec077f
3 changed files with 4 additions and 84 deletions

View File

@ -2,7 +2,7 @@ import picocolors from 'picocolors';
import { $$fetch, defaultRequestInit, ResponseError } from './fetch-retry';
import { waitWithAbort } from 'foxts/wait';
import { nullthrow } from 'foxts/guard';
import { TextLineStream } from './text-line-transform-stream';
import { TextLineStream } from 'foxts/text-line-stream';
import { ProcessLineStream } from './process-line';
// eslint-disable-next-line sukka/unicorn/custom-error-definition -- typescript is better
@ -32,7 +32,7 @@ export async function fetchAssets(url: string, fallbackUrls: null | undefined |
}
const res = await $$fetch(url, { signal: controller.signal, ...defaultRequestInit });
let stream = nullthrow(res.body, url + ' has an empty body').pipeThrough(new TextDecoderStream()).pipeThrough(new TextLineStream());
let stream = nullthrow(res.body, url + ' has an empty body').pipeThrough(new TextDecoderStream()).pipeThrough(new TextLineStream({ skipEmptyLines: processLine }));
if (processLine) {
stream = stream.pipeThrough(new ProcessLineStream());
}

View File

@ -1,7 +1,7 @@
import fs from 'node:fs';
import readline from 'node:readline';
import { TextLineStream } from './text-line-transform-stream';
import { TextLineStream } from 'foxts/text-line-stream';
import type { ReadableStream } from 'node:stream/web';
import { TextDecoderStream } from 'node:stream/web';
import { processLine, ProcessLineStream } from './process-line';
@ -31,7 +31,7 @@ export const createReadlineInterfaceFromResponse: ((resp: UndiciResponseData | U
const resultStream = webStream
.pipeThrough(new TextDecoderStream())
.pipeThrough(new TextLineStream());
.pipeThrough(new TextLineStream({ skipEmptyLines: processLine }));
if (processLine) {
return resultStream.pipeThrough(new ProcessLineStream());

View File

@ -1,80 +0,0 @@
// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
// This module is browser compatible.
// Modified by Sukka (https://skk.moe) to increase compatibility and performance with Bun.
import { TransformStream } from 'node:stream/web';
interface TextLineStreamOptions {
/** Allow splitting by solo \r */
allowCR?: boolean
}
/** Transform a stream into a stream where each chunk is divided by a newline,
* be it `\n` or `\r\n`. `\r` can be enabled via the `allowCR` option.
*
* ```ts
* const res = await fetch('https://example.com');
* const lines = res.body!
* .pipeThrough(new TextDecoderStream())
* .pipeThrough(new TextLineStream());
* ```
*/
export class TextLineStream extends TransformStream<string, string> {
// private __buf = '';
constructor({
allowCR = false
}: TextLineStreamOptions = {}) {
let __buf = '';
let chunkIndex = 0;
super({
transform(chunk, controller) {
chunk = __buf + chunk;
chunkIndex = 0;
for (; ;) {
const lfIndex = chunk.indexOf('\n', chunkIndex);
if (allowCR) {
const crIndex = chunk.indexOf('\r', chunkIndex);
if (
crIndex !== -1 && crIndex !== (chunk.length - 1)
&& (lfIndex === -1 || (lfIndex - 1) > crIndex)
) {
controller.enqueue(chunk.slice(chunkIndex, crIndex));
chunkIndex = crIndex + 1;
continue;
}
}
if (lfIndex === -1) {
// we can no longer find a line break in the chunk, break the current loop
break;
}
// enqueue current line, and loop again to find next line
let crOrLfIndex = lfIndex;
if (chunk[lfIndex - 1] === '\r') {
crOrLfIndex--;
}
controller.enqueue(chunk.slice(chunkIndex, crOrLfIndex));
chunkIndex = lfIndex + 1;
continue;
}
__buf = chunk.slice(chunkIndex);
},
flush(controller) {
if (__buf.length > 0) {
// eslint-disable-next-line sukka/string/prefer-string-starts-ends-with -- performance
if (allowCR && __buf[__buf.length - 1] === '\r') {
controller.enqueue(__buf.slice(0, -1));
} else {
controller.enqueue(__buf);
}
}
}
});
}
}