SSE Server-Sent Events实践

// src/index.ts
import { Hono } from 'hono'
 
import { streamText } from 'hono/streaming'
 
const app = new Hono()
 
app.get('/stream', (c) => {
 
  return streamText(c, async (stream) => {
 
    for (const chunk of ['Hello', ' ', 'World', '!']) {
 
      await stream.write(chunk)
 
      await stream.sleep(100)
 
    }
 
  })
 
})
 
export default app

// client.ts
async function consumeStream() {
 
  const response = await fetch('/stream')
 
  // 1. 从 response.body 获取一个 reader（读取器）
 
  const reader = response.body!.getReader()
 
  // 2. TextDecoder 用于把二进制数据转成字符串
 
  const decoder = new TextDecoder()
 
  // 3. 循环读取，每次拿到一块数据
 
  while (true) {
 
    const { done, value } = await reader.read()
 
    // done 为 true 表示流结束了（水龙头关了）
 
    if (done) break
 
    // value 是 Uint8Array（二进制），需要解码成字符串
 
    const text = decoder.decode(value, { stream: true })
 
    // 把拿到的文本追加到页面上
 
    document.getElementById('output')!.textContent += text
 
  }
 
}

// src/index.ts
import { stream } from 'hono/streaming'
 
app.get('/download', (c) => {
 
  return stream(c, async (stream) => {
 
    // 从远程拉取大文件，直接转发给客户端
 
    const res = await fetch('https://example.com/large-file')
 
    await stream.pipe(res.body!)
 
  })
 
})

// src/index.ts
app.get('/download', (c) => {
 
  c.header('Content-Type', 'application/octet-stream')
 
  c.header('Content-Disposition', 'attachment; filename="data.zip"')
 
  return stream(c, async (stream) => {
 
    const res = await fetch('https://example.com/data.zip')
 
    await stream.pipe(res.body!)
 
  })
 
})

// SSE
event: message
 
id: 1
 
data: 你好
 
event: message
 
id: 2
 
data: 世界
 
event: done
 
id: 3
 
data: [DONE]

// src/index.ts
import { streamSSE } from 'hono/streaming'
 
app.get('/sse', (c) => {
 
  return streamSSE(c, async (stream) => {
 
    let id = 0
 
    while (true) {
 
      await stream.writeSSE({
 
        data: JSON.stringify({ time: Date.now() }),
 
        event: 'tick',
 
        id: String(id++),
 
      })
 
      await stream.sleep(1000)
 
    }
 
  })
 
})

// code.ts
event: tick
 
id: 0
 
data: {"time":1714000000000}

// client.ts
// 创建 SSE 连接（和 fetch 不同，EventSource 会自动保持连接）
 
const source = new EventSource('/sse')
 
// 监听 'tick' 类型的事件（对应后端 writeSSE 里的 event 字段）
 
source.addEventListener('tick', (e) => {
 
  const data = JSON.parse(e.data)
 
  console.log('服务端时间:', data.time)
 
})
 
// 连接建立时触发
 
source.addEventListener('open', () => {
 
  console.log('SSE 连接已建立')
 
})
 
// 连接出错时触发（浏览器会自动尝试重连，不需要你处理）
 
source.addEventListener('error', (e) => {
 
  console.log('SSE 连接出错，浏览器将自动重连')
 
})
 
// 不想接收了，手动关闭
 
// source.close()

// code.ts
前端 --POST 消息--> 你的 Hono 服务 --POST 消息--> OpenAI/Claude API
 
前端 <--SSE 转发-- 你的 Hono 服务 <--SSE 响应-- OpenAI/Claude API

// src/index.ts
import { Hono } from 'hono'
 
import { streamSSE } from 'hono/streaming'
 
const app = new Hono<{ Bindings: { OPENAI_API_KEY: string } }>()
 
app.post('/chat', async (c) => {
 
  const { messages } = await c.req.json()
 
  // 调用 OpenAI，开启 stream 模式
 
  const response = await fetch(
 
    'https://api.openai.com/v1/chat/completions',
 
    {
 
      method: 'POST',
 
      headers: {
 
        'Authorization': `Bearer ${c.env.OPENAI_API_KEY}`,
 
        'Content-Type': 'application/json',
 
      },
 
      body: JSON.stringify({
 
        model: 'gpt-4o',
 
        messages,
 
        stream: true,  // 关键：告诉 OpenAI 用流式返回
 
      }),
 
    }
 
  )
 
  // OpenAI 返回的也是 SSE 格式的流
 
  // 我们需要逐行读取，提取文本内容，再转发给前端
 
  return streamSSE(c, async (stream) => {
 
    const reader = response.body!.getReader()
 
    const decoder = new TextDecoder()
 
    let buffer = ''
 
    let id = 0
 
    while (true) {
 
      const { done, value } = await reader.read()
 
      if (done) break
 
      // 把二进制数据解码成文本，追加到 buffer
 
      buffer += decoder.decode(value, { stream: true })
 
      // 按换行符拆分成多行
 
      const lines = buffer.split('\n')
 
      // 最后一行可能不完整（数据还没到齐），留到下一轮处理
 
      buffer = lines.pop() || ''
 
      for (const line of lines) {
 
        // OpenAI 的 SSE 每行以 'data: ' 开头
 
        if (!line.startsWith('data: ')) continue
 
        const data = line.slice(6) // 去掉 'data: ' 前缀
 
        if (data === '[DONE]') return // 结束信号
 
        // 解析 JSON，提取生成的文本片段
 
        const parsed = JSON.parse(data)
 
        const content = parsed.choices[0]?.delta?.content
 
        if (content) {
 
          // 转发给前端
 
          await stream.writeSSE({
 
            data: content,
 
            event: 'message',
 
            id: String(id++),
 
          })
 
        }
 
      }
 
    }
 
  })
 
})

// client.ts
async function chat(messages: Array<{ role: string; content: string }>) {
 
  const response = await fetch('/chat', {
 
    method: 'POST',
 
    headers: { 'Content-Type': 'application/json' },
 
    body: JSON.stringify({ messages }),
 
  })
 
  // 和前面 streamText 的消费方式一样：拿 reader，循环读
 
  const reader = response.body!.getReader()
 
  const decoder = new TextDecoder()
 
  let result = ''
 
  while (true) {
 
    const { done, value } = await reader.read()
 
    if (done) break
 
    const text = decoder.decode(value, { stream: true })
 
    // 手动解析 SSE 格式：找到 'data: ' 开头的行，提取内容
 
    const lines = text.split('\n')
 
    for (const line of lines) {
 
      if (!line.startsWith('data: ')) continue
 
      const content = line.slice(6)
 
      result += content
 
      // 更新页面上的文字，实现"打字机"效果
 
      document.getElementById('chat')!.textContent = result
 
    }
 
  }
 
}