chenh
/
imeeting
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor(realtime): 重构实时会议会话状态以支持恢复配置 

在 LocalRealtimeAsrChannel 中增加恢复配置处理逻辑，
并在 RealtimeMeetingSessionStateService 中新增
rememberResumeConfig 方法，以完善实时会议恢复
配置的保存与传递，确保会议恢复时能正确应用配置。
dev_na
chenhao 2026-07-07 09:04:38 +08:00
parent 9d0edbfa00
commit b06b7585d8
5 changed files with 154 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
backend/src/main/java/com/imeeting/dto/biz/RealtimeMeetingResumeConfig.java
View File

@ -31,4 +31,6 @@ public class RealtimeMeetingResumeConfig {
private Long hotWordGroupId;
@Schema(description = "腾讯说话人上下文 ID")
private String speakerContextId;
@Schema(description = "鏈湴 ASR 浼氳瘽 session_id")
private String upstreamSessionId;
}

2
backend/src/main/java/com/imeeting/service/biz/RealtimeMeetingSessionStateService.java
View File

@ -13,6 +13,8 @@ public interface RealtimeMeetingSessionStateService {
void rememberSpeakerContext(Long meetingId, String speakerContextId);
void rememberUpstreamSessionId(Long meetingId, String upstreamSessionId);
void assertCanOpenSession(Long meetingId);
boolean activate(Long meetingId, String connectionId);

16
backend/src/main/java/com/imeeting/service/biz/impl/RealtimeMeetingSessionStateServiceImpl.java
View File

@ -83,6 +83,22 @@ public class RealtimeMeetingSessionStateServiceImpl implements RealtimeMeetingSe
writeState(state);
}
@Override
public void rememberUpstreamSessionId(Long meetingId, String upstreamSessionId) {
if (meetingId == null || upstreamSessionId == null || upstreamSessionId.isBlank()) {
return;
}
RealtimeMeetingSessionState state = getOrCreateState(meetingId);
RealtimeMeetingResumeConfig resumeConfig = state.getResumeConfig();
if (resumeConfig == null) {
resumeConfig = new RealtimeMeetingResumeConfig();
state.setResumeConfig(resumeConfig);
}
resumeConfig.setUpstreamSessionId(upstreamSessionId.trim());
state.setUpdatedAt(System.currentTimeMillis());
writeState(state);
}
@Override
public void assertCanOpenSession(Long meetingId) {
RealtimeMeetingSessionStatusVO status = getStatus(meetingId);

11
backend/src/main/java/com/imeeting/service/biz/impl/RealtimeMeetingSocketSessionServiceImpl.java
View File

@ -3,6 +3,7 @@ package com.imeeting.service.biz.impl;
import com.imeeting.common.MeetingConstants;
import com.imeeting.dto.biz.AiModelVO;
import com.imeeting.dto.biz.RealtimeMeetingResumeConfig;
import com.imeeting.dto.biz.RealtimeMeetingSessionStatusVO;
import com.imeeting.dto.biz.RealtimeSocketSessionData;
import com.imeeting.dto.biz.RealtimeSocketSessionVO;
import com.imeeting.entity.biz.Meeting;
@ -64,6 +65,9 @@ public class RealtimeMeetingSocketSessionServiceImpl implements RealtimeMeetingS
throw new RuntimeException("ASR 模型未配置 WebSocket 地址");
}
RealtimeMeetingSessionStatusVO existingStatus = realtimeMeetingSessionStateService.getStatus(meetingId);
RealtimeMeetingResumeConfig existingConfig = existingStatus == null ? null : existingStatus.getResumeConfig();
RealtimeMeetingResumeConfig resumeConfig = new RealtimeMeetingResumeConfig();
resumeConfig.setAsrModelId(asrModelId);
resumeConfig.setMode(mode);
@ -73,9 +77,10 @@ public class RealtimeMeetingSocketSessionServiceImpl implements RealtimeMeetingS
resumeConfig.setEnableItn(enableItn);
resumeConfig.setEnableTextRefine(enableTextRefine);
resumeConfig.setSaveAudio(saveAudio);
RealtimeMeetingResumeConfig existingConfig = realtimeMeetingSessionStateService.getStatus(meetingId) == null
? null
: realtimeMeetingSessionStateService.getStatus(meetingId).getResumeConfig();
if (existingConfig != null) {
resumeConfig.setSpeakerContextId(existingConfig.getSpeakerContextId());
resumeConfig.setUpstreamSessionId(existingConfig.getUpstreamSessionId());
}
List<Map<String, Object>> effectiveHotwords = (hotwords == null || hotwords.isEmpty())
? (existingConfig == null ? List.of() : existingConfig.getHotwords())
: hotwords;

127
backend/src/main/java/com/imeeting/service/realtime/impl/LocalRealtimeAsrChannel.java
View File

