SseEmitterUtil.java
package com.yumu.noveltranslator.util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
/**
* SSE 发射器工具类。
* 提供静态方法用于基本的 SSE 操作，同时作为 Spring Bean 提供 Redis Stream 支持以实现事件重放。
*/
@Component
public class SseEmitterUtil {
private static final Logger log = LoggerFactory.getLogger(SseEmitterUtil.class);
private static final long DEFAULT_TIMEOUT_MS = 300_000L; // 5 分钟，对齐 application.yaml
private static final int MAX_ACTIVE_SSE = Integer.getInteger("noveltrans.max.sse", 100);
/**
* Emitter 注册表：key=唯一ID，value=可取消的 future
*/
private static final Map<String, CompletableFuture<Void>> EMITTER_REGISTRY = new ConcurrentHashMap<>();
private static final AtomicInteger ACTIVE_COUNT = new AtomicInteger(0);
private final StringRedisTemplate stringRedisTemplate;
public SseEmitterUtil(StringRedisTemplate stringRedisTemplate) {
this.stringRedisTemplate = stringRedisTemplate;
}
// ==================== 静态方法（保持向后兼容） ====================
/**
* 注册 emitter，返回唯一 ID（用于心跳循环）
* @throws IllegalStateException 如果超过最大连接数
*/
public static String registerEmitter(SseEmitter emitter) {
int count = ACTIVE_COUNT.get();
if (count >= MAX_ACTIVE_SSE) {
throw new IllegalStateException("SSE 连接数已达上限: " + MAX_ACTIVE_SSE);
}
String emitterId = UUID.randomUUID().toString();
CompletableFuture<Void> cancellationSignal = new CompletableFuture<>();
EMITTER_REGISTRY.put(emitterId, cancellationSignal);
ACTIVE_COUNT.incrementAndGet();
emitter.onCompletion(() -> {
unregisterEmitter(emitterId);
});
emitter.onTimeout(() -> {
unregisterEmitter(emitterId);
emitter.complete();
});
emitter.onError((ex) -> {
if (ex != null) {
String msg = ex.getMessage();
if (msg != null &&
!(msg.contains("aborted") ||
msg.contains("中止") ||
msg.contains("reset") ||
msg.contains("broken pipe") ||
msg.contains("client abort"))) {
log.warn("SSE emitter error: {}", msg);
}
}
unregisterEmitter(emitterId);
emitter.complete();
});
return emitterId;
}
/**
* 检查 emitter 是否仍然活跃（心跳循环用）
*/
public static boolean isEmitterActive(String emitterId) {
CompletableFuture<Void> future = EMITTER_REGISTRY.get(emitterId);
return future != null && !future.isCancelled() && !future.isDone();
}
/**
* 注销 emitter（内部调用）
*/
private static void unregisterEmitter(String emitterId) {
CompletableFuture<Void> future = EMITTER_REGISTRY.remove(emitterId);
if (future != null) {
future.complete(null);
}
ACTIVE_COUNT.decrementAndGet();
}
/**
* 获取当前活跃的 SSE 连接数（监控用）
*/
public static int getActiveCount() {
return ACTIVE_COUNT.get();
}
/**
* 创建 SSE emitter（简单版，不注册到注册表）
* 适用于不需要心跳管理的场景
*/
public static SseEmitter createSseEmitter(Long timeout) {
SseEmitter emitter = new SseEmitter(timeout != null ? timeout : DEFAULT_TIMEOUT_MS);
emitter.onCompletion(() -> {});
emitter.onTimeout(emitter::complete);
emitter.onError((ex) -> {
if (ex != null) {
String msg = ex.getMessage();
if (msg != null &&
!(msg.contains("aborted") ||
msg.contains("中止") ||
msg.contains("reset") ||
msg.contains("broken pipe") ||
msg.contains("client abort"))) {
log.warn("SSE emitter error: {}", msg);
}
}
emitter.complete();
});
return emitter;
}
public static void sendError(SseEmitter emitter, String msg) {
try {
emitter.send(SseEmitter.event().data("ERROR: " + msg));
} catch (Exception ignored) {}
}
public static void sendDone(SseEmitter emitter) {
try {
emitter.send(SseEmitter.event().data("[DONE]"));
} catch (Exception ignored) {}
}
public static void sendData(SseEmitter emitter, String data) {
try {
emitter.send(SseEmitter.event().data(data));
} catch (Exception ignored) {}
}
/**
* 发送心跳事件，防止长翻译过程中 SSE 连接因超时而断开。
* 使用 SSE comment 格式（冒号前缀），前端会自动忽略此类注释行。
*/
public static void sendHeartbeat(SseEmitter emitter) {
try {
emitter.send(SseEmitter.event().comment("heartbeat"));
} catch (Exception ignored) {}
}
public static void complete(SseEmitter emitter) {
emitter.complete();
}
public static void completeWithError(SseEmitter emitter, Exception ex) {
emitter.completeWithError(ex);
}
// ==================== 实例方法（Redis Stream 支持） ====================
/**
* XADD Lua 脚本：向 Redis Stream 添加事件，带 MAXLEN 限制防止无限增长
*/
private static final String XADD_LUA =
"redis.call('XADD', KEYS[1], 'MAXLEN', '~', '10000', '*', 'event', ARGV[1]); return 'OK'";
private static final DefaultRedisScript<String> XADD_SCRIPT =
new DefaultRedisScript<>(XADD_LUA, String.class);
/**
* XRANGE Lua 脚本：从 Redis Stream 获取 lastEventId 之后的所有事件
*/
private static final String XRANGE_LUA =
"local results = redis.call('XRANGE', KEYS[1], '(' .. ARGV[1], '+'); " +
"local out = {}; " +
"for i = 1, #results, 1 do " +
" out[i] = results[i][2][2]; " +
"end; " +
"return out";
private static final DefaultRedisScript<List> XRANGE_SCRIPT =
new DefaultRedisScript<>(XRANGE_LUA, List.class);
/**
* 发布协作事件到 Redis Stream，支持断线重连后补发。
*
* @param projectId 项目ID
* @param eventType 事件类型（如 chapter.updated, comment.added）
* @param payload 事件载荷（JSON 字符串）
* @return 生成的事件 ID
*/
public String publishCollabEvent(String projectId, String eventType, String payload) {
String streamKey = "collab:events:" + projectId;
String eventId = UUID.randomUUID().toString();
long timestamp = System.currentTimeMillis();
String eventJson = String.format(
"{\"eventId\":\"%s\",\"type\":\"%s\",\"payload\":%s,\"timestamp\":%d}",
eventId, eventType, payload, timestamp);
stringRedisTemplate.execute(XADD_SCRIPT, List.of(streamKey), eventJson);
log.debug("发布协作事件: projectId={}, type={}, eventId={}", projectId, eventType, eventId);
return eventId;
}
/**
* 从 Redis Stream 重放指定项目自 lastEventId 之后的遗漏事件。
*
* @param projectId 项目ID
* @param lastEventId 上次收到的事件 ID（Redis Stream 条目 ID），为 null 时不重放
* @param emitter SSE emitter，用于发送事件
*/
public void replayMissedEvents(String projectId, String lastEventId, SseEmitter emitter) {
if (lastEventId == null || lastEventId.isBlank()) {
log.debug("无 lastEventId，跳过事件重放: projectId={}", projectId);
return;
}
String streamKey = "collab:events:" + projectId;
@SuppressWarnings("unchecked")
List<String> events = (List<String>) stringRedisTemplate.execute(XRANGE_SCRIPT, List.of(streamKey), lastEventId);
if (events == null || events.isEmpty()) {
log.debug("无遗漏事件需要重放: projectId={}, lastEventId={}", projectId, lastEventId);
return;
}
int sent = 0;
for (String eventJson : events) {
if (eventJson == null || eventJson.isEmpty()) {
continue;
}
try {
emitter.send(SseEmitter.event()
.name("replay")
.data(eventJson));
sent++;
} catch (Exception e) {
log.warn("重放事件失败: projectId={}, error={}", projectId, e.getMessage());
break;
}
}
log.info("重放完成: projectId={}, sent={} events", projectId, sent);
}
}