Skip to content

S
sdream-ai-fe
Commit 8bfc5f27 by weiyudumei
Options

refactor: 统一使用 Promise 判断流结束，移除 END 消息判断

parent 6390aa55
Showing with 123 additions and 102 deletions
src/api/chat.ts
......@@ -7,108 +7,118 @@ import { http } from '@/utils/request'
export function fetchCheckTokenApi() {
return http.post('/user/api/user_center/mobile/v1/check_token', {})
}
export function fetchStreamResponse(url: string, body: Record<string, any>, onMessage: (msg: any) => void, signal?: AbortSignal) {
export function fetchStreamResponse(url: string, body: Record<string, any>, onMessage: (msg: any) => void, signal?: AbortSignal): Promise<void> {
body.stream = true
const decoder = new TextDecoder('utf-8')
let buffer = ''
let dataMsgBuffer = ''
function processMessage(reader: any) {
reader.read().then((content: any) => {
// 检查是否已被中止
if (signal?.aborted) {
return
}
buffer += decoder.decode(content.value, { stream: !content.done })
const lines = buffer.split('\n')
buffer = lines.pop() as string
lines.forEach((line) => {
if (line === '') { // 读取到空行，一个数据块发送完成
try {
return new Promise<void>((resolve, reject) => {
function processMessage(reader: any) {
reader.read().then((content: any) => {
// 检查是否已被中止
if (signal?.aborted) {
resolve() // 中止时也 resolve，表示流已结束
return
}
buffer += decoder.decode(content.value, { stream: !content.done })
const lines = buffer.split('\n')
buffer = lines.pop() as string
lines.forEach((line) => {
if (line === '') { // 读取到空行，一个数据块发送完成
try {
onMessage({
type: 'DATA',
content: JSON.parse(dataMsgBuffer),
})
}
catch {
// 解析失败时静默处理，避免报错
}
dataMsgBuffer = ''
return
}
const [type] = line.split(':', 1)
const content = line.substring(type.length + 1)
if (type === 'data') { // 数据块没有收到空行之前放入buffer中
// SSE 格式中，data: 后面的内容前面可能有一个空格，trim 可以安全处理
dataMsgBuffer += content.trim()
}
else if (type === '' && content !== '') { // 服务端发送的注释，用于保证链接不断开
onMessage({
type: 'DATA',
content: JSON.parse(dataMsgBuffer),
type: 'COMMENT',
content: content.trim(),
})
}
catch {
// 解析失败时静默处理，避免报错
else {
onMessage({
type,
content: content.trim(),
})
}
dataMsgBuffer = ''
return
}
const [type] = line.split(':', 1)
const content = line.substring(type.length + 1)
if (type === 'data') { // 数据块没有收到空行之前放入buffer中
// SSE 格式中，data: 后面的内容前面可能有一个空格，trim 可以安全处理
dataMsgBuffer += content.trim()
}
else if (type === '' && content !== '') { // 服务端发送的注释，用于保证链接不断开
onMessage({
type: 'COMMENT',
content: content.trim(),
})
})
if (!content.done) {
processMessage(reader)
}
else {
onMessage({
type,
content: content.trim(),
})
// 流结束，resolve Promise（不再发送 END 消息）
resolve()
}
}).catch((error: unknown) => {
// 如果是 AbortError，resolve 而不是 reject（因为这是主动取消）
if (error instanceof Error && error.name === 'AbortError') {
resolve()
return
}
})
if (!content.done) {
processMessage(reader)
}
else {
onMessage({
type: 'END',
type: 'ERROR',
content: error,
})
}
}).catch((error: unknown) => {
// 如果是 AbortError，不发送错误消息
if (error instanceof Error && error.name === 'AbortError') {
return
}
onMessage({
type: 'ERROR',
content: error,
reject(error) // 其他错误 reject Promise
})
})
}
const tokenStr = window.localStorage.getItem('__TOKEN__') || '""'
let token = ''
try {
token = JSON.parse(tokenStr)
}
catch {
token = ''
}
}
const tokenStr = window.localStorage.getItem('__TOKEN__') || '""'
let token = ''
try {
token = JSON.parse(tokenStr)
}
catch {
token = ''
}
fetch(url, {
headers: {
'accept': 'text/event-stream',
'Content-Type': 'application/json',
'X-Token': token,
},
method: 'POST',
body: JSON.stringify(body),
signal,
})
.then((response) => {
return response.body?.getReader()
fetch(url, {
headers: {
'accept': 'text/event-stream',
'Content-Type': 'application/json',
'X-Token': token,
},
method: 'POST',
body: JSON.stringify(body),
signal,
})
.then((reader) => {
return processMessage(reader)
})
.catch((error: unknown) => {
// 如果是 AbortError，不发送错误消息
if (error instanceof Error && error.name === 'AbortError') {
return
}
onMessage({
type: 'ERROR',
content: error,
.then((response) => {
return response.body?.getReader()
})
})
.then((reader) => {
if (!reader) {
reject(new Error('Failed to get reader from response'))
return
}
return processMessage(reader)
})
.catch((error: unknown) => {
// 如果是 AbortError，resolve 而不是 reject（因为这是主动取消）
if (error instanceof Error && error.name === 'AbortError') {
resolve()
return
}
onMessage({
type: 'ERROR',
content: error,
})
reject(error)
})
})
}
/**
......
src/pages/Chat/Chat.tsx
......@@ -298,20 +298,24 @@ export const Chat: React.FC = () => {
}
if (msg?.type === 'DATA' && msg?.content?.code === '01010005') {
handleChatMaxCount(msg, question)
return
}
if (msg.type === 'END') {
// 流式请求正常结束时重置 isAsking 状态
dispatch(setIsAsking(false))
if (isNew) {
setTimeout(() => {
dispatch(fetchConversations())
}, 2000)
}
}
},
abortControllerRef.current.signal,
)
.then(() => {
// 流结束时执行（通过 Promise resolve 判断）
dispatch(setIsAsking(false))
if (isNew) {
setTimeout(() => {
dispatch(fetchConversations())
}, 2000)
}
})
.catch((error) => {
// 流式请求出错时的处理
console.error('[Chat] 流式请求失败:', error)
dispatch(setIsAsking(false))
})
}
/** 获取qa记录 */
......
src/pages/ChatTactics/TacticsChat.tsx
......@@ -524,18 +524,25 @@ export const TacticsChat: React.FC = () => {
}
if (msg?.type === 'DATA' && msg?.content?.code === '01010005') {
handleChatMaxCount(msg, question || '')
return
}
if (msg.type === 'END') {
if (isNew) {
setTimeout(() => {
dispatch(fetchTacticsConversations())
}, 2000)
}
}
},
abortControllerRef.current.signal,
)
.then(() => {
// 流结束时执行（通过 Promise resolve 判断，而不是 END 消息）
console.log('[TacticsChat] 流式响应已完成')
dispatch(setIsAsking(false))
if (isNew) {
setTimeout(() => {
dispatch(fetchTacticsConversations())
}, 2000)
}
})
.catch((error) => {
// 流式请求出错时的处理
console.error('[TacticsChat] 流式请求失败:', error)
dispatch(setIsAsking(false))
})
}
/** 获取qa记录 */
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment