diff --git a/nacos_config/rc-service-external/application.yml b/nacos_config/rc-service-external/application.yml index beeb3923..199f98c0 100644 --- a/nacos_config/rc-service-external/application.yml +++ b/nacos_config/rc-service-external/application.yml @@ -49,7 +49,7 @@ red-circle: noticeAccount: ${LIKEI_IM_NOTICE_ACCOUNT} newsletterAccount: ${LIKEI_IM_NEWSLETTER_ACCOUNT} agora: - appId: ${LIKEI_AGORA_APP_ID} + appId: ${LIKEI_RTC_APP_ID} appCertificate: ${LIKEI_AGORA_APP_CERTIFICATE} tokenExpireSecond: ${LIKEI_AGORA_TOKEN_EXPIRE_SECOND} privilegeExpireSecond: ${LIKEI_AGORA_PRIVILEGE_EXPIRE_SECOND} diff --git a/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/command/mic/LiveMicGoUpCmdExe.java b/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/command/mic/LiveMicGoUpCmdExe.java index f8fa717f..0a270e97 100644 --- a/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/command/mic/LiveMicGoUpCmdExe.java +++ b/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/command/mic/LiveMicGoUpCmdExe.java @@ -4,13 +4,15 @@ import com.red.circle.component.redis.service.RedisService; import com.red.circle.external.inner.endpoint.message.TrtcClient; import com.red.circle.framework.core.asserts.ResponseAssert; import com.red.circle.framework.core.response.CommonErrorCode; +import com.red.circle.framework.dto.ResultResponse; import com.red.circle.live.app.convertor.LiveMicAppConvertor; import com.red.circle.live.app.dto.clientobject.LiveMicrophoneCO; import com.red.circle.live.app.dto.cmd.MicUpDownCmd; -import com.red.circle.live.domain.gateway.LiveMicrophoneGateway; -import com.red.circle.live.domain.live.dto.LiveMicrophoneNoticeDTO; -import com.red.circle.live.infra.database.cache.service.UserAgoraTokenCacheService; -import com.red.circle.live.infra.database.mongo.service.LiveMicSettingService; +import com.red.circle.live.domain.gateway.LiveMicrophoneGateway; +import com.red.circle.live.domain.live.dto.LiveMicrophoneNoticeDTO; +import com.red.circle.live.infra.database.cache.service.LiveMicCacheService; +import com.red.circle.live.infra.database.cache.service.UserAgoraTokenCacheService; +import com.red.circle.live.infra.database.mongo.service.LiveMicSettingService; import com.red.circle.live.infra.util.DateTimeAsiaRiyadhUtils; import com.red.circle.live.inner.error.LiveMicErrorCode; import com.red.circle.other.inner.endpoint.live.RoomManagerClient; @@ -43,41 +45,95 @@ public class LiveMicGoUpCmdExe { private static final Logger log = LoggerFactory.getLogger(LiveMicGoUpCmdExe.class); private final LiveMicAppConvertor liveMicAppConvertor; private final LiveMicSettingService liveMicSettingService; - private final LiveMicrophoneGateway liveMicrophoneGateway; - private final TrtcClient trtcClient; - private final UserAgoraTokenCacheService userAgoraTokenCacheService; + private final LiveMicrophoneGateway liveMicrophoneGateway; + private final LiveMicCacheService liveMicCacheService; + private final TrtcClient trtcClient; + private final UserAgoraTokenCacheService userAgoraTokenCacheService; private final RedisService redisService; private final RoomDailyTaskClient roomDailyTaskClient; private final RoomManagerClient roomManagerClient; private final RedisTemplate redisTemplate; private final RoomMemberClient roomMemberClient; - public LiveMicrophoneCO execute(MicUpDownCmd cmd) { - checkPermissions(cmd); - LiveMicrophoneNoticeDTO noticeDTO = liveMicrophoneGateway.goUp( + public LiveMicrophoneCO execute(MicUpDownCmd cmd) { + boolean targetMicAlreadySelf = false; + boolean goUpCompleted = false; + try { + checkPermissions(cmd); + targetMicAlreadySelf = checkTargetMicAvailableOrSelf(cmd); + String roomToken = createPublisherToken(cmd); + + LiveMicrophoneNoticeDTO noticeDTO = liveMicrophoneGateway.goUp( + cmd.getRoomId(), + cmd.getMickIndex(), + cmd.requiredReqUserId() + ); + goUpCompleted = true; + + cachePublisherToken(cmd, roomToken); + LiveMicrophoneCO notice = returnMicrophoneCO(cmd, noticeDTO); + notice.setRoomToken(roomToken); + + // 房主邀请成员上麦 + handleRoomDailyTask(cmd); + return notice; + } catch (RuntimeException e) { + if (goUpCompleted) { + rollbackPublisherToken(cmd); + if (!targetMicAlreadySelf) { + rollbackMicSeat(cmd, e); + } + } + throw e; + } + } + + private String createPublisherToken(MicUpDownCmd cmd) { + ResultResponse response = trtcClient.createAgoraTokenGoUp( + cmd.requiredReqUserId(), + cmd.getRoomId().toString() + ); + String roomToken = Objects.nonNull(response) ? response.getBody() : null; + ResponseAssert.notNull(CommonErrorCode.THIRD_PARTY_SERVER_ERROR, roomToken); + return roomToken; + } + + private void cachePublisherToken(MicUpDownCmd cmd, String roomToken) { + userAgoraTokenCacheService.remove(cmd.requiredReqUserId(), cmd.getRoomId()); + userAgoraTokenCacheService.save(roomToken, cmd.requiredReqUserId(), cmd.getRoomId()); + } + + private boolean checkTargetMicAvailableOrSelf(MicUpDownCmd cmd) { + boolean targetMicAlreadySelf = liveMicCacheService.checkEqSelf( cmd.getRoomId(), cmd.getMickIndex(), cmd.requiredReqUserId() ); - LiveMicrophoneCO notice = returnMicrophoneCO(cmd, noticeDTO); - notice.setRoomToken(refreshPublisherToken(cmd)); - //notice.setRoomToken(createAgoraTokenGoUp(cmd)); - - // 房主邀请成员上麦 - handleRoomDailyTask(cmd); - - return notice; + if (targetMicAlreadySelf) { + return true; + } + ResponseAssert.isTrue(LiveMicErrorCode.MICROPHONE_OCCUPIED, + liveMicCacheService.checkAvailable(cmd.getRoomId(), cmd.getMickIndex())); + return false; } - private String refreshPublisherToken(MicUpDownCmd cmd) { - String roomToken = trtcClient.createAgoraTokenGoUp( - cmd.getReqUserId(), - cmd.getRoomId().toString() - ).getBody(); - ResponseAssert.notNull(CommonErrorCode.THIRD_PARTY_SERVER_ERROR, roomToken); - userAgoraTokenCacheService.remove(cmd.getReqUserId(), cmd.getRoomId()); - userAgoraTokenCacheService.save(roomToken, cmd.getReqUserId(), cmd.getRoomId()); - return roomToken; + private void rollbackPublisherToken(MicUpDownCmd cmd) { + try { + userAgoraTokenCacheService.remove(cmd.requiredReqUserId(), cmd.getRoomId()); + } catch (RuntimeException rollbackException) { + log.warn("上麦失败回滚 publisher token 失败 roomId={}, userId={}", + cmd.getRoomId(), cmd.requiredReqUserId(), rollbackException); + } + } + + private void rollbackMicSeat(MicUpDownCmd cmd, RuntimeException cause) { + try { + liveMicrophoneGateway.goDown(cmd.getRoomId(), cmd.getMickIndex(), cmd.requiredReqUserId()); + } catch (RuntimeException rollbackException) { + log.warn("上麦失败回滚麦位失败 roomId={}, micIndex={}, userId={}, cause={}", + cmd.getRoomId(), cmd.getMickIndex(), cmd.requiredReqUserId(), cause.getMessage(), + rollbackException); + } } private void handleRoomDailyTask(MicUpDownCmd cmd) { diff --git a/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/listener/UserHeartbeatListener.java b/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/listener/UserHeartbeatListener.java index 31a86f6f..27979aef 100644 --- a/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/listener/UserHeartbeatListener.java +++ b/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/listener/UserHeartbeatListener.java @@ -10,6 +10,7 @@ import com.red.circle.component.redis.service.RedisService; import com.red.circle.external.inner.model.dto.AgoraChannelStateDTO; import com.red.circle.framework.dto.ResultResponse; import com.red.circle.live.app.dto.cmd.HeartbeatStatusEnum; +import com.red.circle.live.app.service.AgoraMicMissingConfirmService; import com.red.circle.live.app.service.AgoraMicStateCleanupService; import com.red.circle.live.app.service.AgoraRoomStateService; import com.red.circle.live.domain.gateway.LiveMicrophoneGateway; @@ -51,8 +52,7 @@ import java.util.stream.Collectors; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.context.annotation.Profile; -import org.springframework.data.redis.core.RedisTemplate; -import org.springframework.stereotype.Component; +import org.springframework.stereotype.Component; /** * 用户心跳. @@ -78,6 +78,7 @@ public class UserHeartbeatListener implements MessageListener { private final RedisService redisService; private final RoomMemberClient roomMemberClient; private final AgoraRoomStateService agoraRoomStateService; + private final AgoraMicMissingConfirmService agoraMicMissingConfirmService; private final AgoraMicStateCleanupService agoraMicStateCleanupService; private static final int HEARTBEAT_INTERVAL_SECONDS = 60; @@ -214,10 +215,11 @@ public class UserHeartbeatListener implements MessageListener { UserProfileDTO actualUserProfile = ensureUserProfile(userProfile); AgoraChannelStateDTO channelState = agoraRoomStateService.getChannelState(roomId); if (channelState.querySuccess() && !isUserInAgoraChannel(channelState, actualUserProfile)) { - agoraMicStateCleanupService.cleanupUser(roomId, actualUserProfile.getId(), + handleAgoraUserMissing(roomId, actualUserProfile.getId(), channelState, "heartbeat agora user missing"); return; } + clearAgoraMissingAfterPresent(channelState, roomId, actualUserProfile.getId()); // 添加房间在线用户 addRoomOnlineUser(actualUserProfile, roomProfile); @@ -314,10 +316,11 @@ public class UserHeartbeatListener implements MessageListener { } if (!isUserInAgoraChannel(channelState, userProfile)) { - agoraMicStateCleanupService.cleanupUser(roomProfile.getId(), userProfile.getId(), + handleAgoraUserMissing(roomProfile.getId(), userProfile.getId(), channelState, "heartbeat agora user missing"); return; } + clearAgoraMissingAfterPresent(channelState, roomProfile.getId(), userProfile.getId()); refreshMicActiveTimeWithoutAgoraCheck(event, roomProfile, liveMicrophone); } @@ -370,4 +373,28 @@ public class UserHeartbeatListener implements MessageListener { && channelState.getAudience().contains(agoraUid)); } + private void handleAgoraUserMissing(Long roomId, + Long userId, + AgoraChannelStateDTO channelState, + String reason) { + if (!agoraMicMissingConfirmService.markMissingAndShouldCleanup(roomId, userId, + appendChannelStateReason(channelState, reason))) { + return; + } + agoraMicStateCleanupService.cleanupUser(roomId, userId, reason); + } + + private void clearAgoraMissingAfterPresent(AgoraChannelStateDTO channelState, Long roomId, Long userId) { + if (Objects.nonNull(channelState) && channelState.querySuccess()) { + agoraMicMissingConfirmService.clearMissing(roomId, userId); + } + } + + private String appendChannelStateReason(AgoraChannelStateDTO channelState, String reason) { + if (Objects.isNull(channelState)) { + return reason; + } + return reason + ", channelExist=" + channelState.channelExist(); + } + } diff --git a/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTask.java b/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTask.java index b6359dfd..4465914f 100644 --- a/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTask.java +++ b/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTask.java @@ -2,6 +2,7 @@ package com.red.circle.live.app.scheduler; import com.red.circle.component.redis.service.RedisService; import com.red.circle.external.inner.model.dto.AgoraChannelStateDTO; +import com.red.circle.live.app.service.AgoraMicMissingConfirmService; import com.red.circle.live.app.service.AgoraMicStateCleanupService; import com.red.circle.live.app.service.AgoraRoomStateService; import com.red.circle.live.domain.live.LiveMicrophone; @@ -35,6 +36,7 @@ public class AgoraMicStateReconcileTask { private final RedisTemplate redisTemplate; private final LiveMicCacheService liveMicCacheService; private final AgoraRoomStateService agoraRoomStateService; + private final AgoraMicMissingConfirmService agoraMicMissingConfirmService; private final AgoraMicStateCleanupService agoraMicStateCleanupService; @Scheduled(fixedDelayString = "${red-circle.live.agora-mic-reconcile-delay-ms:30000}") @@ -62,13 +64,15 @@ public class AgoraMicStateReconcileTask { return; } - if (!state.channelExist()) { - agoraMicStateCleanupService.cleanupRoom(roomId, "reconcile agora channel not exist"); + List microphones = liveMicCacheService.listLiveMicrophone(roomId); + if (CollectionUtils.isEmpty(microphones)) { return; } - List microphones = liveMicCacheService.listLiveMicrophone(roomId); - if (CollectionUtils.isEmpty(microphones)) { + if (!state.channelExist()) { + for (LiveMicrophone microphone : microphones) { + cleanupMissingMicrophoneUser(roomId, microphone, state, "reconcile agora channel not exist"); + } return; } @@ -85,12 +89,30 @@ public class AgoraMicStateReconcileTask { } Long agoraUid = DataTypeUtils.toLong(microphone.getUser().getAccount()); if (Objects.isNull(agoraUid) || !channelUids.contains(agoraUid)) { - agoraMicStateCleanupService.cleanupUser(roomId, microphone.getUser().getId(), - "reconcile agora user missing"); + cleanupMissingMicrophoneUser(roomId, microphone, state, "reconcile agora user missing"); + } else { + agoraMicMissingConfirmService.clearMissing(roomId, microphone.getUser().getId()); } } } + private void cleanupMissingMicrophoneUser(Long roomId, + LiveMicrophone microphone, + AgoraChannelStateDTO state, + String reason) { + if (Objects.isNull(microphone) + || Objects.isNull(microphone.getUser()) + || Objects.isNull(microphone.getUser().getId())) { + return; + } + Long userId = microphone.getUser().getId(); + if (!agoraMicMissingConfirmService.markMissingAndShouldCleanup(roomId, userId, + reason + ", channelExist=" + state.channelExist())) { + return; + } + agoraMicStateCleanupService.cleanupUser(roomId, userId, reason); + } + private Set scanSeatRoomIds() { Set roomIds = new HashSet<>(); ScanOptions options = ScanOptions.scanOptions().match(SEAT_KEY_PREFIX + "*").count(200).build(); diff --git a/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/service/AgoraMicMissingConfirmService.java b/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/service/AgoraMicMissingConfirmService.java new file mode 100644 index 00000000..e24ef69a --- /dev/null +++ b/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/service/AgoraMicMissingConfirmService.java @@ -0,0 +1,102 @@ +package com.red.circle.live.app.service; + +import com.red.circle.component.redis.service.RedisService; +import java.util.Objects; +import java.util.concurrent.TimeUnit; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; + +/** + * 声网 missing 状态二次确认. + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class AgoraMicMissingConfirmService { + + private static final String KEY_PREFIX = "agora:mic:missing:"; + private static final int MISSING_TTL_SECONDS = 10; + private static final int CONFIRM_MISSING_COUNT = 2; + private static final long CONFIRM_MISSING_AFTER_MILLIS = 5_000L; + + private final RedisService redisService; + + public boolean markMissingAndShouldCleanup(Long roomId, Long userId, String reason) { + if (Objects.isNull(roomId) || Objects.isNull(userId)) { + return false; + } + + String key = missingKey(roomId, userId); + long nowMillis = System.currentTimeMillis(); + MissingState nextState = MissingState.next(redisService.getString(key), nowMillis); + if (nextState.shouldCleanup(nowMillis)) { + redisService.delete(key); + log.warn("声网 missing 已二次确认，允许清理 roomId={}, userId={}, count={}, firstMissingAt={}, reason={}", + roomId, userId, nextState.count, nextState.firstMissingAtMillis, reason); + return true; + } + + redisService.setString(key, nextState.serialize(), MISSING_TTL_SECONDS, TimeUnit.SECONDS); + log.warn("声网 missing 首次缓冲，跳过本次清理和刷新 roomId={}, userId={}, count={}, firstMissingAt={}, reason={}", + roomId, userId, nextState.count, nextState.firstMissingAtMillis, reason); + return false; + } + + public void clearMissing(Long roomId, Long userId) { + if (Objects.isNull(roomId) || Objects.isNull(userId)) { + return; + } + redisService.delete(missingKey(roomId, userId)); + } + + private String missingKey(Long roomId, Long userId) { + return KEY_PREFIX + roomId + ":" + userId; + } + + static class MissingState { + + private final int count; + private final long firstMissingAtMillis; + + private MissingState(int count, long firstMissingAtMillis) { + this.count = count; + this.firstMissingAtMillis = firstMissingAtMillis; + } + + static MissingState next(String rawValue, long nowMillis) { + MissingState current = parse(rawValue, nowMillis); + long firstMissingAtMillis = current.count <= 0 ? nowMillis : current.firstMissingAtMillis; + return new MissingState(current.count + 1, firstMissingAtMillis); + } + + boolean shouldCleanup(long nowMillis) { + return count >= CONFIRM_MISSING_COUNT + || nowMillis - firstMissingAtMillis >= CONFIRM_MISSING_AFTER_MILLIS; + } + + String serialize() { + return count + ":" + firstMissingAtMillis; + } + + private static MissingState parse(String rawValue, long nowMillis) { + if (Objects.isNull(rawValue) || rawValue.isBlank()) { + return new MissingState(0, nowMillis); + } + String[] parts = rawValue.split(":", 2); + if (parts.length != 2) { + return new MissingState(0, nowMillis); + } + try { + int count = Integer.parseInt(parts[0]); + long firstMissingAtMillis = Long.parseLong(parts[1]); + if (count <= 0 || firstMissingAtMillis <= 0 || firstMissingAtMillis > nowMillis) { + return new MissingState(0, nowMillis); + } + return new MissingState(count, firstMissingAtMillis); + } catch (NumberFormatException e) { + return new MissingState(0, nowMillis); + } + } + } +} diff --git a/rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/command/mic/LiveMicGoUpCmdExeTest.java b/rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/command/mic/LiveMicGoUpCmdExeTest.java new file mode 100644 index 00000000..62c6ee73 --- /dev/null +++ b/rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/command/mic/LiveMicGoUpCmdExeTest.java @@ -0,0 +1,222 @@ +package com.red.circle.live.app.command.mic; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.fail; + +import com.red.circle.external.inner.endpoint.message.TrtcClient; +import com.red.circle.framework.dto.ResultResponse; +import com.red.circle.live.app.convertor.LiveMicAppConvertor; +import com.red.circle.live.app.dto.clientobject.LiveMicUserCO; +import com.red.circle.live.app.dto.clientobject.LiveMicrophoneCO; +import com.red.circle.live.app.dto.cmd.MicUpDownCmd; +import com.red.circle.live.domain.gateway.LiveMicrophoneGateway; +import com.red.circle.live.domain.live.LiveMicUser; +import com.red.circle.live.domain.live.LiveMicrophone; +import com.red.circle.live.domain.live.dto.LiveMicrophoneDTO; +import com.red.circle.live.domain.live.dto.LiveMicrophoneNoticeDTO; +import com.red.circle.live.infra.database.cache.service.LiveMicCacheService; +import com.red.circle.live.infra.database.cache.service.UserAgoraTokenCacheService; +import com.red.circle.live.infra.database.mongo.entity.LiveMicSetting; +import com.red.circle.live.infra.database.mongo.entity.LiveMicSettingSeat; +import com.red.circle.live.infra.database.mongo.service.LiveMicSettingService; +import com.red.circle.other.inner.endpoint.live.RoomManagerClient; +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.Method; +import java.lang.reflect.Proxy; +import java.util.ArrayList; +import java.util.List; +import org.junit.Test; + +public class LiveMicGoUpCmdExeTest { + + private static final Long ROOM_ID = 1001L; + private static final Long USER_ID = 2002L; + private static final Integer MIC_INDEX = 1; + + @Test + public void executeShouldCreateTokenBeforeGoUpAndCacheAfterGoUp() { + Fixture fixture = new Fixture(); + + LiveMicrophoneCO result = fixture.exe.execute(command()); + + assertEquals("publisher-token", result.getRoomToken()); + assertEquals(List.of("checkEqSelf", "checkAvailable", "createToken", "goUp", "removeToken", + "saveToken", "convert", "getRoom"), fixture.calls); + } + + @Test + public void executeShouldRollbackMicAndTokenWhenCacheFailsAfterGoUp() { + Fixture fixture = new Fixture(); + RuntimeException cacheException = new RuntimeException("cache failed"); + fixture.cacheException = cacheException; + + try { + fixture.exe.execute(command()); + fail("Expected cache exception"); + } catch (RuntimeException e) { + assertSame(cacheException, e); + } + + assertEquals(List.of("checkEqSelf", "checkAvailable", "createToken", "goUp", "removeToken", + "saveToken", "removeToken", "goDown"), fixture.calls); + } + + private static MicUpDownCmd command() { + MicUpDownCmd cmd = new MicUpDownCmd() + .setRoomId(ROOM_ID) + .setMickIndex(MIC_INDEX); + cmd.setReqUserId(USER_ID); + return cmd; + } + + private static LiveMicrophoneNoticeDTO notice() { + return new LiveMicrophoneNoticeDTO() + .setMics(List.of(new LiveMicrophoneDTO().setRoomId(ROOM_ID).setMicIndex(MIC_INDEX))); + } + + @SuppressWarnings("unchecked") + private static T proxy(Class type, InvocationHandler handler) { + return (T) Proxy.newProxyInstance(type.getClassLoader(), new Class[] {type}, + (Object proxy, Method method, Object[] args) -> { + if (method.getDeclaringClass().equals(Object.class)) { + return method.invoke(handler, args); + } + return handler.invoke(proxy, method, args); + }); + } + + private static class Fixture { + + private final List calls = new ArrayList<>(); + private RuntimeException cacheException; + + private final LiveMicAppConvertor liveMicAppConvertor = new LiveMicAppConvertor() { + @Override + public List toListLiveMicrophoneCO(List liveMicrophones) { + return null; + } + + @Override + public LiveMicrophoneCO toLiveMicrophoneCO(LiveMicrophoneDTO dto) { + calls.add("convert"); + return new LiveMicrophoneCO(); + } + + @Override + public LiveMicUserCO toLiveMicUserCO(LiveMicUser liveMicUser) { + return null; + } + }; + + private final LiveMicSettingService liveMicSettingService = new LiveMicSettingService() { + @Override + public LiveMicSetting updateMicSize(Long roomId, int micSize) { + return null; + } + + @Override + public void updateMicSeat(Long roomId, LiveMicSettingSeat settingSeat) { + } + + @Override + public LiveMicSettingSeat getMicSeat(Long roomId, Integer micIndex) { + return null; + } + + @Override + public List listMicSeat(Long roomId) { + return null; + } + + @Override + public boolean checkLock(Long roomId, Integer micIndex) { + return false; + } + + @Override + public boolean exists(Long roomId) { + return false; + } + }; + + private final LiveMicrophoneGateway liveMicrophoneGateway = proxy(LiveMicrophoneGateway.class, + (Object proxy, Method method, Object[] args) -> { + if ("goUp".equals(method.getName())) { + calls.add("goUp"); + return notice(); + } + if ("goDown".equals(method.getName())) { + calls.add("goDown"); + return null; + } + throw new UnsupportedOperationException(method.getName()); + }); + + private final LiveMicCacheService liveMicCacheService = proxy(LiveMicCacheService.class, + (Object proxy, Method method, Object[] args) -> { + if ("checkEqSelf".equals(method.getName())) { + calls.add("checkEqSelf"); + return false; + } + if ("checkAvailable".equals(method.getName())) { + calls.add("checkAvailable"); + return true; + } + throw new UnsupportedOperationException(method.getName()); + }); + + private final TrtcClient trtcClient = proxy(TrtcClient.class, + (Object proxy, Method method, Object[] args) -> { + if ("createAgoraTokenGoUp".equals(method.getName())) { + calls.add("createToken"); + return ResultResponse.success("publisher-token"); + } + throw new UnsupportedOperationException(method.getName()); + }); + + private final UserAgoraTokenCacheService userAgoraTokenCacheService = + new UserAgoraTokenCacheService() { + @Override + public void remove(Long userId, Long roomId) { + calls.add("removeToken"); + } + + @Override + public void save(String agoraToken, Long userId, Long roomId) { + calls.add("saveToken"); + if (cacheException != null) { + throw cacheException; + } + } + + @Override + public String getUserAgoraToken(Long userId, Long roomId) { + return null; + } + }; + + private final RoomManagerClient roomManagerClient = proxy(RoomManagerClient.class, + (Object proxy, Method method, Object[] args) -> { + if ("getById".equals(method.getName())) { + calls.add("getRoom"); + return ResultResponse.success(null); + } + throw new UnsupportedOperationException(method.getName()); + }); + + private final LiveMicGoUpCmdExe exe = new LiveMicGoUpCmdExe( + liveMicAppConvertor, + liveMicSettingService, + liveMicrophoneGateway, + liveMicCacheService, + trtcClient, + userAgoraTokenCacheService, + null, + null, + roomManagerClient, + null, + null + ); + } +} diff --git a/rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/service/AgoraMicMissingConfirmServiceTest.java b/rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/service/AgoraMicMissingConfirmServiceTest.java new file mode 100644 index 00000000..a7b567a6 --- /dev/null +++ b/rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/service/AgoraMicMissingConfirmServiceTest.java @@ -0,0 +1,42 @@ +package com.red.circle.live.app.service; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import com.red.circle.live.app.service.AgoraMicMissingConfirmService.MissingState; +import org.junit.Test; + +public class AgoraMicMissingConfirmServiceTest { + + @Test + public void firstMissingShouldOnlyRecordState() { + MissingState state = MissingState.next(null, 1_000L); + + assertFalse(state.shouldCleanup(1_000L)); + assertEquals("1:1000", state.serialize()); + } + + @Test + public void secondConsecutiveMissingShouldCleanup() { + MissingState state = MissingState.next("1:1000", 2_000L); + + assertTrue(state.shouldCleanup(2_000L)); + assertEquals("2:1000", state.serialize()); + } + + @Test + public void delayedMissingShouldCleanupAfterConfirmWindow() { + MissingState state = MissingState.next("1:1000", 6_000L); + + assertTrue(state.shouldCleanup(6_000L)); + } + + @Test + public void malformedStateShouldRestartConfirmWindow() { + MissingState state = MissingState.next("bad-value", 3_000L); + + assertFalse(state.shouldCleanup(3_000L)); + assertEquals("1:3000", state.serialize()); + } +} diff --git a/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/repository/LiveMicrophoneGatewayImpl.java b/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/repository/LiveMicrophoneGatewayImpl.java index 34dd20e9..69638178 100644 --- a/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/repository/LiveMicrophoneGatewayImpl.java +++ b/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/repository/LiveMicrophoneGatewayImpl.java @@ -164,6 +164,7 @@ public class LiveMicrophoneGatewayImpl implements LiveMicrophoneGateway { micOpsLock(roomId, micIndex); // RLock lock = redissonClient.getLock(roomId.toString()+micIndex.toString()+"room"); // lock.lock(); + boolean micUpdated = false; try { // 验证用户是否真实存在 ResponseAssert.notNull(UserErrorCode.USER_INFO_NOT_FOUND, userProfileClient.getByUserId(userId)); @@ -204,12 +205,24 @@ public class LiveMicrophoneGatewayImpl implements LiveMicrophoneGateway { // 上麦克风 liveMicCacheService.goUp(roomId, micIndex, liveMicrophone); + micUpdated = true; // 清理活跃用户 liveMicCacheService.refreshNotActiveUser(roomId); // 发送通知 return micUserChangeNotice(roomProfile); + } catch (RuntimeException e) { + if (micUpdated) { + try { + liveMicCacheService.goDown(roomId, micIndex); + liveMicCacheService.refreshNotActiveUser(roomId); + } catch (RuntimeException rollbackException) { + log.warn("上麦写入后异常，回滚麦位失败 roomId={}, micIndex={}, userId={}", + roomId, micIndex, userId, rollbackException); + } + } + throw e; } finally { // lock.unlock(); micOpsUnlock(roomId, micIndex);