SSE实现消息实时推送，前端渐进式学习、实践，真香 - qidao123.com技术社区-IT企服评测·应用市场

const eventSourceRef = useRef<any>(null)
const contact = async (messageData: any) => {
eventSourceRef.current = new EventSource(
`${API_BASE}/v1/model/stream?id=${id}`,
)
if (!eventSourceRef.current) return
// 监听 SSE 事件，因为后端定义了具名事件，所以这儿要用addEventListener监听，而不是onmessage
eventSourceRef.current.addEventListener('add', function (e: any) {
// 处理数据展示
})
eventSourceRef.current.addEventListener('finish', function (e: any) {
// 结束标识finish
eventSourceRef.current.close() // 关闭连接
})
eventSourceRef.current.addEventListener('error', function (e: any) {
if (e.status === 401) {
// 用户登录状态失效处理
}
// error报错处理
console.log('Error occurred:', e)
// 关闭连接
eventSourceRef.current.close()
})
}

复制代码

const eventSourceRef = useRef<any>(null)
const [abortController, setAbortController] = useState(new AbortController())
// 通信事件
const contact = async (messageData: any) => {
messageData = { ...messageData, do_stream: modelArg.do_stream } // 请求参数
receivedDataRef.current = ''
const token: string = getLocal('AUTHCODE') || ''
fetchEventSource(`${MAAS_API_BASE}/v1/model_api/invoke`, {
method: 'POST',
// 添加请求头
headers: {
Authorization: token,
'Content-Type': 'application/json',
},
// 传参必须保证是json
body: JSON.stringify(messageData),
// abortController.signal 提供了一个信号对象给 fetchEventSource 函数。
// 如果在任何时候你想取消正在进行的 fetch 操作，你可以调用
// abortController.abort()。这会发出关联任务的信号，你可以使用
// AbortController 的信号来检查异步操作是否已被取消。
signal: abortController.signal,
openWhenHidden: true, // 切换标签页时连接不关闭
async onopen(resp) {
// 处理登录失效
if (resp.status === 401) {
message.warning('登录过期')
return
}
},
onmessage(msg: any) {
const eventType = msg.event // 监听event的具名事件
switch (eventType) {
case 'add':
// 流式输出事件，add每次会返回具体字符，前端负责拼接展示
break
case 'finish':
setStatu('finish') // 结束标识
break
case 'error':
if (msg.status === 401) {
message.warning('登录过期')
}
console.log('Error occurred:', e)
break
}
},
onerror(err) {
throw err // 连接遇到http错误时，如跨域等，必须要throw才能停止，不然会一直重连
},
onclose() {},
})
}
// 终止连接方法，比如在切换模型时，你可能有必要终止上一次连接来避免问答串联
const closeSSE = () => {
abortController.abort()
setAbortController(new AbortController())
}

复制代码

fetchEventSource(sseUrl, {
method: 'POST',
headers,
signal: abortController.signal,
body: JSON.stringify(customInferData),
openWhenHidden: true,
/**
*在onopen阶段处理
第一步：判断resp.headers.get('content-type')，如果不包含text/event-stream，
则代表非流式
第二步：需要在onopen阶段处理非流式返回，即json返回,读取json返回并渲染，注意异常也要处理
第三步：
*/
async onopen(resp) {
const contentype = resp.headers.get('content-type') || ''
console.log('contentype =>', contentype)
console.log('resp.ok =>', resp.ok)
if (resp.ok && !contentype.includes('text/event-stream')) {
// 读取json数据
const responseData = await resp.json()
if (responseData.code !== 0) {
// 报错处理+关闭连接
} else {
//处理数据渲染+关闭连接
stopSession()
}
} else if (resp.status === 401) {
message.warning('登录过期')
// 报错处理+关闭连接
stopSession()
}
},
onmessage(msg: any) {
const eventType = msg.event
const messages: any = cloneDeep(chatState.sessionMessages)
let lastMessage: any = messages[messages.length - 1] || {}
switch (eventType) {
case 'add':
lastMessage = {
...lastMessage,
text: `${lastMessage.text}${msg.data || ' '}`,
loading: false,
}
messages.splice(messages.length - 1, 1, lastMessage)
chatAction.updateSessionMessages(messages)
break
case 'finish':
console.log('finish lastMessage =>', lastMessage)
chatAction.updateSessionStatu(SessionStatuTypes.ready)
chatAction.updateContext(msg.data)
break
case 'info':
{
const messages: any = cloneDeep(chatState.sessionMessages)
let lastMessage: any = messages[messages.length - 1] || {}
lastMessage = {
referenceDocs: JSON.parse(msg.data).reference_by_docs,
...lastMessage,
}
messages.splice(messages.length - 1, 1, lastMessage)
chatAction.updateSessionMessages(messages)
}
break
case 'error':
if (msg.status === 401) {
chatAction.updateSessionStatu(SessionStatuTypes.ready)
} else {
errorItemFn(msg?.msg || msg?.data || '抱歉，暂无法回答问题')
}
break
}
},
onerror(err: any) {
errorItemFn(err?.msg || '抱歉，暂无法回答该问题')
console.log('eventSource error: ', `${err}`)
throw err // 连接遇到http错误时，如跨域等，必须要throw才能停止，不然会一直重连
},
onclose() {
console.log('eventSource close')
},
})
// 终止会话
const stopSession = () => {
abortController.abort()
setAbortController(new AbortController())
}

复制代码

data: Message 1
data: Message 2
data: Message 3
data: of two lines

复制代码

const stopSession = () => {
abortController.abort()
setAbortController(new AbortController())
}

复制代码