@ -5,6 +5,8 @@ import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.imeeting.dto.biz.AiModelVO;
import com.imeeting.dto.biz.RealtimeMeetingResumeConfig;
import com.imeeting.dto.biz.RealtimeMeetingSessionStatusVO;
import com.imeeting.dto.biz.RealtimeMeetingTranscriptCacheItem;
import com.imeeting.enums.ModelProviderEnum;
import com.imeeting.service.biz.RealtimeMeetingSessionStateService;
@ -23,6 +25,7 @@ import java.util.ArrayList;
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.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
@ -39,6 +42,12 @@ public class LocalRealtimeAsrChannel implements RealtimeAsrChannel {
private static final String STATE_UPSTREAM_SEND_CHAIN = "upstreamSendChain";
private static final String STATE_START_MESSAGE_SENT = "startMessageSent";
private static final String STATE_PENDING_AUDIO_FRAMES = "pendingAudioFrames";
private static final String STATE_LAST_START_MESSAGE = "lastStartMessage";
private static final String STATE_UPSTREAM_SESSION_ID = "upstreamSessionId";
private static final String STATE_RECONNECTING = "reconnecting";
private static final String STATE_RECONNECT_ATTEMPT = "reconnectAttempt";
private static final int MAX_RECONNECT_ATTEMPTS = 3;
private static final int MAX_RECONNECT_BUFFER_FRAMES = 40;
private static final CompletableFuture<Void> COMPLETED = CompletableFuture.completedFuture(null);
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
private static final String STOP_MESSAGE = "{\"type\":\"stop\"}";
@ -121,10 +130,12 @@ public class LocalRealtimeAsrChannel implements RealtimeAsrChannel {
}
initializeFrontendState(context);
if (looksLikeStartMessage(payload)) {
String startMessage = ensureStartMessageSessionId(context, payload);
context.getFrontendState().put(STATE_START_MESSAGE_SENT, Boolean.TRUE);
context.getChannelState().put(STATE_LAST_START_MESSAGE, startMessage);
if (!Boolean.TRUE.equals(context.getChannelState().get(STATE_START_MESSAGE_FORWARDED))) {
context.getChannelState().put(STATE_START_MESSAGE_FORWARDED, Boolean.TRUE);
sendUpstreamOrdered(context, () -> upstreamSocket.sendText(payload, true), "text-start");
sendUpstreamOrdered(context, () -> upstreamSocket.sendText(startMessage, true), "text-start");
}
flushPendingAudioFrames(context, upstreamSocket);
return;
@ -142,6 +153,10 @@ public class LocalRealtimeAsrChannel implements RealtimeAsrChannel {
return;
}
initializeFrontendState(context);
if (Boolean.TRUE.equals(context.getChannelState().get(STATE_RECONNECTING))) {
queuePendingAudioFrame(context, payload);
return;
}
if (!Boolean.TRUE.equals(context.getFrontendState().get(STATE_START_MESSAGE_SENT))) {
queuePendingAudioFrame(context, payload);
return;
@ -258,6 +273,9 @@ public class LocalRealtimeAsrChannel implements RealtimeAsrChannel {
frontendState.put(STATE_PENDING_AUDIO_FRAMES, pendingFrames);
}
pendingFrames.add(payload);
while (pendingFrames.size() > MAX_RECONNECT_BUFFER_FRAMES) {
pendingFrames.remove(0);
}
}
}
@ -283,6 +301,106 @@ public class LocalRealtimeAsrChannel implements RealtimeAsrChannel {
return context.getRawSession() == null ? null : context.getRawSession().getId();
}
private String ensureStartMessageSessionId(RealtimeAsrChannelContext context, String payload) {
try {
JsonNode root = OBJECT_MAPPER.readTree(payload);
ObjectNode mutableRoot = root.isObject() ? (ObjectNode) root : OBJECT_MAPPER.createObjectNode();
JsonNode payloadNode = mutableRoot.path("payload");
ObjectNode mutablePayload = payloadNode.isObject() ? (ObjectNode) payloadNode : mutableRoot.putObject("payload");
String upstreamSessionId = resolveUpstreamSessionId(context);
mutablePayload.put("session_id", upstreamSessionId);
return OBJECT_MAPPER.writeValueAsString(mutableRoot);
} catch (Exception ex) {
String upstreamSessionId = resolveUpstreamSessionId(context);
log.warn("本地 ASR start 消息解析失败，将保留原始消息：meetingId={}, upstreamSessionId={}",
context.getMeetingId(), upstreamSessionId, ex);
return payload;
}
}
private String resolveUpstreamSessionId(RealtimeAsrChannelContext context) {
Object existing = context.getChannelState().get(STATE_UPSTREAM_SESSION_ID);
if (existing instanceof String value && !value.isBlank()) {
return value;
}
RealtimeMeetingSessionStatusVO status = realtimeMeetingSessionStateService.getStatus(context.getMeetingId());
RealtimeMeetingResumeConfig resumeConfig = status == null ? null : status.getResumeConfig();
if (resumeConfig != null && resumeConfig.getUpstreamSessionId() != null && !resumeConfig.getUpstreamSessionId().isBlank()) {
String persistedSessionId = resumeConfig.getUpstreamSessionId().trim();
context.getChannelState().put(STATE_UPSTREAM_SESSION_ID, persistedSessionId);
return persistedSessionId;
}
String generated = "local-" + context.getMeetingId() + "-" + UUID.randomUUID().toString().replace("-", "");
context.getChannelState().put(STATE_UPSTREAM_SESSION_ID, generated);
realtimeMeetingSessionStateService.rememberUpstreamSessionId(context.getMeetingId(), generated);
log.info("生成本地 ASR 上游会话标识：meetingId={}, upstreamSessionId={}", context.getMeetingId(), generated);
return generated;
}
private boolean isFrontendOpen(RealtimeAsrChannelContext context) {
return context.getFrontendSession() != null
&& context.getFrontendSession().isOpen()
&& context.getRawSession() != null
&& context.getRawSession().isOpen();
}
private boolean tryReconnect(RealtimeAsrChannelContext context, String reason) {
if (Boolean.TRUE.equals(context.getChannelState().get(STATE_CLOSE_AFTER_END)) || !isFrontendOpen(context)) {
return false;
}
String lastStartMessage = context.getChannelState().get(STATE_LAST_START_MESSAGE) instanceof String value ? value : null;
if (lastStartMessage == null || lastStartMessage.isBlank()) {
return false;
}
int attempt = nextReconnectAttempt(context);
if (attempt > MAX_RECONNECT_ATTEMPTS) {
log.warn("本地 ASR 上游重连预算耗尽：meetingId={}, connectionId={}, reason={}",
context.getMeetingId(), currentConnectionId(context), reason);
return false;
}
context.getChannelState().put(STATE_RECONNECTING, Boolean.TRUE);
context.getChannelState().remove(STATE_UPSTREAM_SOCKET);
context.getChannelState().remove(STATE_START_MESSAGE_FORWARDED);
log.warn("本地 ASR 上游断开，尝试自动重连：meetingId={}, connectionId={}, attempt={}, reason={}",
context.getMeetingId(), currentConnectionId(context), attempt, reason);
try {
connect(context);
java.net.http.WebSocket reconnectedSocket = getUpstreamSocket(context);
if (reconnectedSocket == null) {
return false;
}
context.getChannelState().put(STATE_START_MESSAGE_FORWARDED, Boolean.TRUE);
sendUpstreamOrdered(context, () -> reconnectedSocket.sendText(lastStartMessage, true), "text-start-reconnect");
flushPendingAudioFrames(context, reconnectedSocket);
context.getChannelState().put(STATE_RECONNECTING, Boolean.FALSE);
context.getChannelState().put(STATE_RECONNECT_ATTEMPT, 0);
log.info("本地 ASR 上游自动重连成功：meetingId={}, connectionId={}, attempt={}",
context.getMeetingId(), currentConnectionId(context), attempt);
return true;
} catch (Exception ex) {
log.warn("本地 ASR 上游自动重连失败：meetingId={}, connectionId={}, attempt={}",
context.getMeetingId(), currentConnectionId(context), attempt, ex);
if (attempt < MAX_RECONNECT_ATTEMPTS) {
return tryReconnect(context, reason);
}
return false;
} finally {
if (getUpstreamSocket(context) == null) {
context.getChannelState().put(STATE_RECONNECTING, Boolean.FALSE);
}
}
}
private int nextReconnectAttempt(RealtimeAsrChannelContext context) {
Object value = context.getChannelState().get(STATE_RECONNECT_ATTEMPT);
int current = value instanceof Number number ? number.intValue() : 0;
int next = current + 1;
context.getChannelState().put(STATE_RECONNECT_ATTEMPT, next);
return next;
}
private static ByteBuffer copyBuffer(ByteBuffer source) {
ByteBuffer duplicate = source.asReadOnlyBuffer();
byte[] bytes = new byte[duplicate.remaining()];
@ -513,10 +631,14 @@ public class LocalRealtimeAsrChannel implements RealtimeAsrChannel {
log.info("上游 ASR websocket 已关闭：meetingId={}, sessionId={}, code={}, reason={}",
context.getMeetingId(), currentConnectionId(context), statusCode, reason);
context.getChannelState().remove(STATE_UPSTREAM_SOCKET);
if (tryReconnect(context, reason)) {
return COMPLETED;
}
context.getCallback().sendFrontendError(context.getMeetingId(),
"REALTIME_UPSTREAM_CLOSED",
reason == null || reason.isBlank() ? "上游 ASR WebSocket 已断开" : "上游 ASR WebSocket 已断开: " + reason);
context.getCallback().closeFrontend(context.getMeetingId(), new CloseStatus(statusCode, reason));
context.getCallback().removeMeetingSession(context.getMeetingId());
return COMPLETED;
}
@ -525,6 +647,9 @@ public class LocalRealtimeAsrChannel implements RealtimeAsrChannel {
log.error("上游 ASR websocket 异常：meetingId={}, sessionId={}, upstream={}",
context.getMeetingId(), currentConnectionId(context), context.getTargetWsUrl(), error);
context.getChannelState().remove(STATE_UPSTREAM_SOCKET);
if (tryReconnect(context, error == null ? null : error.getMessage())) {
return;
}
context.getCallback().sendFrontendError(context.getMeetingId(),
"REALTIME_UPSTREAM_ERROR",
error == null || error.getMessage() == null || error.getMessage().isBlank()