From 36c1461c814eb0a4efadeb1906cdcab3dbceb446 Mon Sep 17 00:00:00 2001 From: chenhao Date: Mon, 14 Sep 2026 16:15:19 +0800 Subject: [PATCH] =?UTF-8?q?feat(realtime):=20=E5=AE=9E=E7=8E=B0=E5=AE=9E?= =?UTF-8?q?=E6=97=B6=E4=BC=9A=E8=AE=AE=E8=AE=A2=E9=98=85=E4=B8=8EASR?= =?UTF-8?q?=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 后端新增实时会议事件分发、订阅限流及WebSocket处理器，完善Redis缓存机制与实时ASR通道支持。前端同步更新实时ASR会话组件及会议预览页面，完成前后端实时数据链路对接。 --- .../java/com/imeeting/common/RedisKeys.java | 16 + .../RealtimeMeetingWebSocketConfig.java | 4 + .../controller/biz/MeetingController.java | 19 +- .../biz/MeetingPublicPreviewController.java | 39 ++- ...ealtimeMeetingSubscriptionSessionData.java | 8 + .../RealtimeMeetingSubscriptionSessionVO.java | 10 + .../AndroidRealtimeMeetingGrpcService.java | 308 +++++++++++++++--- .../service/android/AndroidAuthService.java | 5 + .../android/impl/AndroidAuthServiceImpl.java | 25 ++ .../impl/AndroidMeetingPushServiceImpl.java | 8 +- .../RealtimeMeetingSessionStateService.java | 2 + ...timeMeetingSubscriptionSessionService.java | 10 + ...ealtimeMeetingSessionStateServiceImpl.java | 31 +- ...altimeMeetingSocketSessionServiceImpl.java | 3 + ...MeetingSubscriptionSessionServiceImpl.java | 77 +++++ .../realtime/RealtimeMeetingEventHub.java | 73 +++++ .../realtime/RealtimeMeetingEventRelay.java | 62 ++++ .../realtime/RealtimeMeetingMetrics.java | 44 +++ .../RealtimeMeetingSubscriptionQuota.java | 55 ++++ ...ealtimeMeetingSubscriptionRateLimiter.java | 30 ++ .../impl/LocalRealtimeAsrChannel.java | 30 +- .../impl/TencentRealtimeAsrChannel.java | 24 +- .../com/imeeting/support/RedisSupport.java | 99 ++++++ .../redis/RealtimeMeetingPublisherLease.java | 72 ++++ ...altimeMeetingSubscriptionSessionCache.java | 39 +++ .../RealtimeMeetingProxyWebSocketHandler.java | 124 ++++++- ...meMeetingSubscriptionWebSocketHandler.java | 133 ++++++++ frontend/src/api/business/meeting.ts | 21 +- frontend/src/pages/business/MeetingDetail.tsx | 26 +- frontend/src/pages/business/Meetings.tsx | 9 - .../src/pages/business/RealtimeAsrSession.tsx | 187 +++++++++-- imeeting-h5/src/App.tsx | 2 + imeeting-h5/src/api/meeting.ts | 27 ++ .../src/pages/meeting-preview/index.tsx | 33 +- imeeting-h5/src/types/index.ts | 16 + imeeting-h5/tsconfig.tsbuildinfo | 2 +- 36 files changed, 1546 insertions(+), 127 deletions(-) create mode 100644 backend/src/main/java/com/imeeting/dto/biz/RealtimeMeetingSubscriptionSessionData.java create mode 100644 backend/src/main/java/com/imeeting/dto/biz/RealtimeMeetingSubscriptionSessionVO.java create mode 100644 backend/src/main/java/com/imeeting/service/biz/RealtimeMeetingSubscriptionSessionService.java create mode 100644 backend/src/main/java/com/imeeting/service/biz/impl/RealtimeMeetingSubscriptionSessionServiceImpl.java create mode 100644 backend/src/main/java/com/imeeting/service/realtime/RealtimeMeetingEventHub.java create mode 100644 backend/src/main/java/com/imeeting/service/realtime/RealtimeMeetingEventRelay.java create mode 100644 backend/src/main/java/com/imeeting/service/realtime/RealtimeMeetingMetrics.java create mode 100644 backend/src/main/java/com/imeeting/service/realtime/RealtimeMeetingSubscriptionQuota.java create mode 100644 backend/src/main/java/com/imeeting/service/realtime/RealtimeMeetingSubscriptionRateLimiter.java create mode 100644 backend/src/main/java/com/imeeting/support/redis/RealtimeMeetingPublisherLease.java create mode 100644 backend/src/main/java/com/imeeting/support/redis/RealtimeMeetingSubscriptionSessionCache.java create mode 100644 backend/src/main/java/com/imeeting/websocket/RealtimeMeetingSubscriptionWebSocketHandler.java diff --git a/backend/src/main/java/com/imeeting/common/RedisKeys.java b/backend/src/main/java/com/imeeting/common/RedisKeys.java index 68b15ce..18c7329 100644 --- a/backend/src/main/java/com/imeeting/common/RedisKeys.java +++ b/backend/src/main/java/com/imeeting/common/RedisKeys.java @@ -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:"; } diff --git a/backend/src/main/java/com/imeeting/config/RealtimeMeetingWebSocketConfig.java b/backend/src/main/java/com/imeeting/config/RealtimeMeetingWebSocketConfig.java index ddada53..68e8521 100644 --- a/backend/src/main/java/com/imeeting/config/RealtimeMeetingWebSocketConfig.java +++ b/backend/src/main/java/com/imeeting/config/RealtimeMeetingWebSocketConfig.java @@ -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("*"); } /** diff --git a/backend/src/main/java/com/imeeting/controller/biz/MeetingController.java b/backend/src/main/java/com/imeeting/controller/biz/MeetingController.java index 288e465..d823737 100644 --- a/backend/src/main/java/com/imeeting/controller/biz/MeetingController.java +++ b/backend/src/main/java/com/imeeting/controller/biz/MeetingController.java @@ -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 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 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); diff --git a/backend/src/main/java/com/imeeting/controller/biz/MeetingPublicPreviewController.java b/backend/src/main/java/com/imeeting/controller/biz/MeetingPublicPreviewController.java index bdc104f..02a6965 100644 --- a/backend/src/main/java/com/imeeting/controller/biz/MeetingPublicPreviewController.java +++ b/backend/src/main/java/com/imeeting/controller/biz/MeetingPublicPreviewController.java @@ -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 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 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()); + } + } } diff --git a/backend/src/main/java/com/imeeting/dto/biz/RealtimeMeetingSubscriptionSessionData.java b/backend/src/main/java/com/imeeting/dto/biz/RealtimeMeetingSubscriptionSessionData.java new file mode 100644 index 0000000..93230e2 --- /dev/null +++ b/backend/src/main/java/com/imeeting/dto/biz/RealtimeMeetingSubscriptionSessionData.java @@ -0,0 +1,8 @@ +package com.imeeting.dto.biz; + +import lombok.Data; + +@Data +public class RealtimeMeetingSubscriptionSessionData { + private Long meetingId; +} diff --git a/backend/src/main/java/com/imeeting/dto/biz/RealtimeMeetingSubscriptionSessionVO.java b/backend/src/main/java/com/imeeting/dto/biz/RealtimeMeetingSubscriptionSessionVO.java new file mode 100644 index 0000000..4f1e28f --- /dev/null +++ b/backend/src/main/java/com/imeeting/dto/biz/RealtimeMeetingSubscriptionSessionVO.java @@ -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; +} diff --git a/backend/src/main/java/com/imeeting/grpc/realtime/AndroidRealtimeMeetingGrpcService.java b/backend/src/main/java/com/imeeting/grpc/realtime/AndroidRealtimeMeetingGrpcService.java index 290739b..ff3eb7f 100644 --- a/backend/src/main/java/com/imeeting/grpc/realtime/AndroidRealtimeMeetingGrpcService.java +++ b/backend/src/main/java/com/imeeting/grpc/realtime/AndroidRealtimeMeetingGrpcService.java @@ -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 sessions = new ConcurrentHashMap<>(); + private final Map 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 stream(StreamObserver 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 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 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 eventConsumer; + private String subscriptionMember; private GrpcSession(Long meetingId, String connectionId, RealtimeAsrChannel channel, - StreamObserver observer) { + StreamObserver 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 diff --git a/backend/src/main/java/com/imeeting/service/android/AndroidAuthService.java b/backend/src/main/java/com/imeeting/service/android/AndroidAuthService.java index 9032a88..821ac51 100644 --- a/backend/src/main/java/com/imeeting/service/android/AndroidAuthService.java +++ b/backend/src/main/java/com/imeeting/service/android/AndroidAuthService.java @@ -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); diff --git a/backend/src/main/java/com/imeeting/service/android/impl/AndroidAuthServiceImpl.java b/backend/src/main/java/com/imeeting/service/android/impl/AndroidAuthServiceImpl.java index f27ee41..70e656c 100644 --- a/backend/src/main/java/com/imeeting/service/android/impl/AndroidAuthServiceImpl.java +++ b/backend/src/main/java/com/imeeting/service/android/impl/AndroidAuthServiceImpl.java @@ -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); diff --git a/backend/src/main/java/com/imeeting/service/android/impl/AndroidMeetingPushServiceImpl.java b/backend/src/main/java/com/imeeting/service/android/impl/AndroidMeetingPushServiceImpl.java index b723924..f950f41 100644 --- a/backend/src/main/java/com/imeeting/service/android/impl/AndroidMeetingPushServiceImpl.java +++ b/backend/src/main/java/com/imeeting/service/android/impl/AndroidMeetingPushServiceImpl.java @@ -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) { diff --git a/backend/src/main/java/com/imeeting/service/biz/RealtimeMeetingSessionStateService.java b/backend/src/main/java/com/imeeting/service/biz/RealtimeMeetingSessionStateService.java index eb60d8d..3f7cb1c 100644 --- a/backend/src/main/java/com/imeeting/service/biz/RealtimeMeetingSessionStateService.java +++ b/backend/src/main/java/com/imeeting/service/biz/RealtimeMeetingSessionStateService.java @@ -19,6 +19,8 @@ public interface RealtimeMeetingSessionStateService { boolean activate(Long meetingId, String connectionId); + boolean isActiveConnection(Long meetingId, String connectionId); + RealtimeMeetingSessionStatusVO getStatus(Long meetingId); Map getStatuses(List meetingIds); diff --git a/backend/src/main/java/com/imeeting/service/biz/RealtimeMeetingSubscriptionSessionService.java b/backend/src/main/java/com/imeeting/service/biz/RealtimeMeetingSubscriptionSessionService.java new file mode 100644 index 0000000..d6d5ab0 --- /dev/null +++ b/backend/src/main/java/com/imeeting/service/biz/RealtimeMeetingSubscriptionSessionService.java @@ -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); +} diff --git a/backend/src/main/java/com/imeeting/service/biz/impl/RealtimeMeetingSessionStateServiceImpl.java b/backend/src/main/java/com/imeeting/service/biz/impl/RealtimeMeetingSessionStateServiceImpl.java index 29e7ae1..103bb22 100644 --- a/backend/src/main/java/com/imeeting/service/biz/impl/RealtimeMeetingSessionStateServiceImpl.java +++ b/backend/src/main/java/com/imeeting/service/biz/impl/RealtimeMeetingSessionStateServiceImpl.java @@ -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 diff --git a/backend/src/main/java/com/imeeting/service/biz/impl/RealtimeMeetingSocketSessionServiceImpl.java b/backend/src/main/java/com/imeeting/service/biz/impl/RealtimeMeetingSocketSessionServiceImpl.java index 3c06a50..85b4843 100644 --- a/backend/src/main/java/com/imeeting/service/biz/impl/RealtimeMeetingSocketSessionServiceImpl.java +++ b/backend/src/main/java/com/imeeting/service/biz/impl/RealtimeMeetingSocketSessionServiceImpl.java @@ -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); diff --git a/backend/src/main/java/com/imeeting/service/biz/impl/RealtimeMeetingSubscriptionSessionServiceImpl.java b/backend/src/main/java/com/imeeting/service/biz/impl/RealtimeMeetingSubscriptionSessionServiceImpl.java new file mode 100644 index 0000000..32bf602 --- /dev/null +++ b/backend/src/main/java/com/imeeting/service/biz/impl/RealtimeMeetingSubscriptionSessionServiceImpl.java @@ -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(); + } +} diff --git a/backend/src/main/java/com/imeeting/service/realtime/RealtimeMeetingEventHub.java b/backend/src/main/java/com/imeeting/service/realtime/RealtimeMeetingEventHub.java new file mode 100644 index 0000000..a397bd2 --- /dev/null +++ b/backend/src/main/java/com/imeeting/service/realtime/RealtimeMeetingEventHub.java @@ -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>> subscribers = new ConcurrentHashMap<>(); + + public void subscribe(Long meetingId, Consumer consumer) { + if (meetingId == null || consumer == null) { + return; + } + subscribers.computeIfAbsent(meetingId, ignored -> new CopyOnWriteArrayList<>()).add(consumer); + } + + public boolean trySubscribe(Long meetingId, Consumer consumer, int maxSubscribers) { + if (meetingId == null || consumer == null || maxSubscribers < 1) { + return false; + } + CopyOnWriteArrayList> 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 consumer) { + if (meetingId == null || consumer == null) { + return; + } + List> 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> meetingSubscribers = subscribers.get(meetingId); + if (meetingSubscribers == null) { + return; + } + for (Consumer subscriber : meetingSubscribers) { + try { + subscriber.accept(payload); + } catch (RuntimeException ignored) { + RealtimeMeetingMetrics.subscriberDeliveryFailed(); + // 单个失效订阅者不能阻塞同一会议的其他订阅者。 + } + } + } +} diff --git a/backend/src/main/java/com/imeeting/service/realtime/RealtimeMeetingEventRelay.java b/backend/src/main/java/com/imeeting/service/realtime/RealtimeMeetingEventRelay.java new file mode 100644 index 0000000..1639b3c --- /dev/null +++ b/backend/src/main/java/com/imeeting/service/realtime/RealtimeMeetingEventRelay.java @@ -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 监听线程。 + } + } +} diff --git a/backend/src/main/java/com/imeeting/service/realtime/RealtimeMeetingMetrics.java b/backend/src/main/java/com/imeeting/service/realtime/RealtimeMeetingMetrics.java new file mode 100644 index 0000000..690297f --- /dev/null +++ b/backend/src/main/java/com/imeeting/service/realtime/RealtimeMeetingMetrics.java @@ -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(); + } +} diff --git a/backend/src/main/java/com/imeeting/service/realtime/RealtimeMeetingSubscriptionQuota.java b/backend/src/main/java/com/imeeting/service/realtime/RealtimeMeetingSubscriptionQuota.java new file mode 100644 index 0000000..584f573 --- /dev/null +++ b/backend/src/main/java/com/imeeting/service/realtime/RealtimeMeetingSubscriptionQuota.java @@ -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); + } +} diff --git a/backend/src/main/java/com/imeeting/service/realtime/RealtimeMeetingSubscriptionRateLimiter.java b/backend/src/main/java/com/imeeting/service/realtime/RealtimeMeetingSubscriptionRateLimiter.java new file mode 100644 index 0000000..c5ea13a --- /dev/null +++ b/backend/src/main/java/com/imeeting/service/realtime/RealtimeMeetingSubscriptionRateLimiter.java @@ -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("实时订阅请求过于频繁，请稍后重试"); + } + } +} diff --git a/backend/src/main/java/com/imeeting/service/realtime/impl/LocalRealtimeAsrChannel.java b/backend/src/main/java/com/imeeting/service/realtime/impl/LocalRealtimeAsrChannel.java index 6779b80..e8636e0 100644 --- a/backend/src/main/java/com/imeeting/service/realtime/impl/LocalRealtimeAsrChannel.java +++ b/backend/src/main/java/com/imeeting/service/realtime/impl/LocalRealtimeAsrChannel.java @@ -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()); diff --git a/backend/src/main/java/com/imeeting/service/realtime/impl/TencentRealtimeAsrChannel.java b/backend/src/main/java/com/imeeting/service/realtime/impl/TencentRealtimeAsrChannel.java index 69d1ffb..368f42d 100644 --- a/backend/src/main/java/com/imeeting/service/realtime/impl/TencentRealtimeAsrChannel.java +++ b/backend/src/main/java/com/imeeting/service/realtime/impl/TencentRealtimeAsrChannel.java @@ -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 pendingFrames = (List) 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; diff --git a/backend/src/main/java/com/imeeting/support/RedisSupport.java b/backend/src/main/java/com/imeeting/support/RedisSupport.java index 7393bea..5b52005 100644 --- a/backend/src/main/java/com/imeeting/support/RedisSupport.java +++ b/backend/src/main/java/com/imeeting/support/RedisSupport.java @@ -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; diff --git a/backend/src/main/java/com/imeeting/support/redis/RealtimeMeetingPublisherLease.java b/backend/src/main/java/com/imeeting/support/redis/RealtimeMeetingPublisherLease.java new file mode 100644 index 0000000..bcbfeb7 --- /dev/null +++ b/backend/src/main/java/com/imeeting/support/redis/RealtimeMeetingPublisherLease.java @@ -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); + } +} diff --git a/backend/src/main/java/com/imeeting/support/redis/RealtimeMeetingSubscriptionSessionCache.java b/backend/src/main/java/com/imeeting/support/redis/RealtimeMeetingSubscriptionSessionCache.java new file mode 100644 index 0000000..80740e1 --- /dev/null +++ b/backend/src/main/java/com/imeeting/support/redis/RealtimeMeetingSubscriptionSessionCache.java @@ -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(); + } +} diff --git a/backend/src/main/java/com/imeeting/websocket/RealtimeMeetingProxyWebSocketHandler.java b/backend/src/main/java/com/imeeting/websocket/RealtimeMeetingProxyWebSocketHandler.java index c5aba29..c1e83a3 100644 --- a/backend/src/main/java/com/imeeting/websocket/RealtimeMeetingProxyWebSocketHandler.java +++ b/backend/src/main/java/com/imeeting/websocket/RealtimeMeetingProxyWebSocketHandler.java @@ -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 meetingSessions = new ConcurrentHashMap<>(); private final ConcurrentMap 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 diff --git a/backend/src/main/java/com/imeeting/websocket/RealtimeMeetingSubscriptionWebSocketHandler.java b/backend/src/main/java/com/imeeting/websocket/RealtimeMeetingSubscriptionWebSocketHandler.java new file mode 100644 index 0000000..3a89005 --- /dev/null +++ b/backend/src/main/java/com/imeeting/websocket/RealtimeMeetingSubscriptionWebSocketHandler.java @@ -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 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 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 consumer, String member) { + } +} diff --git a/frontend/src/api/business/meeting.ts b/frontend/src/api/business/meeting.ts index a95b6f9..a264449 100644 --- a/frontend/src/api/business/meeting.ts +++ b/frontend/src/api/business/meeting.ts @@ -230,6 +230,12 @@ export interface RealtimeSocketSessionVO { startMessage: Record; } +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}` diff --git a/frontend/src/pages/business/MeetingDetail.tsx b/frontend/src/pages/business/MeetingDetail.tsx index fd3302d..52b91af 100644 --- a/frontend/src/pages/business/MeetingDetail.tsx +++ b/frontend/src/pages/business/MeetingDetail.tsx @@ -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 = () => { )} >
- {meeting.status === 0 || meeting.status === 1 ? ( + {(meeting.status === 0 || meeting.status === 1) ? ( <>
{ 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)) { diff --git a/frontend/src/pages/business/RealtimeAsrSession.tsx b/frontend/src/pages/business/RealtimeAsrSession.tsx index 7e44f1d..87818b9 100644 --- a/frontend/src/pages/business/RealtimeAsrSession.tsx +++ b/frontend/src/pages/business/RealtimeAsrSession.tsx @@ -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 = { - 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) { 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(null); const transcriptRef = useRef(null); const wsRef = useRef(null); + const subscriptionWsRef = useRef(null); const wsHeartbeatRef = useRef | null>(null); const audioContextRef = useRef(null); const processorRef = useRef(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 | 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={ {statusText}} /> - {sessionDraft ? 字数 {totalTranscriptChars} : null} + 字数 {totalTranscriptChars} } contentClassName="realtime-session-section-content" >
- {!sessionDraft ? ( + {canControlCurrentMeeting && !sessionDraft ? (
) : ( @@ -899,7 +1018,7 @@ export function RealtimeAsrSession() { )}
-
+ {canControlCurrentMeeting &&
)} -
+
}
)}
diff --git a/imeeting-h5/src/App.tsx b/imeeting-h5/src/App.tsx index a89580d..5a870ed 100644 --- a/imeeting-h5/src/App.tsx +++ b/imeeting-h5/src/App.tsx @@ -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() { } /> }/> } /> + }/> }); }; +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`); }; diff --git a/imeeting-h5/src/pages/meeting-preview/index.tsx b/imeeting-h5/src/pages/meeting-preview/index.tsx index d74e309..f8c264b 100644 --- a/imeeting-h5/src/pages/meeting-preview/index.tsx +++ b/imeeting-h5/src/pages/meeting-preview/index.tsx @@ -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("访问密码错误"); diff --git a/imeeting-h5/src/types/index.ts b/imeeting-h5/src/types/index.ts index 83e0dd7..4b1528f 100644 --- a/imeeting-h5/src/types/index.ts +++ b/imeeting-h5/src/types/index.ts @@ -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; diff --git a/imeeting-h5/tsconfig.tsbuildinfo b/imeeting-h5/tsconfig.tsbuildinfo index cbb1692..85e7dd3 100644 --- a/imeeting-h5/tsconfig.tsbuildinfo +++ b/imeeting-h5/tsconfig.tsbuildinfo @@ -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"} \ No newline at end of file +{"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"} \ No newline at end of file