-
Notifications
You must be signed in to change notification settings - Fork 21
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: add eventSource and fetch Stream
- Loading branch information
Showing
12 changed files
with
251 additions
and
38 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,41 @@ | ||
| import type { Context } from "@/app/types/index"; | ||
|
|
||
| export default class IMController { | ||
| dialog(ctx: Context): void { | ||
| try { | ||
| const { res, req, request } = ctx; | ||
| const { chat_id, question } = request.body; | ||
| console.log('dialog:', chat_id, question); | ||
| const headers = { | ||
| 'Content-Type': 'text/event-stream', | ||
| 'Cache-Control': 'no-cache', | ||
| 'Connection': 'keep-alive', | ||
| } | ||
| // 如果请求 /events 路径，建立 SSE 连接 | ||
| res.writeHead(200, headers) | ||
| // 每隔 1 秒发送一条消息 | ||
| let id = 0 | ||
| const intervalId = setInterval(() => { | ||
| res.write(`event: customEvent\n`) | ||
| res.write(`id: ${id}\n`) | ||
| res.write(`retry: 30000\n`) | ||
| const data = { id, time: new Date().toISOString(), question } | ||
| res.write(`data: ${JSON.stringify(data)}\n\n`) | ||
| id++ | ||
| if (id >= 10) { | ||
| clearInterval(intervalId) | ||
| res.end() | ||
| } | ||
| }, 1000) | ||
| // 当客户端关闭连接时停止发送消息 | ||
| req.on('close', () => { | ||
| clearInterval(intervalId) | ||
| id = 0 | ||
| res.end() | ||
| }) | ||
| } catch (error) { | ||
| console.log('dialog:', error); | ||
| // ctx.errorHandler({ error }); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,127 @@ | ||
| interface RequestOption { | ||
| onopen?: () => void | ||
| onmessage?: (data: string) => void | ||
| onclose?: () => void | ||
| onerror?: (e: Error) => void | ||
| } | ||
|
|
||
| enum EventSourceStatus { | ||
| CONNECTING, | ||
| OPEN, | ||
| CLOSED, | ||
| ERROR | ||
| } | ||
|
|
||
| export class EventStreamFetch { | ||
| controller?: AbortController; | ||
| status: EventSourceStatus; | ||
| constructor() { | ||
| this.controller = undefined | ||
| this.status = EventSourceStatus.CLOSED | ||
| } | ||
| // 建立 FETCH-SSE 连接 | ||
| public fetchStream = (url: string, body: Record<string, string> = {}): Promise<void> => { | ||
| this.controller = new AbortController() | ||
| const headers = { | ||
| 'content-type': 'application/json', | ||
| 'Accept': 'text/event-stream', | ||
| 'Connection': 'keep-alive', | ||
| } | ||
| return this.fetchEventSource(url, { | ||
| method: 'POST', | ||
| headers, | ||
| body: JSON.stringify(body), | ||
| signal: this.controller.signal, | ||
| onopen: () => { | ||
| this.status = EventSourceStatus.OPEN | ||
| console.log('FETCH 连接打开'); | ||
| }, | ||
| onclose: () => { | ||
| this.status = EventSourceStatus.CLOSED | ||
| console.log('FETCH 连接关闭'); | ||
| }, | ||
| onmessage: (event) => { | ||
| this.status = EventSourceStatus.CONNECTING | ||
| const data = JSON.parse(event) | ||
| console.log('FETCH 接收到消息：', data); | ||
| }, | ||
| onerror: (e) => { | ||
| this.status = EventSourceStatus.ERROR | ||
| console.log(e) | ||
| } | ||
| }) | ||
| } | ||
| // 断开 FETCH-SSE 连接 | ||
| public close = (): void => { | ||
| if (this.controller) { | ||
| this.status = EventSourceStatus.CLOSED | ||
| this.controller.abort() | ||
| this.controller = undefined | ||
| } | ||
| } | ||
| private fetchEventSource = (url: string, options: RequestOption & RequestInit): Promise<void> => { | ||
| return fetch(url, options) | ||
| .then(response => { | ||
| if (response.status === 200) { | ||
| options.onopen && options.onopen() | ||
| return response.body | ||
| } | ||
| }) | ||
| .then(rb => { | ||
| // eslint-disable-next-line n/no-unsupported-features/node-builtins | ||
| const reader = rb?.pipeThrough(new TextDecoderStream()).getReader() | ||
| const push = (): Promise<void> | undefined => { | ||
| // done 为数据流是否接收完成，boolean | ||
| // value 为返回数据，Uint8Array | ||
| return reader?.read().then(({ done, value }) => { | ||
| if (done) { | ||
| options.onclose && options.onclose() | ||
| return | ||
| } | ||
| options.onmessage && options.onmessage(value) | ||
| // 持续读取流信息 | ||
| return push() | ||
| }) | ||
| } | ||
| // 开始读取流信息 | ||
| return push() | ||
| }) | ||
| .catch((e) => { | ||
| options.onerror && options.onerror(e) | ||
| }) | ||
| } | ||
| } | ||
|
|
||
|
|
||
| export class EventStreamSource { | ||
| // eslint-disable-next-line n/no-unsupported-features/node-builtins | ||
| eventSource?: EventSource; | ||
| status: EventSourceStatus; | ||
| constructor() { | ||
| this.status = EventSourceStatus.CLOSED | ||
| } | ||
| // 建立 SSE 连接 | ||
| public connectSSE = (url: string): void => { | ||
| // eslint-disable-next-line n/no-unsupported-features/node-builtins | ||
| this.eventSource = new EventSource(url) | ||
| this.eventSource.onopen = () => { | ||
| this.status = EventSourceStatus.OPEN | ||
| console.log('SSE 连接打开'); | ||
| } | ||
| this.eventSource.onerror = () => { | ||
| this.status = EventSourceStatus.ERROR | ||
| console.log('SSE 连接错误'); | ||
| } | ||
| this.eventSource.onmessage = (event) => { | ||
| this.status = EventSourceStatus.CONNECTING | ||
| const data = JSON.parse(event.data) | ||
| console.log('SSE 接收到消息：', data); | ||
| } | ||
| } | ||
|
|
||
| // 断开 SSE 连接 | ||
| public closeSSE = (): void => { | ||
| this.status = EventSourceStatus.CLOSED | ||
| this.eventSource?.close() | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -67,3 +67,4 @@ export const randomGeneratePolygon = (number: number = 3, maxSides: number = 10) | |
| }; | ||
| }); | ||
| }; | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,7 +1,29 @@ | ||
| import React from 'react'; | ||
| import React, { } from 'react'; | ||
| import { EventStreamFetch } from '@/client/lib/eventSource'; | ||
|
|
||
|
|
||
| export const Home = (): React.JSX.Element => { | ||
| return <div>Home</div>; | ||
|
|
||
| const { fetchStream } = new EventStreamFetch() | ||
|
|
||
| const sendMessage = (params: Record<string, string> = {}) => { | ||
| fetchStream('/api/im/dialog', params) | ||
| } | ||
|
|
||
| const click = () => { | ||
| sendMessage({ chat_id: '1', question: 'hello' }) | ||
| } | ||
|
|
||
| return ( | ||
| <div> | ||
| <h1>Home</h1> | ||
| <div>输入消息</div> | ||
| <input type="text" /> | ||
| <button onClick={click}>发送消息</button> | ||
| <div>回答</div> | ||
| <textarea></textarea> | ||
| </div> | ||
| ); | ||
| }; | ||
|
|
||
| export default Home; |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters