diff --git a/backend/src/main/java/com/imeeting/common/RedisKeys.java b/backend/src/main/java/com/imeeting/common/RedisKeys.java index 18c7329..792234d 100644 --- a/backend/src/main/java/com/imeeting/common/RedisKeys.java +++ b/backend/src/main/java/com/imeeting/common/RedisKeys.java @@ -107,6 +107,14 @@ public final class RedisKeys { return "biz:meeting:realtime:publisher:" + meetingId; } + public static String realtimeMeetingLiveMeetingsKey() { + return "biz:meeting:realtime:live-meetings"; + } + + public static String realtimeMeetingPublisherDeviceKey(Long meetingId) { + return "biz:meeting:realtime:publisher-device:" + meetingId; + } + public static String realtimeMeetingResumeTimeoutPrefix() { return "biz:meeting:realtime:resume-timeout:"; } diff --git a/backend/src/main/java/com/imeeting/grpc/push/AndroidPushGrpcService.java b/backend/src/main/java/com/imeeting/grpc/push/AndroidPushGrpcService.java index 9451917..8e86d4a 100644 --- a/backend/src/main/java/com/imeeting/grpc/push/AndroidPushGrpcService.java +++ b/backend/src/main/java/com/imeeting/grpc/push/AndroidPushGrpcService.java @@ -7,6 +7,7 @@ import com.imeeting.service.android.AndroidDeviceSessionService; import com.imeeting.service.android.AndroidGatewayPushService; import com.imeeting.service.android.AndroidPushMessageService; import com.imeeting.service.biz.DeviceOnlineManagementService; +import com.imeeting.service.realtime.RealtimeMeetingPushSubscriptionManager; import com.unisbase.common.exception.BusinessException; import io.grpc.BindableService; import io.grpc.stub.StreamObserver; @@ -29,6 +30,7 @@ public class AndroidPushGrpcService extends PushServiceGrpc.PushServiceImplBase private final AndroidGatewayPushService androidGatewayPushService; private final AndroidPushMessageService androidPushMessageService; private final DeviceOnlineManagementService deviceOnlineManagementService; + private final RealtimeMeetingPushSubscriptionManager realtimeMeetingPushSubscriptionManager; @Override public StreamObserver communicate(StreamObserver responseObserver) { @@ -151,6 +153,20 @@ public class AndroidPushGrpcService extends PushServiceGrpc.PushServiceImplBase .setMessage(connectionId) .build()) .build()); + attachRealtimePushSubscriptions(authContext); + } + + /** + * 推送长连接即实时会议订阅入口，只挂接该用户作为创建人或参会人的进行中实时会议。 + */ + private void attachRealtimePushSubscriptions(AndroidAuthContext authContext) { + try { + realtimeMeetingPushSubscriptionManager.onPushConnectionOpened( + authContext.getTenantId(), authContext.getUserId(), deviceId, connectionId); + } catch (Exception ex) { + log.warn("Failed to attach realtime push subscriptions, connectionId={}, deviceId={}", + connectionId, deviceId, ex); + } } private void handleHeartbeat(HeartbeatRequest request) { @@ -208,6 +224,7 @@ public class AndroidPushGrpcService extends PushServiceGrpc.PushServiceImplBase return; } AndroidDeviceSessionState state = androidDeviceSessionService.getByConnectionId(connectionId); + releaseRealtimePushSubscriptions(connectionId); androidGatewayPushService.unregister(connectionId); androidDeviceSessionService.closeSession(connectionId); deviceOnlineManagementService.recordDisconnected(deviceId, state == null ? null : state.getLastSeenAt()); @@ -217,6 +234,14 @@ public class AndroidPushGrpcService extends PushServiceGrpc.PushServiceImplBase platform = null; connected = false; } + + private void releaseRealtimePushSubscriptions(String targetConnectionId) { + try { + realtimeMeetingPushSubscriptionManager.onPushConnectionClosed(targetConnectionId); + } catch (Exception ex) { + log.warn("Failed to release realtime push subscriptions, connectionId={}", targetConnectionId, ex); + } + } }; } 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 bae0191..2146d0f 100644 --- a/backend/src/main/java/com/imeeting/grpc/realtime/AndroidRealtimeMeetingGrpcService.java +++ b/backend/src/main/java/com/imeeting/grpc/realtime/AndroidRealtimeMeetingGrpcService.java @@ -53,7 +53,6 @@ public class AndroidRealtimeMeetingGrpcService private final RealtimeMeetingEventRelay eventRelay; private final RealtimeMeetingPushSubscriptionManager pushSubscriptionManager; private final Map sessions = new ConcurrentHashMap<>(); - private final Map subscriberSessions = new ConcurrentHashMap<>(); @Override public StreamObserver stream(StreamObserver responseObserver) { @@ -178,8 +177,8 @@ public class AndroidRealtimeMeetingGrpcService 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; + pushSubscriptionManager.markPublisherDevice(session.meetingId, session.deviceId); session.cleanupAction = this::cleanup; try { RealtimeMeetingMetrics.publisherOpened("grpc"); @@ -253,9 +252,7 @@ public class AndroidRealtimeMeetingGrpcService if (session.publisher) { publisherLease.release(session.meetingId, session.publisherFenceToken); RealtimeMeetingMetrics.publisherClosed("grpc"); - } - if (!session.publisher) { - detachSubscriber(session); + pushSubscriptionManager.clearPublisherDevice(session.meetingId, session.deviceId); } if (session.publisher) { sessionStateService.pauseByDisconnect(session.meetingId, session.publisherFenceToken); @@ -265,14 +262,6 @@ public class AndroidRealtimeMeetingGrpcService } } - private void detachSubscriber(GrpcSession target) { - if (target.pushSubscribed) { - target.pushSubscribed = false; - pushSubscriptionManager.unsubscribe(target.connectionId); - } - 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); @@ -295,122 +284,29 @@ public class AndroidRealtimeMeetingGrpcService 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 (authContext.getUserId() == null || !authContext.getUserId().equals(meeting.getCreatorId())) { + throw new SecurityException("实时会议 gRPC 只接受发布端连接，订阅方请使用推送长连接"); } + meetingAuthorizationService.assertCanControlRealtimeMeeting( + meeting, authContext, MeetingTerminalEnum.CUSTOM_TERMINAL.getCode()); if (!MeetingConstants.TYPE_REALTIME.equalsIgnoreCase(meeting.getMeetingType())) { throw new IllegalStateException("当前会议不是实时会议"); } 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; - String subscriberDeviceId = connect.getDeviceId().trim(); - String rejectReason = pushSubscriptionManager.subscribe( - meetingId, connectionId, subscriberDeviceId, meeting.getTitle()); - if (rejectReason != null) { - throw new IllegalStateException(rejectReason); - } - next.pushSubscribed = true; - try { - observer.onNext(ServerMessage.newBuilder() - .setConnectAck(ConnectResponse.newBuilder() - .setSuccess(true) - .setMessage("实时转写订阅成功，转写经推送通道下发") - .build()) - .build()); - pushSubscriptionManager.replayHistory(meetingId, subscriberDeviceId, meeting.getTitle()); - subscriberSessions.put(connectionId, next); - return next; - } catch (RuntimeException ex) { - detachSubscriber(next); - throw ex; - } - } - - RealtimeMeetingSessionStatusVO status = sessionStateService.getStatus(meetingId); - RealtimeMeetingResumeConfig config = status == null ? null : status.getResumeConfig(); - if (config == null || config.getAsrModelId() == null) { - throw new IllegalStateException("实时会议缺少 ASR 运行参数"); - } - AiModelVO model = aiModelService.getModelById(config.getAsrModelId(), "ASR"); - if (model == null) { - throw new IllegalStateException("实时 ASR 模型不存在"); - } - RealtimeAsrChannel channel = channelFactory.getRequired(model.getProvider()); - 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 (!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) { - 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( - model, config.getMode(), config.getLanguage(), config.getUseSpkId(), - config.getEnablePunctuation(), config.getEnableItn(), config.getEnableTextRefine(), - config.getSaveAudio(), config.getHotwords())); - channel.handleFrontendText(next.context, start); - observer.onNext(ServerMessage.newBuilder() - .setConnectAck(ConnectResponse.newBuilder().setSuccess(true).setMessage("实时会议连接成功")) - .build()); - return next; - } catch (Exception ex) { - sessions.remove(meetingId, next); - publisherLease.release(meetingId, publisherFenceToken); - try { - channel.closeMeeting(next.context); - } catch (Exception closeEx) { - log.debug("Failed to close realtime upstream after handshake failure", closeEx); - } - next.context.closeTransport(); - audioStorageService.closeSession(connectionId); - sessionStateService.pauseByDisconnect(meetingId, publisherFenceToken); - throw ex; - } + GrpcSession next = new GrpcSession(meetingId, connectionId, null, observer, null); + next.deviceId = connect.getDeviceId().trim(); + next.publisher = false; + next.canPublish = true; + session = next; + next.cleanupAction = this::cleanup; + observer.onNext(ServerMessage.newBuilder() + .setConnectAck(ConnectResponse.newBuilder().setSuccess(true).setMessage("实时会议连接成功")) + .build()); + return next; } }; } @@ -449,8 +345,8 @@ public class AndroidRealtimeMeetingGrpcService private volatile boolean cleaned; private volatile boolean publisher; private boolean canPublish; + private String deviceId; private Runnable cleanupAction; - private volatile boolean pushSubscribed; private GrpcSession(Long meetingId, String connectionId, RealtimeAsrChannel channel, StreamObserver observer, String publisherFenceToken) { diff --git a/backend/src/main/java/com/imeeting/service/android/AndroidGatewayPushService.java b/backend/src/main/java/com/imeeting/service/android/AndroidGatewayPushService.java index 464d2bf..1ab906b 100644 --- a/backend/src/main/java/com/imeeting/service/android/AndroidGatewayPushService.java +++ b/backend/src/main/java/com/imeeting/service/android/AndroidGatewayPushService.java @@ -1,10 +1,13 @@ package com.imeeting.service.android; import com.imeeting.dto.android.AndroidGrpcConnectionSnapshotVO; +import com.imeeting.dto.android.AndroidGrpcConnectionDetailVO; import com.imeeting.grpc.push.PushMessage; import com.imeeting.grpc.push.ServerMessage; import io.grpc.stub.StreamObserver; +import java.util.List; + public interface AndroidGatewayPushService { String register(String connectionId, String deviceId, @@ -20,6 +23,11 @@ public interface AndroidGatewayPushService { int pushToUser(Long tenantId, Long userId, PushMessage message); + /** + * 查询某用户在本实例上的全部推送长连接，只按 user_id 命中，不依赖设备授权租户；同一用户的多设备连接全部返回。 + */ + List listUserConnections(Long userId); + String disconnectDevice(String deviceId); AndroidGrpcConnectionSnapshotVO snapshotConnections(); 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 70e656c..bfa9cd8 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 @@ -50,22 +50,31 @@ public class AndroidAuthServiceImpl implements AndroidAuthService { assertDeviceEnabled(device); Long requestedUserId = parseOptionalLong(userId, "Android gRPC userId"); Long requestedTenantId = parseOptionalLong(tenantId, "Android gRPC tenantId"); - boolean anonymous = requestedUserId == null; - AndroidAuthContext context = buildContext(anonymous ? "NONE" : "GRPC_USER", anonymous, deviceId, null, appVersion, platform, null, null, null, null); - Long resolvedTenantId = requestedTenantId != null ? requestedTenantId : license.getTenantId(); - if (requestedUserId != null) { - if (requestedTenantId != null && !requestedTenantId.equals(license.getTenantId())) { - throw new RuntimeException("登录租户与授权租户不一致,无法绑定grpc"); - } - context.setUserId(requestedUserId); - context.setTenantId(resolvedTenantId); - return context; + if (requestedUserId != null && requestedTenantId != null && !requestedTenantId.equals(license.getTenantId())) { + throw new RuntimeException("登录租户与授权租户不一致,无法绑定grpc"); } - context.setUserId(null); - context.setTenantId(resolvedTenantId); + // 客户端未带 user_id 时回落到设备绑定用户，推送长连接必须具备用户身份，按用户下发的推送才能命中。 + Long resolvedUserId = requestedUserId != null ? requestedUserId : device.getUserId(); + boolean anonymous = resolvedUserId == null; + AndroidAuthContext context = buildContext(anonymous ? "NONE" : "GRPC_USER", anonymous, deviceId, null, appVersion, platform, null, null, null, null); + context.setUserId(resolvedUserId); + context.setTenantId(resolveGrpcTenantId(requestedTenantId, resolvedUserId, device, license)); return context; } + /** + * 推送长连接的租户优先级：客户端显式租户、绑定用户所属设备租户、设备授权租户。 + */ + private Long resolveGrpcTenantId(Long requestedTenantId, Long resolvedUserId, DeviceInfoEntity device, LicenseEntity license) { + if (requestedTenantId != null) { + return requestedTenantId; + } + if (resolvedUserId != null && resolvedUserId.equals(device.getUserId()) && device.getTenantId() != null) { + return device.getTenantId(); + } + return license.getTenantId(); + } + @Override public AndroidAuthContext authenticateRealtimeGrpc(String deviceId, String appVersion, String platform, String userId, String tenantId, String accessToken) { diff --git a/backend/src/main/java/com/imeeting/service/android/impl/AndroidGatewayPushServiceImpl.java b/backend/src/main/java/com/imeeting/service/android/impl/AndroidGatewayPushServiceImpl.java index b3f45f8..6584bb8 100644 --- a/backend/src/main/java/com/imeeting/service/android/impl/AndroidGatewayPushServiceImpl.java +++ b/backend/src/main/java/com/imeeting/service/android/impl/AndroidGatewayPushServiceImpl.java @@ -21,6 +21,7 @@ public class AndroidGatewayPushServiceImpl implements AndroidGatewayPushService private final Map byConnectionId = new ConcurrentHashMap<>(); private final Map connectionByDeviceId = new ConcurrentHashMap<>(); private final Map> connectionsByUserKey = new ConcurrentHashMap<>(); + private final Map> connectionsByUserId = new ConcurrentHashMap<>(); @Override public String register(String connectionId, @@ -103,6 +104,20 @@ public class AndroidGatewayPushServiceImpl implements AndroidGatewayPushService return successCount; } + @Override + public List listUserConnections(Long userId) { + if (userId == null) { + return List.of(); + } + Map bindings = connectionsByUserId.get(userId); + if (bindings == null || bindings.isEmpty()) { + return List.of(); + } + return bindings.entrySet().stream() + .map(entry -> toDetail(entry.getKey(), entry.getValue())) + .toList(); + } + @Override public String disconnectDevice(String deviceId) { String connectionId = connectionByDeviceId.get(deviceId); @@ -142,6 +157,11 @@ public class AndroidGatewayPushServiceImpl implements AndroidGatewayPushService } private void addUserIndex(String connectionId, Binding binding) { + if (binding.userId() != null) { + connectionsByUserId + .computeIfAbsent(binding.userId(), ignored -> new ConcurrentHashMap<>()) + .put(connectionId, binding); + } String userKey = buildUserKey(binding.tenantId(), binding.userId()); if (userKey == null) { return; @@ -156,6 +176,12 @@ public class AndroidGatewayPushServiceImpl implements AndroidGatewayPushService } private void removeUserIndex(String connectionId, Binding binding) { + if (binding.userId() != null) { + connectionsByUserId.computeIfPresent(binding.userId(), (ignored, bindings) -> { + bindings.remove(connectionId); + return bindings.isEmpty() ? null : bindings; + }); + } String userKey = buildUserKey(binding.tenantId(), binding.userId()); if (userKey == null) { return; diff --git a/backend/src/main/java/com/imeeting/service/biz/impl/DeviceOnlineManagementServiceImpl.java b/backend/src/main/java/com/imeeting/service/biz/impl/DeviceOnlineManagementServiceImpl.java index 195331c..cbf35be 100644 --- a/backend/src/main/java/com/imeeting/service/biz/impl/DeviceOnlineManagementServiceImpl.java +++ b/backend/src/main/java/com/imeeting/service/biz/impl/DeviceOnlineManagementServiceImpl.java @@ -69,8 +69,10 @@ public class DeviceOnlineManagementServiceImpl implements DeviceOnlineManagement // existing.setTerminalType(normalizeTerminalType(authContext.getPlatform())); existing.setTerminalVersion(normalize(authContext.getAppVersion())); existing.setLastOnlineAt(now); - existing.setUserId(authContext.getUserId()); - existing.setTenantId(authContext.getTenantId()); + // 匿名连接只刷新在线信息，保留既有绑定用户，否则按用户下发的实时推送会全部失效。 + if (authContext.getTenantId() != null) { + existing.setTenantId(authContext.getTenantId()); + } deviceInfoMapper.updateConnectionInfoByIdIgnoreTenant(existing); } diff --git a/backend/src/main/java/com/imeeting/service/biz/impl/MeetingCommandServiceImpl.java b/backend/src/main/java/com/imeeting/service/biz/impl/MeetingCommandServiceImpl.java index 889f32c..79e3211 100644 --- a/backend/src/main/java/com/imeeting/service/biz/impl/MeetingCommandServiceImpl.java +++ b/backend/src/main/java/com/imeeting/service/biz/impl/MeetingCommandServiceImpl.java @@ -472,8 +472,8 @@ public class MeetingCommandServiceImpl implements MeetingCommandService { long transcriptCount = transcriptMapper.selectCount(new LambdaQueryWrapper() .eq(MeetingTranscript::getMeetingId, meetingId)); if (transcriptCount <= 0) { - realtimeMeetingSessionStateService.pause(meetingId); - throw new RuntimeException("当前还没有转录内容，无法结束会议。请先开始识别，或直接离开页面稍后继续。"); + completeRealtimeMeetingWithoutTranscript(meeting, currentStatus, audioUrl); + return; } if ((audioUrl == null || audioUrl.isBlank()) && (meeting.getAudioUrl() == null || meeting.getAudioUrl().isBlank())) { @@ -497,6 +497,42 @@ public class MeetingCommandServiceImpl implements MeetingCommandService { } } + /** + * 没有最终转写时的完成处理：有录音就转离线识别，既无录音也无转写时按已完成结束，避免会议一直停留在录音中。 + */ + private void completeRealtimeMeetingWithoutTranscript(Meeting meeting, + RealtimeMeetingSessionStatusVO currentStatus, + String audioUrl) { + Long meetingId = meeting.getId(); + boolean audioProvided = (audioUrl != null && !audioUrl.isBlank()) + || (meeting.getAudioUrl() != null && !meeting.getAudioUrl().isBlank()); + if (!audioProvided) { + applyRealtimeAudioFinalizeResult(meeting, realtimeMeetingAudioStorageService.finalizeMeetingAudio(meetingId)); + if (meeting.getAudioUrl() == null || meeting.getAudioUrl().isBlank()) { + // 本来就没有录到音频，不属于音频保存失败，避免终端展示误导性的失败提示 + meeting.setAudioSaveStatus(RealtimeMeetingAudioStorageService.STATUS_NONE); + meeting.setAudioSaveMessage(null); + } + } + + RealtimeMeetingResumeConfig resumeConfig = currentStatus == null ? null : currentStatus.getResumeConfig(); + if (meeting.getAudioUrl() != null && !meeting.getAudioUrl().isBlank() + && resumeConfig != null && resumeConfig.getAsrModelId() != null) { + meetingService.updateById(meeting); + meetingDomainSupport.prewarmPlaybackAudioAfterCommit(meeting.getAudioUrl()); + prepareOfflineReprocessTasks(meetingId, currentStatus); + realtimeMeetingSessionStateService.clear(meetingId); + updateMeetingProgress(meetingId, 0, "正在转入离线音频识别流程...", 0); + aiTaskService.triggerQueuedAsrScheduling(); + return; + } + + realtimeMeetingSessionStateService.clear(meetingId); + meeting.setStatus(MeetingStatusEnum.COMPLETED.getCode()); + meetingService.updateById(meeting); + updateMeetingProgress(meetingId, 100, "未识别到有效录音数据，会议已结束", 0); + } + @Override @Transactional(rollbackFor = Exception.class) public void finishOfflineMeeting(Long meetingId, String finishStage) { 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 103bb22..b19e76b 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 @@ -8,6 +8,8 @@ import com.imeeting.dto.biz.RealtimeMeetingTranscriptCacheState; import com.imeeting.entity.biz.Meeting; import com.imeeting.entity.biz.MeetingTranscript; import com.imeeting.enums.MeetingStatusEnum; +import com.imeeting.event.RealtimeMeetingSessionActivatedEvent; +import com.imeeting.event.RealtimeMeetingSessionClosedEvent; import com.imeeting.mapper.biz.MeetingMapper; import com.imeeting.mapper.biz.MeetingTranscriptMapper; import com.imeeting.service.biz.RealtimeMeetingSessionStateService; @@ -18,6 +20,7 @@ import com.imeeting.support.redis.RealtimeMeetingTranscriptCache; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.stereotype.Service; import java.time.Duration; @@ -36,6 +39,7 @@ public class RealtimeMeetingSessionStateServiceImpl implements RealtimeMeetingSe private final MeetingMapper meetingMapper; private final RealtimeMeetingTranscriptCache realtimeMeetingTranscriptCache; private final RealtimeMeetingPublisherLease publisherLease; + private final ApplicationEventPublisher eventPublisher; @Value("${imeeting.realtime.resume-window-minutes:30}") private String resumeWindowMinutesValue; @@ -163,6 +167,8 @@ public class RealtimeMeetingSessionStateServiceImpl implements RealtimeMeetingSe sessionCache.clearResumeTimeout(meetingId); sessionCache.clearEmptyTimeout(meetingId); + // 发布激活事件：推送订阅挂载由监听器完成，避免会话状态服务反向依赖推送层形成 Bean 循环依赖 + eventPublisher.publishEvent(new RealtimeMeetingSessionActivatedEvent(meetingId)); return true; } @@ -322,6 +328,8 @@ public class RealtimeMeetingSessionStateServiceImpl implements RealtimeMeetingSe @Override public void clear(Long meetingId) { sessionCache.clearAll(meetingId); + // 发布会话结束事件：推送订阅回收由监听器完成，避免会话状态服务反向依赖推送层形成 Bean 循环依赖 + eventPublisher.publishEvent(new RealtimeMeetingSessionClosedEvent(meetingId)); } private RealtimeMeetingSessionStatusVO pauseState(Long meetingId, RealtimeMeetingSessionState state) { diff --git a/backend/src/main/java/com/imeeting/support/RedisSupport.java b/backend/src/main/java/com/imeeting/support/RedisSupport.java index 5b52005..23150dc 100644 --- a/backend/src/main/java/com/imeeting/support/RedisSupport.java +++ b/backend/src/main/java/com/imeeting/support/RedisSupport.java @@ -11,6 +11,7 @@ import org.springframework.stereotype.Component; import java.time.Duration; import java.util.Collection; +import java.util.List; @Component @Slf4j @@ -202,6 +203,47 @@ public class RedisSupport { } } + /** + * 写入短期成员，分数为 now+ttl，顺带清理已过期成员，用于跨实例可枚举的短期索引。 + */ + public boolean touchExpiringMember(String key, String member, Duration ttl) { + if (key == null || key.isBlank() || member == null || member.isBlank() || ttl == null) { + return false; + } + try { + Long written = commands().eval( + "local now = tonumber(ARGV[1]); local ttl = tonumber(ARGV[2]); " + + "redis.call('zremrangebyscore', KEYS[1], '-inf', now); " + + "redis.call('zadd', KEYS[1], now + ttl, ARGV[3]); redis.call('pexpire', KEYS[1], ttl); return 1", + ScriptOutputType.INTEGER, + new String[]{key}, + String.valueOf(System.currentTimeMillis()), + String.valueOf(ttl.toMillis()), + member + ); + return written != null && written > 0; + } catch (Exception ex) { + log.warn("Touch Redis set member failed, key={}", key, ex); + return false; + } + } + + /** + * 读取未过期的全部成员，读取前先剪掉已过期成员。 + */ + public List listUnexpiredMembers(String key) { + if (key == null || key.isBlank()) { + return List.of(); + } + try { + commands().zremrangebyscore(key, Double.NEGATIVE_INFINITY, (double) System.currentTimeMillis()); + List members = commands().zrange(key, 0, -1); + return members == null ? List.of() : members; + } catch (Exception ex) { + log.warn("List Redis set members failed, key={}", key, ex); + return List.of(); + } + } 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/RealtimeMeetingSessionCache.java b/backend/src/main/java/com/imeeting/support/redis/RealtimeMeetingSessionCache.java index 0634e04..afe21ec 100644 --- a/backend/src/main/java/com/imeeting/support/redis/RealtimeMeetingSessionCache.java +++ b/backend/src/main/java/com/imeeting/support/redis/RealtimeMeetingSessionCache.java @@ -7,6 +7,7 @@ import lombok.RequiredArgsConstructor; import org.springframework.stereotype.Component; import java.time.Duration; +import java.util.List; @Component @RequiredArgsConstructor @@ -52,5 +53,67 @@ public class RealtimeMeetingSessionCache { redisSupport.deleteQuietly(RedisKeys.realtimeMeetingResumeTimeoutKey(meetingId)); redisSupport.deleteQuietly(RedisKeys.realtimeMeetingEmptyTimeoutKey(meetingId)); redisSupport.deleteQuietly(RedisKeys.realtimeMeetingEventSeqKey(meetingId)); + redisSupport.deleteQuietly(RedisKeys.realtimeMeetingPublisherDeviceKey(meetingId)); + redisSupport.removeExpiringSlot(RedisKeys.realtimeMeetingLiveMeetingsKey(), String.valueOf(meetingId)); + } + + /** + * 登记进行中的实时会议，供其它实例发现并按参会人挂接推送订阅。 + */ + public void markLive(Long meetingId, Duration ttl) { + if (meetingId == null) { + return; + } + redisSupport.touchExpiringMember(RedisKeys.realtimeMeetingLiveMeetingsKey(), String.valueOf(meetingId), ttl); + } + + public void clearLive(Long meetingId) { + if (meetingId == null) { + return; + } + redisSupport.removeExpiringSlot(RedisKeys.realtimeMeetingLiveMeetingsKey(), String.valueOf(meetingId)); + } + + public List listLiveMeetingIds() { + return redisSupport.listUnexpiredMembers(RedisKeys.realtimeMeetingLiveMeetingsKey()).stream() + .map(RealtimeMeetingSessionCache::parseMeetingId) + .filter(RealtimeMeetingSessionCache::nonNullId) + .toList(); + } + + /** + * 记录正在发布音频的设备，跨实例抑制发布端重复收流。 + */ + public void savePublisherDevice(Long meetingId, String deviceId, Duration ttl) { + if (meetingId == null || deviceId == null || deviceId.isBlank()) { + return; + } + redisSupport.setString(RedisKeys.realtimeMeetingPublisherDeviceKey(meetingId), deviceId.trim(), ttl); + } + + public void clearPublisherDevice(Long meetingId, String deviceId) { + if (meetingId == null || deviceId == null || deviceId.isBlank()) { + return; + } + redisSupport.deleteIfValue(RedisKeys.realtimeMeetingPublisherDeviceKey(meetingId), deviceId.trim()); + } + + public String getPublisherDevice(Long meetingId) { + if (meetingId == null) { + return null; + } + return redisSupport.getStringQuietly(RedisKeys.realtimeMeetingPublisherDeviceKey(meetingId)); + } + + private static Long parseMeetingId(String value) { + try { + return Long.valueOf(value); + } catch (NumberFormatException ex) { + return null; + } + } + + private static boolean nonNullId(Long meetingId) { + return meetingId != null; } } diff --git a/frontend/src/pages/business/RealtimeAsrSession.tsx b/frontend/src/pages/business/RealtimeAsrSession.tsx index 87818b9..ef9361d 100644 --- a/frontend/src/pages/business/RealtimeAsrSession.tsx +++ b/frontend/src/pages/business/RealtimeAsrSession.tsx @@ -31,6 +31,7 @@ import { const SAMPLE_RATE = 16000; const CHUNK_SIZE = 1280; const CURRENT_PLATFORM = "WEB" as const; +const MEETING_STATUS_COMPLETED = 3; type WsSpeaker = string | { name?: string; user_id?: string | number } | undefined; type WsMessage = { @@ -884,31 +885,29 @@ export function RealtimeAsrSession() { } catch { // 会议完成已成功提交，详情刷新失败不应反向标记为结束失败。 } + // 没有转写时后端可能直接按已完成结束会议，不再进入总结流程。 + const completedWithoutData = savedMeeting?.status === MEETING_STATUS_COMPLETED; sessionStorage.removeItem(getSessionKey(meetingId)); - setSessionStatus((prev) => prev ? { ...prev, status: "COMPLETING", canResume: false, activeConnection: false } : prev); - setStatusText("已提交总结任务"); + setSessionStatus((prev) => prev ? { + ...prev, + status: completedWithoutData ? "COMPLETED" : "COMPLETING", + canResume: false, + activeConnection: false, + } : prev); + setStatusText(completedWithoutData ? "未识别到有效录音数据，会议已结束" : "已提交总结任务"); if (savedMeeting?.audioSaveStatus === "FAILED") { message.warning(savedMeeting.audioSaveMessage || "实时会议已完成，但音频保存失败，当前无法播放会议录音。转写和总结不受影响。"); + } else if (completedWithoutData) { + message.success("未识别到有效录音数据，会议已结束"); } else { message.success("实时会议已结束，正在生成总结"); } if (navigateAfterStop) { navigate(`/meetings/${meetingId}`); } - } catch (error) { + } catch { completeOnceRef.current = false; - const errorMessage = error instanceof Error ? error.message : "结束会议失败"; - if (errorMessage.includes("当前还没有转录内容")) { - try { - const statusRes = await getRealtimeMeetingSessionStatus(meetingId); - setSessionStatus(statusRes.data.data); - } catch { - // ignore status refresh failure - } - setStatusText("当前还没有转录内容，可继续识别"); - } else { - setStatusText("结束失败"); - } + setStatusText("结束失败"); } finally { setRecording(false); setFinishing(false);