From bd706e4a5be3644df8599a7e865db00b4e8573fd Mon Sep 17 00:00:00 2001 From: Peanut Date: Wed, 22 Jul 2026 22:57:29 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=20H5=20=E6=A8=A1?= =?UTF-8?q?=E5=BC=8F=20stream=20=E8=AF=B7=E6=B1=82=2060=20=E7=A7=92?= =?UTF-8?q?=E8=B6=85=E6=97=B6=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 在 H5 环境使用 fetch + ReadableStream + AbortController - 支持自定义 300 秒超时 - 小程序环境保持原有 uni.request 逻辑不变 - 添加 CRLF 兼容性、主动中止与超时区分、资源清理 --- mini-program/src/services/shortNovel.js | 160 +++++++++++++++++++++++- 1 file changed, 158 insertions(+), 2 deletions(-) diff --git a/mini-program/src/services/shortNovel.js b/mini-program/src/services/shortNovel.js index e0d3b4a..b444582 100644 --- a/mini-program/src/services/shortNovel.js +++ b/mini-program/src/services/shortNovel.js @@ -1,5 +1,11 @@ import { getApiBaseUrl, getAuthHeader } from './request.js' +// 环境检测：H5 模式下存在 window.fetch +const isH5 = typeof window !== 'undefined' && typeof window.fetch === 'function' + +// SSE 请求超时时间（毫秒） +const SSE_TIMEOUT_MS = 300000 + /** * 短篇小说外部服务 API 封装 * 通过后端 SSE 代理调用外部 short-novel-service @@ -14,6 +20,17 @@ import { getApiBaseUrl, getAuthHeader } from './request.js' * @returns {Object} uni.request 任务对象（用于 abort） */ export const startNovelStream = ({ query, onEvent, onError }) => { + // H5 环境：使用原生 fetch + ReadableStream + if (isH5) { + return h5NovelStream( + `${getApiBaseUrl()}/shortNovel/stream`, + { query }, + onEvent, + onError + ) + } + + // 小程序环境：使用 uni.request let chunkProcessed = false const task = uni.request({ url: `${getApiBaseUrl()}/shortNovel/stream`, @@ -25,7 +42,7 @@ export const startNovelStream = ({ query, onEvent, onError }) => { ...getAuthHeader() }, enableChunked: true, - timeout: 300000, + timeout: SSE_TIMEOUT_MS, success: (res) => { if (res.statusCode >= 400) { onError?.(res.data?.message || '请求失败') @@ -66,6 +83,17 @@ export const startNovelStream = ({ query, onEvent, onError }) => { * @returns {Object} uni.request 任务对象 */ export const followupStream = ({ sessionId, action, payload, originalQuery, onEvent, onError }) => { + // H5 环境：使用原生 fetch + ReadableStream + if (isH5) { + return h5NovelStream( + `${getApiBaseUrl()}/shortNovel/followup`, + { sessionId, action, payload, originalQuery }, + onEvent, + onError + ) + } + + // 小程序环境：使用 uni.request let chunkProcessed = false const task = uni.request({ url: `${getApiBaseUrl()}/shortNovel/followup`, @@ -77,7 +105,7 @@ export const followupStream = ({ sessionId, action, payload, originalQuery, onEv ...getAuthHeader() }, enableChunked: true, - timeout: 300000, + timeout: SSE_TIMEOUT_MS, success: (res) => { if (res.statusCode >= 400) { onError?.(res.data?.message || '请求失败') @@ -139,6 +167,134 @@ function decodeChunk(chunk) { return result } +/** + * H5 环境 SSE 事件解析 + * 解析单个 SSE 事件块（已经被 \r\n\r\n 或 \n\n 分隔） + * 事件格式：data: {"type":"status","session_id":"xxx","payload":{...}} + * 注意：本函数不依赖块内空行作为结束标记，由调用方负责按双换行切分事件 + */ +function h5ConsumeSseText(block, onEvent, onError) { + const lines = block.split(/\r?\n/) + let dataBuffer = '' + + for (const line of lines) { + // 移除行尾的 \r（CRLF 兼容） + const cleanLine = line.replace(/\r$/, '') + if (cleanLine.startsWith('data:')) { + dataBuffer += cleanLine.slice(5).trim() + } + // 注意：不再依赖空行作为事件结束标记 + // 因为块已经被调用方按 \n\n 切分好了 + } + + // 遍历完所有行后，直接处理累积的数据 + if (dataBuffer) { + if (dataBuffer === '[DONE]') return + try { + const event = JSON.parse(dataBuffer) + onEvent?.(event) + } catch (e) { + onError?.(`SSE 解析失败: ${e.message}`) + } + } +} + +/** + * H5 环境 SSE 流式读取核心实现 + * 使用浏览器原生 fetch + ReadableStream + AbortController + * @param {string} url - 请求 URL + * @param {Object} body - 请求体 + * @param {Function} onEvent - 事件回调 + * @param {Function} onError - 错误回调 + * @returns {Object} 包含 abort 方法的对象 + */ +function h5NovelStream(url, body, onEvent, onError) { + // 1. 创建 AbortController 用于超时控制 + const controller = new AbortController() + let abortedByUser = false // 标记是否用户主动中止 + let reader = null // 引用 reader 以便主动中止时取消 + const timeoutId = setTimeout(() => controller.abort(), SSE_TIMEOUT_MS) + + // 2. 统一的错误处理函数 + const handleError = (err) => { + clearTimeout(timeoutId) + if (abortedByUser) { + // 用户主动中止，不报错 + return + } + if (err && err.name === 'AbortError') { + onError?.('请求超时（300秒）') + } else { + onError?.(err?.message || '网络请求失败') + } + } + + // 3. 发起 fetch 请求 + fetch(url, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'Accept': 'text/event-stream', + ...getAuthHeader() + }, + body: JSON.stringify(body), + signal: controller.signal + }).then(response => { + // 4. 检查响应状态 + if (!response.ok) { + clearTimeout(timeoutId) + onError?.(`请求失败: HTTP ${response.status}`) + return + } + + // 5. 获取 ReadableStream 读取器 + reader = response.body.getReader() + const decoder = new TextDecoder('utf-8') + let buffer = '' + + // 6. 循环读取流数据 + function pump() { + return reader.read().then(({ done, value }) => { + if (done) { + clearTimeout(timeoutId) + // 处理缓冲区残留数据 + if (buffer.trim()) h5ConsumeSseText(buffer, onEvent, onError) + return + } + + // 7. 解码二进制块为文本 + buffer += decoder.decode(value, { stream: true }) + + // 8. 解析完整的事件（兼容 CRLF） + const events = buffer.split(/\r?\n\r?\n/) + buffer = events.pop() // 最后一个可能不完整，保留到下次 + + for (const event of events) { + h5ConsumeSseText(event, onEvent, onError) + } + + // 9. 继续读取 + return pump() + }) + } + + return pump() + }).catch(handleError) + + // 10. 返回 abort 接口 + return { + abort: () => { + clearTimeout(timeoutId) + abortedByUser = true + // 主动中止时取消 reader，释放底层资源 + if (reader) { + reader.cancel().catch(() => {}) + } + controller.abort() + } + } +} + /** * 解析 SSE 文本 * 外部服务事件格式：data: {"type":"status","session_id":"xxx","payload":{...}}