erpnew/docs/sse.md

SSE (Server-Sent Events) 模块

基于 Django 和 Django REST Framework 的服务器推送事件实现，支持多商户隔离。

功能特性

• ✅ 使用 DRF 处理请求和响应（非流式端点）
• ✅ 支持多种请求格式（JSON、Form Data、Multipart）
• ✅ 自动数据验证和序列化
• ✅ 异步支持，高并发处理
• ✅ 心跳机制保持连接活跃
• ✅ 自动清理断开的连接
• ✅ 连接状态监控
• ✅ 多商户消息隔离
• ✅ JWT认证和商户验证

API 端点

1. 订阅 SSE 事件流

端点: GET /sse/

客户端连接此端点保持长连接，接收服务器推送的事件。

认证要求: 需要JWT认证，且用户必须有关联的商户

示例:

# 需要提供JWT token
curl -N -H "Authorization: Bearer YOUR_JWT_TOKEN" http://localhost:8000/sse/

JavaScript 示例:

// 需要在连接时提供认证头
const eventSource = new EventSource('/sse/', {
    headers: {
        'Authorization': `Bearer ${yourJwtToken}`
    }
});

eventSource.onmessage = function(event) {
    const data = JSON.parse(event.data);
    console.log('收到消息:', data);
    console.log('商户ID:', data.merchant_id);
};

注意: 浏览器原生的EventSource不支持自定义请求头，因此推荐使用EventSource polyfill或fetch实现。

---

2. 推送事件到当前商户的客户端

端点: POST /sse/push/

向当前用户所属商户的所有已连接客户端广播消息。

认证要求: 需要JWT认证，且用户必须有关联的商户

请求参数:

• 自动使用当前用户的商户ID
• 固定发送测试消息：'订单已支付'
• 固定事件类型：'order_paid'
• 固定对象ID：12345

请求示例:

curl -X POST http://localhost:8000/sse/push/ \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_JWT_TOKEN"

响应示例:

{
    "status": "ok",
    "message": "Test event broadcasted to your merchant",
    "merchant_id": 1,
    "clients": 3
}

---

3. 获取连接状态

端点: GET /sse/status/

查询当前 SSE 服务器的状态和连接数。

认证要求: 需要JWT认证，且用户必须有关联的商户

示例:

curl -H "Authorization: Bearer YOUR_JWT_TOKEN" http://localhost:8000/sse/status/

响应示例:

{
    "status": "running",
    "total_clients": 10,
    "merchant_clients": 3,
    "merchant_id": 1,
    "message": "SSE server is running with 10 total connections, 3 for your merchant"
}

4. 关闭商户连接

端点: POST /sse/shutdown/

关闭当前用户所属商户的所有SSE连接。

认证要求: 需要JWT认证，且用户必须有关联的商户

示例:

curl -X POST http://localhost:8000/sse/shutdown/ \
  -H "Authorization: Bearer YOUR_JWT_TOKEN"

响应示例:

{
    "status": "ok",
    "message": "Shutdown signal sent to your merchant's SSE connections",
    "merchant_id": 1,
    "clients": 3
}

启动服务器

使用 Uvicorn (ASGI 服务器) 启动：

# 开发环境
uvicorn flower.asgi:application --reload --host 0.0.0.0 --port 8000

# 生产环境
uvicorn flower.asgi:application --host 0.0.0.0 --port 8000 --workers 4

测试

打开 sse_test.html 在浏览器中测试：

1. 点击"连接 SSE"建立连接
2. 输入消息
3. 点击"发送 (JSON)"或"发送 (Form Data)"测试不同格式
4. 点击"获取连接状态"查看当前连接数
5. 打开多个浏览器标签测试广播功能

Python 客户端示例

import requests
import sseclient  # pip install sseclient-py

# 订阅事件
response = requests.get('http://localhost:8000/sse/', stream=True)
client = sseclient.SSEClient(response)

for event in client.events():
    print(f'收到消息: {event.data}')

技术实现

• 流式响应: 使用 StreamingHttpResponse 实现 SSE 连接
• 队列机制: 每个连接对应一个 queue.Queue，按商户组织
• 心跳: 30 秒超时，自动发送心跳保持连接
• DRF 集成: 非流式端点使用 DRF 的 @api_view 和序列化器
• 多格式支持: 自动解析 JSON、Form Data、Multipart 等格式
• 商户隔离: 所有消息按商户隔离，确保数据安全
• JWT认证: 使用 DRF Simple JWT 进行身份验证

消息结构

连接成功消息

{
  "type": "connected",
  "message": "SSE connection established",
  "merchant_id": 1
}

业务事件消息

{
  "mode": "simple_message",
  "type": "order_paid",
  "message": "订单已支付",
  "object_id": 12345,
  "merchant_id": 1
}

服务器关闭消息

{
  "type": "server_shutdown",
  "message": "Server shutting down your connections, please reconnect later"
}

注意事项

1. 必须使用 ASGI 服务器（如 Uvicorn、Daphne）运行
2. 不支持使用传统的 WSGI 服务器（如 Gunicorn + WSGI）
3. 如果使用 Nginx，需要禁用缓冲：X-Accel-Buffering: no
4. 所有SSE端点都需要JWT认证，且用户必须有关联的商户
5. 浏览器原生的EventSource不支持自定义请求头，推荐使用polyfill或fetch实现
6. SSE 使用 GET 请求，注意 CORS 配置