xieyinlu/ruoyi-ai
1
0
Fork 0
mirror of https://gitcode.com/ageerle/ruoyi-ai.git synced 2026-11-15 07:20:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: enhance coding harness, model security and system features

Browse Source
This commit is contained in:
ageerle
2026-09-08 23:33:29 +08:00
parent d4ed2b8672
commit bcff4fd9e3
259 changed files with 15183 additions and 2713 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
ruoyi-common/ruoyi-common-sse/pom.xml
View File

@@ -32,5 +32,10 @@
<groupId>org.ruoyi</groupId>
<artifactId>ruoyi-common-json</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

81
ruoyi-common/ruoyi-common-sse/src/main/java/org/ruoyi/common/sse/core/SseEmitterManager.java
View File

@@ -97,7 +97,8 @@ public class SseEmitterManager {
try {
emitter.send(SseEmitter.event().comment("disconnected"));
} catch (Exception exception) {
log.error(exception.getMessage());
log.error("sse_connection operation=DISCONNECT routeType=SESSION status=FAILED errorType={}",
errorType(exception));
}
emitter.complete();
}
@@ -115,16 +116,18 @@ public class SseEmitterManager {
}
SseEmitter emitter = SESSION_EMITTERS.get(sessionId);
if (emitter == null) {
log.warn("【SSE发送失败】sessionId: {} 没有活跃的SSE连接", sessionId);
log.warn("sse_delivery routeType=SESSION status=SKIPPED reason=NO_ACTIVE_CONNECTION activeSessionCount={}",
SESSION_EMITTERS.size());
return;
}
try {
log.debug("【SSE发送】sessionId: {}, event: {}", sessionId, eventDto.getEvent());
log.debug("sse_delivery routeType=SESSION status=SENDING eventType={} payloadChars={}",
safeEventType(eventDto), eventPayloadLength(eventDto));
emitter.send(SseEmitter.event()
.name(eventDto.getEvent())
.data(JSONUtil.toJsonStr(eventDto)));
} catch (Exception e) {
log.error("【SSE发送失败】sessionId: {}, error: {}", sessionId, e.getMessage());
log.error("sse_delivery routeType=SESSION status=FAILED errorType={}", errorType(e));
removeSessionEmitter(sessionId, emitter);
}
}
@@ -212,7 +215,8 @@ public class SseEmitterManager {
sseEmitter.send(SseEmitter.event().comment("disconnected"));
//sseEmitter.complete();
} catch (Exception exception) {
log.error(exception.getMessage());
log.error("sse_connection operation=DISCONNECT routeType=USER status=FAILED errorType={}",
errorType(exception));
}
emitters.remove(token);
} else {
@@ -321,14 +325,15 @@ public class SseEmitterManager {
public void sendEvent(Long userId, SseEventDto eventDto) {
Map<String, SseEmitter> emitters = USER_TOKEN_EMITTERS.get(userId);
if (MapUtil.isNotEmpty(emitters)) {
log.debug("【SSE发送】userId: {}, emitter数量: {}, event: {}", userId, emitters.size(), eventDto.getEvent());
log.debug("sse_delivery routeType=USER status=SENDING connectionCount={} eventType={} payloadChars={}",
emitters.size(), safeEventType(eventDto), eventPayloadLength(eventDto));
for (Map.Entry<String, SseEmitter> entry : emitters.entrySet()) {
try {
entry.getValue().send(SseEmitter.event()
.name(eventDto.getEvent())
.data(JSONUtil.toJsonStr(eventDto)));
} catch (Exception e) {
log.error("【SSE发送失败】userId: {}, token: {}, error: {}", userId, entry.getKey(), e.getMessage());
log.error("{}", sendFailureSummary(e));
SseEmitter remove = emitters.remove(entry.getKey());
if (remove != null) {
remove.complete();
@@ -336,11 +341,17 @@ public class SseEmitterManager {
}
}
} else {
log.warn("【SSE发送失败】userId: {} 没有活跃的SSE连接, 当前连接用户: {}", userId, USER_TOKEN_EMITTERS.keySet());
log.warn("sse_delivery routeType=USER status=SKIPPED reason=NO_ACTIVE_CONNECTION activeUserCount={}",
USER_TOKEN_EMITTERS.size());
USER_TOKEN_EMITTERS.remove(userId);
}
}
/** This boundary accepts neither routing identifiers nor raw exception text. */
static String sendFailureSummary(Exception failure) {
return "sse_delivery routeType=USER status=FAILED errorType=" + errorType(failure);
}
/**
* 本机全用户会话发送消息
@@ -365,8 +376,7 @@ public class SseEmitterManager {
broadcastMessage.setSessionId(sseMessageDto.getSessionId());
broadcastMessage.setEventDto(sseMessageDto.getEventDto());
RedisUtils.publish(SSE_TOPIC, broadcastMessage, consumer -> {
log.info("SSE发送主题订阅消息topic:{} session:{} session keys:{} message:{}",
SSE_TOPIC, sseMessageDto.getSessionId(), sseMessageDto.getUserIds(), sseMessageDto.getMessage());
log.info("{}", publishSummary(sseMessageDto));
});
}
@@ -379,7 +389,56 @@ public class SseEmitterManager {
SseMessageDto broadcastMessage = new SseMessageDto();
broadcastMessage.setMessage(message);
RedisUtils.publish(SSE_TOPIC, broadcastMessage, consumer -> {
log.info("SSE发送主题订阅消息topic:{} message:{}", SSE_TOPIC, message);
log.info("sse_publish routeType=BROADCAST recipientCount=-1 payloadChars={} eventType=none",
textLength(message));
});
}
static String publishSummary(SseMessageDto message) {
if (message == null) {
return "sse_publish routeType=UNKNOWN recipientCount=0 payloadChars=0 eventType=none";
}
String routeType;
int recipientCount;
if (message.getSessionId() != null && !message.getSessionId().isBlank()) {
routeType = "SESSION";
recipientCount = 1;
} else if (message.getUserIds() != null && !message.getUserIds().isEmpty()) {
routeType = "USER";
recipientCount = message.getUserIds().size();
} else {
routeType = "BROADCAST";
recipientCount = -1;
}
return "sse_publish routeType=" + routeType
+ " recipientCount=" + recipientCount
+ " payloadChars=" + (textLength(message.getMessage()) + eventPayloadLength(message.getEventDto()))
+ " eventType=" + safeEventType(message.getEventDto());
}
static String safeEventType(SseEventDto event) {
if (event == null || event.getEvent() == null) {
return "none";
}
return switch (event.getEvent()) {
case "content", "reasoning", "done", "error", "mcp_tool", "message" -> event.getEvent();
default -> "custom";
};
}
private static int eventPayloadLength(SseEventDto event) {
if (event == null) {
return 0;
}
return textLength(event.getContent()) + textLength(event.getReasoningContent())
+ textLength(event.getError());
}
private static int textLength(String value) {
return value == null ? 0 : value.length();
}
private static String errorType(Throwable failure) {
return failure == null ? "unknown" : failure.getClass().getName();
}
}

43
ruoyi-common/ruoyi-common-sse/src/main/java/org/ruoyi/common/sse/listener/SseTopicListener.java
View File

@@ -30,8 +30,7 @@ public class SseTopicListener implements ApplicationRunner, Ordered {
@Override
public void run(ApplicationArguments args) throws Exception {
sseEmitterManager.subscribeMessage((message) -> {
log.info("SSE主题订阅收到消息session:{} session keys={} message={}",
message.getSessionId(), message.getUserIds(), message.getMessage());
log.info("{}", deliverySummary(message));
// 优先按会话路由（对话流式响应）
if (StrUtil.isNotBlank(message.getSessionId())) {
if (message.getEventDto() != null) {
@@ -59,4 +58,44 @@ public class SseTopicListener implements ApplicationRunner, Ordered {
public int getOrder() {
return -1;
}
static String deliverySummary(SseMessageDto message) {
if (message == null) {
return "sse_subscription status=RECEIVED routeType=UNKNOWN recipientCount=0 payloadChars=0 eventType=none";
}
String routeType;
int recipientCount;
if (StrUtil.isNotBlank(message.getSessionId())) {
routeType = "SESSION";
recipientCount = 1;
} else if (CollUtil.isNotEmpty(message.getUserIds())) {
routeType = "USER";
recipientCount = message.getUserIds().size();
} else {
routeType = "BROADCAST";
recipientCount = -1;
}
int payloadChars = textLength(message.getMessage());
String eventType = "none";
if (message.getEventDto() != null) {
payloadChars += textLength(message.getEventDto().getContent())
+ textLength(message.getEventDto().getReasoningContent())
+ textLength(message.getEventDto().getError());
String rawEventType = message.getEventDto().getEvent();
if (rawEventType != null) {
eventType = switch (rawEventType) {
case "content", "reasoning", "done", "error", "mcp_tool", "message" -> rawEventType;
default -> "custom";
};
}
}
return "sse_subscription status=RECEIVED routeType=" + routeType
+ " recipientCount=" + recipientCount
+ " payloadChars=" + payloadChars
+ " eventType=" + eventType;
}
private static int textLength(String value) {
return value == null ? 0 : value.length();
}
}

5
ruoyi-common/ruoyi-common-sse/src/main/java/org/ruoyi/common/sse/utils/SseMessageUtils.java
View File

@@ -18,6 +18,7 @@ import org.ruoyi.common.sse.dto.SseMessageDto;
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class SseMessageUtils {
private static final String SAFE_ERROR_MESSAGE = "请求处理失败，请稍后重试";
private final static Boolean SSE_ENABLE = SpringUtils.getProperty("sse.enabled", Boolean.class, true);
private static SseEmitterManager MANAGER;
@@ -196,7 +197,7 @@ public class SseMessageUtils {
* @param error 错误信息
*/
public static void sendError(Long userId, String error) {
sendEvent(userId, SseEventDto.error(error));
sendEvent(userId, SseEventDto.error(SAFE_ERROR_MESSAGE));
}
/**
@@ -206,7 +207,7 @@ public class SseMessageUtils {
* @param error 错误信息
*/
public static void sendError(String sessionId, String error) {
sendEvent(sessionId, SseEventDto.error(error));
sendEvent(sessionId, SseEventDto.error(SAFE_ERROR_MESSAGE));
}
/**