feat(realtime): 实现实时会议订阅与ASR功能
后端新增实时会议事件分发、订阅限流及WebSocket处理器，完善Redis缓存机制与实时ASR通道支持。前端同步更新实时ASR会话组件及会议预览页面，完成前后端实时数据链路对接。dev_na
chenhao
parent
907a9755c0
commit
36c1461c81
|
|
@ -71,6 +71,18 @@ public final class RedisKeys {
|
|||
return "biz:meeting:realtime:socket:" + sessionToken;
|
||||
}
|
||||
|
||||
public static String realtimeMeetingSubscriptionSessionKey(String sessionToken) {
|
||||
return "biz:meeting:realtime:subscribe:" + sessionToken;
|
||||
}
|
||||
|
||||
public static String realtimeMeetingSubscriptionRateLimitKey(String scope) {
|
||||
return "biz:meeting:realtime:subscribe:rate:" + scope;
|
||||
}
|
||||
|
||||
public static String realtimeMeetingSubscriberSetKey(Long meetingId) {
|
||||
return "biz:meeting:realtime:subscribers:" + meetingId;
|
||||
}
|
||||
|
||||
public static String realtimeMeetingSessionStateKey(Long meetingId) {
|
||||
return "biz:meeting:realtime:state:" + meetingId;
|
||||
}
|
||||
|
|
@ -91,6 +103,10 @@ public final class RedisKeys {
|
|||
return "biz:meeting:realtime:timeout:lock:" + meetingId;
|
||||
}
|
||||
|
||||
public static String realtimeMeetingPublisherLeaseKey(Long meetingId) {
|
||||
return "biz:meeting:realtime:publisher:" + meetingId;
|
||||
}
|
||||
|
||||
public static String realtimeMeetingResumeTimeoutPrefix() {
|
||||
return "biz:meeting:realtime:resume-timeout:";
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
package com.imeeting.config;
|
||||
|
||||
import com.imeeting.websocket.RealtimeMeetingProxyWebSocketHandler;
|
||||
import com.imeeting.websocket.RealtimeMeetingSubscriptionWebSocketHandler;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.springframework.boot.web.embedded.tomcat.TomcatServletWebServerFactory;
|
||||
import org.springframework.boot.web.server.WebServerFactoryCustomizer;
|
||||
|
|
@ -20,11 +21,14 @@ public class RealtimeMeetingWebSocketConfig implements WebSocketConfigurer {
|
|||
private static final String WS_TEXT_BUFFER_SIZE = "1048576";
|
||||
|
||||
private final RealtimeMeetingProxyWebSocketHandler realtimeMeetingProxyWebSocketHandler;
|
||||
private final RealtimeMeetingSubscriptionWebSocketHandler realtimeMeetingSubscriptionWebSocketHandler;
|
||||
|
||||
@Override
|
||||
public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
|
||||
registry.addHandler(realtimeMeetingProxyWebSocketHandler, "/ws/meeting/realtime")
|
||||
.setAllowedOriginPatterns("*");
|
||||
registry.addHandler(realtimeMeetingSubscriptionWebSocketHandler, "/ws/meeting/realtime/subscribe")
|
||||
.setAllowedOriginPatterns("*");
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@ import com.imeeting.dto.biz.MeetingVO;
|
|||
import com.imeeting.dto.biz.OpenRealtimeSocketSessionCommand;
|
||||
import com.imeeting.dto.biz.RealtimeMeetingCompleteDTO;
|
||||
import com.imeeting.dto.biz.RealtimeMeetingSessionStatusVO;
|
||||
import com.imeeting.dto.biz.RealtimeMeetingSubscriptionSessionVO;
|
||||
import com.imeeting.dto.biz.RealtimeSocketSessionVO;
|
||||
import com.imeeting.dto.biz.RealtimeTranscriptItemDTO;
|
||||
import com.imeeting.dto.biz.UpdateMeetingBasicCommand;
|
||||
|
|
@ -37,6 +38,7 @@ import com.imeeting.service.biz.MeetingUnifiedStatusService;
|
|||
import com.imeeting.service.biz.PromptTemplateService;
|
||||
import com.imeeting.service.biz.RealtimeMeetingSessionStateService;
|
||||
import com.imeeting.service.biz.RealtimeMeetingSocketSessionService;
|
||||
import com.imeeting.service.biz.RealtimeMeetingSubscriptionSessionService;
|
||||
import com.imeeting.service.biz.impl.MeetingAudioUploadSupport;
|
||||
import com.unisbase.common.ApiResponse;
|
||||
import com.unisbase.common.annotation.Log;
|
||||
|
|
@ -46,6 +48,7 @@ import com.unisbase.service.SysParamService;
|
|||
import io.swagger.v3.oas.annotations.Operation;
|
||||
import io.swagger.v3.oas.annotations.tags.Tag;
|
||||
import jakarta.validation.Valid;
|
||||
import jakarta.servlet.http.HttpServletRequest;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.http.HttpHeaders;
|
||||
|
|
@ -85,6 +88,7 @@ public class MeetingController {
|
|||
private final MeetingTranscriptFileService meetingTranscriptFileService;
|
||||
private final PromptTemplateService promptTemplateService;
|
||||
private final RealtimeMeetingSocketSessionService realtimeMeetingSocketSessionService;
|
||||
private final RealtimeMeetingSubscriptionSessionService realtimeMeetingSubscriptionSessionService;
|
||||
private final RealtimeMeetingSessionStateService realtimeMeetingSessionStateService;
|
||||
private final MeetingAudioUploadSupport meetingAudioUploadSupport;
|
||||
private final MeetingProgressService meetingProgressService;
|
||||
|
|
@ -103,6 +107,7 @@ public class MeetingController {
|
|||
MeetingTranscriptFileService meetingTranscriptFileService,
|
||||
PromptTemplateService promptTemplateService,
|
||||
RealtimeMeetingSocketSessionService realtimeMeetingSocketSessionService,
|
||||
RealtimeMeetingSubscriptionSessionService realtimeMeetingSubscriptionSessionService,
|
||||
RealtimeMeetingSessionStateService realtimeMeetingSessionStateService,
|
||||
MeetingAudioUploadSupport meetingAudioUploadSupport,
|
||||
MeetingProgressService meetingProgressService,
|
||||
|
|
@ -116,6 +121,7 @@ public class MeetingController {
|
|||
this.meetingTranscriptFileService = meetingTranscriptFileService;
|
||||
this.promptTemplateService = promptTemplateService;
|
||||
this.realtimeMeetingSocketSessionService = realtimeMeetingSocketSessionService;
|
||||
this.realtimeMeetingSubscriptionSessionService = realtimeMeetingSubscriptionSessionService;
|
||||
this.realtimeMeetingSessionStateService = realtimeMeetingSessionStateService;
|
||||
this.meetingAudioUploadSupport = meetingAudioUploadSupport;
|
||||
this.meetingProgressService = meetingProgressService;
|
||||
|
|
@ -303,6 +309,15 @@ public class MeetingController {
|
|||
return ApiResponse.ok(meetingQueryService.getDetail(id));
|
||||
}
|
||||
|
||||
@Operation(summary = "创建实时转写订阅会话")
|
||||
@PostMapping("/{id}/realtime/subscription-session")
|
||||
@PreAuthorize("isAuthenticated()")
|
||||
public ApiResponse<RealtimeMeetingSubscriptionSessionVO> createRealtimeSubscriptionSession(@PathVariable Long id,
|
||||
HttpServletRequest request) {
|
||||
return ApiResponse.ok(realtimeMeetingSubscriptionSessionService.createForUser(
|
||||
id, currentLoginUser(), request.getRemoteAddr()));
|
||||
}
|
||||
|
||||
@Operation(summary = "导出会议摘要")
|
||||
@GetMapping("/{id}/summary/export")
|
||||
@PreAuthorize("isAuthenticated()")
|
||||
|
|
@ -371,7 +386,7 @@ public class MeetingController {
|
|||
public ApiResponse<RealtimeMeetingSessionStatusVO> getRealtimeSessionStatus(@PathVariable Long id) {
|
||||
LoginUser loginUser = currentLoginUser();
|
||||
Meeting meeting = meetingAccessService.requireMeeting(id);
|
||||
meetingAccessService.assertCanManageRealtimeMeeting(meeting, loginUser);
|
||||
meetingAccessService.assertCanViewMeeting(meeting, loginUser);
|
||||
return ApiResponse.ok(realtimeMeetingSessionStateService.getStatus(id));
|
||||
}
|
||||
|
||||
|
|
@ -392,7 +407,7 @@ public class MeetingController {
|
|||
}
|
||||
try {
|
||||
Meeting meeting = meetingAccessService.requireMeeting(id);
|
||||
meetingAccessService.assertCanManageRealtimeMeeting(meeting, loginUser);
|
||||
meetingAccessService.assertCanViewMeeting(meeting, loginUser);
|
||||
RealtimeMeetingSessionStatusVO status = statuses.get(id);
|
||||
if (status != null) {
|
||||
result.put(id, status);
|
||||
|
|
|
|||
|
|
@ -2,17 +2,23 @@ package com.imeeting.controller.biz;
|
|||
|
||||
import com.imeeting.dto.biz.MeetingPreviewAccessVO;
|
||||
import com.imeeting.dto.biz.PublicMeetingPreviewVO;
|
||||
import com.imeeting.dto.biz.RealtimeMeetingSessionStatusVO;
|
||||
import com.imeeting.dto.biz.RealtimeMeetingSubscriptionSessionVO;
|
||||
import com.imeeting.entity.biz.Meeting;
|
||||
import com.imeeting.service.biz.MeetingAccessService;
|
||||
import com.imeeting.service.biz.MeetingQueryService;
|
||||
import com.imeeting.service.biz.RealtimeMeetingSessionStateService;
|
||||
import com.imeeting.service.biz.RealtimeMeetingSubscriptionSessionService;
|
||||
import com.unisbase.common.ApiResponse;
|
||||
import io.swagger.v3.oas.annotations.Operation;
|
||||
import io.swagger.v3.oas.annotations.tags.Tag;
|
||||
import org.springframework.web.bind.annotation.GetMapping;
|
||||
import org.springframework.web.bind.annotation.PostMapping;
|
||||
import org.springframework.web.bind.annotation.PathVariable;
|
||||
import org.springframework.web.bind.annotation.RequestMapping;
|
||||
import org.springframework.web.bind.annotation.RequestParam;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
import jakarta.servlet.http.HttpServletRequest;
|
||||
|
||||
@Tag(name = "会议公开预览")
|
||||
@RestController
|
||||
|
|
@ -21,11 +27,17 @@ public class MeetingPublicPreviewController {
|
|||
|
||||
private final MeetingQueryService meetingQueryService;
|
||||
private final MeetingAccessService meetingAccessService;
|
||||
private final RealtimeMeetingSessionStateService realtimeMeetingSessionStateService;
|
||||
private final RealtimeMeetingSubscriptionSessionService realtimeMeetingSubscriptionSessionService;
|
||||
|
||||
public MeetingPublicPreviewController(MeetingQueryService meetingQueryService,
|
||||
MeetingAccessService meetingAccessService) {
|
||||
MeetingAccessService meetingAccessService,
|
||||
RealtimeMeetingSessionStateService realtimeMeetingSessionStateService,
|
||||
RealtimeMeetingSubscriptionSessionService realtimeMeetingSubscriptionSessionService) {
|
||||
this.meetingQueryService = meetingQueryService;
|
||||
this.meetingAccessService = meetingAccessService;
|
||||
this.realtimeMeetingSessionStateService = realtimeMeetingSessionStateService;
|
||||
this.realtimeMeetingSubscriptionSessionService = realtimeMeetingSubscriptionSessionService;
|
||||
}
|
||||
|
||||
@Operation(summary = "查询会议预览访问要求")
|
||||
|
|
@ -59,4 +71,29 @@ public class MeetingPublicPreviewController {
|
|||
return ApiResponse.error(ex.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
@Operation(summary = "查询公开预览实时会议状态")
|
||||
@GetMapping("/{id}/preview/realtime/session-status")
|
||||
public ApiResponse<RealtimeMeetingSessionStatusVO> getRealtimeSessionStatus(
|
||||
@PathVariable Long id, @RequestParam(required = false) String accessPassword) {
|
||||
try {
|
||||
Meeting meeting = meetingAccessService.requireMeetingIgnoreTenant(id);
|
||||
meetingAccessService.assertCanPreviewMeeting(meeting, accessPassword);
|
||||
return ApiResponse.ok(realtimeMeetingSessionStateService.getStatus(id));
|
||||
} catch (RuntimeException ex) {
|
||||
return ApiResponse.error(ex.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
@Operation(summary = "创建公开实时转写订阅会话")
|
||||
@PostMapping("/{id}/preview/realtime-subscription-session")
|
||||
public ApiResponse<RealtimeMeetingSubscriptionSessionVO> createRealtimeSubscriptionSession(
|
||||
@PathVariable Long id, @RequestParam(required = false) String accessPassword, HttpServletRequest request) {
|
||||
try {
|
||||
return ApiResponse.ok(realtimeMeetingSubscriptionSessionService.createForPreview(
|
||||
id, accessPassword, request.getRemoteAddr()));
|
||||
} catch (RuntimeException ex) {
|
||||
return ApiResponse.error(ex.getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,8 @@
|
|||
package com.imeeting.dto.biz;
|
||||
|
||||
import lombok.Data;
|
||||
|
||||
@Data
|
||||
public class RealtimeMeetingSubscriptionSessionData {
|
||||
private Long meetingId;
|
||||
}
|
||||
|
|
@ -0,0 +1,10 @@
|
|||
package com.imeeting.dto.biz;
|
||||
|
||||
import lombok.Data;
|
||||
|
||||
@Data
|
||||
public class RealtimeMeetingSubscriptionSessionVO {
|
||||
private String sessionToken;
|
||||
private String path;
|
||||
private Long expiresInSeconds;
|
||||
}
|
||||
|
|
@ -2,7 +2,6 @@ package com.imeeting.grpc.realtime;
|
|||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.imeeting.common.MeetingConstants;
|
||||
import com.imeeting.config.grpc.AndroidGrpcAuthProperties;
|
||||
import com.imeeting.dto.android.AndroidAuthContext;
|
||||
import com.imeeting.dto.biz.AiModelVO;
|
||||
import com.imeeting.dto.biz.RealtimeMeetingResumeConfig;
|
||||
|
|
@ -13,19 +12,29 @@ import com.imeeting.service.biz.AiModelService;
|
|||
import com.imeeting.service.biz.MeetingAccessService;
|
||||
import com.imeeting.service.biz.MeetingAuthorizationService;
|
||||
import com.imeeting.service.biz.RealtimeMeetingSessionStateService;
|
||||
import com.imeeting.service.android.AndroidAuthService;
|
||||
import com.imeeting.service.realtime.RealtimeAsrChannel;
|
||||
import com.imeeting.service.realtime.RealtimeAsrChannelContext;
|
||||
import com.imeeting.service.realtime.RealtimeAsrChannelFactory;
|
||||
import com.imeeting.service.realtime.RealtimeMeetingAudioStorageService;
|
||||
import com.imeeting.service.realtime.RealtimeMeetingTranscriptCacheService;
|
||||
import com.imeeting.service.realtime.RealtimeMeetingEventHub;
|
||||
import com.imeeting.service.realtime.RealtimeMeetingEventRelay;
|
||||
import com.imeeting.service.realtime.RealtimeMeetingSubscriptionQuota;
|
||||
import com.imeeting.service.realtime.RealtimeMeetingMetrics;
|
||||
import com.imeeting.support.redis.RealtimeMeetingPublisherLease;
|
||||
import io.grpc.Status;
|
||||
import io.grpc.stub.StreamObserver;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.web.socket.CloseStatus;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.function.Consumer;
|
||||
|
||||
/**
|
||||
* Android 实时会议音频双向流。会议完成仍由 REST complete 接口负责。
|
||||
|
|
@ -38,14 +47,32 @@ public class AndroidRealtimeMeetingGrpcService
|
|||
|
||||
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
|
||||
|
||||
private final AndroidGrpcAuthProperties authProperties;
|
||||
private final AndroidAuthService androidAuthService;
|
||||
private final MeetingAccessService meetingAccessService;
|
||||
private final MeetingAuthorizationService meetingAuthorizationService;
|
||||
private final RealtimeMeetingSessionStateService sessionStateService;
|
||||
private final RealtimeAsrChannelFactory channelFactory;
|
||||
private final RealtimeMeetingAudioStorageService audioStorageService;
|
||||
private final AiModelService aiModelService;
|
||||
private final RealtimeMeetingPublisherLease publisherLease;
|
||||
private final RealtimeMeetingEventHub eventHub;
|
||||
private final RealtimeMeetingEventRelay eventRelay;
|
||||
private final RealtimeMeetingSubscriptionQuota subscriptionQuota;
|
||||
private final RealtimeMeetingTranscriptCacheService transcriptCacheService;
|
||||
private final Map<Long, GrpcSession> sessions = new ConcurrentHashMap<>();
|
||||
private final Map<String, GrpcSession> subscriberSessions = new ConcurrentHashMap<>();
|
||||
|
||||
@Value("${imeeting.realtime.max-subscribers-per-meeting:200}")
|
||||
private int maxSubscribersPerMeeting = 200;
|
||||
|
||||
@Scheduled(fixedDelay = 60_000)
|
||||
public void renewSubscriberQuotaSlots() {
|
||||
subscriberSessions.values().forEach(session -> {
|
||||
if (!session.cleaned) {
|
||||
subscriptionQuota.renew(session.meetingId, session.subscriptionMember);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public StreamObserver<ClientMessage> stream(StreamObserver<ServerMessage> responseObserver) {
|
||||
|
|
@ -102,18 +129,98 @@ public class AndroidRealtimeMeetingGrpcService
|
|||
if (frame.getPcm().isEmpty()) {
|
||||
return;
|
||||
}
|
||||
if (!session.publisher) {
|
||||
try {
|
||||
if (!activatePublisher()) {
|
||||
return;
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
throw new IllegalStateException("Realtime publisher initialization failed", ex);
|
||||
}
|
||||
}
|
||||
long sequence = frame.getSequence();
|
||||
if (session.lastSequence >= 0 && sequence <= session.lastSequence) {
|
||||
fail("REALTIME_AUDIO_SEQUENCE_INVALID", "音频帧 sequence 必须严格递增", false);
|
||||
return;
|
||||
}
|
||||
session.lastSequence = sequence;
|
||||
if (!publisherLease.renew(session.meetingId, session.publisherFenceToken)
|
||||
|| !sessionStateService.isActiveConnection(session.meetingId, session.publisherFenceToken)) {
|
||||
fail("REALTIME_PUBLISHER_LEASE_LOST", "发布连接已失效，请重新连接", true);
|
||||
return;
|
||||
}
|
||||
byte[] pcm = frame.getPcm().toByteArray();
|
||||
audioStorageService.append(session.connectionId, pcm);
|
||||
session.channel.handleFrontendBinary(session.context, pcm);
|
||||
}
|
||||
|
||||
private boolean activatePublisher() throws Exception {
|
||||
if (!session.canPublish) {
|
||||
fail("REALTIME_AUDIO_PUBLISH_FORBIDDEN", "Subscriber cannot publish audio", false);
|
||||
return false;
|
||||
}
|
||||
sessionStateService.assertCanOpenSession(session.meetingId);
|
||||
RealtimeMeetingSessionStatusVO status = sessionStateService.getStatus(session.meetingId);
|
||||
RealtimeMeetingResumeConfig config = status == null ? null : status.getResumeConfig();
|
||||
if (config == null || config.getAsrModelId() == null) {
|
||||
throw new IllegalStateException("Realtime meeting ASR configuration is missing");
|
||||
}
|
||||
AiModelVO model = aiModelService.getModelById(config.getAsrModelId(), "ASR");
|
||||
if (model == null) {
|
||||
throw new IllegalStateException("Realtime ASR model does not exist");
|
||||
}
|
||||
RealtimeAsrChannel channel = channelFactory.getRequired(model.getProvider());
|
||||
String publisherFenceToken = java.util.UUID.randomUUID().toString();
|
||||
if (!publisherLease.tryAcquire(session.meetingId, publisherFenceToken)) {
|
||||
RealtimeMeetingMetrics.publisherRejected("grpc");
|
||||
fail("REALTIME_PUBLISHER_EXISTS", "Realtime publisher already exists", false);
|
||||
return false;
|
||||
}
|
||||
if (!sessionStateService.activate(session.meetingId, publisherFenceToken)) {
|
||||
publisherLease.release(session.meetingId, publisherFenceToken);
|
||||
fail("REALTIME_PUBLISHER_EXISTS", "Realtime publisher already exists", false);
|
||||
return false;
|
||||
}
|
||||
if (sessions.putIfAbsent(session.meetingId, session) != null) {
|
||||
publisherLease.release(session.meetingId, publisherFenceToken);
|
||||
sessionStateService.pauseByDisconnect(session.meetingId, publisherFenceToken);
|
||||
fail("REALTIME_PUBLISHER_EXISTS", "Realtime publisher already exists", false);
|
||||
return false;
|
||||
}
|
||||
session.publisherFenceToken = publisherFenceToken;
|
||||
session.channel = channel;
|
||||
session.context.setMeetingId(session.meetingId);
|
||||
session.context.setProvider(channelFactory.normalizeProvider(model.getProvider()));
|
||||
session.context.setTargetWsUrl(channel.resolveTargetWsUrl(model));
|
||||
session.context.bindTransport(session.connectionId);
|
||||
session.context.setConnectionId(publisherFenceToken);
|
||||
session.context.getChannelState().put("modelCode", model.getModelCode());
|
||||
session.context.getChannelState().put("mediaConfig", model.getMediaConfig());
|
||||
session.context.setCallback(new GrpcChannelCallback(session));
|
||||
detachSubscriber(session);
|
||||
session.publisher = true;
|
||||
session.cleanupAction = this::cleanup;
|
||||
try {
|
||||
RealtimeMeetingMetrics.publisherOpened("grpc");
|
||||
audioStorageService.openSession(session.meetingId, session.connectionId);
|
||||
channel.connect(session.context);
|
||||
String start = OBJECT_MAPPER.writeValueAsString(channel.buildStartMessage(
|
||||
model, config.getMode(), config.getLanguage(), config.getUseSpkId(),
|
||||
config.getEnablePunctuation(), config.getEnableItn(), config.getEnableTextRefine(),
|
||||
config.getSaveAudio(), config.getHotwords()));
|
||||
channel.handleFrontendText(session.context, start);
|
||||
return true;
|
||||
} catch (Exception ex) {
|
||||
cleanup();
|
||||
throw ex;
|
||||
}
|
||||
}
|
||||
|
||||
private void handleStop(StopRequest stop) {
|
||||
if (!session.publisher) {
|
||||
fail("REALTIME_STOP_PUBLISH_FORBIDDEN", "Subscriber cannot stop publisher audio", false);
|
||||
return;
|
||||
}
|
||||
if (!stop.getConnectionId().isBlank()
|
||||
&& !session.connectionId.equals(stop.getConnectionId())) {
|
||||
fail("REALTIME_CONNECTION_ID_MISMATCH", "stop 的 connection_id 与连接不一致", false);
|
||||
|
|
@ -146,24 +253,50 @@ public class AndroidRealtimeMeetingGrpcService
|
|||
if (session == null || !session.cleaned) {
|
||||
if (session != null) {
|
||||
session.cleaned = true;
|
||||
try {
|
||||
session.channel.closeMeeting(session.context);
|
||||
} catch (Exception ex) {
|
||||
log.debug("Failed to close realtime upstream channel", ex);
|
||||
}
|
||||
try {
|
||||
session.channel.onFrontendDetached(session.context);
|
||||
} catch (Exception ex) {
|
||||
log.debug("Failed to detach realtime gRPC channel", ex);
|
||||
if (session.publisher) {
|
||||
try {
|
||||
session.channel.closeMeeting(session.context);
|
||||
} catch (Exception ex) {
|
||||
log.debug("Failed to close realtime upstream channel", ex);
|
||||
}
|
||||
try {
|
||||
session.channel.onFrontendDetached(session.context);
|
||||
} catch (Exception ex) {
|
||||
log.debug("Failed to detach realtime gRPC channel", ex);
|
||||
}
|
||||
}
|
||||
session.context.closeTransport();
|
||||
audioStorageService.closeSession(session.connectionId);
|
||||
sessionStateService.pauseByDisconnect(session.meetingId, session.connectionId);
|
||||
sessions.remove(session.meetingId, session);
|
||||
if (session.publisher) {
|
||||
audioStorageService.closeSession(session.connectionId);
|
||||
}
|
||||
if (session.publisher) {
|
||||
publisherLease.release(session.meetingId, session.publisherFenceToken);
|
||||
RealtimeMeetingMetrics.publisherClosed("grpc");
|
||||
}
|
||||
if (!session.publisher) {
|
||||
detachSubscriber(session);
|
||||
}
|
||||
if (session.publisher) {
|
||||
sessionStateService.pauseByDisconnect(session.meetingId, session.publisherFenceToken);
|
||||
sessions.remove(session.meetingId, session);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void detachSubscriber(GrpcSession target) {
|
||||
if (target.eventConsumer != null) {
|
||||
eventHub.unsubscribe(target.meetingId, target.eventConsumer);
|
||||
target.eventConsumer = null;
|
||||
}
|
||||
if (target.subscriptionMember != null) {
|
||||
subscriptionQuota.release(target.meetingId, target.subscriptionMember);
|
||||
target.subscriptionMember = null;
|
||||
RealtimeMeetingMetrics.subscriberClosed("grpc");
|
||||
}
|
||||
subscriberSessions.remove(target.connectionId, target);
|
||||
}
|
||||
|
||||
private GrpcSession open(ConnectRequest connect, StreamObserver<ServerMessage> observer) throws Exception {
|
||||
if (connect.getDeviceId().isBlank()) {
|
||||
fail("REALTIME_DEVICE_ID_REQUIRED", "device_id 不能为空", false);
|
||||
|
|
@ -177,25 +310,70 @@ public class AndroidRealtimeMeetingGrpcService
|
|||
// fail("REALTIME_PLATFORM_UNSUPPORTED", "实时会议 gRPC 当前仅支持 Android 平台", false);
|
||||
// return null;
|
||||
// }
|
||||
if (authProperties.isEnabled() && !authProperties.isAllowAnonymous()) {
|
||||
fail("REALTIME_AUTH_REQUIRED", "当前 gRPC 配置要求认证", false);
|
||||
return null;
|
||||
}
|
||||
|
||||
Long meetingId = parseId(connect.getMeetingId(), "meeting_id");
|
||||
Meeting meeting = meetingAccessService.requireMeetingIgnoreTenant(meetingId);
|
||||
Long requestedTenantId = parseOptionalId(connect.getTenantId());
|
||||
if (requestedTenantId != null && !requestedTenantId.equals(meeting.getTenantId())) {
|
||||
throw new SecurityException("tenant_id 与会议租户不一致");
|
||||
}
|
||||
AndroidAuthContext authContext = buildAuthContext(connect);
|
||||
meetingAuthorizationService.assertCanControlRealtimeMeeting(
|
||||
meeting, authContext, MeetingTerminalEnum.CUSTOM_TERMINAL.getCode());
|
||||
AndroidAuthContext authContext = androidAuthService.authenticateGrpc(
|
||||
connect.getDeviceId(), connect.getAppVersion(), "android",
|
||||
connect.getUserId(), connect.getTenantId());
|
||||
boolean publisher = false;
|
||||
boolean canPublish = authContext.getUserId() != null && authContext.getUserId().equals(meeting.getCreatorId());
|
||||
if (publisher) {
|
||||
if (authContext.getUserId() == null || !authContext.getUserId().equals(meeting.getCreatorId())) {
|
||||
throw new SecurityException("只有会议创建人可以发布实时音频");
|
||||
}
|
||||
meetingAuthorizationService.assertCanControlRealtimeMeeting(
|
||||
meeting, authContext, MeetingTerminalEnum.CUSTOM_TERMINAL.getCode());
|
||||
} else {
|
||||
meetingAuthorizationService.assertCanViewMeeting(meeting, authContext);
|
||||
RealtimeMeetingSessionStatusVO currentStatus = sessionStateService.getStatus(meetingId);
|
||||
if ((currentStatus == null || !"ACTIVE".equals(currentStatus.getStatus())) && !canPublish) {
|
||||
throw new IllegalStateException("会议尚未开始，暂不能订阅实时转写");
|
||||
}
|
||||
}
|
||||
if (!MeetingConstants.TYPE_REALTIME.equalsIgnoreCase(meeting.getMeetingType())) {
|
||||
throw new IllegalStateException("当前会议不是实时会议");
|
||||
}
|
||||
sessionStateService.initSessionIfAbsent(meetingId, meeting.getTenantId(), parseOptionalId(connect.getUserId()));
|
||||
sessionStateService.assertCanOpenSession(meetingId);
|
||||
sessionStateService.initSessionIfAbsent(meetingId, meeting.getTenantId(), authContext.getUserId());
|
||||
if (publisher) {
|
||||
sessionStateService.assertCanOpenSession(meetingId);
|
||||
}
|
||||
|
||||
String connectionId = connect.getConnectionId().isBlank()
|
||||
? "grpc-" + java.util.UUID.randomUUID().toString().replace("-", "")
|
||||
: connect.getConnectionId().trim();
|
||||
if (!publisher) {
|
||||
GrpcSession next = new GrpcSession(meetingId, connectionId, null, observer, null);
|
||||
next.publisher = false;
|
||||
next.canPublish = canPublish;
|
||||
next.subscriptionMember = subscriptionQuota.member(connectionId);
|
||||
if (!subscriptionQuota.tryAcquire(meetingId, next.subscriptionMember)) {
|
||||
RealtimeMeetingMetrics.subscriberRejected("grpc");
|
||||
throw new IllegalStateException("会议实时订阅人数已达上限");
|
||||
}
|
||||
next.eventConsumer = next::sendTranscript;
|
||||
if (!eventHub.trySubscribe(meetingId, next.eventConsumer, maxSubscribersPerMeeting)) {
|
||||
subscriptionQuota.release(meetingId, next.subscriptionMember);
|
||||
RealtimeMeetingMetrics.subscriberRejected("grpc");
|
||||
throw new IllegalStateException("会议实时订阅人数已达上限");
|
||||
}
|
||||
try {
|
||||
observer.onNext(ServerMessage.newBuilder()
|
||||
.setConnectAck(ConnectResponse.newBuilder().setSuccess(true).setMessage("实时转写订阅成功"))
|
||||
.build());
|
||||
replayCachedTranscripts(meetingId, next);
|
||||
subscriberSessions.put(connectionId, next);
|
||||
RealtimeMeetingMetrics.subscriberOpened("grpc");
|
||||
return next;
|
||||
} catch (RuntimeException ex) {
|
||||
eventHub.unsubscribe(meetingId, next.eventConsumer);
|
||||
subscriptionQuota.release(meetingId, next.subscriptionMember);
|
||||
throw ex;
|
||||
}
|
||||
}
|
||||
|
||||
RealtimeMeetingSessionStatusVO status = sessionStateService.getStatus(meetingId);
|
||||
RealtimeMeetingResumeConfig config = status == null ? null : status.getResumeConfig();
|
||||
|
|
@ -207,26 +385,35 @@ public class AndroidRealtimeMeetingGrpcService
|
|||
throw new IllegalStateException("实时 ASR 模型不存在");
|
||||
}
|
||||
RealtimeAsrChannel channel = channelFactory.getRequired(model.getProvider());
|
||||
String connectionId = connect.getConnectionId().isBlank()
|
||||
? "grpc-" + java.util.UUID.randomUUID().toString().replace("-", "")
|
||||
: connect.getConnectionId().trim();
|
||||
GrpcSession next = new GrpcSession(meetingId, connectionId, channel, observer);
|
||||
String publisherFenceToken = java.util.UUID.randomUUID().toString();
|
||||
GrpcSession next = new GrpcSession(meetingId, connectionId, channel, observer, publisherFenceToken);
|
||||
next.publisher = publisher;
|
||||
next.context.setMeetingId(meetingId);
|
||||
next.context.setProvider(channelFactory.normalizeProvider(model.getProvider()));
|
||||
next.context.setTargetWsUrl(channel.resolveTargetWsUrl(model));
|
||||
next.context.bindTransport(connectionId);
|
||||
next.context.setConnectionId(publisherFenceToken);
|
||||
next.context.getChannelState().put("modelCode", model.getModelCode());
|
||||
next.context.getChannelState().put("mediaConfig", model.getMediaConfig());
|
||||
next.context.setCallback(new GrpcChannelCallback(next));
|
||||
if (!sessionStateService.activate(meetingId, connectionId)) {
|
||||
if (!publisherLease.tryAcquire(meetingId, publisherFenceToken)) {
|
||||
RealtimeMeetingMetrics.publisherRejected("grpc");
|
||||
throw new IllegalStateException("当前会议已有活跃发布连接");
|
||||
}
|
||||
if (!sessionStateService.activate(meetingId, publisherFenceToken)) {
|
||||
publisherLease.release(meetingId, publisherFenceToken);
|
||||
throw new IllegalStateException("当前会议已有活跃连接");
|
||||
}
|
||||
GrpcSession previous = sessions.putIfAbsent(meetingId, next);
|
||||
if (previous != null) {
|
||||
sessionStateService.pauseByDisconnect(meetingId, connectionId);
|
||||
publisherLease.release(meetingId, publisherFenceToken);
|
||||
sessionStateService.pauseByDisconnect(meetingId, publisherFenceToken);
|
||||
throw new IllegalStateException("当前会议已有活跃连接");
|
||||
}
|
||||
try {
|
||||
session = next;
|
||||
next.cleanupAction = this::cleanup;
|
||||
RealtimeMeetingMetrics.publisherOpened("grpc");
|
||||
audioStorageService.openSession(meetingId, connectionId);
|
||||
channel.connect(next.context);
|
||||
String start = OBJECT_MAPPER.writeValueAsString(channel.buildStartMessage(
|
||||
|
|
@ -240,6 +427,8 @@ public class AndroidRealtimeMeetingGrpcService
|
|||
return next;
|
||||
} catch (Exception ex) {
|
||||
sessions.remove(meetingId, next);
|
||||
publisherLease.release(meetingId, publisherFenceToken);
|
||||
eventHub.unsubscribe(meetingId, next.eventConsumer);
|
||||
try {
|
||||
channel.closeMeeting(next.context);
|
||||
} catch (Exception closeEx) {
|
||||
|
|
@ -247,23 +436,13 @@ public class AndroidRealtimeMeetingGrpcService
|
|||
}
|
||||
next.context.closeTransport();
|
||||
audioStorageService.closeSession(connectionId);
|
||||
sessionStateService.pauseByDisconnect(meetingId, connectionId);
|
||||
sessionStateService.pauseByDisconnect(meetingId, publisherFenceToken);
|
||||
throw ex;
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
private AndroidAuthContext buildAuthContext(ConnectRequest connect) {
|
||||
AndroidAuthContext context = new AndroidAuthContext();
|
||||
context.setDeviceId(connect.getDeviceId());
|
||||
context.setUserId(parseOptionalId(connect.getUserId()));
|
||||
context.setTenantId(parseOptionalId(connect.getTenantId()));
|
||||
context.setPlatform("android");
|
||||
context.setAnonymous(true);
|
||||
return context;
|
||||
}
|
||||
|
||||
private Long parseId(String value, String field) {
|
||||
Long id = parseOptionalId(value);
|
||||
if (id == null) {
|
||||
|
|
@ -287,22 +466,54 @@ public class AndroidRealtimeMeetingGrpcService
|
|||
return message == null || message.isBlank() ? fallback : message;
|
||||
}
|
||||
|
||||
private void replayCachedTranscripts(Long meetingId, GrpcSession session) {
|
||||
try {
|
||||
for (var item : transcriptCacheService.listOrderedItems(meetingId)) {
|
||||
session.send(ServerMessage.newBuilder()
|
||||
.setTranscript(TextMessage.newBuilder()
|
||||
.setText(com.imeeting.service.realtime.impl.LocalRealtimeAsrChannel.buildFrontendTranscriptMessage(item)))
|
||||
.build());
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
log.warn("Failed to replay realtime transcripts to subscriber, meetingId={}", meetingId, ex);
|
||||
}
|
||||
}
|
||||
|
||||
private static final class GrpcSession {
|
||||
private final Long meetingId;
|
||||
private final String connectionId;
|
||||
private final RealtimeAsrChannel channel;
|
||||
private String publisherFenceToken;
|
||||
private RealtimeAsrChannel channel;
|
||||
private final StreamObserver<ServerMessage> observer;
|
||||
private final RealtimeAsrChannelContext context = new RealtimeAsrChannelContext();
|
||||
private volatile long lastSequence = -1;
|
||||
private volatile boolean cleaned;
|
||||
private volatile boolean publisher;
|
||||
private boolean canPublish;
|
||||
private Runnable cleanupAction;
|
||||
private Consumer<String> eventConsumer;
|
||||
private String subscriptionMember;
|
||||
|
||||
private GrpcSession(Long meetingId, String connectionId, RealtimeAsrChannel channel,
|
||||
StreamObserver<ServerMessage> observer) {
|
||||
StreamObserver<ServerMessage> observer, String publisherFenceToken) {
|
||||
this.meetingId = meetingId;
|
||||
this.connectionId = connectionId;
|
||||
this.channel = channel;
|
||||
this.publisherFenceToken = publisherFenceToken;
|
||||
this.observer = observer;
|
||||
}
|
||||
|
||||
private void send(ServerMessage message) {
|
||||
if (!cleaned && message != null) {
|
||||
observer.onNext(message);
|
||||
}
|
||||
}
|
||||
|
||||
private void sendTranscript(String payload) {
|
||||
send(ServerMessage.newBuilder()
|
||||
.setTranscript(TextMessage.newBuilder().setText(payload == null ? "" : payload))
|
||||
.build());
|
||||
}
|
||||
}
|
||||
|
||||
private final class GrpcChannelCallback implements com.imeeting.service.realtime.RealtimeAsrChannelCallback {
|
||||
|
|
@ -319,11 +530,8 @@ public class AndroidRealtimeMeetingGrpcService
|
|||
|
||||
@Override
|
||||
public void sendFrontendText(Long meetingId, String payload) {
|
||||
if (!session.cleaned) {
|
||||
session.observer.onNext(ServerMessage.newBuilder()
|
||||
.setTranscript(TextMessage.newBuilder().setText(payload == null ? "" : payload))
|
||||
.build());
|
||||
}
|
||||
session.sendTranscript(payload);
|
||||
eventRelay.publish(meetingId, payload == null ? "" : payload);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -342,7 +550,9 @@ public class AndroidRealtimeMeetingGrpcService
|
|||
|
||||
@Override
|
||||
public void removeMeetingSession(Long meetingId) {
|
||||
sessions.remove(meetingId, session);
|
||||
if (session.cleanupAction != null) {
|
||||
session.cleanupAction.run();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
|
|||
|
|
@ -6,6 +6,11 @@ import jakarta.servlet.http.HttpServletRequest;
|
|||
public interface AndroidAuthService {
|
||||
AndroidAuthContext authenticateGrpc(String deviceId, String appVersion, String platform, String userId, String tenantId);
|
||||
|
||||
default AndroidAuthContext authenticateRealtimeGrpc(String deviceId, String appVersion, String platform,
|
||||
String userId, String tenantId, String accessToken) {
|
||||
return authenticateGrpc(deviceId, appVersion, platform, userId, tenantId);
|
||||
}
|
||||
|
||||
AndroidAuthContext authenticateHttp(HttpServletRequest request);
|
||||
|
||||
AndroidAuthContext authenticateHttp(HttpServletRequest request, boolean requireRegistered);
|
||||
|
|
|
|||
|
|
@ -66,6 +66,31 @@ public class AndroidAuthServiceImpl implements AndroidAuthService {
|
|||
return context;
|
||||
}
|
||||
|
||||
@Override
|
||||
public AndroidAuthContext authenticateRealtimeGrpc(String deviceId, String appVersion, String platform,
|
||||
String userId, String tenantId, String accessToken) {
|
||||
LicenseEntity license = licenseService.requireValidBoundLicense(deviceId);
|
||||
DeviceInfoEntity device = requireRegisteredDevice(deviceId);
|
||||
assertDeviceEnabled(device);
|
||||
if (!StringUtils.hasText(accessToken)) {
|
||||
throw new BusinessException(ErrorCodeEnum.UNAUTHORIZED.getCode(), "实时会议 gRPC 缺少访问令牌");
|
||||
}
|
||||
InternalAuthCheckResponse authResult = validateToken(accessToken);
|
||||
if (!license.getTenantId().equals(authResult.getTenantId())) {
|
||||
throw new BusinessException(ErrorCodeEnum.UNAUTHORIZED.getCode(), "登录租户与设备授权租户不一致");
|
||||
}
|
||||
Long requestedUserId = parseOptionalLong(userId, "实时会议 gRPC userId");
|
||||
Long requestedTenantId = parseOptionalLong(tenantId, "实时会议 gRPC tenantId");
|
||||
if (requestedUserId != null && !requestedUserId.equals(authResult.getUserId())) {
|
||||
throw new BusinessException(ErrorCodeEnum.UNAUTHORIZED.getCode(), "实时会议 gRPC userId 与登录用户不一致");
|
||||
}
|
||||
if (requestedTenantId != null && !requestedTenantId.equals(authResult.getTenantId())) {
|
||||
throw new BusinessException(ErrorCodeEnum.UNAUTHORIZED.getCode(), "实时会议 gRPC tenantId 与登录租户不一致");
|
||||
}
|
||||
return buildContext("USER_JWT", false, deviceId, null, appVersion, platform, accessToken,
|
||||
null, authResult, null);
|
||||
}
|
||||
|
||||
@Override
|
||||
public AndroidAuthContext authenticateHttp(HttpServletRequest request) {
|
||||
return authenticateHttp(request, true, false, false);
|
||||
|
|
|
|||
|
|
@ -90,7 +90,7 @@ public class AndroidMeetingPushServiceImpl implements AndroidMeetingPushService
|
|||
return;
|
||||
}
|
||||
Meeting meeting = meetingMapper.selectByIdIgnoreTenant(meetingId);
|
||||
if (meeting == null || meeting.getSourceDeviceCode() == null || meeting.getSourceDeviceCode().isBlank()) {
|
||||
if (meeting == null || meeting.getTenantId() == null || meeting.getCreatorId() == null) {
|
||||
return;
|
||||
}
|
||||
PushMessage message = PushMessage.newBuilder()
|
||||
|
|
@ -101,9 +101,9 @@ public class AndroidMeetingPushServiceImpl implements AndroidMeetingPushService
|
|||
.setContent(buildStatusChangedContent(meetingId, statusCode))
|
||||
.setNeedAck(false)
|
||||
.build();
|
||||
int pushed = androidGatewayPushService.pushToDevice(meeting.getSourceDeviceCode(), message);
|
||||
log.info("Android meeting status change push finished, meetingId={}, deviceId={}, statusCode={}, pushedConnections={}",
|
||||
meetingId, meeting.getSourceDeviceCode(), statusCode, pushed);
|
||||
int pushed = androidGatewayPushService.pushToUser(meeting.getTenantId(), meeting.getCreatorId(), message);
|
||||
log.info("Android meeting status change push finished, meetingId={}, tenantId={}, userId={}, statusCode={}, pushedConnections={}",
|
||||
meetingId, meeting.getTenantId(), meeting.getCreatorId(), statusCode, pushed);
|
||||
}
|
||||
|
||||
private String resolvePendingTitle(Meeting meeting) {
|
||||
|
|
|
|||
|
|
@ -19,6 +19,8 @@ public interface RealtimeMeetingSessionStateService {
|
|||
|
||||
boolean activate(Long meetingId, String connectionId);
|
||||
|
||||
boolean isActiveConnection(Long meetingId, String connectionId);
|
||||
|
||||
RealtimeMeetingSessionStatusVO getStatus(Long meetingId);
|
||||
|
||||
Map<Long, RealtimeMeetingSessionStatusVO> getStatuses(List<Long> meetingIds);
|
||||
|
|
|
|||
|
|
@ -0,0 +1,10 @@
|
|||
package com.imeeting.service.biz;
|
||||
|
||||
import com.imeeting.dto.biz.RealtimeMeetingSubscriptionSessionVO;
|
||||
import com.unisbase.security.LoginUser;
|
||||
|
||||
public interface RealtimeMeetingSubscriptionSessionService {
|
||||
RealtimeMeetingSubscriptionSessionVO createForUser(Long meetingId, LoginUser loginUser, String clientIp);
|
||||
|
||||
RealtimeMeetingSubscriptionSessionVO createForPreview(Long meetingId, String accessPassword, String clientIp);
|
||||
}
|
||||
|
|
@ -12,6 +12,7 @@ import com.imeeting.mapper.biz.MeetingMapper;
|
|||
import com.imeeting.mapper.biz.MeetingTranscriptMapper;
|
||||
import com.imeeting.service.biz.RealtimeMeetingSessionStateService;
|
||||
import com.imeeting.support.redis.MeetingLockCache;
|
||||
import com.imeeting.support.redis.RealtimeMeetingPublisherLease;
|
||||
import com.imeeting.support.redis.RealtimeMeetingSessionCache;
|
||||
import com.imeeting.support.redis.RealtimeMeetingTranscriptCache;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
|
|
@ -34,6 +35,7 @@ public class RealtimeMeetingSessionStateServiceImpl implements RealtimeMeetingSe
|
|||
private final MeetingTranscriptMapper transcriptMapper;
|
||||
private final MeetingMapper meetingMapper;
|
||||
private final RealtimeMeetingTranscriptCache realtimeMeetingTranscriptCache;
|
||||
private final RealtimeMeetingPublisherLease publisherLease;
|
||||
|
||||
@Value("${imeeting.realtime.resume-window-minutes:30}")
|
||||
private String resumeWindowMinutesValue;
|
||||
|
|
@ -114,6 +116,13 @@ public class RealtimeMeetingSessionStateServiceImpl implements RealtimeMeetingSe
|
|||
throw new RuntimeException("Realtime meeting is already completed");
|
||||
}
|
||||
if ("ACTIVE".equals(currentStatus) || Boolean.TRUE.equals(status.getActiveConnection())) {
|
||||
RealtimeMeetingSessionState state = readState(meetingId);
|
||||
String activeConnectionId = state == null ? null : state.getActiveConnectionId();
|
||||
if (activeConnectionId != null && !activeConnectionId.isBlank()
|
||||
&& !publisherLease.isHeldBy(meetingId, activeConnectionId)) {
|
||||
pauseState(meetingId, state);
|
||||
return;
|
||||
}
|
||||
throw new RuntimeException("Realtime meeting already has an active connection");
|
||||
}
|
||||
if ("PAUSED_RESUMABLE".equals(currentStatus) && !Boolean.TRUE.equals(status.getCanResume())) {
|
||||
|
|
@ -126,6 +135,9 @@ public class RealtimeMeetingSessionStateServiceImpl implements RealtimeMeetingSe
|
|||
if (meetingId == null || connectionId == null || connectionId.isBlank()) {
|
||||
return false;
|
||||
}
|
||||
if (!publisherLease.isHeldBy(meetingId, connectionId)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
RealtimeMeetingSessionState state = getOrCreateState(meetingId);
|
||||
if ("COMPLETING".equals(state.getStatus()) || "COMPLETED".equals(state.getStatus())) {
|
||||
|
|
@ -154,6 +166,18 @@ public class RealtimeMeetingSessionStateServiceImpl implements RealtimeMeetingSe
|
|||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isActiveConnection(Long meetingId, String connectionId) {
|
||||
if (meetingId == null || connectionId == null || connectionId.isBlank()) {
|
||||
return false;
|
||||
}
|
||||
RealtimeMeetingSessionState state = readState(meetingId);
|
||||
return state != null
|
||||
&& "ACTIVE".equals(state.getStatus())
|
||||
&& connectionId.equals(state.getActiveConnectionId())
|
||||
&& publisherLease.isHeldBy(meetingId, connectionId);
|
||||
}
|
||||
|
||||
@Override
|
||||
public RealtimeMeetingSessionStatusVO getStatus(Long meetingId) {
|
||||
RealtimeMeetingSessionState state = readState(meetingId);
|
||||
|
|
@ -193,7 +217,12 @@ public class RealtimeMeetingSessionStateServiceImpl implements RealtimeMeetingSe
|
|||
if ("COMPLETING".equals(state.getStatus()) || "COMPLETED".equals(state.getStatus())) {
|
||||
return toStatusVO(state);
|
||||
}
|
||||
return pauseState(meetingId, state);
|
||||
String publisherFenceToken = state.getActiveConnectionId();
|
||||
RealtimeMeetingSessionStatusVO status = pauseState(meetingId, state);
|
||||
if (publisherFenceToken != null && !publisherFenceToken.isBlank()) {
|
||||
publisherLease.release(meetingId, publisherFenceToken);
|
||||
}
|
||||
return status;
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
|
|||
|
|
@ -59,6 +59,9 @@ public class RealtimeMeetingSocketSessionServiceImpl implements RealtimeMeetingS
|
|||
|
||||
Meeting meeting = meetingAccessService.requireMeeting(meetingId);
|
||||
meetingAccessService.assertCanControlRealtimeMeeting(meeting, loginUser, MeetingTerminalEnum.WEB.getCode());
|
||||
if (meeting.getCreatorId() == null || !meeting.getCreatorId().equals(loginUser.getUserId())) {
|
||||
throw new RuntimeException("只有会议创建人可以发布实时音频");
|
||||
}
|
||||
|
||||
realtimeMeetingSessionStateService.initSessionIfAbsent(meetingId, loginUser.getTenantId(), loginUser.getUserId());
|
||||
realtimeMeetingSessionStateService.assertCanOpenSession(meetingId);
|
||||
|
|
|
|||
|
|
@ -0,0 +1,77 @@
|
|||
package com.imeeting.service.biz.impl;
|
||||
|
||||
import com.imeeting.common.MeetingConstants;
|
||||
import com.imeeting.dto.biz.RealtimeMeetingSessionStatusVO;
|
||||
import com.imeeting.dto.biz.RealtimeMeetingSubscriptionSessionData;
|
||||
import com.imeeting.dto.biz.RealtimeMeetingSubscriptionSessionVO;
|
||||
import com.imeeting.entity.biz.Meeting;
|
||||
import com.imeeting.service.biz.MeetingAccessService;
|
||||
import com.imeeting.service.biz.RealtimeMeetingSessionStateService;
|
||||
import com.imeeting.service.biz.RealtimeMeetingSubscriptionSessionService;
|
||||
import com.imeeting.service.realtime.RealtimeMeetingSubscriptionRateLimiter;
|
||||
import com.imeeting.support.redis.RealtimeMeetingSubscriptionSessionCache;
|
||||
import com.unisbase.security.LoginUser;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
|
||||
import java.util.UUID;
|
||||
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class RealtimeMeetingSubscriptionSessionServiceImpl implements RealtimeMeetingSubscriptionSessionService {
|
||||
|
||||
private static final String SUBSCRIPTION_PATH = "/ws/meeting/realtime/subscribe";
|
||||
|
||||
private final MeetingAccessService meetingAccessService;
|
||||
private final RealtimeMeetingSessionStateService sessionStateService;
|
||||
private final RealtimeMeetingSubscriptionSessionCache subscriptionSessionCache;
|
||||
private final RealtimeMeetingSubscriptionRateLimiter rateLimiter;
|
||||
|
||||
@Value("${imeeting.realtime.max-subscription-requests-per-user-per-minute:5}")
|
||||
private int maxSubscriptionRequestsPerUser = 5;
|
||||
|
||||
@Value("${imeeting.realtime.max-subscription-requests-per-ip-per-minute:20}")
|
||||
private int maxSubscriptionRequestsPerIp = 20;
|
||||
|
||||
@Override
|
||||
public RealtimeMeetingSubscriptionSessionVO createForUser(Long meetingId, LoginUser loginUser, String clientIp) {
|
||||
rateLimiter.check("user:" + loginUser.getUserId(), maxSubscriptionRequestsPerUser);
|
||||
rateLimiter.check("ip:" + normalizeIp(clientIp), maxSubscriptionRequestsPerIp);
|
||||
Meeting meeting = meetingAccessService.requireMeeting(meetingId);
|
||||
meetingAccessService.assertCanViewMeeting(meeting, loginUser);
|
||||
return create(meeting);
|
||||
}
|
||||
|
||||
@Override
|
||||
public RealtimeMeetingSubscriptionSessionVO createForPreview(Long meetingId, String accessPassword, String clientIp) {
|
||||
rateLimiter.check("ip:" + normalizeIp(clientIp), maxSubscriptionRequestsPerIp);
|
||||
Meeting meeting = meetingAccessService.requireMeetingIgnoreTenant(meetingId);
|
||||
meetingAccessService.assertCanPreviewMeeting(meeting, accessPassword);
|
||||
return create(meeting);
|
||||
}
|
||||
|
||||
private RealtimeMeetingSubscriptionSessionVO create(Meeting meeting) {
|
||||
if (!MeetingConstants.TYPE_REALTIME.equalsIgnoreCase(meeting.getMeetingType())) {
|
||||
throw new RuntimeException("当前会议不是实时会议");
|
||||
}
|
||||
RealtimeMeetingSessionStatusVO status = sessionStateService.getStatus(meeting.getId());
|
||||
if (status == null || !"ACTIVE".equals(status.getStatus())) {
|
||||
throw new RuntimeException("会议尚未开始实时转写");
|
||||
}
|
||||
String token = UUID.randomUUID().toString().replace("-", "");
|
||||
RealtimeMeetingSubscriptionSessionData data = new RealtimeMeetingSubscriptionSessionData();
|
||||
data.setMeetingId(meeting.getId());
|
||||
subscriptionSessionCache.save(token, data);
|
||||
|
||||
RealtimeMeetingSubscriptionSessionVO result = new RealtimeMeetingSubscriptionSessionVO();
|
||||
result.setSessionToken(token);
|
||||
result.setPath(SUBSCRIPTION_PATH);
|
||||
result.setExpiresInSeconds(subscriptionSessionCache.getSessionTtlSeconds());
|
||||
return result;
|
||||
}
|
||||
|
||||
private String normalizeIp(String clientIp) {
|
||||
return clientIp == null || clientIp.isBlank() ? "unknown" : clientIp.trim();
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,73 @@
|
|||
package com.imeeting.service.realtime;
|
||||
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.CopyOnWriteArrayList;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
import java.util.function.Consumer;
|
||||
|
||||
/**
|
||||
* 会议级文本事件总线。订阅端与传输协议解耦，单个订阅者失败不影响其他连接。
|
||||
*/
|
||||
@Component
|
||||
public class RealtimeMeetingEventHub {
|
||||
|
||||
private final ConcurrentMap<Long, CopyOnWriteArrayList<Consumer<String>>> subscribers = new ConcurrentHashMap<>();
|
||||
|
||||
public void subscribe(Long meetingId, Consumer<String> consumer) {
|
||||
if (meetingId == null || consumer == null) {
|
||||
return;
|
||||
}
|
||||
subscribers.computeIfAbsent(meetingId, ignored -> new CopyOnWriteArrayList<>()).add(consumer);
|
||||
}
|
||||
|
||||
public boolean trySubscribe(Long meetingId, Consumer<String> consumer, int maxSubscribers) {
|
||||
if (meetingId == null || consumer == null || maxSubscribers < 1) {
|
||||
return false;
|
||||
}
|
||||
CopyOnWriteArrayList<Consumer<String>> meetingSubscribers = subscribers.computeIfAbsent(
|
||||
meetingId, ignored -> new CopyOnWriteArrayList<>());
|
||||
synchronized (meetingSubscribers) {
|
||||
if (meetingSubscribers.size() >= maxSubscribers) {
|
||||
return false;
|
||||
}
|
||||
meetingSubscribers.add(consumer);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
public void unsubscribe(Long meetingId, Consumer<String> consumer) {
|
||||
if (meetingId == null || consumer == null) {
|
||||
return;
|
||||
}
|
||||
List<Consumer<String>> meetingSubscribers = subscribers.get(meetingId);
|
||||
if (meetingSubscribers == null) {
|
||||
return;
|
||||
}
|
||||
meetingSubscribers.remove(consumer);
|
||||
if (meetingSubscribers.isEmpty()) {
|
||||
subscribers.remove(meetingId, meetingSubscribers);
|
||||
}
|
||||
}
|
||||
|
||||
public void publish(Long meetingId, String payload) {
|
||||
if (meetingId == null || payload == null) {
|
||||
return;
|
||||
}
|
||||
RealtimeMeetingMetrics.transcriptBroadcast();
|
||||
List<Consumer<String>> meetingSubscribers = subscribers.get(meetingId);
|
||||
if (meetingSubscribers == null) {
|
||||
return;
|
||||
}
|
||||
for (Consumer<String> subscriber : meetingSubscribers) {
|
||||
try {
|
||||
subscriber.accept(payload);
|
||||
} catch (RuntimeException ignored) {
|
||||
RealtimeMeetingMetrics.subscriberDeliveryFailed();
|
||||
// 单个失效订阅者不能阻塞同一会议的其他订阅者。
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,62 @@
|
|||
package com.imeeting.service.realtime;
|
||||
|
||||
import com.imeeting.support.RedisSupport;
|
||||
import jakarta.annotation.PostConstruct;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.springframework.data.redis.connection.Message;
|
||||
import org.springframework.data.redis.connection.MessageListener;
|
||||
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
|
||||
import org.springframework.data.redis.listener.ChannelTopic;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.Base64;
|
||||
import java.util.UUID;
|
||||
|
||||
/**
|
||||
* Redis Pub/Sub 仅负责跨实例实时分发；缺失事件由转写 REST 查询补偿。
|
||||
*/
|
||||
@Component
|
||||
@RequiredArgsConstructor
|
||||
public class RealtimeMeetingEventRelay implements MessageListener {
|
||||
|
||||
private static final String CHANNEL = "biz:meeting:realtime:events";
|
||||
|
||||
private final RedisSupport redisSupport;
|
||||
private final RealtimeMeetingEventHub eventHub;
|
||||
private final RedisMessageListenerContainer listenerContainer;
|
||||
private final String instanceId = UUID.randomUUID().toString();
|
||||
|
||||
@PostConstruct
|
||||
public void subscribe() {
|
||||
listenerContainer.addMessageListener(this, ChannelTopic.of(CHANNEL));
|
||||
}
|
||||
|
||||
public void publish(Long meetingId, String payload) {
|
||||
eventHub.publish(meetingId, payload);
|
||||
if (meetingId == null || payload == null) {
|
||||
return;
|
||||
}
|
||||
String encodedPayload = Base64.getEncoder().encodeToString(payload.getBytes(StandardCharsets.UTF_8));
|
||||
redisSupport.publish(CHANNEL, instanceId + "|" + meetingId + "|" + encodedPayload);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onMessage(Message message, byte[] pattern) {
|
||||
if (message == null || message.getBody() == null) {
|
||||
return;
|
||||
}
|
||||
String raw = new String(message.getBody(), StandardCharsets.UTF_8);
|
||||
String[] parts = raw.split("\\|", 3);
|
||||
if (parts.length != 3 || instanceId.equals(parts[0])) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
Long meetingId = Long.valueOf(parts[1]);
|
||||
String payload = new String(Base64.getDecoder().decode(parts[2]), StandardCharsets.UTF_8);
|
||||
eventHub.publish(meetingId, payload);
|
||||
} catch (IllegalArgumentException ignored) {
|
||||
// 非法广播消息不应影响 Redis 监听线程。
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,44 @@
|
|||
package com.imeeting.service.realtime;
|
||||
|
||||
import io.micrometer.core.instrument.Metrics;
|
||||
|
||||
/**
|
||||
* 实时会议连接与广播指标，避免使用 meetingId 等高基数标签。
|
||||
*/
|
||||
public final class RealtimeMeetingMetrics {
|
||||
|
||||
private RealtimeMeetingMetrics() {
|
||||
}
|
||||
|
||||
public static void publisherOpened(String transport) {
|
||||
Metrics.counter("imeeting.realtime.publisher.opened", "transport", transport).increment();
|
||||
}
|
||||
|
||||
public static void publisherRejected(String transport) {
|
||||
Metrics.counter("imeeting.realtime.publisher.rejected", "transport", transport).increment();
|
||||
}
|
||||
|
||||
public static void publisherClosed(String transport) {
|
||||
Metrics.counter("imeeting.realtime.publisher.closed", "transport", transport).increment();
|
||||
}
|
||||
|
||||
public static void subscriberOpened(String transport) {
|
||||
Metrics.counter("imeeting.realtime.subscriber.opened", "transport", transport).increment();
|
||||
}
|
||||
|
||||
public static void subscriberRejected(String transport) {
|
||||
Metrics.counter("imeeting.realtime.subscriber.rejected", "transport", transport).increment();
|
||||
}
|
||||
|
||||
public static void subscriberClosed(String transport) {
|
||||
Metrics.counter("imeeting.realtime.subscriber.closed", "transport", transport).increment();
|
||||
}
|
||||
|
||||
public static void transcriptBroadcast() {
|
||||
Metrics.counter("imeeting.realtime.transcript.broadcast").increment();
|
||||
}
|
||||
|
||||
public static void subscriberDeliveryFailed() {
|
||||
Metrics.counter("imeeting.realtime.subscriber.delivery.failed").increment();
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,55 @@
|
|||
package com.imeeting.service.realtime;
|
||||
|
||||
import com.imeeting.common.RedisKeys;
|
||||
import com.imeeting.support.RedisSupport;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.UUID;
|
||||
|
||||
/**
|
||||
* 跨实例的会议订阅连接配额；Redis 只保存短期连接占位，不保存订阅内容。
|
||||
*/
|
||||
@Component
|
||||
@RequiredArgsConstructor
|
||||
public class RealtimeMeetingSubscriptionQuota {
|
||||
|
||||
private final RedisSupport redisSupport;
|
||||
private final String instanceId = UUID.randomUUID().toString();
|
||||
|
||||
@Value("${imeeting.realtime.max-subscribers-per-meeting:200}")
|
||||
private int maxSubscribersPerMeeting = 200;
|
||||
|
||||
@Value("${imeeting.realtime.global-subscriber-slot-ttl-seconds:120}")
|
||||
private long slotTtlSeconds = 120;
|
||||
|
||||
public String member(String connectionId) {
|
||||
return instanceId + ":" + connectionId;
|
||||
}
|
||||
|
||||
public boolean tryAcquire(Long meetingId, String member) {
|
||||
if (meetingId == null || member == null || member.isBlank()) {
|
||||
return false;
|
||||
}
|
||||
return redisSupport.tryAcquireExpiringSlot(
|
||||
RedisKeys.realtimeMeetingSubscriberSetKey(meetingId),
|
||||
member,
|
||||
maxSubscribersPerMeeting,
|
||||
Duration.ofSeconds(slotTtlSeconds)
|
||||
);
|
||||
}
|
||||
|
||||
public boolean release(Long meetingId, String member) {
|
||||
if (meetingId == null || member == null || member.isBlank()) {
|
||||
return false;
|
||||
}
|
||||
return redisSupport.removeExpiringSlot(
|
||||
RedisKeys.realtimeMeetingSubscriberSetKey(meetingId), member);
|
||||
}
|
||||
|
||||
public boolean renew(Long meetingId, String member) {
|
||||
return tryAcquire(meetingId, member);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,30 @@
|
|||
package com.imeeting.service.realtime;
|
||||
|
||||
import com.imeeting.common.RedisKeys;
|
||||
import com.imeeting.support.RedisSupport;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
|
||||
import java.time.Duration;
|
||||
|
||||
@Component
|
||||
@RequiredArgsConstructor
|
||||
public class RealtimeMeetingSubscriptionRateLimiter {
|
||||
|
||||
private final RedisSupport redisSupport;
|
||||
|
||||
@Value("${imeeting.realtime.subscription-rate-window-seconds:60}")
|
||||
private long windowSeconds = 60;
|
||||
|
||||
public void check(String scope, int maxRequests) {
|
||||
if (scope == null || scope.isBlank() || maxRequests < 1) {
|
||||
throw new IllegalArgumentException("Invalid realtime subscription rate limit scope");
|
||||
}
|
||||
long current = redisSupport.incrementWithTtl(
|
||||
RedisKeys.realtimeMeetingSubscriptionRateLimitKey(scope), Duration.ofSeconds(windowSeconds));
|
||||
if (current > maxRequests) {
|
||||
throw new RuntimeException("实时订阅请求过于频繁，请稍后重试");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -124,6 +124,9 @@ public class LocalRealtimeAsrChannel implements RealtimeAsrChannel {
|
|||
|
||||
@Override
|
||||
public void handleFrontendText(RealtimeAsrChannelContext context, String payload) {
|
||||
if (!looksLikeStopMessage(payload) && !isActivePublisher(context)) {
|
||||
return;
|
||||
}
|
||||
java.net.http.WebSocket upstreamSocket = getUpstreamSocket(context);
|
||||
if (upstreamSocket == null) {
|
||||
return;
|
||||
|
|
@ -148,6 +151,9 @@ public class LocalRealtimeAsrChannel implements RealtimeAsrChannel {
|
|||
|
||||
@Override
|
||||
public void handleFrontendBinary(RealtimeAsrChannelContext context, byte[] payload) {
|
||||
if (!isActivePublisher(context)) {
|
||||
return;
|
||||
}
|
||||
java.net.http.WebSocket upstreamSocket = getUpstreamSocket(context);
|
||||
if (upstreamSocket == null) {
|
||||
return;
|
||||
|
|
@ -293,6 +299,9 @@ public class LocalRealtimeAsrChannel implements RealtimeAsrChannel {
|
|||
log.info("start 后开始补发排队音频帧：meetingId={}, sessionId={}, frameCount={}",
|
||||
context.getMeetingId(), currentConnectionId(context), pendingFrames.size());
|
||||
for (byte[] frame : pendingFrames) {
|
||||
if (!isActivePublisher(context)) {
|
||||
return;
|
||||
}
|
||||
sendUpstreamOrdered(context, () -> upstreamSocket.sendBinary(ByteBuffer.wrap(frame), true), "binary-flush");
|
||||
}
|
||||
}
|
||||
|
|
@ -301,6 +310,12 @@ public class LocalRealtimeAsrChannel implements RealtimeAsrChannel {
|
|||
return context.getConnectionId();
|
||||
}
|
||||
|
||||
private boolean isActivePublisher(RealtimeAsrChannelContext context) {
|
||||
String connectionId = currentConnectionId(context);
|
||||
return connectionId != null
|
||||
&& realtimeMeetingSessionStateService.isActiveConnection(context.getMeetingId(), connectionId);
|
||||
}
|
||||
|
||||
private String ensureStartMessageSessionId(RealtimeAsrChannelContext context, String payload) {
|
||||
try {
|
||||
JsonNode root = OBJECT_MAPPER.readTree(payload);
|
||||
|
|
@ -342,7 +357,8 @@ public class LocalRealtimeAsrChannel implements RealtimeAsrChannel {
|
|||
}
|
||||
|
||||
private boolean tryReconnect(RealtimeAsrChannelContext context, String reason) {
|
||||
if (Boolean.TRUE.equals(context.getChannelState().get(STATE_CLOSE_AFTER_END)) || !isFrontendOpen(context)) {
|
||||
if (Boolean.TRUE.equals(context.getChannelState().get(STATE_CLOSE_AFTER_END))
|
||||
|| !isFrontendOpen(context) || !isActivePublisher(context)) {
|
||||
return false;
|
||||
}
|
||||
String lastStartMessage = context.getChannelState().get(STATE_LAST_START_MESSAGE) instanceof String value ? value : null;
|
||||
|
|
@ -538,7 +554,7 @@ public class LocalRealtimeAsrChannel implements RealtimeAsrChannel {
|
|||
log.info("上游 ASR websocket 已打开：meetingId={}, sessionId={}, upstream={}",
|
||||
context.getMeetingId(), currentConnectionId(context), context.getTargetWsUrl());
|
||||
String connectionId = currentConnectionId(context);
|
||||
if (connectionId == null || !realtimeMeetingSessionStateService.activate(context.getMeetingId(), connectionId)) {
|
||||
if (connectionId == null || !isActivePublisher(context)) {
|
||||
context.getCallback().sendFrontendError(context.getMeetingId(), "REALTIME_ACTIVE_CONNECTION_EXISTS", "当前会议无法激活这条前端连接");
|
||||
webSocket.sendClose(CloseStatus.POLICY_VIOLATION.getCode(), "当前会议无法激活这条前端连接");
|
||||
context.getCallback().closeFrontend(context.getMeetingId(), CloseStatus.POLICY_VIOLATION.withReason("当前会议无法激活这条前端连接"));
|
||||
|
|
@ -558,6 +574,11 @@ public class LocalRealtimeAsrChannel implements RealtimeAsrChannel {
|
|||
public java.util.concurrent.CompletionStage<?> onText(java.net.http.WebSocket webSocket, CharSequence data, boolean last) {
|
||||
textBuffer.append(data);
|
||||
if (last) {
|
||||
if (!isActivePublisher(context)) {
|
||||
textBuffer.setLength(0);
|
||||
webSocket.request(1);
|
||||
return COMPLETED;
|
||||
}
|
||||
int count = upstreamTextCount.incrementAndGet();
|
||||
String upstreamPayload = textBuffer.toString();
|
||||
realtimeMeetingTranscriptCacheService.mergeUpstreamMessage(context.getMeetingId(), upstreamPayload);
|
||||
|
|
@ -588,6 +609,11 @@ public class LocalRealtimeAsrChannel implements RealtimeAsrChannel {
|
|||
data.get(chunk);
|
||||
binaryBuffer.writeBytes(chunk);
|
||||
if (last) {
|
||||
if (!isActivePublisher(context)) {
|
||||
binaryBuffer.reset();
|
||||
webSocket.request(1);
|
||||
return COMPLETED;
|
||||
}
|
||||
int count = upstreamBinaryCount.incrementAndGet();
|
||||
try {
|
||||
context.getCallback().sendFrontendBinary(context.getMeetingId(), binaryBuffer.toByteArray());
|
||||
|
|
|
|||
|
|
@ -91,7 +91,7 @@ public class TencentRealtimeAsrChannel implements RealtimeAsrChannel {
|
|||
@Override
|
||||
public void connect(RealtimeAsrChannelContext context) throws Exception {
|
||||
String connectionId = currentConnectionId(context);
|
||||
if (connectionId == null || !realtimeMeetingSessionStateService.activate(context.getMeetingId(), connectionId)) {
|
||||
if (connectionId == null || !isActivePublisher(context)) {
|
||||
context.getCallback().sendFrontendError(context.getMeetingId(), "REALTIME_ACTIVE_CONNECTION_EXISTS", "当前会议无法激活这条前端连接");
|
||||
context.getCallback().closeFrontend(context.getMeetingId(), CloseStatus.POLICY_VIOLATION.withReason("当前会议无法激活这条前端连接"));
|
||||
return;
|
||||
|
|
@ -107,6 +107,9 @@ public class TencentRealtimeAsrChannel implements RealtimeAsrChannel {
|
|||
|
||||
@Override
|
||||
public void handleFrontendText(RealtimeAsrChannelContext context, String payload) {
|
||||
if (!looksLikeStopMessage(payload) && !isActivePublisher(context)) {
|
||||
return;
|
||||
}
|
||||
if (looksLikeStartMessage(payload)) {
|
||||
startRecognizerIfNecessary(context);
|
||||
return;
|
||||
|
|
@ -119,6 +122,9 @@ public class TencentRealtimeAsrChannel implements RealtimeAsrChannel {
|
|||
|
||||
@Override
|
||||
public void handleFrontendBinary(RealtimeAsrChannelContext context, byte[] payload) {
|
||||
if (!isActivePublisher(context)) {
|
||||
return;
|
||||
}
|
||||
SpeakerRecognizer recognizer = getRecognizer(context);
|
||||
if (payload == null || payload.length == 0) {
|
||||
return;
|
||||
|
|
@ -184,6 +190,9 @@ public class TencentRealtimeAsrChannel implements RealtimeAsrChannel {
|
|||
}
|
||||
|
||||
private void startRecognizerIfNecessary(RealtimeAsrChannelContext context) {
|
||||
if (!isActivePublisher(context)) {
|
||||
return;
|
||||
}
|
||||
synchronized (context.getChannelState()) {
|
||||
if (Boolean.TRUE.equals(context.getChannelState().get(STATE_STARTED))) {
|
||||
return;
|
||||
|
|
@ -350,6 +359,9 @@ public class TencentRealtimeAsrChannel implements RealtimeAsrChannel {
|
|||
if (response == null || response.getSentences() == null || response.getSentences().getSentenceList() == null) {
|
||||
return;
|
||||
}
|
||||
if (!isActivePublisher(context)) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
rememberSpeakerContext(context, response);
|
||||
String cachePayload = buildCachePayload(response, forceFinal);
|
||||
|
|
@ -414,6 +426,10 @@ public class TencentRealtimeAsrChannel implements RealtimeAsrChannel {
|
|||
}
|
||||
List<byte[]> pendingFrames = (List<byte[]>) list;
|
||||
for (byte[] frame : pendingFrames) {
|
||||
if (!isActivePublisher(context)) {
|
||||
pendingFrames.clear();
|
||||
return;
|
||||
}
|
||||
if (frame != null && frame.length > 0) {
|
||||
recognizer.write(frame);
|
||||
}
|
||||
|
|
@ -454,6 +470,12 @@ public class TencentRealtimeAsrChannel implements RealtimeAsrChannel {
|
|||
return context.getConnectionId();
|
||||
}
|
||||
|
||||
private boolean isActivePublisher(RealtimeAsrChannelContext context) {
|
||||
String connectionId = currentConnectionId(context);
|
||||
return connectionId != null
|
||||
&& realtimeMeetingSessionStateService.isActiveConnection(context.getMeetingId(), connectionId);
|
||||
}
|
||||
|
||||
private static boolean looksLikeStartMessage(String payload) {
|
||||
if (payload == null || payload.isBlank()) {
|
||||
return false;
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ package com.imeeting.support;
|
|||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import io.lettuce.core.SetArgs;
|
||||
import io.lettuce.core.ScriptOutputType;
|
||||
import io.lettuce.core.api.StatefulRedisConnection;
|
||||
import io.lettuce.core.api.sync.RedisCommands;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
|
|
@ -103,6 +104,104 @@ public class RedisSupport {
|
|||
}
|
||||
}
|
||||
|
||||
public boolean deleteIfValue(String key, String expectedValue) {
|
||||
if (key == null || key.isBlank() || expectedValue == null) {
|
||||
return false;
|
||||
}
|
||||
try {
|
||||
Long deleted = commands().eval(
|
||||
"if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end",
|
||||
ScriptOutputType.INTEGER,
|
||||
new String[]{key},
|
||||
expectedValue
|
||||
);
|
||||
return deleted != null && deleted > 0;
|
||||
} catch (Exception ex) {
|
||||
log.warn("Conditional Redis delete failed, key={}", key, ex);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
public boolean expireIfValue(String key, String expectedValue, Duration ttl) {
|
||||
if (key == null || key.isBlank() || expectedValue == null || ttl == null) {
|
||||
return false;
|
||||
}
|
||||
try {
|
||||
Long updated = commands().eval(
|
||||
"if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('pexpire', KEYS[1], ARGV[2]) else return 0 end",
|
||||
ScriptOutputType.INTEGER,
|
||||
new String[]{key},
|
||||
expectedValue,
|
||||
String.valueOf(ttl.toMillis())
|
||||
);
|
||||
return updated != null && updated > 0;
|
||||
} catch (Exception ex) {
|
||||
log.warn("Conditional Redis expire failed, key={}", key, ex);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
public void publish(String channel, String message) {
|
||||
try {
|
||||
commands().publish(channel, message);
|
||||
} catch (Exception ex) {
|
||||
log.warn("Publish Redis message failed, channel={}", channel, ex);
|
||||
}
|
||||
}
|
||||
|
||||
public long incrementWithTtl(String key, Duration ttl) {
|
||||
try {
|
||||
Long value = commands().eval(
|
||||
"local value = redis.call('incr', KEYS[1]); if value == 1 then redis.call('pexpire', KEYS[1], ARGV[1]); end; return value",
|
||||
ScriptOutputType.INTEGER,
|
||||
new String[]{key},
|
||||
String.valueOf(ttl.toMillis())
|
||||
);
|
||||
return value == null ? 0L : value;
|
||||
} catch (Exception ex) {
|
||||
throw new RuntimeException("Redis rate limit increment failed, key=" + key, ex);
|
||||
}
|
||||
}
|
||||
|
||||
public boolean tryAcquireExpiringSlot(String key, String member, int maxMembers, Duration ttl) {
|
||||
if (key == null || key.isBlank() || member == null || member.isBlank() || maxMembers < 1 || ttl == null) {
|
||||
return false;
|
||||
}
|
||||
try {
|
||||
Long acquired = commands().eval(
|
||||
"local now = tonumber(ARGV[1]); local ttl = tonumber(ARGV[2]); "
|
||||
+ "redis.call('zremrangebyscore', KEYS[1], '-inf', now); "
|
||||
+ "if redis.call('zscore', KEYS[1], ARGV[4]) then "
|
||||
+ "redis.call('zadd', KEYS[1], now + ttl, ARGV[4]); redis.call('pexpire', KEYS[1], ttl); return 1 end; "
|
||||
+ "if redis.call('zcard', KEYS[1]) >= tonumber(ARGV[3]) then return 0 end; "
|
||||
+ "redis.call('zadd', KEYS[1], now + ttl, ARGV[4]); redis.call('pexpire', KEYS[1], ttl); return 1",
|
||||
ScriptOutputType.INTEGER,
|
||||
new String[]{key},
|
||||
String.valueOf(System.currentTimeMillis()),
|
||||
String.valueOf(ttl.toMillis()),
|
||||
String.valueOf(maxMembers),
|
||||
member
|
||||
);
|
||||
return acquired != null && acquired > 0;
|
||||
} catch (Exception ex) {
|
||||
log.warn("Acquire Redis set slot failed, key={}", key, ex);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
public boolean removeExpiringSlot(String key, String member) {
|
||||
if (key == null || key.isBlank() || member == null || member.isBlank()) {
|
||||
return false;
|
||||
}
|
||||
try {
|
||||
Long removed = commands().zrem(key, member);
|
||||
return removed != null && removed > 0;
|
||||
} catch (Exception ex) {
|
||||
log.warn("Remove Redis set member failed, key={}", key, ex);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
public void removeFromSetQuietly(String key, String... members) {
|
||||
if (members == null || members.length == 0) {
|
||||
return;
|
||||
|
|
|
|||
|
|
@ -0,0 +1,72 @@
|
|||
package com.imeeting.support.redis;
|
||||
|
||||
import com.imeeting.common.RedisKeys;
|
||||
import com.imeeting.support.RedisSupport;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.time.Duration;
|
||||
|
||||
/**
|
||||
* 会议级发布者租约。只允许一个 Web 或 gRPC 发布连接持有音频上行资格。
|
||||
*/
|
||||
@Component
|
||||
public class RealtimeMeetingPublisherLease {
|
||||
|
||||
private static final long DEFAULT_LEASE_TTL_SECONDS = 120L;
|
||||
|
||||
private final RedisSupport redisSupport;
|
||||
|
||||
@Value("${imeeting.realtime.publisher-lease-ttl-seconds:120}")
|
||||
private long leaseTtlSeconds = DEFAULT_LEASE_TTL_SECONDS;
|
||||
|
||||
public RealtimeMeetingPublisherLease(RedisSupport redisSupport) {
|
||||
this.redisSupport = redisSupport;
|
||||
}
|
||||
|
||||
public boolean tryAcquire(Long meetingId, String publisherFenceToken) {
|
||||
if (meetingId == null || publisherFenceToken == null || publisherFenceToken.isBlank()) {
|
||||
return false;
|
||||
}
|
||||
return redisSupport.setIfAbsentOrThrow(
|
||||
RedisKeys.realtimeMeetingPublisherLeaseKey(meetingId),
|
||||
publisherFenceToken,
|
||||
leaseTtl()
|
||||
);
|
||||
}
|
||||
|
||||
public boolean release(Long meetingId, String publisherFenceToken) {
|
||||
if (meetingId == null || publisherFenceToken == null || publisherFenceToken.isBlank()) {
|
||||
return false;
|
||||
}
|
||||
return redisSupport.deleteIfValue(
|
||||
RedisKeys.realtimeMeetingPublisherLeaseKey(meetingId),
|
||||
publisherFenceToken
|
||||
);
|
||||
}
|
||||
|
||||
public boolean renew(Long meetingId, String publisherFenceToken) {
|
||||
if (meetingId == null || publisherFenceToken == null || publisherFenceToken.isBlank()) {
|
||||
return false;
|
||||
}
|
||||
return redisSupport.expireIfValue(
|
||||
RedisKeys.realtimeMeetingPublisherLeaseKey(meetingId),
|
||||
publisherFenceToken,
|
||||
leaseTtl()
|
||||
);
|
||||
}
|
||||
|
||||
public boolean isHeldBy(Long meetingId, String publisherFenceToken) {
|
||||
if (meetingId == null || publisherFenceToken == null || publisherFenceToken.isBlank()) {
|
||||
return false;
|
||||
}
|
||||
return publisherFenceToken.equals(redisSupport.getStringQuietly(
|
||||
RedisKeys.realtimeMeetingPublisherLeaseKey(meetingId)
|
||||
));
|
||||
}
|
||||
|
||||
private Duration leaseTtl() {
|
||||
long seconds = leaseTtlSeconds > 0 ? leaseTtlSeconds : DEFAULT_LEASE_TTL_SECONDS;
|
||||
return Duration.ofSeconds(seconds);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,39 @@
|
|||
package com.imeeting.support.redis;
|
||||
|
||||
import com.imeeting.common.RedisKeys;
|
||||
import com.imeeting.dto.biz.RealtimeMeetingSubscriptionSessionData;
|
||||
import com.imeeting.support.RedisSupport;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.time.Duration;
|
||||
|
||||
@Component
|
||||
@RequiredArgsConstructor
|
||||
public class RealtimeMeetingSubscriptionSessionCache {
|
||||
|
||||
private static final Duration SESSION_TTL = Duration.ofMinutes(10);
|
||||
|
||||
private final RedisSupport redisSupport;
|
||||
|
||||
public void save(String sessionToken, RealtimeMeetingSubscriptionSessionData data) {
|
||||
if (sessionToken == null || sessionToken.isBlank() || data == null || data.getMeetingId() == null) {
|
||||
return;
|
||||
}
|
||||
redisSupport.setJson(RedisKeys.realtimeMeetingSubscriptionSessionKey(sessionToken), data, SESSION_TTL);
|
||||
}
|
||||
|
||||
public RealtimeMeetingSubscriptionSessionData get(String sessionToken) {
|
||||
if (sessionToken == null || sessionToken.isBlank()) {
|
||||
return null;
|
||||
}
|
||||
return redisSupport.getJsonQuietly(
|
||||
RedisKeys.realtimeMeetingSubscriptionSessionKey(sessionToken),
|
||||
RealtimeMeetingSubscriptionSessionData.class
|
||||
);
|
||||
}
|
||||
|
||||
public long getSessionTtlSeconds() {
|
||||
return SESSION_TTL.toSeconds();
|
||||
}
|
||||
}
|
||||
|
|
@ -10,8 +10,12 @@ import com.imeeting.service.realtime.RealtimeAsrChannelCallback;
|
|||
import com.imeeting.service.realtime.RealtimeAsrChannelContext;
|
||||
import com.imeeting.service.realtime.RealtimeAsrChannelFactory;
|
||||
import com.imeeting.service.realtime.RealtimeMeetingAudioStorageService;
|
||||
import com.imeeting.service.realtime.RealtimeMeetingEventHub;
|
||||
import com.imeeting.service.realtime.RealtimeMeetingEventRelay;
|
||||
import com.imeeting.service.realtime.RealtimeMeetingMetrics;
|
||||
import com.imeeting.service.realtime.RealtimeMeetingTranscriptCacheService;
|
||||
import com.imeeting.service.realtime.impl.LocalRealtimeAsrChannel;
|
||||
import com.imeeting.support.redis.RealtimeMeetingPublisherLease;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
|
@ -31,6 +35,7 @@ import java.time.Duration;
|
|||
import java.util.Arrays;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
|
@ -52,6 +57,9 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
|
|||
private final RealtimeMeetingAudioStorageService realtimeMeetingAudioStorageService;
|
||||
private final RealtimeMeetingTranscriptCacheService realtimeMeetingTranscriptCacheService;
|
||||
private final RealtimeAsrChannelFactory realtimeAsrChannelFactory;
|
||||
private final RealtimeMeetingPublisherLease publisherLease;
|
||||
private final RealtimeMeetingEventHub eventHub;
|
||||
private final RealtimeMeetingEventRelay eventRelay;
|
||||
private final ConcurrentMap<Long, MeetingChannelSession> meetingSessions = new ConcurrentHashMap<>();
|
||||
private final ConcurrentMap<Long, Object> meetingLocks = new ConcurrentHashMap<>();
|
||||
|
||||
|
|
@ -81,14 +89,19 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
|
|||
|
||||
@Override
|
||||
protected void handleTextMessage(WebSocketSession session, TextMessage message) {
|
||||
MeetingChannelSession meetingSession = getMeetingSession(session);
|
||||
// 过滤前端发来的心跳保活消息，不转发给上游 ASR 服务
|
||||
if (looksLikeKeepaliveMessage(message.getPayload())) {
|
||||
if (!renewAndCheckActivePublisher(session, meetingSession)) {
|
||||
closeStalePublisher(session);
|
||||
return;
|
||||
}
|
||||
log.debug("Frontend keepalive received, ignored: meetingId={}, sessionId={}",
|
||||
session.getAttributes().get(ATTR_MEETING_ID), session.getId());
|
||||
return;
|
||||
}
|
||||
MeetingChannelSession meetingSession = getMeetingSession(session);
|
||||
if (meetingSession == null || !meetingSession.isChannelOpen()) {
|
||||
if (meetingSession == null || !meetingSession.isChannelOpen()
|
||||
|| !renewAndCheckActivePublisher(session, meetingSession)) {
|
||||
log.warn("前端文本消息已忽略：上游 ASR 连接不可用，meetingId={}, sessionId={}",
|
||||
session.getAttributes().get(ATTR_MEETING_ID), session.getId());
|
||||
return;
|
||||
|
|
@ -103,7 +116,8 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
|
|||
@Override
|
||||
protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) {
|
||||
MeetingChannelSession meetingSession = getMeetingSession(session);
|
||||
if (meetingSession == null || !meetingSession.isChannelOpen()) {
|
||||
if (meetingSession == null || !meetingSession.isChannelOpen()
|
||||
|| !renewAndCheckActivePublisher(session, meetingSession)) {
|
||||
log.warn("前端音频帧已忽略：上游 ASR 连接不可用，meetingId={}, sessionId={}",
|
||||
session.getAttributes().get(ATTR_MEETING_ID), session.getId());
|
||||
return;
|
||||
|
|
@ -145,7 +159,7 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
|
|||
Object meetingIdValue = session.getAttributes().get(ATTR_MEETING_ID);
|
||||
if (meetingIdValue instanceof Long meetingId) {
|
||||
detachFrontend(meetingId, session.getId());
|
||||
realtimeMeetingSessionStateService.pauseByDisconnect(meetingId, session.getId());
|
||||
RealtimeMeetingMetrics.publisherClosed("websocket");
|
||||
}
|
||||
realtimeMeetingAudioStorageService.closeSession(session.getId());
|
||||
}
|
||||
|
|
@ -159,6 +173,7 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
|
|||
return;
|
||||
}
|
||||
meetingSession.channel.closeMeeting(meetingSession.context);
|
||||
publisherLease.release(meetingId, meetingSession.publisherFenceToken);
|
||||
}
|
||||
|
||||
private void attachFrontendSession(RealtimeSocketSessionData sessionData,
|
||||
|
|
@ -173,23 +188,36 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
|
|||
String previousSessionId = meetingSession.context.getRawSession() == null
|
||||
? null
|
||||
: meetingSession.context.getRawSession().getId();
|
||||
String previousFenceToken = meetingSession.publisherFenceToken;
|
||||
meetingSession.clearFrontendIfClosed();
|
||||
if (previousSessionId != null && !meetingSession.hasOpenFrontend()) {
|
||||
realtimeMeetingSessionStateService.pauseByDisconnect(meetingId, previousSessionId);
|
||||
realtimeMeetingSessionStateService.pauseByDisconnect(meetingId, previousFenceToken);
|
||||
publisherLease.release(meetingId, previousFenceToken);
|
||||
}
|
||||
if (meetingSession.hasOpenFrontend()) {
|
||||
RealtimeMeetingMetrics.publisherRejected("websocket");
|
||||
sendFrontendError(frontendSession, "REALTIME_ACTIVE_CONNECTION_EXISTS", "当前会议已有活跃前端连接");
|
||||
frontendSession.close(CloseStatus.POLICY_VIOLATION.withReason("已存在活跃的前端连接"));
|
||||
realtimeMeetingAudioStorageService.closeSession(rawSession.getId());
|
||||
return;
|
||||
}
|
||||
if (!realtimeMeetingSessionStateService.activate(meetingId, rawSession.getId())) {
|
||||
String publisherFenceToken = UUID.randomUUID().toString();
|
||||
if (!publisherLease.tryAcquire(meetingId, publisherFenceToken)) {
|
||||
RealtimeMeetingMetrics.publisherRejected("websocket");
|
||||
sendFrontendError(frontendSession, "REALTIME_ACTIVE_CONNECTION_EXISTS", "当前会议已有活跃发布连接");
|
||||
frontendSession.close(CloseStatus.POLICY_VIOLATION.withReason("已有活跃发布连接"));
|
||||
realtimeMeetingAudioStorageService.closeSession(rawSession.getId());
|
||||
return;
|
||||
}
|
||||
if (!realtimeMeetingSessionStateService.activate(meetingId, publisherFenceToken)) {
|
||||
publisherLease.release(meetingId, publisherFenceToken);
|
||||
sendFrontendError(frontendSession, "REALTIME_ACTIVE_CONNECTION_REJECTED", "当前状态下无法继续会议");
|
||||
frontendSession.close(CloseStatus.POLICY_VIOLATION.withReason("当前状态下无法继续会议"));
|
||||
realtimeMeetingAudioStorageService.closeSession(rawSession.getId());
|
||||
return;
|
||||
}
|
||||
meetingSession.bindFrontend(rawSession, frontendSession);
|
||||
meetingSession.bindFrontend(rawSession, frontendSession, publisherFenceToken);
|
||||
RealtimeMeetingMetrics.publisherOpened("websocket");
|
||||
reused = true;
|
||||
} else {
|
||||
RealtimeAsrChannel channel = realtimeAsrChannelFactory.getRequired(sessionData.getProvider());
|
||||
|
|
@ -197,12 +225,31 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
|
|||
context.setMeetingId(meetingId);
|
||||
context.setProvider(realtimeAsrChannelFactory.normalizeProvider(sessionData.getProvider()));
|
||||
context.setTargetWsUrl(sessionData.getTargetWsUrl());
|
||||
context.setCallback(new HandlerChannelCallback());
|
||||
context.setCallback(new HandlerChannelCallback(context));
|
||||
context.bindFrontendSession(rawSession, frontendSession);
|
||||
String publisherFenceToken = UUID.randomUUID().toString();
|
||||
context.setConnectionId(publisherFenceToken);
|
||||
context.getChannelState().put("modelCode", sessionData.getModelCode());
|
||||
context.getChannelState().put("mediaConfig", sessionData.getMediaConfig());
|
||||
meetingSession = new MeetingChannelSession(meetingId, channel, context);
|
||||
meetingSession = new MeetingChannelSession(meetingId, channel, context, publisherFenceToken);
|
||||
meetingSessions.put(meetingId, meetingSession);
|
||||
if (!publisherLease.tryAcquire(meetingId, publisherFenceToken)) {
|
||||
meetingSessions.remove(meetingId, meetingSession);
|
||||
RealtimeMeetingMetrics.publisherRejected("websocket");
|
||||
sendFrontendError(frontendSession, "REALTIME_ACTIVE_CONNECTION_EXISTS", "当前会议已有活跃发布连接");
|
||||
frontendSession.close(CloseStatus.POLICY_VIOLATION.withReason("已有活跃发布连接"));
|
||||
realtimeMeetingAudioStorageService.closeSession(rawSession.getId());
|
||||
return;
|
||||
}
|
||||
if (!realtimeMeetingSessionStateService.activate(meetingId, publisherFenceToken)) {
|
||||
publisherLease.release(meetingId, publisherFenceToken);
|
||||
meetingSessions.remove(meetingId, meetingSession);
|
||||
sendFrontendError(frontendSession, "REALTIME_ACTIVE_CONNECTION_REJECTED", "当前状态下无法继续会议");
|
||||
frontendSession.close(CloseStatus.POLICY_VIOLATION.withReason("当前状态下无法继续会议"));
|
||||
realtimeMeetingAudioStorageService.closeSession(rawSession.getId());
|
||||
return;
|
||||
}
|
||||
RealtimeMeetingMetrics.publisherOpened("websocket");
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -217,12 +264,16 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
|
|||
} catch (InterruptedException ex) {
|
||||
Thread.currentThread().interrupt();
|
||||
removeMeetingSession(meetingId, meetingSession);
|
||||
publisherLease.release(meetingId, meetingSession.publisherFenceToken);
|
||||
realtimeMeetingSessionStateService.pauseByDisconnect(meetingId, meetingSession.publisherFenceToken);
|
||||
log.error("连接上游 ASR websocket 时被中断：meetingId={}, sessionId={}", meetingId, rawSession.getId(), ex);
|
||||
sendFrontendError(frontendSession, "REALTIME_UPSTREAM_CONNECT_INTERRUPTED", "连接上游 ASR 服务时被中断");
|
||||
realtimeMeetingAudioStorageService.closeSession(rawSession.getId());
|
||||
frontendSession.close(CloseStatus.SERVER_ERROR.withReason("连接上游服务时被中断"));
|
||||
} catch (Exception ex) {
|
||||
removeMeetingSession(meetingId, meetingSession);
|
||||
publisherLease.release(meetingId, meetingSession.publisherFenceToken);
|
||||
realtimeMeetingSessionStateService.pauseByDisconnect(meetingId, meetingSession.publisherFenceToken);
|
||||
log.warn("连接上游 ASR websocket 失败：meetingId={}, provider={}, target={}",
|
||||
meetingId, sessionData.getProvider(), sessionData.getTargetWsUrl(), ex);
|
||||
sendFrontendError(frontendSession, "REALTIME_UPSTREAM_CONNECT_FAILED", "连接上游 ASR 服务失败");
|
||||
|
|
@ -265,8 +316,10 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
|
|||
synchronized (lockForMeeting(meetingId)) {
|
||||
if (meetingSession.context.getRawSession() != null && meetingSession.context.getRawSession().getId().equals(sessionId)) {
|
||||
meetingSession.channel.onFrontendDetached(meetingSession.context);
|
||||
publisherLease.release(meetingId, meetingSession.publisherFenceToken);
|
||||
realtimeMeetingSessionStateService.pauseByDisconnect(meetingId, meetingSession.publisherFenceToken);
|
||||
meetingSession.detachFrontend(sessionId);
|
||||
}
|
||||
meetingSession.detachFrontend(sessionId);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -278,6 +331,26 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
|
|||
return meetingSessions.get(meetingId);
|
||||
}
|
||||
|
||||
private boolean renewAndCheckActivePublisher(WebSocketSession session, MeetingChannelSession meetingSession) {
|
||||
if (meetingSession == null || meetingSession.context.getRawSession() == null
|
||||
|| !session.getId().equals(meetingSession.context.getRawSession().getId())) {
|
||||
return false;
|
||||
}
|
||||
return publisherLease.renew(meetingSession.meetingId, meetingSession.publisherFenceToken)
|
||||
&& realtimeMeetingSessionStateService.isActiveConnection(
|
||||
meetingSession.meetingId, meetingSession.publisherFenceToken);
|
||||
}
|
||||
|
||||
private void closeStalePublisher(WebSocketSession session) {
|
||||
try {
|
||||
if (session.isOpen()) {
|
||||
session.close(CloseStatus.POLICY_VIOLATION.withReason("发布连接已失效"));
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
log.debug("关闭失效发布连接失败：sessionId={}", session.getId(), ex);
|
||||
}
|
||||
}
|
||||
|
||||
void removeMeetingSession(Long meetingId) {
|
||||
synchronized (lockForMeeting(meetingId)) {
|
||||
meetingSessions.remove(meetingId);
|
||||
|
|
@ -290,6 +363,18 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
|
|||
}
|
||||
}
|
||||
|
||||
private void removeMeetingSession(Long meetingId, RealtimeAsrChannelContext context) {
|
||||
synchronized (lockForMeeting(meetingId)) {
|
||||
MeetingChannelSession meetingSession = meetingSessions.get(meetingId);
|
||||
if (meetingSession == null || meetingSession.context != context) {
|
||||
return;
|
||||
}
|
||||
realtimeMeetingSessionStateService.pauseByDisconnect(meetingId, meetingSession.publisherFenceToken);
|
||||
publisherLease.release(meetingId, meetingSession.publisherFenceToken);
|
||||
meetingSessions.remove(meetingId, meetingSession);
|
||||
}
|
||||
}
|
||||
|
||||
private Object lockForMeeting(Long meetingId) {
|
||||
return meetingLocks.computeIfAbsent(meetingId, ignored -> new Object());
|
||||
}
|
||||
|
|
@ -363,15 +448,21 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
|
|||
private final Long meetingId;
|
||||
private final RealtimeAsrChannel channel;
|
||||
private final RealtimeAsrChannelContext context;
|
||||
private String publisherFenceToken;
|
||||
|
||||
private MeetingChannelSession(Long meetingId, RealtimeAsrChannel channel, RealtimeAsrChannelContext context) {
|
||||
private MeetingChannelSession(Long meetingId, RealtimeAsrChannel channel,
|
||||
RealtimeAsrChannelContext context, String publisherFenceToken) {
|
||||
this.meetingId = meetingId;
|
||||
this.channel = channel;
|
||||
this.context = context;
|
||||
this.publisherFenceToken = publisherFenceToken;
|
||||
}
|
||||
|
||||
private void bindFrontend(WebSocketSession rawSession, ConcurrentWebSocketSessionDecorator frontendSession) {
|
||||
private void bindFrontend(WebSocketSession rawSession, ConcurrentWebSocketSessionDecorator frontendSession,
|
||||
String publisherFenceToken) {
|
||||
context.bindFrontendSession(rawSession, frontendSession);
|
||||
context.setConnectionId(publisherFenceToken);
|
||||
this.publisherFenceToken = publisherFenceToken;
|
||||
}
|
||||
|
||||
private void detachFrontend(String sessionId) {
|
||||
|
|
@ -399,6 +490,12 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
|
|||
}
|
||||
|
||||
private final class HandlerChannelCallback implements RealtimeAsrChannelCallback {
|
||||
private final RealtimeAsrChannelContext context;
|
||||
|
||||
private HandlerChannelCallback(RealtimeAsrChannelContext context) {
|
||||
this.context = context;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onChannelOpen(Long meetingId) throws Exception {
|
||||
MeetingChannelSession meetingSession = meetingSessions.get(meetingId);
|
||||
|
|
@ -421,6 +518,7 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
|
|||
if (frontendSession != null && frontendSession.isOpen()) {
|
||||
frontendSession.sendMessage(new TextMessage(payload));
|
||||
}
|
||||
eventRelay.publish(meetingId, payload == null ? "" : payload);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -449,7 +547,7 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
|
|||
|
||||
@Override
|
||||
public void removeMeetingSession(Long meetingId) {
|
||||
RealtimeMeetingProxyWebSocketHandler.this.removeMeetingSession(meetingId);
|
||||
RealtimeMeetingProxyWebSocketHandler.this.removeMeetingSession(meetingId, context);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
|
|||
|
|
@ -0,0 +1,133 @@
|
|||
package com.imeeting.websocket;
|
||||
|
||||
import com.imeeting.dto.biz.RealtimeMeetingSessionStatusVO;
|
||||
import com.imeeting.dto.biz.RealtimeMeetingSubscriptionSessionData;
|
||||
import com.imeeting.service.biz.RealtimeMeetingSessionStateService;
|
||||
import com.imeeting.service.realtime.RealtimeMeetingEventHub;
|
||||
import com.imeeting.service.realtime.RealtimeMeetingTranscriptCacheService;
|
||||
import com.imeeting.service.realtime.RealtimeMeetingSubscriptionQuota;
|
||||
import com.imeeting.service.realtime.RealtimeMeetingMetrics;
|
||||
import com.imeeting.service.realtime.impl.LocalRealtimeAsrChannel;
|
||||
import com.imeeting.support.redis.RealtimeMeetingSubscriptionSessionCache;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.web.socket.CloseStatus;
|
||||
import org.springframework.web.socket.TextMessage;
|
||||
import org.springframework.web.socket.WebSocketSession;
|
||||
import org.springframework.web.socket.handler.AbstractWebSocketHandler;
|
||||
import org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorator;
|
||||
|
||||
import java.net.URI;
|
||||
import java.net.URLDecoder;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.Arrays;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.function.Consumer;
|
||||
|
||||
@Component
|
||||
@RequiredArgsConstructor
|
||||
public class RealtimeMeetingSubscriptionWebSocketHandler extends AbstractWebSocketHandler {
|
||||
|
||||
private final RealtimeMeetingSubscriptionSessionCache subscriptionSessionCache;
|
||||
private final RealtimeMeetingSessionStateService sessionStateService;
|
||||
private final RealtimeMeetingTranscriptCacheService transcriptCacheService;
|
||||
private final RealtimeMeetingEventHub eventHub;
|
||||
private final RealtimeMeetingSubscriptionQuota subscriptionQuota;
|
||||
private final Map<String, Subscription> subscriptions = new ConcurrentHashMap<>();
|
||||
|
||||
@Value("${imeeting.realtime.max-subscribers-per-meeting:200}")
|
||||
private int maxSubscribersPerMeeting = 200;
|
||||
|
||||
@Override
|
||||
public void afterConnectionEstablished(WebSocketSession session) throws Exception {
|
||||
String token = extractQueryParam(session.getUri(), "sessionToken");
|
||||
RealtimeMeetingSubscriptionSessionData data = subscriptionSessionCache.get(token);
|
||||
if (data == null || data.getMeetingId() == null) {
|
||||
session.close(CloseStatus.POLICY_VIOLATION.withReason("实时订阅会话无效"));
|
||||
return;
|
||||
}
|
||||
RealtimeMeetingSessionStatusVO status = sessionStateService.getStatus(data.getMeetingId());
|
||||
if (status == null || !"ACTIVE".equals(status.getStatus())) {
|
||||
session.close(CloseStatus.POLICY_VIOLATION.withReason("会议尚未开始"));
|
||||
return;
|
||||
}
|
||||
ConcurrentWebSocketSessionDecorator decorated = new ConcurrentWebSocketSessionDecorator(session, 15_000, 1_048_576);
|
||||
Consumer<String> consumer = payload -> send(decorated, payload);
|
||||
String member = subscriptionQuota.member(session.getId());
|
||||
if (!subscriptionQuota.tryAcquire(data.getMeetingId(), member)) {
|
||||
RealtimeMeetingMetrics.subscriberRejected("websocket");
|
||||
session.close(CloseStatus.POLICY_VIOLATION.withReason("会议实时订阅人数已达上限"));
|
||||
return;
|
||||
}
|
||||
if (!eventHub.trySubscribe(data.getMeetingId(), consumer, maxSubscribersPerMeeting)) {
|
||||
subscriptionQuota.release(data.getMeetingId(), member);
|
||||
RealtimeMeetingMetrics.subscriberRejected("websocket");
|
||||
session.close(CloseStatus.POLICY_VIOLATION.withReason("会议实时订阅人数已达上限"));
|
||||
return;
|
||||
}
|
||||
Subscription subscription = new Subscription(data.getMeetingId(), consumer, member);
|
||||
subscriptions.put(session.getId(), subscription);
|
||||
RealtimeMeetingMetrics.subscriberOpened("websocket");
|
||||
send(decorated, "{\"type\":\"state\",\"data\":{\"status\":\"ACTIVE\"}}");
|
||||
replayFinalTranscripts(data.getMeetingId(), decorated);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
|
||||
Subscription subscription = subscriptions.remove(session.getId());
|
||||
if (subscription != null) {
|
||||
eventHub.unsubscribe(subscription.meetingId(), subscription.consumer());
|
||||
subscriptionQuota.release(subscription.meetingId(), subscription.member());
|
||||
RealtimeMeetingMetrics.subscriberClosed("websocket");
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void handleTextMessage(WebSocketSession session, TextMessage message) {
|
||||
// 只读订阅通道不处理任何客户端业务消息。
|
||||
}
|
||||
|
||||
@Scheduled(fixedDelay = 60_000)
|
||||
public void renewGlobalQuotaSlots() {
|
||||
subscriptions.values().forEach(subscription ->
|
||||
subscriptionQuota.renew(subscription.meetingId(), subscription.member()));
|
||||
}
|
||||
|
||||
private void replayFinalTranscripts(Long meetingId, ConcurrentWebSocketSessionDecorator session) {
|
||||
try {
|
||||
for (var item : transcriptCacheService.listOrderedItems(meetingId)) {
|
||||
send(session, LocalRealtimeAsrChannel.buildFrontendTranscriptMessage(item));
|
||||
}
|
||||
} catch (Exception ignored) {
|
||||
// 历史回放失败不影响后续实时事件订阅。
|
||||
}
|
||||
}
|
||||
|
||||
private void send(ConcurrentWebSocketSessionDecorator session, String payload) {
|
||||
try {
|
||||
if (session.isOpen()) {
|
||||
session.sendMessage(new TextMessage(payload == null ? "" : payload));
|
||||
}
|
||||
} catch (Exception ignored) {
|
||||
// 后续关闭回调会移除失效订阅者。
|
||||
}
|
||||
}
|
||||
|
||||
private String extractQueryParam(URI uri, String key) {
|
||||
if (uri == null || uri.getQuery() == null || uri.getQuery().isBlank()) {
|
||||
return null;
|
||||
}
|
||||
return Arrays.stream(uri.getQuery().split("&"))
|
||||
.map(item -> item.split("=", 2))
|
||||
.filter(parts -> parts.length == 2 && key.equals(parts[0]))
|
||||
.map(parts -> URLDecoder.decode(parts[1], StandardCharsets.UTF_8))
|
||||
.findFirst()
|
||||
.orElse(null);
|
||||
}
|
||||
|
||||
private record Subscription(Long meetingId, Consumer<String> consumer, String member) {
|
||||
}
|
||||
}
|
||||
|
|
@ -230,6 +230,12 @@ export interface RealtimeSocketSessionVO {
|
|||
startMessage: Record<string, any>;
|
||||
}
|
||||
|
||||
export interface RealtimeMeetingSubscriptionSessionVO {
|
||||
sessionToken: string;
|
||||
path: string;
|
||||
expiresInSeconds: number;
|
||||
}
|
||||
|
||||
export interface RealtimeSocketSessionRequest {
|
||||
asrModelId: number;
|
||||
mode?: string;
|
||||
|
|
@ -261,9 +267,10 @@ export const createRealtimeMeeting = (data: CreateRealtimeMeetingCommand) => {
|
|||
};
|
||||
|
||||
|
||||
export const getRealtimeMeetingSessionStatus = (meetingId: number) => {
|
||||
export const getRealtimeMeetingSessionStatus = (meetingId: number, options?: { suppressErrorToast?: boolean }) => {
|
||||
return http.get<{ code: string; data: RealtimeMeetingSessionStatus; msg: string }>(
|
||||
`/api/biz/meeting/${meetingId}/realtime/session-status`
|
||||
`/api/biz/meeting/${meetingId}/realtime/session-status`,
|
||||
{suppressErrorToast: options?.suppressErrorToast}
|
||||
);
|
||||
};
|
||||
|
||||
|
|
@ -298,6 +305,16 @@ export const completeRealtimeMeeting = (meetingId: number, data?: { audioUrl?: s
|
|||
);
|
||||
};
|
||||
|
||||
export const createRealtimeMeetingSubscriptionSession = (meetingId: number, options?: {
|
||||
suppressErrorToast?: boolean
|
||||
}) => {
|
||||
return http.post<{ code: string; data: RealtimeMeetingSubscriptionSessionVO; msg: string }>(
|
||||
`/api/biz/meeting/${meetingId}/realtime/subscription-session`,
|
||||
{},
|
||||
{suppressErrorToast: options?.suppressErrorToast},
|
||||
);
|
||||
};
|
||||
|
||||
export const deleteMeeting = (id: number) => {
|
||||
return http.delete<{ code: string; data: boolean; msg: string }>(
|
||||
`/api/biz/meeting/${id}`
|
||||
|
|
|
|||
|
|
@ -30,6 +30,7 @@ import {
|
|||
getMeetingChapters,
|
||||
getMeetingProgress,
|
||||
getMeetingShareConfig,
|
||||
getRealtimeMeetingSessionStatus,
|
||||
getTranscripts,
|
||||
MeetingChapterVO,
|
||||
MeetingProgress,
|
||||
|
|
@ -1438,7 +1439,6 @@ const MeetingDetail: React.FC = () => {
|
|||
}
|
||||
return false;
|
||||
}, [meeting]);
|
||||
|
||||
const aiCatalogEnabled = meeting?.aiCatalogEnabled !== false;
|
||||
const canRetrySummary = isOwner
|
||||
&& transcripts.length > 0
|
||||
|
|
@ -1597,6 +1597,28 @@ const MeetingDetail: React.FC = () => {
|
|||
loadUsers();
|
||||
}, [id, fetchData]);
|
||||
|
||||
useEffect(() => {
|
||||
if (!meeting || meeting.meetingType !== 'REALTIME') {
|
||||
return;
|
||||
}
|
||||
let disposed = false;
|
||||
const redirectToRealtimeSession = async () => {
|
||||
try {
|
||||
const response = await getRealtimeMeetingSessionStatus(meeting.id, {suppressErrorToast: true});
|
||||
const status = response.data.data?.status;
|
||||
if (!disposed && status !== 'COMPLETING' && status !== 'COMPLETED') {
|
||||
navigate(`/meeting-live-session/${meeting.id}`, {replace: true});
|
||||
}
|
||||
} catch {
|
||||
// Redis 状态暂时不可用时保留详情页，避免错误跳转。
|
||||
}
|
||||
};
|
||||
void redirectToRealtimeSession();
|
||||
return () => {
|
||||
disposed = true;
|
||||
};
|
||||
}, [meeting?.id, meeting?.meetingType, navigate]);
|
||||
|
||||
useEffect(() => {
|
||||
setSelectedKeywords((current) => current.filter((item) => analysis.keywords.includes(item)));
|
||||
}, [analysis.keywords]);
|
||||
|
|
@ -2488,7 +2510,7 @@ const MeetingDetail: React.FC = () => {
|
|||
)}
|
||||
>
|
||||
<div className="meeting-detail-workspace">
|
||||
{meeting.status === 0 || meeting.status === 1 ? (
|
||||
{(meeting.status === 0 || meeting.status === 1) ? (
|
||||
<>
|
||||
<div style={{ display: 'none' }}>
|
||||
<MeetingProgressDisplay
|
||||
|
|
|
|||
|
|
@ -776,19 +776,10 @@ const Meetings: React.FC = () => {
|
|||
navigate("/meetings/" + meeting.id);
|
||||
return;
|
||||
}
|
||||
if (!canControlRealtimeFromCurrentPlatform(meeting)) {
|
||||
message.info(`该实时会议需在${getRealtimeSourceLabel(meeting)}继续，当前仅支持查看详情`);
|
||||
navigate("/meetings/" + meeting.id);
|
||||
return;
|
||||
}
|
||||
if (canOpenRealtimeSession(meeting.realtimeSessionStatus)) {
|
||||
navigate("/meeting-live-session/" + meeting.id);
|
||||
return;
|
||||
}
|
||||
if (!canManageMeeting(meeting)) {
|
||||
navigate("/meetings/" + meeting.id);
|
||||
return;
|
||||
}
|
||||
try {
|
||||
const res = await getRealtimeMeetingSessionStatus(meeting.id);
|
||||
if (canOpenRealtimeSession(res.data?.data?.status)) {
|
||||
|
|
|
|||
|
|
@ -13,9 +13,11 @@ import {
|
|||
import { useNavigate, useParams } from "react-router-dom";
|
||||
import PageContainer from "@/components/shared/PageContainer";
|
||||
import SectionCard from "@/components/shared/SectionCard";
|
||||
import {isRealtimeMeetingPublisher} from "../../utils/realtimePublisher";
|
||||
import "./RealtimeAsrSession.css";
|
||||
import {
|
||||
completeRealtimeMeeting,
|
||||
createRealtimeMeetingSubscriptionSession,
|
||||
getMeetingDetail,
|
||||
getRealtimeMeetingSessionStatus,
|
||||
getTranscripts,
|
||||
|
|
@ -29,16 +31,6 @@ import {
|
|||
const SAMPLE_RATE = 16000;
|
||||
const CHUNK_SIZE = 1280;
|
||||
const CURRENT_PLATFORM = "WEB" as const;
|
||||
const MEETING_SOURCE_LABELS: Record<string, string> = {
|
||||
WINDOWS: "Windows",
|
||||
MACOS: "macOS",
|
||||
KYLIN: "麒麟",
|
||||
UOS: "统信",
|
||||
HARMONYOS: "鸿蒙",
|
||||
WEB: "Web端",
|
||||
CUSTOM_TERMINAL: "定制终端",
|
||||
ANDROID: "定制终端",
|
||||
};
|
||||
|
||||
type WsSpeaker = string | { name?: string; user_id?: string | number } | undefined;
|
||||
type WsMessage = {
|
||||
|
|
@ -193,7 +185,16 @@ function buildTranscriptCardId(sentenceKey?: string, sentenceId?: number) {
|
|||
return `sentence-${sentenceId}`;
|
||||
}
|
||||
|
||||
function buildRealtimeProxyWsUrl(socketSession: RealtimeSocketSessionVO) {
|
||||
function getCurrentUserId() {
|
||||
try {
|
||||
const profileStr = sessionStorage.getItem("userProfile");
|
||||
return profileStr ? JSON.parse(profileStr).userId : undefined;
|
||||
} catch {
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
function buildRealtimeProxyWsUrl(socketSession: Pick<RealtimeSocketSessionVO, "path" | "sessionToken">) {
|
||||
const protocol = window.location.protocol === "https:" ? "wss" : "ws";
|
||||
return `${protocol}://${window.location.host}${socketSession.path}?sessionToken=${encodeURIComponent(socketSession.sessionToken)}`;
|
||||
}
|
||||
|
|
@ -250,6 +251,7 @@ export function RealtimeAsrSession() {
|
|||
const [sessionStatus, setSessionStatus] = useState<RealtimeMeetingSessionStatus | null>(null);
|
||||
const transcriptRef = useRef<HTMLDivElement | null>(null);
|
||||
const wsRef = useRef<WebSocket | null>(null);
|
||||
const subscriptionWsRef = useRef<WebSocket | null>(null);
|
||||
const wsHeartbeatRef = useRef<ReturnType<typeof setInterval> | null>(null);
|
||||
const audioContextRef = useRef<AudioContext | null>(null);
|
||||
const processorRef = useRef<ScriptProcessorNode | null>(null);
|
||||
|
|
@ -267,7 +269,8 @@ export function RealtimeAsrSession() {
|
|||
);
|
||||
const statusColor = recording ? "#1677ff" : connecting || finishing ? "#faad14" : "#94a3b8";
|
||||
const hasRemoteActiveConnection = Boolean(sessionStatus?.activeConnection) && !recording && !connecting;
|
||||
const canControlCurrentMeeting = !meeting?.meetingSource || meeting.meetingSource === CURRENT_PLATFORM;
|
||||
const canControlCurrentMeeting = isRealtimeMeetingPublisher(meeting?.creatorId, getCurrentUserId())
|
||||
&& (!meeting?.meetingSource || meeting.meetingSource === CURRENT_PLATFORM);
|
||||
|
||||
useEffect(() => {
|
||||
if (!meetingId || Number.isNaN(meetingId)) {
|
||||
|
|
@ -280,22 +283,20 @@ export function RealtimeAsrSession() {
|
|||
const stored = sessionStorage.getItem(getSessionKey(meetingId));
|
||||
const parsedDraft = stored ? JSON.parse(stored) : null;
|
||||
|
||||
const [detailRes, transcriptRes, statusRes] = await Promise.all([
|
||||
const [detailRes, transcriptRes] = await Promise.all([
|
||||
getMeetingDetail(meetingId),
|
||||
getTranscripts(meetingId),
|
||||
getRealtimeMeetingSessionStatus(meetingId),
|
||||
]);
|
||||
const detail = detailRes.data.data;
|
||||
const realtimeStatus = statusRes.data.data;
|
||||
if (detail.meetingType && detail.meetingType !== "REALTIME") {
|
||||
message.warning("当前会议不是实时会议，无法进入实时控制页");
|
||||
navigate(`/meetings/${meetingId}`);
|
||||
return;
|
||||
}
|
||||
if (detail.meetingSource && detail.meetingSource !== CURRENT_PLATFORM) {
|
||||
const sourceLabel = MEETING_SOURCE_LABELS[detail.meetingSource] ?? detail.meetingSource;
|
||||
message.warning(`该实时会议需在${sourceLabel}继续，当前仅支持查看详情`);
|
||||
navigate(`/meetings/${meetingId}`);
|
||||
const statusRes = await getRealtimeMeetingSessionStatus(meetingId);
|
||||
const realtimeStatus = statusRes.data.data;
|
||||
if (realtimeStatus?.status === "COMPLETING" || realtimeStatus?.status === "COMPLETED") {
|
||||
navigate(`/meetings/${meetingId}`, {replace: true});
|
||||
return;
|
||||
}
|
||||
setMeeting(detail);
|
||||
|
|
@ -306,7 +307,15 @@ export function RealtimeAsrSession() {
|
|||
if (resolvedDraft) {
|
||||
sessionStorage.setItem(getSessionKey(meetingId), JSON.stringify(resolvedDraft));
|
||||
}
|
||||
if (realtimeStatus?.status === "PAUSED_RESUMABLE") {
|
||||
const canControl = isRealtimeMeetingPublisher(detail.creatorId, getCurrentUserId())
|
||||
&& (!detail.meetingSource || detail.meetingSource === CURRENT_PLATFORM);
|
||||
if (!canControl && realtimeStatus?.status === "ACTIVE") {
|
||||
setStatusText("实时转写中");
|
||||
} else if (!canControl && (realtimeStatus?.status === "PAUSED_RESUMABLE" || realtimeStatus?.status === "PAUSED_EMPTY")) {
|
||||
setStatusText("实时转写已暂停，等待主持人继续");
|
||||
} else if (!canControl && realtimeStatus?.status === "IDLE") {
|
||||
setStatusText("等待主持人开始实时转写");
|
||||
} else if (realtimeStatus?.status === "PAUSED_RESUMABLE") {
|
||||
setStatusText(`已暂停，可在 ${Math.max(1, Math.ceil((realtimeStatus.remainingSeconds || 0) / 60))} 分钟内继续`);
|
||||
} else if (realtimeStatus?.status === "PAUSED_EMPTY") {
|
||||
setStatusText("已暂停，可继续识别");
|
||||
|
|
@ -356,6 +365,9 @@ export function RealtimeAsrSession() {
|
|||
}, [streamingText, transcripts]);
|
||||
|
||||
useEffect(() => {
|
||||
if (!canControlCurrentMeeting) {
|
||||
return;
|
||||
}
|
||||
const handlePageHide = () => {
|
||||
if (!meetingId || completeOnceRef.current) {
|
||||
return;
|
||||
|
|
@ -377,7 +389,7 @@ export function RealtimeAsrSession() {
|
|||
|
||||
window.addEventListener("pagehide", handlePageHide);
|
||||
return () => window.removeEventListener("pagehide", handlePageHide);
|
||||
}, [meetingId]);
|
||||
}, [canControlCurrentMeeting, meetingId]);
|
||||
|
||||
// 组件卸载（切换路由）时统一清理所有资源，防止定时器、WebSocket、音频管道泄漏
|
||||
useEffect(() => {
|
||||
|
|
@ -449,18 +461,123 @@ export function RealtimeAsrSession() {
|
|||
endTime: normalized.endTime,
|
||||
final: true,
|
||||
};
|
||||
if (normalized.sentenceKey || (normalized.sentenceId !== undefined && normalized.sentenceId !== null)) {
|
||||
const index = next.findIndex((item) => item.id === cardId);
|
||||
if (index >= 0) {
|
||||
next[index] = {...next[index], ...nextCard};
|
||||
return next;
|
||||
}
|
||||
const index = next.findIndex((item) =>
|
||||
item.id === cardId
|
||||
|| (item.final
|
||||
&& item.text === nextCard.text
|
||||
&& item.startTime === nextCard.startTime
|
||||
&& item.endTime === nextCard.endTime),
|
||||
);
|
||||
if (index >= 0) {
|
||||
next[index] = {...next[index], ...nextCard, id: next[index].id};
|
||||
} else {
|
||||
next.push(nextCard);
|
||||
}
|
||||
next.push(nextCard);
|
||||
return next;
|
||||
return next.sort((left, right) => (left.startTime ?? Number.MAX_SAFE_INTEGER) - (right.startTime ?? Number.MAX_SAFE_INTEGER));
|
||||
});
|
||||
};
|
||||
|
||||
useEffect(() => {
|
||||
if (!meeting || canControlCurrentMeeting) {
|
||||
return;
|
||||
}
|
||||
|
||||
let disposed = false;
|
||||
let timer: ReturnType<typeof window.setTimeout> | null = null;
|
||||
let subscriptionOpening = false;
|
||||
const closeSubscription = () => {
|
||||
const socket = subscriptionWsRef.current;
|
||||
if (socket) {
|
||||
socket.onclose = null;
|
||||
socket.close();
|
||||
subscriptionWsRef.current = null;
|
||||
}
|
||||
};
|
||||
const connect = async () => {
|
||||
if (subscriptionOpening || subscriptionWsRef.current?.readyState === WebSocket.OPEN
|
||||
|| subscriptionWsRef.current?.readyState === WebSocket.CONNECTING) {
|
||||
return;
|
||||
}
|
||||
subscriptionOpening = true;
|
||||
try {
|
||||
const response = await createRealtimeMeetingSubscriptionSession(meeting.id, {suppressErrorToast: true});
|
||||
const session = response.data.data;
|
||||
if (disposed || !session?.sessionToken) {
|
||||
return;
|
||||
}
|
||||
const socket = new WebSocket(buildRealtimeProxyWsUrl(session));
|
||||
subscriptionWsRef.current = socket;
|
||||
socket.onmessage = (event) => {
|
||||
try {
|
||||
const normalized = normalizeWsMessage(JSON.parse(String(event.data)) as WsMessage);
|
||||
if (!normalized) {
|
||||
return;
|
||||
}
|
||||
const speaker = resolveSpeaker(normalized.speaker);
|
||||
if (normalized.isFinal) {
|
||||
upsertTranscriptCard(normalized, speaker);
|
||||
setStreamingText("");
|
||||
setStreamingSpeaker("Unknown");
|
||||
} else {
|
||||
setStreamingText(normalized.text);
|
||||
setStreamingSpeaker(speaker.speakerName);
|
||||
}
|
||||
} catch {
|
||||
// 忽略订阅通道的非转写消息与异常数据。
|
||||
}
|
||||
};
|
||||
socket.onclose = () => {
|
||||
if (subscriptionWsRef.current === socket) {
|
||||
subscriptionWsRef.current = null;
|
||||
}
|
||||
};
|
||||
} catch {
|
||||
// 状态轮询会在发布端恢复 ACTIVE 后自动重新订阅。
|
||||
} finally {
|
||||
subscriptionOpening = false;
|
||||
}
|
||||
};
|
||||
const poll = async () => {
|
||||
let status: RealtimeMeetingSessionStatus | undefined;
|
||||
try {
|
||||
const response = await getRealtimeMeetingSessionStatus(meeting.id, {suppressErrorToast: true});
|
||||
status = response.data.data;
|
||||
if (disposed || !status) {
|
||||
return;
|
||||
}
|
||||
setSessionStatus(status);
|
||||
if (status.status === "COMPLETING" || status.status === "COMPLETED") {
|
||||
navigate(`/meetings/${meeting.id}`, {replace: true});
|
||||
return;
|
||||
}
|
||||
if (status.status === "ACTIVE") {
|
||||
setStatusText("实时转写中");
|
||||
await connect();
|
||||
} else {
|
||||
closeSubscription();
|
||||
setStreamingText("");
|
||||
setStatusText(status.status === "IDLE" ? "等待主持人开始实时转写" : "实时转写已暂停，等待主持人继续");
|
||||
}
|
||||
} catch {
|
||||
// 保持只读页面可见，下一次轮询会重试状态和订阅连接。
|
||||
} finally {
|
||||
if (!disposed && status?.status !== "COMPLETING" && status?.status !== "COMPLETED") {
|
||||
timer = window.setTimeout(() => {
|
||||
void poll();
|
||||
}, 3000);
|
||||
}
|
||||
}
|
||||
};
|
||||
void poll();
|
||||
return () => {
|
||||
disposed = true;
|
||||
if (timer) {
|
||||
window.clearTimeout(timer);
|
||||
}
|
||||
closeSubscription();
|
||||
};
|
||||
}, [canControlCurrentMeeting, meeting, navigate]);
|
||||
|
||||
const handleFatalRealtimeError = async (errorMessage: string) => {
|
||||
setConnecting(false);
|
||||
setRecording(false);
|
||||
|
|
@ -838,13 +955,13 @@ export function RealtimeAsrSession() {
|
|||
description={
|
||||
<span className="realtime-session-meta">
|
||||
<Badge color={statusColor} text={<span>{statusText}</span>} />
|
||||
{sessionDraft ? <span>字数 <strong>{totalTranscriptChars}</strong></span> : null}
|
||||
<span>字数 <strong>{totalTranscriptChars}</strong></span>
|
||||
</span>
|
||||
}
|
||||
contentClassName="realtime-session-section-content"
|
||||
>
|
||||
<div className="realtime-session-body">
|
||||
{!sessionDraft ? (
|
||||
{canControlCurrentMeeting && !sessionDraft ? (
|
||||
<div className="realtime-session-alert-wrap">
|
||||
<Alert
|
||||
type="warning"
|
||||
|
|
@ -861,7 +978,9 @@ export function RealtimeAsrSession() {
|
|||
<div className="realtime-session-empty">
|
||||
<Empty
|
||||
image={Empty.PRESENTED_IMAGE_SIMPLE}
|
||||
description={hasRemoteActiveConnection ? "当前会议已有活跃连接，请先关闭旧连接后再继续。" : "会议已就绪，点击下方按钮开始识别。"}
|
||||
description={canControlCurrentMeeting
|
||||
? (hasRemoteActiveConnection ? "当前会议已有活跃连接，请先关闭旧连接后再继续。" : "会议已就绪，点击下方按钮开始识别。")
|
||||
: (sessionStatus?.status === "ACTIVE" ? "正在等待实时转写内容。" : "等待主持人开始或恢复实时转写。")}
|
||||
/>
|
||||
</div>
|
||||
) : (
|
||||
|
|
@ -899,7 +1018,7 @@ export function RealtimeAsrSession() {
|
|||
)}
|
||||
</div>
|
||||
|
||||
<div className="realtime-control-bar">
|
||||
{canControlCurrentMeeting && <div className="realtime-control-bar">
|
||||
<div className="realtime-control-status">
|
||||
<div
|
||||
className={recording ? "realtime-control-orb recording-orb" : "realtime-control-orb"}
|
||||
|
|
@ -950,7 +1069,7 @@ export function RealtimeAsrSession() {
|
|||
/>
|
||||
)}
|
||||
</Space>
|
||||
</div>
|
||||
</div>}
|
||||
</div>
|
||||
)}
|
||||
</div>
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@ const ForgotPasswordPage = lazy(() => import("@/pages/forgot-password"));
|
|||
const MeetingsPage = lazy(() => import("@/pages/meetings"));
|
||||
const MeetingDetailPage = lazy(() => import("@/pages/meeting-detail"));
|
||||
const MeetingPreviewPage = lazy(() => import("@/pages/meeting-preview"));
|
||||
const RealtimeMeetingPreviewPage = lazy(() => import("@/pages/realtime-meeting-preview"));
|
||||
const ProfilePage = lazy(() => import("@/pages/profile"));
|
||||
const PasswordPage = lazy(() => import("@/pages/password"));
|
||||
const ScanConfirmPage = lazy(() => import("@/pages/scan-confirm"));
|
||||
|
|
@ -27,6 +28,7 @@ export default function App() {
|
|||
<Route path="/login" element={<LoginPage />} />
|
||||
<Route path="/forgot-password" element={<ForgotPasswordPage/>}/>
|
||||
<Route path="/meetings/:id/preview" element={<MeetingPreviewPage />} />
|
||||
<Route path="/meetings/:id/realtime-preview" element={<RealtimeMeetingPreviewPage/>}/>
|
||||
|
||||
<Route
|
||||
path="/meetings"
|
||||
|
|
|
|||
|
|
@ -3,6 +3,8 @@ import type {
|
|||
MeetingChapterVO,
|
||||
MeetingPageResult,
|
||||
MeetingPreviewAccessVO,
|
||||
RealtimeMeetingSessionStatusVO,
|
||||
RealtimeMeetingSubscriptionSessionVO,
|
||||
MeetingTranscriptVO,
|
||||
MeetingVO,
|
||||
PublicMeetingPreviewVO,
|
||||
|
|
@ -67,6 +69,31 @@ export const getPublicMeetingPreview = (id: number, accessPassword?: string) =>
|
|||
});
|
||||
};
|
||||
|
||||
export const createPublicRealtimeSubscriptionSession = (
|
||||
id: number,
|
||||
accessPassword?: string,
|
||||
options?: { suppressErrorToast?: boolean },
|
||||
) => {
|
||||
return http.post<{ code: string; data: RealtimeMeetingSubscriptionSessionVO; msg: string }>(
|
||||
`/api/public/meetings/${id}/preview/realtime-subscription-session`,
|
||||
undefined,
|
||||
{
|
||||
params: accessPassword ? {accessPassword} : undefined,
|
||||
suppressErrorToast: options?.suppressErrorToast,
|
||||
},
|
||||
);
|
||||
};
|
||||
|
||||
export const getPublicRealtimeMeetingSessionStatus = (id: number, accessPassword?: string) => {
|
||||
return http.get<{ code: string; data: RealtimeMeetingSessionStatusVO; msg: string }>(
|
||||
`/api/public/meetings/${id}/preview/realtime/session-status`,
|
||||
{
|
||||
params: accessPassword ? {accessPassword} : undefined,
|
||||
suppressErrorToast: true,
|
||||
},
|
||||
);
|
||||
};
|
||||
|
||||
export const createPublicDeviceMeetingBySession = (sessionId: string) => {
|
||||
return http.post<{ code: string; data: boolean; msg: string }>(`/api/biz/public-device-meetings/sessions/${sessionId}/create`);
|
||||
};
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ import { App, Button, Card, Input, Space, Typography } from "antd";
|
|||
import { useEffect, useMemo, useState } from "react";
|
||||
import { useNavigate, useParams, useSearchParams } from "react-router-dom";
|
||||
|
||||
import { getMeetingPreviewAccess, getPublicMeetingPreview } from "@/api/meeting";
|
||||
import {getMeetingPreviewAccess, getPublicMeetingPreview, getPublicRealtimeMeetingSessionStatus} from "@/api/meeting";
|
||||
import LoadingScreen from "@/components/LoadingScreen";
|
||||
import PageHeader from "@/components/PageHeader";
|
||||
import MeetingPreviewView from "@/components/preview/MeetingPreviewView";
|
||||
|
|
@ -36,10 +36,25 @@ export default function MeetingPreviewPage() {
|
|||
|
||||
const loadPreview = async (password?: string) => {
|
||||
const previewResp = await getPublicMeetingPreview(meetingId, password);
|
||||
setMeeting(previewResp.data.data.meeting);
|
||||
setTranscripts(previewResp.data.data.transcripts || []);
|
||||
setChapters(previewResp.data.data.chapters || []);
|
||||
const preview = previewResp.data.data;
|
||||
if (preview.meeting.meetingType === "REALTIME") {
|
||||
try {
|
||||
const statusResp = await getPublicRealtimeMeetingSessionStatus(meetingId, password);
|
||||
const status = statusResp.data.data.status;
|
||||
if (status !== "COMPLETING" && status !== "COMPLETED") {
|
||||
const query = password ? `?accessPassword=${encodeURIComponent(password)}` : "";
|
||||
navigate(`/meetings/${meetingId}/realtime-preview${query}`, {replace: true});
|
||||
return true;
|
||||
}
|
||||
} catch {
|
||||
// 状态查询异常时继续进入原预览页。
|
||||
}
|
||||
}
|
||||
setMeeting(preview.meeting);
|
||||
setTranscripts(preview.transcripts || []);
|
||||
setChapters(preview.chapters || []);
|
||||
setPasswordVerified(true);
|
||||
return false;
|
||||
};
|
||||
|
||||
useEffect(() => {
|
||||
|
|
@ -63,7 +78,10 @@ export default function MeetingPreviewPage() {
|
|||
|
||||
if (presetAccessPassword) {
|
||||
try {
|
||||
await loadPreview(presetAccessPassword);
|
||||
const redirected = await loadPreview(presetAccessPassword);
|
||||
if (redirected) {
|
||||
return;
|
||||
}
|
||||
setAccessPassword(presetAccessPassword);
|
||||
return;
|
||||
} catch {
|
||||
|
|
@ -84,7 +102,10 @@ export default function MeetingPreviewPage() {
|
|||
}
|
||||
setLoading(true);
|
||||
try {
|
||||
await loadPreview(accessPassword.trim());
|
||||
const redirected = await loadPreview(accessPassword.trim());
|
||||
if (redirected) {
|
||||
return;
|
||||
}
|
||||
message.success("访问校验通过");
|
||||
} catch {
|
||||
message.error("访问密码错误");
|
||||
|
|
|
|||
|
|
@ -107,6 +107,22 @@ export interface MeetingPreviewAccessVO {
|
|||
passwordRequired: boolean;
|
||||
}
|
||||
|
||||
export interface RealtimeMeetingSubscriptionSessionVO {
|
||||
sessionToken: string;
|
||||
path: string;
|
||||
expiresInSeconds: number;
|
||||
}
|
||||
|
||||
export interface RealtimeMeetingSessionStatusVO {
|
||||
meetingId: number;
|
||||
status: "IDLE" | "ACTIVE" | "PAUSED_EMPTY" | "PAUSED_RESUMABLE" | "COMPLETING" | "COMPLETED";
|
||||
hasTranscript?: boolean;
|
||||
canResume?: boolean;
|
||||
remainingSeconds?: number;
|
||||
resumeExpireAt?: number;
|
||||
activeConnection?: boolean;
|
||||
}
|
||||
|
||||
export interface UpdateMeetingBasicCommand {
|
||||
meetingId: number;
|
||||
title?: string;
|
||||
|
|
|
|||
|
|
@ -1 +1 @@
|
|||
{"root":["./src/app.tsx","./src/main.tsx","./src/api/auth.ts","./src/api/http.ts","./src/api/meeting.ts","./src/api/platform.ts","./src/api/user.ts","./src/components/bottomnav.tsx","./src/components/loadingscreen.tsx","./src/components/pageheader.tsx","./src/components/platformconfigprovider.tsx","./src/components/preview/meetingpreviewview.tsx","./src/components/preview/meetinganalysis.ts","./src/hooks/usepagetitle.ts","./src/layouts/mainlayout.tsx","./src/pages/forgot-password/index.tsx","./src/pages/login/index.tsx","./src/pages/meeting-detail/index.tsx","./src/pages/meeting-preview/index.tsx","./src/pages/meetings/index.tsx","./src/pages/password/index.tsx","./src/pages/profile/index.tsx","./src/pages/scan-confirm/index.tsx","./src/routes/protectedroute.tsx","./src/types/index.ts","./src/types/platform.ts","./src/utils/auth.ts","./src/utils/meeting.ts","./src/utils/password.ts"],"version":"5.9.3"}
|
||||
{"root":["./src/app.tsx","./src/main.tsx","./src/api/auth.ts","./src/api/http.ts","./src/api/meeting.ts","./src/api/platform.ts","./src/api/user.ts","./src/components/bottomnav.tsx","./src/components/loadingscreen.tsx","./src/components/pageheader.tsx","./src/components/platformconfigprovider.tsx","./src/components/preview/meetingpreviewview.tsx","./src/components/preview/meetinganalysis.ts","./src/hooks/usepagetitle.ts","./src/layouts/mainlayout.tsx","./src/pages/forgot-password/index.tsx","./src/pages/login/index.tsx","./src/pages/meeting-detail/index.tsx","./src/pages/meeting-preview/index.tsx","./src/pages/meetings/index.tsx","./src/pages/password/index.tsx","./src/pages/profile/index.tsx","./src/pages/realtime-meeting-preview/index.tsx","./src/pages/scan-confirm/index.tsx","./src/routes/protectedroute.tsx","./src/types/index.ts","./src/types/platform.ts","./src/utils/auth.ts","./src/utils/meeting.ts","./src/utils/password.ts"],"version":"5.9.3"}
|
||||
Loading…
Reference in New Issue