128 lines
4.8 KiB
Java
128 lines
4.8 KiB
Java
package com.webgame.webgamebackend.sse;
|
||
|
||
import cn.dev33.satoken.stp.StpUtil;
|
||
import com.alibaba.fastjson2.JSON;
|
||
import com.webgame.webgamebackend.common.event.SseEvent;
|
||
import com.webgame.webgamebackend.common.event.SseEventType;
|
||
import lombok.RequiredArgsConstructor;
|
||
import lombok.extern.slf4j.Slf4j;
|
||
import org.springframework.http.MediaType;
|
||
import org.springframework.web.bind.annotation.GetMapping;
|
||
import org.springframework.web.bind.annotation.PostMapping;
|
||
import org.springframework.web.bind.annotation.RequestBody;
|
||
import org.springframework.web.bind.annotation.RequestMapping;
|
||
import org.springframework.web.bind.annotation.RestController;
|
||
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
|
||
|
||
import java.io.IOException;
|
||
import java.util.Map;
|
||
import java.util.concurrent.Executors;
|
||
import java.util.concurrent.ScheduledExecutorService;
|
||
import java.util.concurrent.TimeUnit;
|
||
|
||
/**
|
||
* SSE 连接端点
|
||
*
|
||
* 提供 GET /sse/subscribe 端点，建立持久化 SSE 连接。
|
||
* 连接成功后立即发送 connected 事件确认握手，之后每 30 秒发送心跳保活。
|
||
*
|
||
* 鉴权由 SaTokenConfigure 拦截器统一处理，前端通过 fetch-event-source
|
||
* 的 headers 传 saToken，401 时由前端 err.response.status 直接读取。
|
||
*/
|
||
@Slf4j
|
||
@RestController
|
||
@RequestMapping("/sse")
|
||
@RequiredArgsConstructor
|
||
public class SseController {
|
||
|
||
private final SseConnectionManager connectionManager;
|
||
|
||
/**
|
||
* 建立 SSE 订阅连接
|
||
*
|
||
* 鉴权由拦截器处理，此处直接获取登录用户 ID。
|
||
*
|
||
* @return SseEmitter 实例（30 分钟超时）
|
||
*/
|
||
@GetMapping(value = "/subscribe", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
|
||
public SseEmitter subscribe() {
|
||
String userId = StpUtil.getLoginIdAsString();
|
||
SseEmitter emitter = new SseEmitter(30 * 60 * 1000L); // 30 分钟超时
|
||
|
||
connectionManager.register(userId, emitter);
|
||
|
||
// 发送初始连接确认
|
||
try {
|
||
emitter.send(SseEmitter.event()
|
||
.name("connected")
|
||
.data("{\"message\":\"连接成功\"}"));
|
||
} catch (IOException e) {
|
||
log.warn("[SSE] 发送 connected 事件失败, userId={}", userId);
|
||
}
|
||
|
||
// 启动心跳定时任务
|
||
ScheduledExecutorService heartbeat = Executors.newSingleThreadScheduledExecutor(
|
||
r -> new Thread(r, "sse-heartbeat-" + userId.substring(0, Math.min(8, userId.length())))
|
||
);
|
||
heartbeat.scheduleAtFixedRate(() -> {
|
||
try {
|
||
emitter.send(SseEmitter.event()
|
||
.name("heartbeat")
|
||
.data(""));
|
||
} catch (Exception e) {
|
||
// 客户端已断开（IOException）或响应已不可用（AsyncRequestNotUsableException）
|
||
connectionManager.remove(userId);
|
||
heartbeat.shutdown();
|
||
}
|
||
}, 30, 30, TimeUnit.SECONDS);
|
||
|
||
// 连接关闭时停止心跳
|
||
emitter.onCompletion(heartbeat::shutdown);
|
||
emitter.onTimeout(heartbeat::shutdown);
|
||
emitter.onError(e -> heartbeat.shutdown());
|
||
|
||
log.info("[SSE] 新连接建立, userId={}, 当前在线={}", userId, connectionManager.getOnlineCount());
|
||
return emitter;
|
||
}
|
||
|
||
// ==================== 测试接口 ====================
|
||
|
||
/**
|
||
* 查看当前 SSE 在线连接数
|
||
*
|
||
* GET /sse/test/online
|
||
*/
|
||
@GetMapping("/test/online")
|
||
public Map<String, Object> onlineCount() {
|
||
return Map.of("onlineCount", connectionManager.getOnlineCount());
|
||
}
|
||
|
||
/**
|
||
* 手动发送广播事件（测试用）
|
||
*
|
||
* POST /sse/test/broadcast
|
||
* Body: { "event": "room_created", "data": { "roomId": "test", ... } }
|
||
*
|
||
* event 可选值: room_created / room_updated / room_removed / kicked / game_started / balance_chg
|
||
*/
|
||
@PostMapping("/test/broadcast")
|
||
public Map<String, Object> testBroadcast(@RequestBody Map<String, Object> body) {
|
||
String eventType = (String) body.get("event");
|
||
Object data = body.get("data");
|
||
|
||
if (eventType == null || data == null) {
|
||
return Map.of("success", false, "error", "event 和 data 字段为必填");
|
||
}
|
||
|
||
try {
|
||
SseEventType type = SseEventType.valueOf(eventType.toUpperCase());
|
||
SseEvent event = SseEvent.broadcast(type, JSON.toJSONString(data));
|
||
connectionManager.broadcast(event);
|
||
log.info("[SSE] 测试广播, type={}, data={}", eventType, JSON.toJSONString(data));
|
||
return Map.of("success", true, "eventType", eventType, "onlineCount", connectionManager.getOnlineCount());
|
||
} catch (IllegalArgumentException e) {
|
||
return Map.of("success", false, "error", "无效的事件类型: " + eventType);
|
||
}
|
||
}
|
||
}
|