PR-Helper/docs/06-sse-streaming.md

SSE 流式传输

概述

Server-Sent Events (SSE) 是 PR-Helper 的核心通信机制，用于实时推送克隆进度、PR 生成和代码审查结果。

为什么选择 SSE

对比 WebSocket

特性	SSE	WebSocket
方向	单向（服务器→客户端）	双向
协议	HTTP	独立协议
实现	简单	复杂
重连	自动	手动
兼容性	好	需要升级

适用场景

• 服务器推送数据到客户端
• 不需要客户端频繁发送数据
• 需要 HTTP 兼容性
• 需要自动重连

SSE 协议

基本格式

event: message_type
data: {"key": "value"}

• event: 事件类型（可选）
• data: 事件数据（JSON 字符串）
• 空行分隔事件

多行数据

event: content
data: {"content": "第一行\n第二行"}

后端实现

设置响应头

c.Header("Content-Type", "text/event-stream")
c.Header("Cache-Control", "no-cache")
c.Header("Connection", "keep-alive")
c.Header("X-Accel-Buffering", "no")
c.Status(http.StatusOK)

发送事件

sendEvent := func(event string, data interface{}) {
    jsonData, err := json.Marshal(data)
    if err != nil {
        jsonData = []byte(`{"error":"序列化事件数据失败"}`)
    }
    fmt.Fprintf(c.Writer, "event: %s\ndata: %s\n\n", event, jsonData)
    flusher.Flush()
}

获取 Flusher

flusher, ok := c.Writer.(http.Flusher)
if !ok {
    c.JSON(http.StatusInternalServerError, gin.H{"error": "服务器不支持流式传输"})
    return
}

前端实现

SSE 客户端

const SSE = {
    async post(url, body, handlers = {}) {
        const controller = new AbortController();

        const run = async () => {
            try {
                const resp = await fetch(url, {
                    method: 'POST',
                    headers: { 'Content-Type': 'application/json' },
                    credentials: 'same-origin',
                    body: JSON.stringify(body),
                    signal: controller.signal,
                });

                if (!resp.ok) {
                    const errText = await resp.text();
                    // 处理错误...
                    return;
                }

                const reader = resp.body.getReader();
                const decoder = new TextDecoder();
                let buffer = '';
                let currentEvent = '';

                while (true) {
                    const { done, value } = await reader.read();
                    if (done) break;

                    buffer += decoder.decode(value, { stream: true });
                    const lines = buffer.split('\n');
                    buffer = lines.pop() || '';

                    for (const line of lines) {
                        if (line.startsWith('event: ')) {
                            currentEvent = line.slice(7).trim();
                        } else if (line.startsWith('data: ')) {
                            const raw = line.slice(6);
                            let data;
                            try {
                                data = JSON.parse(raw);
                            } catch (_) {
                                data = raw;
                            }

                            if (currentEvent && handlers[currentEvent]) {
                                handlers[currentEvent](data);
                            }
                        }
                    }
                }

                if (handlers.done) handlers.done();
            } catch (err) {
                if (err.name === 'AbortError') return;
                if (handlers.error) handlers.error({ message: err.message });
            }
        };

        run();
        return { abort: () => controller.abort() };
    }
};

使用示例

SSE.post('/api/repos/1/review', {
    base: 'main',
    head: 'feature',
    top_n: 20,
    concurrency: 5
}, {
    // 开始事件
    start: (data) => {
        console.log(`开始审查 ${data.total_files} 个文件`);
        updateProgress(0, data.reviewed_files);
    },

    // 文件开始
    file_start: (data) => {
        console.log(`正在审查 ${data.file} (${data.index}/${data.total})`);
        highlightCurrentFile(data.file);
    },

    // 内容流
    content: (data) => {
        appendToOutput(data.content);
    },

    // 审查建议
    suggestion: (data) => {
        showSuggestion(data.file, data.severity, data.content);
    },

    // 文件结束
    file_end: (data) => {
        markFileComplete(data.file);
    },

    // 汇总
    summary: (data) => {
        showSummary(data.score, data.overall, data.findings, data.recommendations);
    },

    // 分析保存
    analysis_saved: (data) => {
        console.log(`分析已保存，ID: ${data.analysis_id}`);
    },

    // 进度
    progress: (data) => {
        if (data.step === 'generating_summary') {
            showSpinner('生成汇总中...');
        }
    },

    // 完成
    done: () => {
        console.log('审查完成');
        hideSpinner();
    },

    // 错误
    error: (data) => {
        console.error('错误:', data.message);
        showError(data.message);
    }
});

