|
@@ -1,20 +1,21 @@
|
|
|
import type { BunFile } from 'bun';
|
|
import type { BunFile } from 'bun';
|
|
|
import { fetchWithRetry, defaultRequestInit } from './fetch-retry';
|
|
import { fetchWithRetry, defaultRequestInit } from './fetch-retry';
|
|
|
|
|
|
|
|
-// import { TextLineStream } from './text-line-transform-stream';
|
|
|
|
|
-// import { PolyfillTextDecoderStream } from './text-decoder-stream';
|
|
|
|
|
|
|
+import { TextLineStream } from './text-line-transform-stream';
|
|
|
|
|
+import { PolyfillTextDecoderStream } from './text-decoder-stream';
|
|
|
import { processLine } from './process-line';
|
|
import { processLine } from './process-line';
|
|
|
-// function createTextLineStreamFromStreamSource(stream: ReadableStream<Uint8Array>) {
|
|
|
|
|
-// return stream
|
|
|
|
|
-// .pipeThrough(new PolyfillTextDecoderStream())
|
|
|
|
|
-// .pipeThrough(new TextLineStream());
|
|
|
|
|
-// }
|
|
|
|
|
|
|
+
|
|
|
|
|
+const enableTextLineStream = !!process.env.ENABLE_TEXT_LINE_STREAM;
|
|
|
|
|
+
|
|
|
|
|
+interface TextLineStreamLike {
|
|
|
|
|
+ [Symbol.asyncIterator](): AsyncIterableIterator<string>
|
|
|
|
|
+}
|
|
|
|
|
|
|
|
const decoder = new TextDecoder('utf-8');
|
|
const decoder = new TextDecoder('utf-8');
|
|
|
async function *createTextLineAsyncGeneratorFromStreamSource(stream: ReadableStream<Uint8Array>): AsyncGenerator<string> {
|
|
async function *createTextLineAsyncGeneratorFromStreamSource(stream: ReadableStream<Uint8Array>): AsyncGenerator<string> {
|
|
|
let buf = '';
|
|
let buf = '';
|
|
|
|
|
|
|
|
- for await (const chunk of stream) {
|
|
|
|
|
|
|
+ for await (const chunk of stream as any) {
|
|
|
const chunkStr = decoder.decode(chunk).replaceAll('\r\n', '\n');
|
|
const chunkStr = decoder.decode(chunk).replaceAll('\r\n', '\n');
|
|
|
for (let i = 0, len = chunkStr.length; i < len; i++) {
|
|
for (let i = 0, len = chunkStr.length; i < len; i++) {
|
|
|
const char = chunkStr[i];
|
|
const char = chunkStr[i];
|
|
@@ -32,25 +33,45 @@ async function *createTextLineAsyncGeneratorFromStreamSource(stream: ReadableStr
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
-export function readFileByLine(file: string | URL | BunFile) {
|
|
|
|
|
- if (typeof file === 'string') {
|
|
|
|
|
- file = Bun.file(file);
|
|
|
|
|
- } else if (!('writer' in file)) {
|
|
|
|
|
- file = Bun.file(file);
|
|
|
|
|
|
|
+export const readFileByLine: ((file: string | URL | BunFile) => TextLineStreamLike) = enableTextLineStream
|
|
|
|
|
+ ? (file: string | URL | BunFile) => {
|
|
|
|
|
+ if (typeof file === 'string') {
|
|
|
|
|
+ file = Bun.file(file);
|
|
|
|
|
+ } else if (!('writer' in file)) {
|
|
|
|
|
+ file = Bun.file(file);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return file.stream().pipeThrough(new PolyfillTextDecoderStream()).pipeThrough(new TextLineStream());
|
|
|
}
|
|
}
|
|
|
|
|
+ : (file: string | URL | BunFile) => {
|
|
|
|
|
+ if (typeof file === 'string') {
|
|
|
|
|
+ file = Bun.file(file);
|
|
|
|
|
+ } else if (!('writer' in file)) {
|
|
|
|
|
+ file = Bun.file(file);
|
|
|
|
|
+ }
|
|
|
|
|
|
|
|
- return createTextLineAsyncGeneratorFromStreamSource(file.stream());
|
|
|
|
|
-}
|
|
|
|
|
|
|
+ return createTextLineAsyncGeneratorFromStreamSource(file.stream()) as any;
|
|
|
|
|
+ };
|
|
|
|
|
|
|
|
-export function createReadlineInterfaceFromResponse(this: void, resp: Response) {
|
|
|
|
|
- if (!resp.body) {
|
|
|
|
|
- throw new Error('Failed to fetch remote text');
|
|
|
|
|
- }
|
|
|
|
|
- if (resp.bodyUsed) {
|
|
|
|
|
- throw new Error('Body has already been consumed.');
|
|
|
|
|
|
|
+export const createReadlineInterfaceFromResponse: ((resp: Response) => TextLineStreamLike) = enableTextLineStream
|
|
|
|
|
+ ? (resp) => {
|
|
|
|
|
+ if (!resp.body) {
|
|
|
|
|
+ throw new Error('Failed to fetch remote text');
|
|
|
|
|
+ }
|
|
|
|
|
+ if (resp.bodyUsed) {
|
|
|
|
|
+ throw new Error('Body has already been consumed.');
|
|
|
|
|
+ }
|
|
|
|
|
+ return resp.body.pipeThrough(new PolyfillTextDecoderStream()).pipeThrough(new TextLineStream());
|
|
|
}
|
|
}
|
|
|
- return createTextLineAsyncGeneratorFromStreamSource(resp.body);
|
|
|
|
|
-}
|
|
|
|
|
|
|
+ : (resp) => {
|
|
|
|
|
+ if (!resp.body) {
|
|
|
|
|
+ throw new Error('Failed to fetch remote text');
|
|
|
|
|
+ }
|
|
|
|
|
+ if (resp.bodyUsed) {
|
|
|
|
|
+ throw new Error('Body has already been consumed.');
|
|
|
|
|
+ }
|
|
|
|
|
+ return createTextLineAsyncGeneratorFromStreamSource(resp.body) as any;
|
|
|
|
|
+ };
|
|
|
|
|
|
|
|
export function fetchRemoteTextByLine(url: string | URL) {
|
|
export function fetchRemoteTextByLine(url: string | URL) {
|
|
|
return fetchWithRetry(url, defaultRequestInit).then(createReadlineInterfaceFromResponse);
|
|
return fetchWithRetry(url, defaultRequestInit).then(createReadlineInterfaceFromResponse);
|