From 0f240361bbc601fa7a4da051fdc8ce541847e54b Mon Sep 17 00:00:00 2001
From: Azure <983547216@qq.com>
Date: Mon, 22 Jun 2026 13:40:56 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=20SseConnectionManag?=
 =?UTF-8?q?er=20SSE=20=E8=BF=9E=E6=8E=A5=E7=AE=A1=E7=90=86=E5=99=A8?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../sse/SseConnectionManager.java             | 134 ++++++++++++++++++
 1 file changed, 134 insertions(+)
 create mode 100644 src/main/java/com/webgame/webgamebackend/sse/SseConnectionManager.java

diff --git a/src/main/java/com/webgame/webgamebackend/sse/SseConnectionManager.java b/src/main/java/com/webgame/webgamebackend/sse/SseConnectionManager.java
new file mode 100644
index 0000000..ee3c9e1
--- /dev/null
+++ b/src/main/java/com/webgame/webgamebackend/sse/SseConnectionManager.java
@@ -0,0 +1,134 @@
+package com.webgame.webgamebackend.sse;
+
+import com.webgame.webgamebackend.common.event.SseEvent;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * SSE 连接管理器
+ *
+ * 维护 userId → SseEmitter 映射，提供广播和定向推送能力。
+ * 使用 ConcurrentHashMap 保证线程安全。
+ */
+@Slf4j
+@Component
+public class SseConnectionManager {
+
+    /** userId → SseEmitter 映射表，线程安全 */
+    private final ConcurrentHashMap<String, SseEmitter> emitters = new ConcurrentHashMap<>();
+
+    /**
+     * 注册新的 SSE 连接
+     *
+     * @param userId  用户 accountId
+     * @param emitter SseEmitter 实例
+     */
+    public void register(String userId, SseEmitter emitter) {
+        // 如果已有旧连接，先关闭再替换
+        SseEmitter old = emitters.put(userId, emitter);
+        if (old != null) {
+            try {
+                old.complete();
+            } catch (Exception ignored) {
+            }
+        }
+        // 注册清理回调
+        emitter.onCompletion(() -> {
+            log.debug("[SSE] 连接正常关闭, userId={}", userId);
+            emitters.remove(userId, emitter);
+        });
+        emitter.onTimeout(() -> {
+            log.debug("[SSE] 连接超时, userId={}", userId);
+            emitters.remove(userId, emitter);
+        });
+        emitter.onError(throwable -> {
+            log.debug("[SSE] 连接异常, userId={}, reason={}", userId, throwable.getMessage());
+            emitters.remove(userId, emitter);
+        });
+
+        log.info("[SSE] 用户连接成功, userId={}, 当前在线={}", userId, emitters.size());
+    }
+
+    /**
+     * 移除连接
+     *
+     * @param userId 用户 accountId
+     */
+    public void remove(String userId) {
+        SseEmitter removed = emitters.remove(userId);
+        if (removed != null) {
+            try {
+                removed.complete();
+            } catch (Exception ignored) {
+            }
+            log.info("[SSE] 用户连接已移除, userId={}, 当前在线={}", userId, emitters.size());
+        }
+    }
+
+    /**
+     * 广播事件给所有在线用户
+     *
+     * @param event SSE 事件
+     */
+    public void broadcast(SseEvent event) {
+        if (emitters.isEmpty()) return;
+
+        log.debug("[SSE] 广播事件, type={}, 接收人数={}", event.type(), emitters.size());
+        for (Map.Entry<String, SseEmitter> entry : emitters.entrySet()) {
+            sendEvent(entry.getValue(), event, entry.getKey());
+        }
+    }
+
+    /**
+     * 发送事件给指定用户
+     *
+     * @param userId 目标用户 accountId
+     * @param event  SSE 事件
+     */
+    public void sendTo(String userId, SseEvent event) {
+        SseEmitter emitter = emitters.get(userId);
+        if (emitter == null) {
+            log.debug("[SSE] 目标用户不在线, userId={}, eventType={}", userId, event.type());
+            return;
+        }
+
+        log.debug("[SSE] 定向推送, userId={}, eventType={}", userId, event.type());
+        sendEvent(emitter, event, userId);
+    }
+
+    /**
+     * 获取当前在线连接数
+     *
+     * @return 在线用户数
+     */
+    public int getOnlineCount() {
+        return emitters.size();
+    }
+
+    /**
+     * 向单个 SseEmitter 发送事件
+     *
+     * 捕获 IOException（客户端已断开但还未触发回调），
+     * 此时安全移除连接。
+     */
+    private void sendEvent(SseEmitter emitter, SseEvent event, String userId) {
+        try {
+            emitter.send(SseEmitter.event()
+                    .id(event.id())
+                    .name(event.type().name().toLowerCase())
+                    .data(event.data()));
+        } catch (IOException e) {
+            log.debug("[SSE] 推送失败（客户端可能已断开）, userId={}, reason={}", userId, e.getMessage());
+            emitters.remove(userId);
+            try {
+                emitter.completeWithError(e);
+            } catch (Exception ignored) {
+            }
+        }
+    }
+}