diff --git a/packages/trpc/src/services/stream.service.ts b/packages/trpc/src/services/stream.service.ts index a9c81b2..6c7d879 100644 --- a/packages/trpc/src/services/stream.service.ts +++ b/packages/trpc/src/services/stream.service.ts @@ -8,6 +8,33 @@ export type RemoteFileStream = { body: ReadableStream | null; }; +type ParsedRange = { start: number; end: number } | "unsatisfiable" | null; + +/** + * 解析单段 bytes Range(RFC 9110): + * - null = 语法非法/不支持的形态 → 调用方忽略并回退 200 全量 + * - "unsatisfiable" = 语义不可满足 → 416 + */ +function parseBytesRange(header: string, size: number): ParsedRange { + const m = /^bytes=(\d*)-(\d*)$/i.exec(header.trim()); + if (!m) return null; + const first = m[1] ?? ""; + const second = m[2] ?? ""; + if (!first && !second) return null; + if (!first) { + // 后缀语法 bytes=-N:最后 N 字节;N=0 不可满足 + const suffix = Number(second); + if (suffix === 0) return "unsatisfiable"; + return { start: Math.max(0, size - suffix), end: size - 1 }; + } + const start = Number(first); + if (start >= size) return "unsatisfiable"; + if (!second) return { start, end: size - 1 }; + const end = Number(second); + if (end < start) return "unsatisfiable"; + return { start, end: Math.min(end, size - 1) }; +} + /** 打开远端媒体的 Range 流;webdav 细节只在本层出现,web 只拿 Headers + ReadableStream。 */ export async function openRemoteFileStream( userId: string, @@ -36,15 +63,21 @@ export async function openRemoteFileStream( let end = item.size > 0 ? item.size - 1 : 0; if (rangeHeader && item.size > 0) { - const m = /bytes=(\d*)-(\d*)/.exec(rangeHeader); - if (m) { - if (m[1]) start = Number(m[1]); - if (m[2]) end = Math.min(Number(m[2]), item.size - 1); + const parsed = parseBytesRange(rangeHeader, item.size); + if (parsed === "unsatisfiable") { + headers.set("Content-Range", `bytes */${item.size}`); + return { status: 416, headers, body: null }; + } + if (parsed) { + start = parsed.start; + end = parsed.end; status = 206; headers.set("Content-Range", `bytes ${start}-${end}/${item.size}`); headers.set("Content-Length", String(end - start + 1)); } - } else if (item.size > 0) { + // parsed === null(非法 Range):忽略并回退 200,下方补 Content-Length + } + if (status === 200 && item.size > 0) { headers.set("Content-Length", String(item.size)); } @@ -53,17 +86,24 @@ export async function openRemoteFileStream( item.size > 0 ? client.createReadStream(abs, { range: { start, end } }) : client.createReadStream(abs); + // ReadableLike 类型未声明 pause/resume,但实现是 Node Readable(data 监听后进入 flowing 模式) + const src = nodeStream as typeof nodeStream & { pause(): void; resume(): void }; + // pull 驱动 + desiredSize 背压:队列满时 pause,pull(消费者取走)时 resume const body = new ReadableStream({ start(controller) { - nodeStream.on("data", (chunk: Buffer | string) => { + src.on("data", (chunk: Buffer | string) => { controller.enqueue(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); + if ((controller.desiredSize ?? 1) <= 0) src.pause(); }); - nodeStream.on("end", () => controller.close()); - nodeStream.on("error", (err: unknown) => controller.error(err)); + src.on("end", () => controller.close()); + src.on("error", (err: unknown) => controller.error(err)); + }, + pull() { + src.resume(); }, cancel() { - nodeStream.destroy(); + src.destroy(); }, });