From 0fecd235b57394075a9569d32b76750e7868cef9 Mon Sep 17 00:00:00 2001 From: hy001 Date: Fri, 15 May 2026 23:50:14 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A3=B0=E7=BD=91=E9=97=AE=E9=A2=98=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../app/listener/UserHeartbeatListener.java | 49 +++++++++--- .../scheduler/AgoraMicStateReconcileTask.java | 17 ++++ .../UserHeartbeatListenerRtcProviderTest.java | 79 +++++++++++++++++++ ...aMicStateReconcileTaskRtcProviderTest.java | 69 ++++++++++++++++ 4 files changed, 204 insertions(+), 10 deletions(-) create mode 100644 rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/listener/UserHeartbeatListenerRtcProviderTest.java create mode 100644 rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTaskRtcProviderTest.java 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 173e354c..71c14bbf 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 @@ -4,9 +4,11 @@ import com.red.circle.component.mq.MessageEventProcess; import com.red.circle.component.mq.MessageEventProcessDescribe; import com.red.circle.component.mq.config.RocketMqMessageListener; import com.red.circle.component.mq.service.Action; -import com.red.circle.component.mq.service.ConsumerMessage; +import com.red.circle.component.mq.service.ConsumerMessage; import com.red.circle.component.mq.service.MessageListener; import com.red.circle.component.redis.service.RedisService; +import com.red.circle.common.business.dto.rtc.RtcProviderEnum; +import com.red.circle.common.business.dto.rtc.RtcVersionSupport; 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; @@ -218,18 +220,23 @@ public class UserHeartbeatListener implements MessageListener { } UserProfileDTO actualUserProfile = ensureUserProfile(userProfile); - AgoraChannelStateDTO channelState = agoraRoomStateService.getChannelState(roomId); - if (channelState.querySuccess() && !isUserInAgoraChannel(channelState, actualUserProfile)) { - handleAgoraUserMissing(roomId, actualUserProfile.getId(), channelState, - "heartbeat agora user missing"); - return; + boolean validateAgoraState = shouldValidateAgoraState(roomId); + AgoraChannelStateDTO channelState = null; + if (validateAgoraState) { + channelState = agoraRoomStateService.getChannelState(roomId); + if (channelState.querySuccess() && !isUserInAgoraChannel(channelState, actualUserProfile)) { + handleAgoraUserMissing(roomId, actualUserProfile.getId(), channelState, + "heartbeat agora user missing"); + return; + } + clearAgoraMissingAfterPresent(channelState, roomId, actualUserProfile.getId()); } - clearAgoraMissingAfterPresent(channelState, roomId, actualUserProfile.getId()); // 添加房间在线用户 addRoomOnlineUser(actualUserProfile, roomProfile); - refreshMicActiveTime(event, roomProfile, actualUserProfile, channelState); + refreshMicActiveTime(event, roomProfile, actualUserProfile, channelState, + validateAgoraState); } private RoomProfileDTO getRoomProfileDto(Long roomId) { @@ -281,13 +288,15 @@ public class UserHeartbeatListener implements MessageListener { private void refreshMicActiveTime(UserHeartbeatEvent event, RoomProfileDTO roomProfile, UserProfileDTO userProfile) { - refreshMicActiveTime(event, roomProfile, userProfile, null); + refreshMicActiveTime(event, roomProfile, userProfile, null, + shouldValidateAgoraState(roomProfile.getId())); } private void refreshMicActiveTime(UserHeartbeatEvent event, RoomProfileDTO roomProfile, UserProfileDTO userProfile, - AgoraChannelStateDTO checkedChannelState) { + AgoraChannelStateDTO checkedChannelState, + boolean validateAgoraState) { // 如果不在麦克风, true&null兼容历史版本 if (Objects.equals(event.getUpMick(), Boolean.FALSE)) { @@ -310,6 +319,12 @@ public class UserHeartbeatListener implements MessageListener { return; } + if (!validateAgoraState) { + agoraMicMissingConfirmService.clearMissing(roomProfile.getId(), userProfile.getId()); + refreshMicActiveTimeWithoutAgoraCheck(event, roomProfile, liveMicrophone); + return; + } + AgoraChannelStateDTO channelState = Objects.nonNull(checkedChannelState) ? checkedChannelState : agoraRoomStateService.getChannelState(roomProfile.getId()); @@ -405,6 +420,20 @@ public class UserHeartbeatListener implements MessageListener { } } + boolean shouldValidateAgoraState(Long roomId) { + if (Objects.isNull(roomId)) { + return true; + } + try { + RtcProviderEnum provider = RtcProviderEnum.of( + redisService.getString(RtcVersionSupport.roomProviderKey(roomId))); + return !Objects.equals(RtcProviderEnum.TRTC, provider); + } catch (Exception e) { + log.warn("读取房间 RTC provider 失败，按 Agora 校验处理 roomId={}", roomId, e); + return true; + } + } + private String appendChannelStateReason(AgoraChannelStateDTO channelState, String reason) { if (Objects.isNull(channelState)) { return reason; 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 4465914f..67fe26fa 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 @@ -1,6 +1,8 @@ package com.red.circle.live.app.scheduler; import com.red.circle.component.redis.service.RedisService; +import com.red.circle.common.business.dto.rtc.RtcProviderEnum; +import com.red.circle.common.business.dto.rtc.RtcVersionSupport; 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; @@ -57,6 +59,10 @@ public class AgoraMicStateReconcileTask { if (Objects.isNull(roomId)) { return; } + if (!shouldReconcileAgoraRoom(roomId)) { + log.debug("声网麦位对账跳过非 Agora 房间 roomId={}", roomId); + return; + } AgoraChannelStateDTO state = agoraRoomStateService.getChannelState(roomId); if (!state.querySuccess()) { @@ -96,6 +102,17 @@ public class AgoraMicStateReconcileTask { } } + boolean shouldReconcileAgoraRoom(Long roomId) { + try { + RtcProviderEnum provider = RtcProviderEnum.of( + redisService.getString(RtcVersionSupport.roomProviderKey(roomId))); + return !Objects.equals(RtcProviderEnum.TRTC, provider); + } catch (Exception e) { + log.warn("读取房间 RTC provider 失败，按 Agora 对账处理 roomId={}", roomId, e); + return true; + } + } + private void cleanupMissingMicrophoneUser(Long roomId, LiveMicrophone microphone, AgoraChannelStateDTO state, diff --git a/rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/listener/UserHeartbeatListenerRtcProviderTest.java b/rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/listener/UserHeartbeatListenerRtcProviderTest.java new file mode 100644 index 00000000..58950385 --- /dev/null +++ b/rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/listener/UserHeartbeatListenerRtcProviderTest.java @@ -0,0 +1,79 @@ +package com.red.circle.live.app.listener; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import com.red.circle.component.redis.service.RedisService; +import java.lang.reflect.Proxy; +import org.junit.Test; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.core.ValueOperations; + +public class UserHeartbeatListenerRtcProviderTest { + + @Test + public void trtcRoomShouldSkipAgoraStateValidation() { + assertFalse(listener("TRTC").shouldValidateAgoraState(1001026L)); + } + + @Test + public void agoraRoomShouldKeepAgoraStateValidation() { + assertTrue(listener("AGORA").shouldValidateAgoraState(1001026L)); + } + + @Test + public void unknownRoomProviderShouldKeepAgoraStateValidation() { + assertTrue(listener(null).shouldValidateAgoraState(1001026L)); + } + + private static UserHeartbeatListener listener(String provider) { + return new UserHeartbeatListener( + null, + null, + null, + null, + null, + null, + null, + null, + null, + null, + redisService(provider), + null, + null, + null, + null, + null); + } + + private static RedisService redisService(String provider) { + return new RedisService(new RedisTemplate<>() { + @Override + public ValueOperations opsForValue() { + return valueOperations(provider); + } + }); + } + + @SuppressWarnings("unchecked") + private static ValueOperations valueOperations(String provider) { + return (ValueOperations) Proxy.newProxyInstance( + ValueOperations.class.getClassLoader(), + new Class[] {ValueOperations.class}, + (proxy, method, args) -> { + if ("get".equals(method.getName())) { + return provider; + } + if ("toString".equals(method.getName())) { + return "ValueOperations(" + provider + ")"; + } + if ("hashCode".equals(method.getName())) { + return System.identityHashCode(proxy); + } + if ("equals".equals(method.getName())) { + return proxy == args[0]; + } + throw new UnsupportedOperationException(method.getName()); + }); + } +} diff --git a/rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTaskRtcProviderTest.java b/rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTaskRtcProviderTest.java new file mode 100644 index 00000000..a69c0000 --- /dev/null +++ b/rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTaskRtcProviderTest.java @@ -0,0 +1,69 @@ +package com.red.circle.live.app.scheduler; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import com.red.circle.component.redis.service.RedisService; +import java.lang.reflect.Proxy; +import org.junit.Test; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.core.ValueOperations; + +public class AgoraMicStateReconcileTaskRtcProviderTest { + + @Test + public void trtcRoomShouldSkipAgoraReconcile() { + assertFalse(task("TRTC").shouldReconcileAgoraRoom(1001026L)); + } + + @Test + public void agoraRoomShouldKeepAgoraReconcile() { + assertTrue(task("AGORA").shouldReconcileAgoraRoom(1001026L)); + } + + @Test + public void unknownRoomProviderShouldKeepAgoraReconcile() { + assertTrue(task(null).shouldReconcileAgoraRoom(1001026L)); + } + + private static AgoraMicStateReconcileTask task(String provider) { + return new AgoraMicStateReconcileTask( + redisService(provider), + null, + null, + null, + null, + null); + } + + private static RedisService redisService(String provider) { + return new RedisService(new RedisTemplate<>() { + @Override + public ValueOperations opsForValue() { + return valueOperations(provider); + } + }); + } + + @SuppressWarnings("unchecked") + private static ValueOperations valueOperations(String provider) { + return (ValueOperations) Proxy.newProxyInstance( + ValueOperations.class.getClassLoader(), + new Class[] {ValueOperations.class}, + (proxy, method, args) -> { + if ("get".equals(method.getName())) { + return provider; + } + if ("toString".equals(method.getName())) { + return "ValueOperations(" + provider + ")"; + } + if ("hashCode".equals(method.getName())) { + return System.identityHashCode(proxy); + } + if ("equals".equals(method.getName())) { + return proxy == args[0]; + } + throw new UnsupportedOperationException(method.getName()); + }); + } +}