143 lines
5.7 KiB
Java
143 lines
5.7 KiB
Java
package com.webgame.webgamebackend.sse;
|
||
|
||
import com.webgame.webgamebackend.common.auth.AuthenticationProvider;
|
||
import com.webgame.webgamebackend.common.dto.ApiResponse;
|
||
import com.webgame.webgamebackend.common.event.SseEventPublisher;
|
||
import com.webgame.webgamebackend.common.event.SseEventType;
|
||
import jakarta.servlet.http.HttpServletResponse;
|
||
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.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;
|
||
|
||
/**
|
||
* SSE 订阅端点
|
||
*
|
||
* 提供 GET /sse/subscribe 端点建立持久化 SSE 连接。
|
||
* 心跳由 SseConnectionManager 统一批量管理，此处不再维护 per-connection 定时器。
|
||
*/
|
||
@Slf4j
|
||
@RestController
|
||
@RequestMapping("/sse")
|
||
@RequiredArgsConstructor
|
||
public class SseController {
|
||
|
||
/** SseEmitter 超时时间（30 分钟），超时后由 SseConnectionManager 的 onTimeout 回调清理 */
|
||
private static final long SSE_TIMEOUT_MS = 30 * 60 * 1000L;
|
||
|
||
private final SseConnectionManager connectionManager;
|
||
private final AuthenticationProvider authProvider;
|
||
private final SseEventPublisher eventPublisher;
|
||
|
||
@GetMapping(value = "/subscribe", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
|
||
public SseEmitter subscribe(
|
||
@RequestHeader(value = "Last-Event-ID", required = false) String lastEventId,
|
||
HttpServletResponse response) throws IOException {
|
||
|
||
// SSE 路径已排除在 AuthInterceptor 之外，此处手动校验登录态
|
||
String token = authProvider.getCurrentToken();
|
||
if (token == null || token.isBlank()) {
|
||
log.warn("[SSE] 缺少认证 token");
|
||
return writeAuthError(response, HttpServletResponse.SC_UNAUTHORIZED, "缺少认证信息，请先登录");
|
||
}
|
||
|
||
String userId = authProvider.getUserIdByToken(token);
|
||
if (userId == null) {
|
||
log.warn("[SSE] token 无效或已过期");
|
||
return writeAuthError(response, HttpServletResponse.SC_UNAUTHORIZED, "登录已过期，请重新登录");
|
||
}
|
||
|
||
// 设置认证上下文，后续 getCurrentUserId() 可用
|
||
authProvider.setContextToken(token, -1);
|
||
SseEmitter emitter = new SseEmitter(SSE_TIMEOUT_MS);
|
||
String connectionId = connectionManager.register(userId, emitter);
|
||
|
||
// 发送初始连接确认
|
||
try {
|
||
emitter.send(SseEmitter.event()
|
||
.name("connected")
|
||
.data("{\"message\":\"连接成功\"}"));
|
||
} catch (IOException | IllegalStateException e) {
|
||
log.warn("[SSE] 发送 connected 事件失败, userId={}, connectionId={}",
|
||
userId, connectionId, e);
|
||
connectionManager.remove(userId, connectionId);
|
||
}
|
||
|
||
if (lastEventId != null && !lastEventId.isBlank()) {
|
||
log.debug("[SSE] 客户端重连, Last-Event-ID={}, userId={}, connectionId={}",
|
||
lastEventId, userId, connectionId);
|
||
}
|
||
|
||
return emitter;
|
||
}
|
||
|
||
/**
|
||
* 测试用消息发布接口
|
||
*
|
||
* 支持广播（不传 userId）和定向推送（传 userId）两种模式。
|
||
* 可传入任意 JSON 数据作为消息体，事件类型固定为 CUSTOM。
|
||
*
|
||
* 请求体示例：
|
||
* <pre>
|
||
* // 广播
|
||
* { "data": { "msg": "hello" } }
|
||
*
|
||
* // 定向推送
|
||
* { "userId": "123", "data": { "msg": "hello" } }
|
||
* </pre>
|
||
*
|
||
* @param body 请求体，包含可选的 userId 和必填的 data
|
||
* @return 统一 API 响应
|
||
*/
|
||
@PostMapping("/publish")
|
||
public ApiResponse<Map<String, Object>> publish(@RequestBody Map<String, Object> body) {
|
||
@SuppressWarnings("unchecked")
|
||
Map<String, Object> data = (Map<String, Object>) body.get("data");
|
||
if (data == null || data.isEmpty()) {
|
||
return ApiResponse.fail(400, "data 字段不能为空");
|
||
}
|
||
|
||
String userId = body.get("userId") != null ? body.get("userId").toString() : null;
|
||
|
||
if (userId != null && !userId.isBlank()) {
|
||
eventPublisher.sendTo(userId, SseEventType.CUSTOM, data);
|
||
log.info("[SSE] 测试定向推送, userId={}, data={}", userId, data);
|
||
} else {
|
||
eventPublisher.broadcast(SseEventType.CUSTOM, data);
|
||
log.info("[SSE] 测试广播推送, data={}", data);
|
||
}
|
||
|
||
return ApiResponse.ok(Map.of(
|
||
"mode", userId != null && !userId.isBlank() ? "targeted" : "broadcast",
|
||
"userId", userId != null ? userId : "all",
|
||
"data", data
|
||
));
|
||
}
|
||
|
||
/**
|
||
* 向 SSE 客户端返回认证错误
|
||
*
|
||
* 直接写入 HttpServletResponse 返回 401 + SSE error 事件，
|
||
* 前端 SseClient.onopen 检测到 response.status === 401 后触发 onUnauthorized，
|
||
* 自动刷新 token 并重连。
|
||
*
|
||
* @return null，通知 Spring MVC 响应已处理完毕，无需 MessageConverter
|
||
*/
|
||
private SseEmitter writeAuthError(HttpServletResponse response, int status, String message) throws IOException {
|
||
response.setStatus(status);
|
||
response.setContentType(MediaType.TEXT_EVENT_STREAM_VALUE);
|
||
response.getWriter().write("event: error\ndata: {\"code\":" + status + ",\"message\":\"" + message + "\"}\n\n");
|
||
response.getWriter().flush();
|
||
return null;
|
||
}
|
||
}
|