事件类型

PR 生成事件

事件	数据	说明
content	{content: string}	Markdown 片段
done	{content: ""}	完成

代码审查事件

事件	数据	说明
start	{total_files, reviewed_files, top_n}	开始
file_start	{file, index, total}	文件开始
content	{content: string}	LLM 输出
suggestion	{file, severity, content}	建议
file_end	{file}	文件结束
summary	{score, overall, findings, recommendations}	汇总
progress	{step: string}	进度
error	{message: string}	错误
analysis_saved	{analysis_id}	保存完成
done	{content: ""}	完成

克隆事件

事件	数据	说明
progress	{step, current, total}	进度
error	{message: string}	错误
done	{repo_id, branches, tags, commit_num}	完成

错误处理

后端错误

if err != nil {
    sendEvent("error", map[string]interface{}{
        "message": err.Error(),
    })
    return
}

前端错误

error: (data) => {
    // 显示错误消息
    showErrorToast(data.message);

    // 重置 UI
    resetProgressBar();
    hideSpinner();
}

网络错误

catch (err) {
    if (err.name === 'AbortError') {
        // 用户中断，忽略
        return;
    }
    if (handlers.error) {
        handlers.error({ message: err.message });
    }
}

中断请求

前端中断

const controller = SSE.post(url, body, handlers);

// 用户点击取消按钮
cancelButton.onclick = () => {
    controller.abort();
};

后端处理

• 客户端断开连接时，Gin 会检测到
• 服务层应检查 context 取消
• 长时间运行的操作应支持取消

性能优化

缓冲控制

X-Accel-Buffering: no

• 禁用 Nginx 缓冲
• 确保事件立即发送

批量发送

• 避免频繁发送小事件
• 合并相关数据

压缩

• SSE 不支持 gzip 压缩
• 数据量大时考虑压缩正文

并发安全

线程安全回调

safeCallback := callback
if callback != nil {
    safeCallback = func(event string, data interface{}) {
        mu.Lock()
        defer mu.Unlock()
        callback(event, data)
    }
}

• 使用互斥锁保护回调
• 避免并发写入

Goroutine 管理

var wg sync.WaitGroup

for i, file := range files {
    wg.Add(1)
    go func(idx int, f FileDiff) {
        defer wg.Done()
        // ...
    }(i, file)
}

wg.Wait()

• 等待所有 goroutine 完成
• 避免资源泄漏

测试

手动测试

curl -N -X POST http://localhost:8080/api/repos/1/review \
  -H "Content-Type: application/json" \
  -d '{"base":"main","head":"feature"}'

自动化测试

func TestSSEStream(t *testing.T) {
    // 创建测试服务器
    // 发送请求
    // 读取事件流
    // 验证事件序列
}

监控

关键指标

• 连接数
• 事件发送速率
• 错误率
• 响应时间

日志

log.Printf("SSE connected: %s", c.ClientIP())
log.Printf("SSE event: %s", event)
log.Printf("SSE disconnected: %s", c.ClientIP())

安全考虑

认证

• 所有 SSE 端点需要认证
• 使用会话 Cookie

速率限制

• 限制并发 SSE 连接数
• 限制事件发送频率

数据验证

• 验证输入参数
• 防止注入攻击

故障排查

常见问题

1. 事件不发送

   • 检查 Flusher 是否可用
   • 确认响应头设置正确
   • 检查 Nginx 缓冲配置

2. 连接断开

   • 检查超时设置
   • 确认网络稳定
   • 查看错误日志

3. 数据乱码

   • 确认 JSON 序列化正确
   • 检查字符编码
   • 验证事件格式

调试工具

• 浏览器开发者工具 Network 面板
• curl 命令行测试
• Wireshark 抓包