@@ -8,11 +8,13 @@ const TRUNCATED_SUFFIX = "... [truncated]";
88type ResponseTextSnippetOptions = {
99maxBytes?: number;
1010maxChars?: number;
11+signal?: AbortSignal;
1112};
12131314type ResponseJsonOptions = {
1415maxBytes?: number;
1516errorPrefix: string;
17+signal?: AbortSignal;
1618};
17191820type ResponsePrefix = {
@@ -28,7 +30,7 @@ export async function readResponseTextSnippet(
2830): Promise<string> {
2931const maxBytes = options.maxBytes ?? DEFAULT_ERROR_BODY_MAX_BYTES;
3032const maxChars = options.maxChars ?? DEFAULT_ERROR_BODY_MAX_CHARS;
31-const prefix = await readResponsePrefix(res, maxBytes);
33+const prefix = await readResponsePrefix(res, maxBytes, options.signal);
3234if (prefix.length === 0) {
3335return "";
3436}
@@ -56,7 +58,7 @@ export async function readResponseJsonWithLimit(
5658throw responseTooLarge(options.errorPrefix, contentLength, maxBytes);
5759}
586059-const text = await readResponseTextWithLimit(res, maxBytes, options.errorPrefix);
61+const text = await readResponseTextWithLimit(res, maxBytes, options.errorPrefix, options.signal);
60626163try {
6264return JSON.parse(text);
@@ -65,7 +67,45 @@ export async function readResponseJsonWithLimit(
6567}
6668}
676968-async function readResponsePrefix(res: Response, maxBytes: number): Promise<ResponsePrefix> {
70+function toAbortError(signal: AbortSignal, fallbackMessage: string): Error {
71+return signal.reason instanceof Error ? signal.reason : new Error(fallbackMessage);
72+}
73+74+async function readChunkWithAbort(
75+reader: ReadableStreamDefaultReader<Uint8Array>,
76+signal: AbortSignal | undefined,
77+fallbackMessage: string,
78+): Promise<ReadableStreamReadResult<Uint8Array>> {
79+if (!signal) {
80+return await reader.read();
81+}
82+if (signal.aborted) {
83+await reader.cancel().catch(() => undefined);
84+throw toAbortError(signal, fallbackMessage);
85+}
86+87+let removeAbortListener: (() => void) | undefined;
88+const abortPromise = new Promise<ReadableStreamReadResult<Uint8Array>>((_resolve, reject) => {
89+const onAbort = () => {
90+void reader.cancel().catch(() => undefined);
91+reject(toAbortError(signal, fallbackMessage));
92+};
93+signal.addEventListener("abort", onAbort, { once: true });
94+removeAbortListener = () => signal.removeEventListener("abort", onAbort);
95+});
96+97+try {
98+return await Promise.race([reader.read(), abortPromise]);
99+} finally {
100+removeAbortListener?.();
101+}
102+}
103+104+async function readResponsePrefix(
105+res: Response,
106+maxBytes: number,
107+signal?: AbortSignal,
108+): Promise<ResponsePrefix> {
69109const body = res.body;
70110if (!body || typeof body.getReader !== "function") {
71111return { bytes: [], length: 0, truncated: false };
@@ -78,7 +118,11 @@ async function readResponsePrefix(res: Response, maxBytes: number): Promise<Resp
7811879119try {
80120while (true) {
81-const { done, value } = await reader.read();
121+const { done, value } = await readChunkWithAbort(
122+reader,
123+signal,
124+"Response snippet body read aborted",
125+);
82126if (done) {
83127break;
84128}
@@ -115,6 +159,7 @@ async function readResponseTextWithLimit(
115159res: Response,
116160maxBytes: number,
117161errorPrefix: string,
162+signal?: AbortSignal,
118163): Promise<string> {
119164const body = res.body;
120165if (!body || typeof body.getReader !== "function") {
@@ -127,7 +172,11 @@ async function readResponseTextWithLimit(
127172128173try {
129174while (true) {
130-const { done, value } = await reader.read();
175+const { done, value } = await readChunkWithAbort(
176+reader,
177+signal,
178+`${errorPrefix}: response body read aborted`,
179+);
131180if (done) {
132181break;
133182}
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。