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: 完善短剧创作、技能目录与媒体任务恢复

Browse Source 
保存原始创作输入，整合剧本打磨、分场规划与流式进度，接入审美及导演技能、角色声音和音乐；保持已批准媒体并完善请求去重与恢复。

统一增量 SQL 目录，补齐新安装字段并修正示例插入列，更新迁移说明；修复测试夹具与当前审阅、重启恢复行为的断言。验证：相关模块编译通过，短剧及 Atlas 回归 337 项、SSE 回归 3 项均通过。
This commit is contained in:
ageerle
2026-10-03 19:11:54 +08:00
parent 91b6b53ce9
commit ca6ef96200
169 changed files with 11185 additions and 1155 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -111,25 +111,36 @@ public class SseEmitterManager {
* @param eventDto SSE事件对象
*/
public void sendEvent(String sessionId, SseEventDto eventDto) {
if (!sendLocalEvent(sessionId, eventDto)) {
log.warn("sse_delivery routeType=SESSION status=SKIPPED reason=NO_ACTIVE_CONNECTION activeSessionCount={}",
SESSION_EMITTERS.size());
}
}
/** Deliver locally in order; a terminal event is flushed before its connection is removed. */
public boolean sendLocalEvent(String sessionId, SseEventDto eventDto) {
if (sessionId == null || eventDto == null) {
return;
return false;
}
SseEmitter emitter = SESSION_EMITTERS.get(sessionId);
if (emitter == null) {
log.warn("sse_delivery routeType=SESSION status=SKIPPED reason=NO_ACTIVE_CONNECTION activeSessionCount={}",
SESSION_EMITTERS.size());
return;
return false;
}
try {
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)));
synchronized (emitter) {
if (SESSION_EMITTERS.get(sessionId) != emitter) return false;
emitter.send(SseEmitter.event()
.name(eventDto.getEvent())
.data(JSONUtil.toJsonStr(eventDto)));
if (Boolean.TRUE.equals(eventDto.getDone())) removeSessionEmitter(sessionId, emitter);
}
} catch (Exception e) {
log.error("sse_delivery routeType=SESSION status=FAILED errorType={}", errorType(e));
removeSessionEmitter(sessionId, emitter);
}
return true;
}
private void removeSessionEmitter(String sessionId, SseEmitter emitter) {

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

@@ -126,6 +126,8 @@ public class SseMessageUtils {
if (!isEnable() || sessionId == null) {
return;
}
// The producer normally owns this emitter. Do not race a Redis delivery against disconnect.
if (MANAGER.sendLocalEvent(sessionId, eventDto)) return;
SseMessageDto dto = new SseMessageDto();
dto.setSessionId(sessionId);
dto.setEventDto(eventDto);
@@ -210,6 +212,13 @@ public class SseMessageUtils {
sendEvent(sessionId, SseEventDto.error(SAFE_ERROR_MESSAGE));
}
/** Final error only: unlike recoverable error events, this flushes and closes the stream. */
public static void sendErrorAndComplete(String sessionId, String error) {
SseEventDto terminal = SseEventDto.error(SAFE_ERROR_MESSAGE);
terminal.setDone(true);
sendEvent(sessionId, terminal);
}
/**
* 是否开启
*/

51
ruoyi-common/ruoyi-common-sse/src/test/java/org/ruoyi/common/sse/core/SseTerminalDeliveryTest.java Normal file
View File

@@ -0,0 +1,51 @@
package org.ruoyi.common.sse.core;
import cn.hutool.extra.spring.SpringUtil;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.ruoyi.common.sse.dto.SseEventDto;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.test.util.ReflectionTestUtils;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.Mockito.*;
@Tag("dev")
class SseTerminalDeliveryTest {
private SseEmitterManager manager;
private Map<String, SseEmitter> emitters;
@BeforeEach @SuppressWarnings("unchecked") void setup() {
var beans = new DefaultListableBeanFactory();
beans.registerSingleton("scheduler", mock(ScheduledExecutorService.class));
new SpringUtil().postProcessBeanFactory(beans);
manager = new SseEmitterManager();
emitters = (Map<String, SseEmitter>) ReflectionTestUtils.getField(SseEmitterManager.class, "SESSION_EMITTERS");
emitters.clear();
}
@Test void doneIsSentBeforeRemovalAndCompletion() throws Exception { assertTerminal(SseEventDto.done(), false); }
@Test void redisTerminalErrorIsSentBeforeRemovalAndCompletion() throws Exception {
var error = SseEventDto.error("safe failure"); error.setDone(true); assertTerminal(error, true);
}
private void assertTerminal(SseEventDto terminal, boolean viaRedisListener) throws Exception {
var emitter = mock(SseEmitter.class); emitters.put("s", emitter);
doAnswer(call -> { assertSame(emitter, emitters.get("s")); return null; }).when(emitter).send(any(SseEmitter.SseEventBuilder.class));
doAnswer(call -> { assertFalse(emitters.containsKey("s")); return null; }).when(emitter).complete();
if (viaRedisListener) manager.sendEvent("s", terminal); else assertTrue(manager.sendLocalEvent("s", terminal));
var ordered = inOrder(emitter); ordered.verify(emitter).send(any(SseEmitter.SseEventBuilder.class)); ordered.verify(emitter).complete();
assertFalse(manager.sendLocalEvent("s", terminal));
}
@Test void normalContentAndRecoverableErrorKeepTheConnectionOpen() throws Exception {
var emitter = mock(SseEmitter.class); emitters.put("s", emitter);
assertTrue(manager.sendLocalEvent("s", SseEventDto.content("正文")));
assertTrue(manager.sendLocalEvent("s", SseEventDto.error("recoverable")));
assertSame(emitter, emitters.get("s")); verify(emitter, never()).complete();
}
}