diff --git a/rc-service/rc-inner-api/external-inner/external-inner-api/src/main/java/com/red/circle/external/inner/endpoint/message/api/AgoraClientApi.java b/rc-service/rc-inner-api/external-inner/external-inner-api/src/main/java/com/red/circle/external/inner/endpoint/message/api/AgoraClientApi.java index b8c27679..fc17aa29 100644 --- a/rc-service/rc-inner-api/external-inner/external-inner-api/src/main/java/com/red/circle/external/inner/endpoint/message/api/AgoraClientApi.java +++ b/rc-service/rc-inner-api/external-inner/external-inner-api/src/main/java/com/red/circle/external/inner/endpoint/message/api/AgoraClientApi.java @@ -1,8 +1,9 @@ package com.red.circle.external.inner.endpoint.message.api; -import com.red.circle.framework.dto.ResultResponse; -import org.springframework.web.bind.annotation.GetMapping; -import org.springframework.web.bind.annotation.RequestParam; +import com.red.circle.framework.dto.ResultResponse; +import com.red.circle.external.inner.model.dto.AgoraChannelStateDTO; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RequestParam; import java.util.List; @@ -37,6 +38,15 @@ public interface AgoraClientApi { * @param cname * @return */ - @GetMapping("/channel/user") - ResultResponse> channelUser(@RequestParam("cname") String cname); -} + @GetMapping("/channel/user") + ResultResponse> channelUser(@RequestParam("cname") String cname); + + /** + * 查询频道状态. + * + * @param cname 频道名 + * @return 频道状态 + */ + @GetMapping("/channel/state") + ResultResponse channelState(@RequestParam("cname") String cname); +} diff --git a/rc-service/rc-inner-api/external-inner/external-inner-model/src/main/java/com/red/circle/external/inner/model/dto/AgoraChannelStateDTO.java b/rc-service/rc-inner-api/external-inner/external-inner-model/src/main/java/com/red/circle/external/inner/model/dto/AgoraChannelStateDTO.java new file mode 100644 index 00000000..c2deef01 --- /dev/null +++ b/rc-service/rc-inner-api/external-inner/external-inner-model/src/main/java/com/red/circle/external/inner/model/dto/AgoraChannelStateDTO.java @@ -0,0 +1,52 @@ +package com.red.circle.external.inner.model.dto; + +import java.io.Serial; +import java.io.Serializable; +import java.util.ArrayList; +import java.util.List; +import lombok.Data; +import lombok.experimental.Accessors; + +/** + * 声网频道状态. + */ +@Data +@Accessors(chain = true) +public class AgoraChannelStateDTO implements Serializable { + + @Serial + private static final long serialVersionUID = 1L; + + /** + * 查询是否成功. + */ + private Boolean querySuccess; + + /** + * 频道是否存在. + */ + private Boolean channelExist; + + /** + * 主播 uid 列表. + */ + private List broadcasters = new ArrayList<>(); + + /** + * 观众 uid 列表. + */ + private List audience = new ArrayList<>(); + + /** + * 原始返回或错误信息. + */ + private String rawResponse; + + public boolean querySuccess() { + return Boolean.TRUE.equals(querySuccess); + } + + public boolean channelExist() { + return Boolean.TRUE.equals(channelExist); + } +} diff --git a/rc-service/rc-service-external/external-inner-endpoint/src/main/java/com/red/circle/external/inner/endpoint/message/AgoraClientEndpoint.java b/rc-service/rc-service-external/external-inner-endpoint/src/main/java/com/red/circle/external/inner/endpoint/message/AgoraClientEndpoint.java index 7d2333b9..ae128b97 100644 --- a/rc-service/rc-service-external/external-inner-endpoint/src/main/java/com/red/circle/external/inner/endpoint/message/AgoraClientEndpoint.java +++ b/rc-service/rc-service-external/external-inner-endpoint/src/main/java/com/red/circle/external/inner/endpoint/message/AgoraClientEndpoint.java @@ -1,6 +1,7 @@ package com.red.circle.external.inner.endpoint.message; -import com.red.circle.external.inner.endpoint.message.api.AgoraClientApi; +import com.red.circle.external.inner.model.dto.AgoraChannelStateDTO; +import com.red.circle.external.inner.endpoint.message.api.AgoraClientApi; import com.red.circle.external.inner.service.message.AgoraClientService; import com.red.circle.framework.dto.ResultResponse; import lombok.RequiredArgsConstructor; @@ -38,7 +39,12 @@ public class AgoraClientEndpoint implements AgoraClient{ } @Override - public ResultResponse> channelUser(String cname) { - return ResultResponse.success(agoraClientService.channelUser(cname)); - } -} + public ResultResponse> channelUser(String cname) { + return ResultResponse.success(agoraClientService.channelUser(cname)); + } + + @Override + public ResultResponse channelState(String cname) { + return ResultResponse.success(agoraClientService.channelState(cname)); + } +} diff --git a/rc-service/rc-service-external/external-inner-endpoint/src/main/java/com/red/circle/external/inner/service/message/AgoraClientService.java b/rc-service/rc-service-external/external-inner-endpoint/src/main/java/com/red/circle/external/inner/service/message/AgoraClientService.java index 3c47ed39..892d78ed 100644 --- a/rc-service/rc-service-external/external-inner-endpoint/src/main/java/com/red/circle/external/inner/service/message/AgoraClientService.java +++ b/rc-service/rc-service-external/external-inner-endpoint/src/main/java/com/red/circle/external/inner/service/message/AgoraClientService.java @@ -1,7 +1,8 @@ package com.red.circle.external.inner.service.message; -import java.util.List; +import com.red.circle.external.inner.model.dto.AgoraChannelStateDTO; +import java.util.List; /** * 声网. @@ -29,5 +30,13 @@ public interface AgoraClientService { * @param cname * @return */ - List channelUser(String cname); -} + List channelUser(String cname); + + /** + * 查询频道状态. + * + * @param cname 频道名 + * @return 频道状态 + */ + AgoraChannelStateDTO channelState(String cname); +} diff --git a/rc-service/rc-service-external/external-inner-endpoint/src/main/java/com/red/circle/external/inner/service/message/impl/AgoraClientServiceImpl.java b/rc-service/rc-service-external/external-inner-endpoint/src/main/java/com/red/circle/external/inner/service/message/impl/AgoraClientServiceImpl.java index d09cfbde..358e27e3 100644 --- a/rc-service/rc-service-external/external-inner-endpoint/src/main/java/com/red/circle/external/inner/service/message/impl/AgoraClientServiceImpl.java +++ b/rc-service/rc-service-external/external-inner-endpoint/src/main/java/com/red/circle/external/inner/service/message/impl/AgoraClientServiceImpl.java @@ -1,9 +1,10 @@ package com.red.circle.external.inner.service.message.impl; -import com.alibaba.fastjson.JSON; -import com.alibaba.fastjson.JSONObject; -import com.red.circle.external.infra.props.RcMessageAgoraProperties; -import com.red.circle.external.inner.service.message.AgoraClientService; +import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.JSONObject; +import com.red.circle.external.infra.props.RcMessageAgoraProperties; +import com.red.circle.external.inner.model.dto.AgoraChannelStateDTO; +import com.red.circle.external.inner.service.message.AgoraClientService; import com.red.circle.tool.core.collection.CollectionUtils; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -92,7 +93,7 @@ public class AgoraClientServiceImpl implements AgoraClientService { @Override - public List channelUser(String cname) { + public List channelUser(String cname) { log.info("rcMessageAgoraProperties配置信息:{}", JSON.toJSONString(rcMessageAgoraProperties)); CloseableHttpClient httpClient = HttpClients.createDefault(); HttpGet httpGet = new HttpGet("https://api.sd-rtn.com/dev/v1/channel/user/" + rcMessageAgoraProperties.getAppId() + "/" + cname); @@ -122,6 +123,48 @@ public class AgoraClientServiceImpl implements AgoraClientService { log.error("声网获取在线用户信息失败:{}", e.getMessage()); return CollectionUtils.newArrayList(); } - } - -} + } + + @Override + public AgoraChannelStateDTO channelState(String cname) { + CloseableHttpClient httpClient = HttpClients.createDefault(); + HttpGet httpGet = new HttpGet("https://api.sd-rtn.com/dev/v1/channel/user/" + rcMessageAgoraProperties.getAppId() + "/" + cname); + httpGet.addHeader("Authorization", rcMessageAgoraProperties.getAuthKey()); + try { + CloseableHttpResponse response = httpClient.execute(httpGet); + String result = EntityUtils.toString(response.getEntity()); + log.info("声网频道状态返回信息 cname={}, result={}", cname, result); + JSONObject resultObject = JSON.parseObject(result); + Boolean success = resultObject.getBoolean("success"); + AgoraChannelStateDTO state = new AgoraChannelStateDTO() + .setQuerySuccess(Boolean.TRUE.equals(success)) + .setChannelExist(Boolean.FALSE) + .setRawResponse(result); + if (!Boolean.TRUE.equals(success)) { + return state; + } + JSONObject dataObject = resultObject.getJSONObject("data"); + if (Objects.isNull(dataObject)) { + return state; + } + state.setChannelExist(Boolean.TRUE.equals(dataObject.getBoolean("channel_exist"))); + if (!state.channelExist()) { + return state; + } + if (Objects.nonNull(dataObject.getJSONArray("broadcasters"))) { + state.setBroadcasters(dataObject.getJSONArray("broadcasters").toJavaList(Long.class)); + } + if (Objects.nonNull(dataObject.getJSONArray("audience"))) { + state.setAudience(dataObject.getJSONArray("audience").toJavaList(Long.class)); + } + return state; + } catch (Exception e) { + log.error("声网获取频道状态失败 cname={}:{}", cname, e.getMessage()); + return new AgoraChannelStateDTO() + .setQuerySuccess(Boolean.FALSE) + .setChannelExist(Boolean.FALSE) + .setRawResponse(e.getMessage()); + } + } + +} diff --git a/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/command/user/LiveQuitRoomExe.java b/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/command/user/LiveQuitRoomExe.java index 7d0c4412..dbf0be88 100644 --- a/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/command/user/LiveQuitRoomExe.java +++ b/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/command/user/LiveQuitRoomExe.java @@ -7,8 +7,9 @@ import com.red.circle.framework.dto.ResultResponse; import com.red.circle.live.domain.gateway.LiveMicrophoneGateway; import com.red.circle.live.domain.live.LiveMicrophone; import com.red.circle.live.domain.live.LiveMusicStatus; -import com.red.circle.live.infra.database.cache.service.LiveMicCacheService; -import com.red.circle.live.infra.database.cache.service.LiveMusicHeartbeatService; +import com.red.circle.live.infra.database.cache.service.LiveMicCacheService; +import com.red.circle.live.infra.database.cache.service.LiveMusicHeartbeatService; +import com.red.circle.live.infra.database.cache.service.UserAgoraTokenCacheService; import com.red.circle.other.inner.endpoint.live.ActiveVoiceRoomClient; import com.red.circle.other.inner.model.dto.live.ActiveVoiceRoomCO; import com.red.circle.tool.core.collection.CollectionUtils; @@ -33,8 +34,9 @@ public class LiveQuitRoomExe { private final LiveMicCacheService liveMicCacheService; private final LiveMicrophoneGateway liveMicrophoneGateway; private final LiveMusicHeartbeatService liveMusicHeartbeatService; - private final ActiveVoiceRoomClient activeVoiceRoomClient; - private final RedisService redisService; + private final ActiveVoiceRoomClient activeVoiceRoomClient; + private final RedisService redisService; + private final UserAgoraTokenCacheService userAgoraTokenCacheService; private static final String EMPTY_ROOM_CACHE_KEY = "empty_room_cache:"; private static final int CACHE_EXPIRE_MINUTES = 10; @@ -70,6 +72,9 @@ public class LiveQuitRoomExe { liveMusicHeartbeatService.removeMickUser(roomId); } + redisService.delete("IN_ROOM:" + userId); + userAgoraTokenCacheService.remove(userId, roomId); + boolean finalMicChanged = micChanged; ThreadPoolManager.getInstance().execute(() -> { if (finalMicChanged) { 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 0cb4c85f..596264c2 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 @@ -5,11 +5,14 @@ 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.MessageListener; -import com.red.circle.component.redis.service.RedisService; -import com.red.circle.framework.dto.ResultResponse; -import com.red.circle.live.app.dto.cmd.HeartbeatStatusEnum; -import com.red.circle.live.domain.gateway.LiveMicrophoneGateway; +import com.red.circle.component.mq.service.MessageListener; +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.AgoraMicStateCleanupService; +import com.red.circle.live.app.service.AgoraRoomStateService; +import com.red.circle.live.domain.gateway.LiveMicrophoneGateway; import com.red.circle.live.domain.live.LiveMicrophone; import com.red.circle.live.domain.online.LiveRoomUser; import com.red.circle.live.infra.database.cache.service.LiveMicCacheService; @@ -72,8 +75,10 @@ public class UserHeartbeatListener implements MessageListener { private final LiveRoomCacheService liveRoomCacheService; private final LiveMicUserCacheService liveMicUserCacheService; private final RoomDailyTaskClient roomDailyTaskClient; - private final RedisService redisService; - private final RoomMemberClient roomMemberClient; + private final RedisService redisService; + private final RoomMemberClient roomMemberClient; + private final AgoraRoomStateService agoraRoomStateService; + private final AgoraMicStateCleanupService agoraMicStateCleanupService; private static final int HEARTBEAT_INTERVAL_SECONDS = 60; @@ -200,17 +205,25 @@ public class UserHeartbeatListener implements MessageListener { return; } - RoomProfileDTO roomProfile = getRoomProfileDto(roomId); - if (Objects.isNull(roomProfile.getRoomAccount())) { - log.warn("没有找到房间ID: {}", event.getSessionId()); - return; - } - - // 添加房间在线用户 - addRoomOnlineUser(userProfile, roomProfile); - - refreshMicActiveTime(event, roomProfile, userProfile); - } + RoomProfileDTO roomProfile = getRoomProfileDto(roomId); + if (Objects.isNull(roomProfile.getRoomAccount())) { + log.warn("没有找到房间ID: {}", event.getSessionId()); + return; + } + + UserProfileDTO actualUserProfile = ensureUserProfile(userProfile); + AgoraChannelStateDTO channelState = agoraRoomStateService.getChannelState(roomId); + if (channelState.querySuccess() && !isUserInAgoraChannel(channelState, actualUserProfile)) { + agoraMicStateCleanupService.cleanupUser(roomId, actualUserProfile.getId(), + "heartbeat agora user missing"); + return; + } + + // 添加房间在线用户 + addRoomOnlineUser(actualUserProfile, roomProfile); + + refreshMicActiveTime(event, roomProfile, actualUserProfile, channelState); + } private RoomProfileDTO getRoomProfileDto(Long roomId) { String roomAccount = liveRoomCacheService.getRoomAccount(roomId); @@ -258,9 +271,16 @@ public class UserHeartbeatListener implements MessageListener { /** * 刷新麦克风, 用户在线时间. */ - private void refreshMicActiveTime(UserHeartbeatEvent event, - RoomProfileDTO roomProfile, - UserProfileDTO userProfile) { + private void refreshMicActiveTime(UserHeartbeatEvent event, + RoomProfileDTO roomProfile, + UserProfileDTO userProfile) { + refreshMicActiveTime(event, roomProfile, userProfile, null); + } + + private void refreshMicActiveTime(UserHeartbeatEvent event, + RoomProfileDTO roomProfile, + UserProfileDTO userProfile, + AgoraChannelStateDTO checkedChannelState) { // 如果不在麦克风, true&null兼容历史版本 if (Objects.equals(event.getUpMick(), Boolean.FALSE)) { @@ -276,9 +296,33 @@ public class UserHeartbeatListener implements MessageListener { return; } - LiveMicrophone liveMicrophone = liveMicrophones.get(0); - - // 刷新活跃时间 + LiveMicrophone liveMicrophone = liveMicrophones.get(0); + if (checkAccountNotAvailable(userProfile)) { + agoraMicStateCleanupService.cleanupUser(roomProfile.getId(), userProfile.getId(), + "heartbeat user account not available"); + return; + } + + AgoraChannelStateDTO channelState = Objects.nonNull(checkedChannelState) + ? checkedChannelState + : agoraRoomStateService.getChannelState(roomProfile.getId()); + if (!channelState.querySuccess()) { + log.warn("心跳刷新麦位跳过，声网查询失败 roomId={}, userId={}, response={}", + roomProfile.getId(), userProfile.getId(), channelState.getRawResponse()); + liveMicrophoneGateway.refreshNotActiveUser(roomProfile); + return; + } + + Long agoraUid = DataTypeUtils.toLong(userProfile.getAccount()); + if (!channelState.channelExist() || Objects.isNull(agoraUid) + || CollectionUtils.isEmpty(channelState.getBroadcasters()) + || !channelState.getBroadcasters().contains(agoraUid)) { + agoraMicStateCleanupService.cleanupUser(roomProfile.getId(), userProfile.getId(), + "heartbeat agora broadcaster missing"); + return; + } + + // 刷新活跃时间 if (liveMicCacheService.checkLockMic(roomProfile.getId(), liveMicrophone.getMicIndex())) { event.setRetryRefreshMick(true); event.setReqTrackId("retry-" + IdWorkerUtils.getIdStr()); @@ -288,7 +332,39 @@ public class UserHeartbeatListener implements MessageListener { liveMicrophone.updateLastActivityTime(); liveMicCacheService.goUp(roomProfile.getId(), liveMicrophone.getMicIndex(), liveMicrophone); - liveMicrophoneGateway.refreshNotActiveUser(roomProfile); - } - -} + liveMicrophoneGateway.refreshNotActiveUser(roomProfile); + } + + private UserProfileDTO ensureUserProfile(UserProfileDTO userProfile) { + if (Objects.nonNull(userProfile) && StringUtils.isNotBlank(userProfile.getAccount())) { + return userProfile; + } + if (Objects.isNull(userProfile) || Objects.isNull(userProfile.getId())) { + return userProfile; + } + try { + return userProfileClient.getByUserId(userProfile.getId()).getBody(); + } catch (Exception e) { + log.error("获取用户信息失败 userId={}", userProfile.getId(), e); + return userProfile; + } + } + + private boolean isUserInAgoraChannel(AgoraChannelStateDTO channelState, UserProfileDTO userProfile) { + if (Objects.isNull(userProfile) || StringUtils.isBlank(userProfile.getAccount())) { + return false; + } + if (!channelState.channelExist()) { + return false; + } + Long agoraUid = DataTypeUtils.toLong(userProfile.getAccount()); + if (Objects.isNull(agoraUid)) { + return false; + } + return (CollectionUtils.isNotEmpty(channelState.getBroadcasters()) + && channelState.getBroadcasters().contains(agoraUid)) + || (CollectionUtils.isNotEmpty(channelState.getAudience()) + && channelState.getAudience().contains(agoraUid)); + } + +} diff --git a/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/listener/agora/AgoraChannelWebhookImplListener.java b/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/listener/agora/AgoraChannelWebhookImplListener.java index 05b88026..2fb4c462 100644 --- a/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/listener/agora/AgoraChannelWebhookImplListener.java +++ b/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/listener/agora/AgoraChannelWebhookImplListener.java @@ -1,16 +1,25 @@ -package com.red.circle.live.app.listener.agora; - -import com.red.circle.live.app.listener.agora.callback.AgoraChannelBase; -import com.red.circle.live.app.listener.agora.callback.AgoraChannelUserJoin; -import com.red.circle.live.app.listener.agora.callback.AgoraChannelUserLeave; -import com.red.circle.live.app.listener.agora.callback.AgoraChannelUserRoleChange; -import com.red.circle.live.app.listener.agora.callback.AgoraChannelWebhook; -import com.red.circle.live.app.listener.agora.callback.AgoraChannelWebhookListener; -import com.red.circle.other.inner.model.dto.live.RoomProfileDTO; -import com.red.circle.other.inner.model.dto.user.UserProfileDTO; -import lombok.Data; -import lombok.RequiredArgsConstructor; -import lombok.experimental.Accessors; +package com.red.circle.live.app.listener.agora; + +import com.red.circle.component.redis.service.RedisService; +import com.red.circle.live.app.listener.agora.callback.AgoraChannelBase; +import com.red.circle.live.app.listener.agora.callback.AgoraChannelUserJoin; +import com.red.circle.live.app.listener.agora.callback.AgoraChannelUserLeave; +import com.red.circle.live.app.listener.agora.callback.AgoraChannelUserRoleChange; +import com.red.circle.live.app.listener.agora.callback.AgoraChannelWebhook; +import com.red.circle.live.app.listener.agora.callback.AgoraChannelWebhookListener; +import com.red.circle.live.app.service.AgoraMicStateCleanupService; +import com.red.circle.other.inner.endpoint.live.RoomManagerClient; +import com.red.circle.other.inner.endpoint.user.user.UserProfileClient; +import com.red.circle.other.inner.model.dto.live.RoomProfileDTO; +import com.red.circle.other.inner.model.dto.user.UserProfileDTO; +import com.red.circle.tool.core.parse.DataTypeUtils; +import com.red.circle.tool.core.regex.RegexConstant; +import com.red.circle.tool.core.text.StringUtils; +import java.util.Objects; +import java.util.concurrent.TimeUnit; +import lombok.Data; +import lombok.RequiredArgsConstructor; +import lombok.experimental.Accessors; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; @@ -22,24 +31,26 @@ import org.springframework.stereotype.Component; @Slf4j @Component @RequiredArgsConstructor -public class AgoraChannelWebhookImplListener implements AgoraChannelWebhookListener { - -// private final RedisService redisService; -// private final LiveMicCommon liveMicCommon; -// private final UserProfileClient userProfileClient; -// private final RoomManagerClient roomManagerClient; -// private final LiveMicCacheService liveMicCacheService; -// private final LiveMicUserCacheService liveMicUserCacheService; +public class AgoraChannelWebhookImplListener implements AgoraChannelWebhookListener { + + private final RedisService redisService; + private final UserProfileClient userProfileClient; + private final RoomManagerClient roomManagerClient; + private final AgoraMicStateCleanupService agoraMicStateCleanupService; @Override public void channelCreate(AgoraChannelWebhook webhook, AgoraChannelBase body) { } - @Override - public void channelDestroy(AgoraChannelWebhook webhook, AgoraChannelBase body) { - - } + @Override + public void channelDestroy(AgoraChannelWebhook webhook, AgoraChannelBase body) { + Long roomId = getRoomId(body); + if (Objects.isNull(roomId) || !checkProcess(webhook, body)) { + return; + } + agoraMicStateCleanupService.cleanupRoom(roomId, "agora channel destroy"); + } @Override public void broadcasterJoinChannel(AgoraChannelWebhook webhook, AgoraChannelUserJoin body) { @@ -54,10 +65,10 @@ public class AgoraChannelWebhookImplListener implements AgoraChannelWebhookListe // userRoomProfile.getUserProfile().getId()); } - @Override - public void broadcasterLeaveChannel(AgoraChannelWebhook webhook, AgoraChannelUserLeave body) { - // leaveChannelProcess(webhook, body); - } + @Override + public void broadcasterLeaveChannel(AgoraChannelWebhook webhook, AgoraChannelUserLeave body) { + leaveChannelProcess(webhook, body, body.getUid(), "agora broadcaster leave"); + } @Override @@ -65,10 +76,10 @@ public class AgoraChannelWebhookImplListener implements AgoraChannelWebhookListe } - @Override - public void audienceLeaveChannel(AgoraChannelWebhook webhook, AgoraChannelUserLeave body) { - // leaveChannelProcess(webhook, body); - } + @Override + public void audienceLeaveChannel(AgoraChannelWebhook webhook, AgoraChannelUserLeave body) { + leaveChannelProcess(webhook, body, body.getUid(), "agora audience leave"); + } @Override public void clientRoleChangeToBroadcaster(AgoraChannelWebhook webhook, @@ -77,105 +88,74 @@ public class AgoraChannelWebhookImplListener implements AgoraChannelWebhookListe } @Override - public void clientRoleChangeToAudience(AgoraChannelWebhook webhook, - AgoraChannelUserRoleChange body) { - - } - - /** - * 退出频道处理. - */ -// private void leaveChannelProcess(AgoraChannelWebhook webhook, AgoraChannelUserLeave body) { -// -// UserRoomProfile userRoomProfile = getUserRoomProfile(webhook, body, body.getUid()); -// -// if (Objects.isNull(userRoomProfile)) { -// return; -// } -// -// RoomProfileDTO roomProfile = userRoomProfile.getRoomProfile(); -// UserProfileDTO userProfile = userRoomProfile.getUserProfile(); -// Long roomId = roomProfile.getId(); -// -// liveMicCacheService.refreshNotActiveUser(roomId); -// liveMicUserCacheService.refreshNotActiveUser(roomId); -// liveMicUserCacheService.removeUser(roomProfile.getId(), userProfile.getId()); -// -// List liveMicrophones = liveMicCacheService.listLiveMicrophone( -// roomId, userProfile.getId()); -// if (CollectionUtils.isEmpty(liveMicrophones)) { -// return; -// } -// -// LiveMicrophone liveMicrophone = liveMicrophones.get(0); -// -// liveMicCommon.tryMicOpsLock(roomProfile.getId(), liveMicrophone.getMicIndex(), -// () -> { -// log.warn("踢下麦克风: roomId={}, micIndex={}", roomProfile.getId(), -// liveMicrophone.getMicIndex()); -// liveMicCacheService.goDown(roomProfile.getId(), liveMicrophone.getMicIndex()); -// liveMicCommon.micUserChangeNotice(roomProfile); -// }); -// } - -// private Long getRoomId(String channelName) { -// -// if (StringUtils.isBlank(channelName) || !channelName.matches(RegexConstant.NUMBER)) { -// log.warn("错误的频道名称: {}", channelName); -// return null; -// } -// return DataTypeUtils.toLong(channelName); -// } -// -// private UserRoomProfile getUserRoomProfile(AgoraChannelWebhook webhook, AgoraChannelBase body, -// Long uid) { -// if (Objects.isNull(webhook) || Objects.isNull(body)) { -// log.warn("数据错误: {}, {}", -// JacksonUtils.toJson(webhook), -// JacksonUtils.toJson(body) -// ); -// return null; -// } -// -// if (Objects.isNull(uid)) { -// log.warn("数据错误, uid: {}", uid); -// return null; -// } -// -// Long roomId = getRoomId(body.getChannelName()); -// if (Objects.isNull(roomId) || roomId <= 0) { -// return null; -// } -// -// if (!checkProcess(webhook, body.getTs())) { -// return null; -// } -// -// // log.warn("leaveChannelProcess: {}", JacksonUtils.toJson(webhook)); -// -// RoomProfileDTO roomProfile = roomManagerClient.getById(roomId).getBody(); -// if (Objects.isNull(roomProfile)) { -// return null; -// } -// -// UserProfileDTO userProfile = userProfileClient.getByAccount(roomProfile.getSysOrigin(), -// Objects.toString(uid)).getBody(); -// if (Objects.isNull(userProfile)) { -// return null; -// } -// -// return new UserRoomProfile() -// .setRoomProfile(roomProfile) -// .setUserProfile(userProfile); -// } - -// /** -// * true 未处理, false 已处理. -// */ -// private boolean checkProcess(AgoraChannelWebhook webhook, Long ts) { -// return redisService.setIfAbsent("AGORA:" + webhook.getEventType() + ":" + ts, 1, -// TimeUnit.MINUTES); -// } + public void clientRoleChangeToAudience(AgoraChannelWebhook webhook, + AgoraChannelUserRoleChange body) { + leaveChannelProcess(webhook, body, body.getUid(), "agora role change to audience"); + } + + private void leaveChannelProcess(AgoraChannelWebhook webhook, AgoraChannelBase body, Long uid, + String reason) { + UserRoomProfile userRoomProfile = getUserRoomProfile(webhook, body, uid); + if (Objects.isNull(userRoomProfile)) { + return; + } + + agoraMicStateCleanupService.cleanupUser( + userRoomProfile.getRoomProfile().getId(), + userRoomProfile.getUserProfile().getId(), + reason + ); + } + + private Long getRoomId(AgoraChannelBase body) { + if (Objects.isNull(body) || StringUtils.isBlank(body.getChannelName()) + || !body.getChannelName().matches(RegexConstant.NUMBER)) { + log.warn("错误的声网频道名称: {}", Objects.nonNull(body) ? body.getChannelName() : null); + return null; + } + return DataTypeUtils.toLong(body.getChannelName()); + } + + private UserRoomProfile getUserRoomProfile(AgoraChannelWebhook webhook, AgoraChannelBase body, + Long uid) { + Long roomId = getRoomId(body); + if (Objects.isNull(roomId) || roomId <= 0 || Objects.isNull(uid)) { + return null; + } + + if (!checkProcess(webhook, body)) { + return null; + } + + RoomProfileDTO roomProfile = roomManagerClient.getById(roomId).getBody(); + if (Objects.isNull(roomProfile)) { + return null; + } + + UserProfileDTO userProfile = userProfileClient.getByAccount(roomProfile.getSysOrigin(), + Objects.toString(uid)).getBody(); + if (Objects.isNull(userProfile)) { + log.warn("声网回调未找到用户 roomId={}, uid={}", roomId, uid); + return null; + } + + return new UserRoomProfile() + .setRoomProfile(roomProfile) + .setUserProfile(userProfile); + } + + /** + * true 未处理, false 已处理. + */ + private boolean checkProcess(AgoraChannelWebhook webhook, AgoraChannelBase body) { + if (Objects.isNull(webhook)) { + return false; + } + String uniqueKey = StringUtils.isNotBlank(webhook.getNoticeId()) + ? webhook.getNoticeId() + : webhook.getEventType() + ":" + (Objects.nonNull(body) ? body.getTs() : webhook.getNotifyMs()); + return redisService.setIfAbsent("AGORA:" + uniqueKey, 1, 1, TimeUnit.MINUTES); + } @Data 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 new file mode 100644 index 00000000..82ab41d7 --- /dev/null +++ b/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTask.java @@ -0,0 +1,111 @@ +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.AgoraMicStateCleanupService; +import com.red.circle.live.app.service.AgoraRoomStateService; +import com.red.circle.live.domain.live.LiveMicrophone; +import com.red.circle.live.infra.database.cache.service.LiveMicCacheService; +import com.red.circle.tool.core.collection.CollectionUtils; +import com.red.circle.tool.core.parse.DataTypeUtils; +import java.util.HashSet; +import java.util.List; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.core.ScanOptions; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +/** + * 声网真实状态和业务麦位状态补偿对账. + */ +@Slf4j +@Component +@RequiredArgsConstructor +public class AgoraMicStateReconcileTask { + + private static final String LOCK_KEY = "agora:mic:reconcile:lock"; + private static final String SEAT_KEY_PREFIX = "LiveMick:SEAT:"; + + private final RedisService redisService; + private final RedisTemplate redisTemplate; + private final LiveMicCacheService liveMicCacheService; + private final AgoraRoomStateService agoraRoomStateService; + private final AgoraMicStateCleanupService agoraMicStateCleanupService; + + @Scheduled(fixedDelayString = "${red-circle.live.agora-mic-reconcile-delay-ms:30000}") + public void reconcileAgoraMicState() { + if (!redisService.setIfAbsent(LOCK_KEY, 1, 25, TimeUnit.SECONDS)) { + return; + } + try { + for (Long roomId : scanSeatRoomIds()) { + reconcileRoom(roomId); + } + } finally { + redisService.delete(LOCK_KEY); + } + } + + private void reconcileRoom(Long roomId) { + if (Objects.isNull(roomId)) { + return; + } + + AgoraChannelStateDTO state = agoraRoomStateService.getChannelState(roomId); + if (!state.querySuccess()) { + log.warn("声网麦位对账跳过，查询失败 roomId={}, response={}", roomId, state.getRawResponse()); + return; + } + + if (!state.channelExist()) { + agoraMicStateCleanupService.cleanupRoom(roomId, "reconcile agora channel not exist"); + return; + } + + List microphones = liveMicCacheService.listLiveMicrophone(roomId); + if (CollectionUtils.isEmpty(microphones)) { + return; + } + + Set broadcasterUids = new HashSet<>(state.getBroadcasters()); + for (LiveMicrophone microphone : microphones) { + if (Objects.isNull(microphone) || Objects.isNull(microphone.getUser())) { + continue; + } + Long agoraUid = DataTypeUtils.toLong(microphone.getUser().getAccount()); + if (Objects.isNull(agoraUid) || !broadcasterUids.contains(agoraUid)) { + agoraMicStateCleanupService.cleanupUser(roomId, microphone.getUser().getId(), + "reconcile agora broadcaster missing"); + } + } + } + + private Set scanSeatRoomIds() { + Set roomIds = new HashSet<>(); + ScanOptions options = ScanOptions.scanOptions().match(SEAT_KEY_PREFIX + "*").count(200).build(); + try (var cursor = redisTemplate.scan(options)) { + while (cursor.hasNext()) { + Object key = cursor.next(); + Long roomId = parseRoomId(Objects.toString(key, null)); + if (Objects.nonNull(roomId)) { + roomIds.add(roomId); + } + } + } catch (Exception e) { + log.error("扫描声网麦位对账 Redis key 失败", e); + } + return roomIds; + } + + private Long parseRoomId(String key) { + if (Objects.isNull(key) || !key.startsWith(SEAT_KEY_PREFIX)) { + return null; + } + return DataTypeUtils.toLong(key.substring(SEAT_KEY_PREFIX.length())); + } +} diff --git a/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/service/AgoraMicStateCleanupService.java b/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/service/AgoraMicStateCleanupService.java new file mode 100644 index 00000000..4bea9b3b --- /dev/null +++ b/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/service/AgoraMicStateCleanupService.java @@ -0,0 +1,84 @@ +package com.red.circle.live.app.service; + +import com.red.circle.component.redis.service.RedisService; +import com.red.circle.live.domain.gateway.LiveMicrophoneGateway; +import com.red.circle.live.domain.live.LiveMicrophone; +import com.red.circle.live.infra.database.cache.service.LiveMicCacheService; +import com.red.circle.live.infra.database.cache.service.LiveMicUserCacheService; +import com.red.circle.live.infra.database.cache.service.LiveMusicHeartbeatService; +import com.red.circle.live.infra.database.cache.service.UserAgoraTokenCacheService; +import com.red.circle.tool.core.collection.CollectionUtils; +import java.util.List; +import java.util.Objects; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; + +/** + * 声网状态反向清理业务麦位和房间在线缓存. + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class AgoraMicStateCleanupService { + + private final LiveMicCacheService liveMicCacheService; + private final LiveMicUserCacheService liveMicUserCacheService; + private final LiveMusicHeartbeatService liveMusicHeartbeatService; + private final LiveMicrophoneGateway liveMicrophoneGateway; + private final UserAgoraTokenCacheService userAgoraTokenCacheService; + private final RedisService redisService; + + public boolean cleanupUser(Long roomId, Long userId, String reason) { + if (Objects.isNull(roomId) || Objects.isNull(userId)) { + return false; + } + + boolean micChanged = false; + List liveMicrophones = liveMicCacheService.listLiveMicrophone(roomId, userId); + if (CollectionUtils.isNotEmpty(liveMicrophones)) { + for (LiveMicrophone mic : liveMicrophones) { + if (Objects.isNull(mic) || Objects.isNull(mic.getMicIndex())) { + continue; + } + final boolean[] removed = {false}; + liveMicrophoneGateway.tryMicOpsLock(roomId, mic.getMicIndex(), () -> { + liveMicCacheService.goDown(roomId, mic.getMicIndex()); + removed[0] = true; + }); + micChanged = micChanged || removed[0]; + } + } + + liveMicUserCacheService.removeUser(roomId, userId); + redisService.delete("IN_ROOM:" + userId); + userAgoraTokenCacheService.remove(userId, roomId); + + if (micChanged) { + notifyMicChanged(roomId, reason); + } + + log.warn("声网状态清理用户 roomId={}, userId={}, micChanged={}, reason={}", + roomId, userId, micChanged, reason); + return micChanged; + } + + public void cleanupRoom(Long roomId, String reason) { + if (Objects.isNull(roomId)) { + return; + } + liveMicCacheService.goDownAll(roomId); + liveMicUserCacheService.remove(roomId); + liveMusicHeartbeatService.removeMickUser(roomId); + notifyMicChanged(roomId, reason); + log.warn("声网状态清理房间 roomId={}, reason={}", roomId, reason); + } + + private void notifyMicChanged(Long roomId, String reason) { + try { + liveMicrophoneGateway.micUserChangeNotice(roomId); + } catch (Exception e) { + log.error("声网状态清理后推送麦位变更失败 roomId={}, reason={}", roomId, reason, e); + } + } +} diff --git a/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/service/AgoraRoomStateService.java b/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/service/AgoraRoomStateService.java new file mode 100644 index 00000000..f0bef77d --- /dev/null +++ b/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/service/AgoraRoomStateService.java @@ -0,0 +1,65 @@ +package com.red.circle.live.app.service; + +import com.red.circle.external.inner.endpoint.message.AgoraClient; +import com.red.circle.external.inner.model.dto.AgoraChannelStateDTO; +import com.red.circle.framework.dto.ResultResponse; +import com.red.circle.tool.core.collection.CollectionUtils; +import java.util.Objects; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; + +/** + * 声网房间状态查询. + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class AgoraRoomStateService { + + private final AgoraClient agoraClient; + + public AgoraChannelStateDTO getChannelState(Long roomId) { + if (Objects.isNull(roomId)) { + return queryFailed("roomId is null"); + } + try { + ResultResponse response = agoraClient.channelState(roomId.toString()); + if (Objects.isNull(response) || !response.checkSuccess()) { + return queryFailed("response not success"); + } + AgoraChannelStateDTO body = response.getBody(); + if (Objects.isNull(body)) { + return queryFailed("body is null"); + } + return body; + } catch (Exception e) { + log.error("查询声网频道状态异常 roomId={}", roomId, e); + return queryFailed(e.getMessage()); + } + } + + public boolean isBroadcasterInChannel(Long roomId, String account) { + AgoraChannelStateDTO state = getChannelState(roomId); + if (!state.querySuccess() || !state.channelExist() || CollectionUtils.isEmpty(state.getBroadcasters())) { + return false; + } + Long agoraUid = parseAccount(account); + return Objects.nonNull(agoraUid) && state.getBroadcasters().contains(agoraUid); + } + + private Long parseAccount(String account) { + try { + return Long.valueOf(account); + } catch (Exception e) { + return null; + } + } + + private AgoraChannelStateDTO queryFailed(String reason) { + return new AgoraChannelStateDTO() + .setQuerySuccess(Boolean.FALSE) + .setChannelExist(Boolean.FALSE) + .setRawResponse(reason); + } +} diff --git a/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/listener/gift/GiveGiftsListener.java b/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/listener/gift/GiveGiftsListener.java index be6c7ff2..c26c7593 100644 --- a/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/listener/gift/GiveGiftsListener.java +++ b/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/listener/gift/GiveGiftsListener.java @@ -8,11 +8,12 @@ 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.MessageListener; -import com.red.circle.external.inner.endpoint.message.ImGroupClient; -import com.red.circle.external.inner.model.cmd.message.CustomGroupMsgBodyCmd; -import com.red.circle.external.inner.model.enums.message.GroupMessageTypeEnum; +import com.red.circle.component.mq.service.ConsumerMessage; +import com.red.circle.component.mq.service.MessageListener; +import com.red.circle.external.inner.endpoint.message.ImGroupClient; +import com.red.circle.external.inner.model.cmd.message.BroadcastGroupMsgBodyCmd; +import com.red.circle.external.inner.model.cmd.message.CustomGroupMsgBodyCmd; +import com.red.circle.external.inner.model.enums.message.GroupMessageTypeEnum; import com.red.circle.framework.core.dto.ReqSysOrigin; import com.red.circle.framework.web.util.ValidatorUtils; import com.red.circle.live.inner.endpoint.api.LiveHeartbeatApi; @@ -61,9 +62,10 @@ import com.red.circle.other.inner.enums.team.TeamPolicyTypeEnum; import com.red.circle.other.inner.model.dto.activity.props.ActivityRewardPropsDTO; import com.red.circle.other.inner.model.dto.user.UserProfileDTO; import com.red.circle.tool.core.collection.CollectionUtils; -import com.red.circle.tool.core.date.TimestampUtils; -import com.red.circle.tool.core.json.JacksonUtils; -import com.red.circle.tool.core.sequence.IdWorkerUtils; +import com.red.circle.tool.core.date.TimestampUtils; +import com.red.circle.tool.core.json.JacksonUtils; +import com.red.circle.tool.core.sequence.IdWorkerUtils; +import com.red.circle.tool.core.text.StringUtils; import java.math.BigDecimal; import java.math.RoundingMode; @@ -372,14 +374,16 @@ public class GiveGiftsListener implements MessageListener { Pair runningWaterBooleanPair = saveRunningWater(event); GiftGiveRunningWater runningWater = runningWaterBooleanPair.getFirst(); - if (!runningWaterBooleanPair.getSecond()) { - log.warn("保存礼物流水失败:{}", runningWater); - return; - } - - if (!SysOriginPlatformEnum.isVoiceSystem(runningWater.getSysOrigin())) { - return; - } + if (!runningWaterBooleanPair.getSecond()) { + log.warn("保存礼物流水失败:{}", runningWater); + return; + } + + sendNoticeGiftBroadcast(event, runningWater); + + if (!SysOriginPlatformEnum.isVoiceSystem(runningWater.getSysOrigin())) { + return; + } // 不是祝福礼物 if (Boolean.FALSE.equals(runningWater.getCheckBlessingsGift())) { @@ -400,11 +404,56 @@ public class GiveGiftsListener implements MessageListener { // 排行榜相关 offlineProcessGiftEvent(OfflineStatisticsGiftEventType.RANK_COUNT, runningWater); - // 火箭统计相关 - offlineProcessGiftEvent(OfflineStatisticsGiftEventType.ROCKET_COUNT, runningWater); - } - - private Pair saveRunningWater(GiveAwayGiftBatchEvent event) { + // 火箭统计相关 + offlineProcessGiftEvent(OfflineStatisticsGiftEventType.ROCKET_COUNT, runningWater); + } + + void sendNoticeGiftBroadcast(GiveAwayGiftBatchEvent event, + GiftGiveRunningWater runningWater) { + if (Objects.isNull(event) || Objects.isNull(event.getGiftConfig()) + || Objects.isNull(runningWater)) { + return; + } + if (event.checkLuckyGift()) { + return; + } + String special = event.getGiftConfig().getSpecial(); + if (StringUtils.isBlank(special) || !special.contains(GiftSpecialEnum.NOTICE.name())) { + return; + } + + RoomProfile roomProfile = Objects.nonNull(event.getRoomId()) + ? roomProfileManagerService.getProfileById(event.getRoomId()) + : null; + SendGiftNotifyCO notify = new SendGiftNotifyCO() + .setGlobalNews(Boolean.TRUE) + .setSysOrigin(event.getSysOrigin().name()) + .setRoomId(event.getRoomId()) + .setRoomAccount(Objects.nonNull(roomProfile) ? roomProfile.getRoomAccount() : null) + .setSendUserId(event.getSendUserId()) + .setAcceptUserIds(event.getAccepts().stream() + .map(GiveAwayGiftRoomAccepts::getAcceptUserId) + .filter(Objects::nonNull) + .distinct() + .collect(Collectors.toList())) + .setGiftId(event.getGiftConfig().getId()) + .setGiftQuantity(event.getQuantity()) + .setGiftCandy(event.getGiftConfig().getGiftCandy()) + .setActualAmount(runningWater.getGiftValue().getActualAmount()) + .setGiftType(runningWater.getGiftValue().getGiftType()); + + imGroupClient.sendMessageBroadcast( + BroadcastGroupMsgBodyCmd.builder() + .toPlatform(event.getSysOrigin()) + .type(GroupMessageTypeEnum.SEND_GIFT) + .data(notify) + .build() + ); + log.warn("NOTICE礼物全服通报成功, trackId={}, giftId={}, sendUserId={}", + event.getTrackId(), event.getGiftConfig().getId(), event.getSendUserId()); + } + + private Pair saveRunningWater(GiveAwayGiftBatchEvent event) { List anchorUserIds = teamMemberService.listMemberIds(event.allUserId()); //直播间送礼后的心动值 diff --git a/rc-service/rc-service-other/other-application/src/test/java/com/red/circle/other/app/listener/gift/GiveGiftsListenerTest.java b/rc-service/rc-service-other/other-application/src/test/java/com/red/circle/other/app/listener/gift/GiveGiftsListenerTest.java new file mode 100644 index 00000000..31384b11 --- /dev/null +++ b/rc-service/rc-service-other/other-application/src/test/java/com/red/circle/other/app/listener/gift/GiveGiftsListenerTest.java @@ -0,0 +1,141 @@ +package com.red.circle.other.app.listener.gift; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.red.circle.common.business.core.enums.SysOriginPlatformEnum; +import com.red.circle.component.mq.MessageEventProcess; +import com.red.circle.external.inner.endpoint.message.ImGroupClient; +import com.red.circle.external.inner.model.cmd.message.BroadcastGroupMsgBodyCmd; +import com.red.circle.external.inner.model.enums.message.GroupMessageTypeEnum; +import com.red.circle.mq.business.model.event.gift.GiveAwayGiftBatchEvent; +import com.red.circle.mq.business.model.event.gift.GiveAwayGiftRoomAccepts; +import com.red.circle.mq.business.model.event.gift.GiveGiftConfig; +import com.red.circle.mq.rocket.business.producer.GameMqMessage; +import com.red.circle.mq.rocket.business.producer.GiftMqMessage; +import com.red.circle.other.app.command.game.barrage.BarrageGameBaishunExe; +import com.red.circle.other.app.command.game.burstcrystal.GameBurstCrystalScheduleTwoFunQueryExe; +import com.red.circle.other.app.command.game.roompk.GameRoomPkUserCmdExe; +import com.red.circle.other.app.command.game.teampk.GameTeamPkBaseInfoCmdExe; +import com.red.circle.other.app.common.gift.GameLuckyGiftCommon; +import com.red.circle.other.app.dto.clientobject.gift.SendGiftNotifyCO; +import com.red.circle.other.domain.gateway.user.ability.UserRegionGateway; +import com.red.circle.other.infra.database.cache.service.other.EnumConfigCacheService; +import com.red.circle.other.infra.database.cache.service.other.LiveHeartbeatCacheService; +import com.red.circle.other.infra.database.mongo.entity.gift.GiftGiveRunningWater; +import com.red.circle.other.infra.database.mongo.entity.gift.GiftValue; +import com.red.circle.other.infra.database.mongo.entity.live.RoomProfile; +import com.red.circle.other.infra.database.mongo.service.gift.GiftCountAnchorDailyRunningWaterService; +import com.red.circle.other.infra.database.mongo.service.gift.GiftGiveRunningWaterService; +import com.red.circle.other.infra.database.mongo.service.live.RoomProfileManagerService; +import com.red.circle.other.infra.database.mongo.service.team.team.TeamBillCycleService; +import com.red.circle.other.infra.database.mongo.service.team.team.TeamMemberService; +import com.red.circle.other.infra.database.mongo.service.team.team.TeamPolicyManagerService; +import com.red.circle.other.inner.enums.material.GiftSpecialEnum; +import java.math.BigDecimal; +import java.util.List; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +class GiveGiftsListenerTest { + + private final ImGroupClient imGroupClient = mock(ImGroupClient.class); + private final RoomProfileManagerService roomProfileManagerService = mock( + RoomProfileManagerService.class); + private final GiveGiftsListener listener = new GiveGiftsListener( + mock(GameMqMessage.class), + mock(GiftMqMessage.class), + mock(TeamMemberService.class), + mock(UserRegionGateway.class), + mock(MessageEventProcess.class), + mock(GameLuckyGiftCommon.class), + mock(TeamBillCycleService.class), + mock(TeamPolicyManagerService.class), + mock(EnumConfigCacheService.class), + mock(LiveHeartbeatCacheService.class), + mock(GiftGiveRunningWaterService.class), + imGroupClient, + roomProfileManagerService, + mock(GameTeamPkBaseInfoCmdExe.class), + mock(GameRoomPkUserCmdExe.class), + mock(BarrageGameBaishunExe.class), + mock(GameBurstCrystalScheduleTwoFunQueryExe.class), + mock(GiftCountAnchorDailyRunningWaterService.class) + ); + + @Test + void sendNoticeGiftBroadcastShouldBroadcastNoticeGift() { + RoomProfile roomProfile = new RoomProfile(); + roomProfile.setRoomAccount("1002001"); + when(roomProfileManagerService.getProfileById(2001L)) + .thenReturn(roomProfile); + + listener.sendNoticeGiftBroadcast( + event(GiftSpecialEnum.NOTICE.name(), "GOLD"), + runningWater() + ); + + ArgumentCaptor captor = ArgumentCaptor.forClass( + BroadcastGroupMsgBodyCmd.class); + verify(imGroupClient).sendMessageBroadcast(captor.capture()); + + BroadcastGroupMsgBodyCmd message = captor.getValue(); + assertEquals(GroupMessageTypeEnum.SEND_GIFT.name(), message.getType()); + assertEquals(SysOriginPlatformEnum.LIKEI, message.getSysOrigin()); + + SendGiftNotifyCO data = (SendGiftNotifyCO) message.getData(); + assertTrue(data.getGlobalNews()); + assertEquals(2001L, data.getRoomId()); + assertEquals("1002001", data.getRoomAccount()); + assertEquals(3001L, data.getSendUserId()); + assertEquals(List.of(4001L), data.getAcceptUserIds()); + assertEquals(5001L, data.getGiftId()); + assertEquals(2, data.getGiftQuantity()); + assertEquals(new BigDecimal("20"), data.getGiftCandy()); + assertEquals(new BigDecimal("40"), data.getActualAmount()); + assertEquals("NORMAL", data.getGiftType()); + } + + @Test + void sendNoticeGiftBroadcastShouldIgnoreNonNoticeGift() { + listener.sendNoticeGiftBroadcast(event("ANIMATION", "GOLD"), runningWater()); + + verify(imGroupClient, never()).sendMessageBroadcast(any()); + } + + @Test + void sendNoticeGiftBroadcastShouldIgnoreLuckyGift() { + listener.sendNoticeGiftBroadcast(event(GiftSpecialEnum.NOTICE.name(), "LUCKY_GIFT"), + runningWater()); + + verify(imGroupClient, never()).sendMessageBroadcast(any()); + } + + private GiveAwayGiftBatchEvent event(String special, String giftTab) { + return new GiveAwayGiftBatchEvent() + .setSysOrigin(SysOriginPlatformEnum.LIKEI) + .setRoomId(2001L) + .setSendUserId(3001L) + .setAccepts(List.of(new GiveAwayGiftRoomAccepts().setAcceptUserId(4001L))) + .setQuantity(2) + .setTrackId(6001L) + .setGiftConfig(new GiveGiftConfig() + .setId(5001L) + .setSpecial(special) + .setType("GOLD") + .setGiftTab(giftTab) + .setGiftCandy(new BigDecimal("20"))); + } + + private GiftGiveRunningWater runningWater() { + return new GiftGiveRunningWater() + .setGiftValue(new GiftValue() + .setGiftType("NORMAL") + .setActualAmount(new BigDecimal("40"))); + } +} diff --git a/rc-service/rc-service-other/other-client/src/main/java/com/red/circle/other/app/dto/clientobject/gift/SendGiftNotifyCO.java b/rc-service/rc-service-other/other-client/src/main/java/com/red/circle/other/app/dto/clientobject/gift/SendGiftNotifyCO.java index 29086615..ef69efc4 100644 --- a/rc-service/rc-service-other/other-client/src/main/java/com/red/circle/other/app/dto/clientobject/gift/SendGiftNotifyCO.java +++ b/rc-service/rc-service-other/other-client/src/main/java/com/red/circle/other/app/dto/clientobject/gift/SendGiftNotifyCO.java @@ -1,12 +1,15 @@ -package com.red.circle.other.app.dto.clientobject.gift; - -import com.red.circle.other.app.dto.clientobject.game.GameBurstCrystalScheduleCO; -import com.red.circle.other.app.dto.clientobject.game.GameIndoorTeamPkCO; -import com.red.circle.other.app.dto.clientobject.game.GameRoomPkUserCO; -import lombok.Data; -import lombok.experimental.Accessors; - -import java.util.List; +package com.red.circle.other.app.dto.clientobject.gift; + +import com.fasterxml.jackson.databind.annotation.JsonSerialize; +import com.fasterxml.jackson.databind.ser.std.ToStringSerializer; +import com.red.circle.other.app.dto.clientobject.game.GameBurstCrystalScheduleCO; +import com.red.circle.other.app.dto.clientobject.game.GameIndoorTeamPkCO; +import com.red.circle.other.app.dto.clientobject.game.GameRoomPkUserCO; +import java.math.BigDecimal; +import lombok.Data; +import lombok.experimental.Accessors; + +import java.util.List; /** * 发送礼物通知. @@ -15,11 +18,69 @@ import java.util.List; */ @Data @Accessors(chain = true) -public class SendGiftNotifyCO { - - /** - * PK详细信息 - */ +public class SendGiftNotifyCO { + + /** + * 是否全局通报. + */ + private Boolean globalNews; + + /** + * 系统. + */ + private String sysOrigin; + + /** + * 房间ID. + */ + @JsonSerialize(using = ToStringSerializer.class) + private Long roomId; + + /** + * 房间账号. + */ + private String roomAccount; + + /** + * 送礼物人用户id. + */ + @JsonSerialize(using = ToStringSerializer.class) + private Long sendUserId; + + /** + * 接收用户ID集合. + */ + private List acceptUserIds; + + /** + * 礼物ID. + */ + @JsonSerialize(using = ToStringSerializer.class) + private Long giftId; + + /** + * 礼物数量. + */ + private Integer giftQuantity; + + /** + * 礼物价值/糖果. + */ + private BigDecimal giftCandy; + + /** + * 本次实际消费. + */ + private BigDecimal actualAmount; + + /** + * 礼物类型. + */ + private String giftType; + + /** + * PK详细信息 + */ private GameIndoorTeamPkCO gameIndoorTeamPk; /**