fix: 暂存

src/main/java/com/webgame/webgamebackend/sse/SseController.java

@@ -1,33 +1,22 @@
 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.scheduling.TaskScheduler;
 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.RequestHeader;
 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;
+import java.time.Duration;
+import java.util.concurrent.ScheduledFuture;
 
 /**
- * SSE 连接端点
- *
- * 提供 GET /sse/subscribe 端点，建立持久化 SSE 连接。
- * 连接成功后立即发送 connected 事件确认握手，之后每 30 秒发送心跳保活。
- *
- * 鉴权由 SaTokenConfigure 拦截器统一处理，前端通过 fetch-event-source
- * 的 headers 传 saToken，401 时由前端 err.response.status 直接读取。
+ * SSE 订阅端点
  */
 @Slf4j
 @RestController
@@ -35,93 +24,51 @@ import java.util.concurrent.TimeUnit;
 @RequiredArgsConstructor
 public class SseController {
 
+    private static final long SSE_TIMEOUT_MS = 30 * 60 * 1000L;
+    private static final Duration HEARTBEAT_INTERVAL = Duration.ofSeconds(30);
+
     private final SseConnectionManager connectionManager;
+    private final TaskScheduler sseTaskScheduler;
 
-    /**
-     * 建立 SSE 订阅连接
-     *
-     * 鉴权由拦截器处理，此处直接获取登录用户 ID。
-     *
-     * @return SseEmitter 实例（30 分钟超时）
-     */
     @GetMapping(value = "/subscribe", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
-    public SseEmitter subscribe() {
+    public SseEmitter subscribe(
+            @RequestHeader(value = "Last-Event-ID", required = false) String lastEventId) {
         String userId = StpUtil.getLoginIdAsString();
-        SseEmitter emitter = new SseEmitter(30 * 60 * 1000L); // 30 分钟超时
+        SseEmitter emitter = new SseEmitter(SSE_TIMEOUT_MS);
+        String connectionId = connectionManager.register(userId, emitter);
 
-        connectionManager.register(userId, emitter);
+        ScheduledFuture<?> heartbeatTask = sseTaskScheduler.scheduleAtFixedRate(() -> {
+            try {
+                emitter.send(SseEmitter.event().name("heartbeat").data(""));
+            } catch (IOException | IllegalStateException e) {
+                log.debug("[SSE] 心跳发送失败, userId={}, connectionId={}, reason={}",
+                        userId, connectionId, e.getMessage());
+                connectionManager.remove(userId, connectionId);
+            }
+        }, HEARTBEAT_INTERVAL);
+
+        Runnable cancelHeartbeat = () -> heartbeatTask.cancel(false);
+        emitter.onCompletion(cancelHeartbeat);
+        emitter.onTimeout(cancelHeartbeat);
+        emitter.onError(error -> cancelHeartbeat.run());
 
-        // 发送初始连接确认
         try {
             emitter.send(SseEmitter.event()
                     .name("connected")
                     .data("{\"message\":\"连接成功\"}"));
-        } catch (IOException e) {
-            log.warn("[SSE] 发送 connected 事件失败, userId={}", userId);
+        } catch (IOException | IllegalStateException e) {
+            log.warn("[SSE] 发送 connected 事件失败, userId={}, connectionId={}",
+                    userId, connectionId, e);
+            connectionManager.remove(userId, connectionId);
         }
 
-        // 启动心跳定时任务
-        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);
+        if (lastEventId != null && !lastEventId.isBlank()) {
+            log.debug("[SSE] 客户端重连, Last-Event-ID={}, userId={}, connectionId={}",
+                    lastEventId, userId, connectionId);
+        }
 
-        // 连接关闭时停止心跳
-        emitter.onCompletion(heartbeat::shutdown);
-        emitter.onTimeout(heartbeat::shutdown);
-        emitter.onError(e -> heartbeat.shutdown());
-
-        log.info("[SSE] 新连接建立, userId={}, 当前在线={}", userId, connectionManager.getOnlineCount());
+        log.info("[SSE] 连接建立成功, userId={}, connectionId={}, onlineConnections={}",
+                userId, connectionId, 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);
-        }
-    }
 }