shayu/cai-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

123

Browse Source
This commit is contained in:
77
2024-05-21 11:19:46 +08:00
parent 5dc1ca2521
commit 640aa3237e
13 changed files with 101 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
ruoyi-websocket-boot/src/main/java/com/ruoyi/consumer/RedisConsumer.java
View File

@@ -26,9 +26,7 @@ public class RedisConsumer {
executorService.execute(() -> {
while (true){
try {
System.out.println();
run();
System.out.println();
}catch (Exception e){
}

3
ruoyi-websocket-boot/src/main/java/com/ruoyi/consumer/RoomSettleDelayMqConsumer.java
View File

@@ -23,7 +23,10 @@ public class RoomSettleDelayMqConsumer {
boolean next = settleService.withholdingFee(Long.valueOf(message));
if(next){
// 1分钟后继续执行
log.info("预扣分完成 1分钟后继续后执行 {}", message);
amqpWsProducer.sendRoomSettleDelay(message,60);
}else{
log.info("预扣费失败 可能余额不足 不在继续执行 {}", message);
}
}catch (Exception e){
log.error("每分钟定时扣费失败!",e);

29
ruoyi-websocket-boot/src/main/java/com/ruoyi/service/RoomCheckJobService.java
View File

@@ -5,7 +5,9 @@ import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.ruoyi.cai.domain.UserCall;
import com.ruoyi.cai.notice.YunxinWsServiceV2;
import com.ruoyi.cai.service.AnchorService;
import com.ruoyi.cai.service.UserCallService;
import com.ruoyi.cai.service.UserService;
import com.ruoyi.cai.ws.bean.Room;
import com.ruoyi.cai.ws.bean.UserData;
import com.ruoyi.cai.ws.cache.RoomCtxCache;
@@ -90,12 +92,15 @@ public class RoomCheckJobService {
UserData receiverUserData = room.getReceiverUserData();
boolean timeOut = false;
WsR hangup = null;
String logMessage = null;
if(isHeartTimeout(callUserData)){
timeOut = true;
hangup = WsRMsgGen.hangup("呼叫方连接中断", roomId, HangUpEnums.FROM.getCode());
logMessage = "呼叫方连接中断";
}else if(isHeartTimeout(receiverUserData)){
timeOut = true;
hangup = WsRMsgGen.hangup("接听方连接中断", roomId, HangUpEnums.TO.getCode());
logMessage = "接听方连接中断";
}
if(timeOut){
boolean nextCreateJob = false;
@@ -109,6 +114,8 @@ public class RoomCheckJobService {
List<String> keys = roomCtxCache.getSessionKeysByRoomId(roomId);
RoomWebSocketUtil.sendSendMessage(keys, hangup);
roomService.closeAllFd(roomId);
updateAnchorVideoStatus(room);
log.info("roomId={} 心跳检测发现{} 关闭房间",roomId,logMessage);
return JobResp.builder().nextCreateJob(nextCreateJob).build();
}
return JobResp.builder().nextCreateJob(true).build();
@@ -148,6 +155,8 @@ public class RoomCheckJobService {
if(!nextCreateJob){
yunxinWsService.sendToCallNotify(callUserData.getId(),receiverUserData.getId(), CallNoticeEnum.TIMEOUT,0L);
}
updateAnchorVideoStatus(room);
log.info("roomId={} 主播两分钟没有接听， 房间关闭 ",room.getRoomId());
return JobResp.builder().nextCreateJob(nextCreateJob).build();
}
return JobResp.builder().nextCreateJob(true).build();
@@ -172,6 +181,26 @@ public class RoomCheckJobService {
return true;
}
@Autowired
private UserService userService;
@Autowired
private AnchorService anchorService;
private void updateAnchorVideoStatus(Room room){
try {
// 释放主播的接听状态
UserData receiverUserData = room.getReceiverUserData();
if(receiverUserData != null && receiverUserData.getConnectTime() != null && receiverUserData.getConnectTime() > 0){
userService.updateVideoStatus(receiverUserData.getId(),0);
anchorService.updateVideoStatus(receiverUserData.getId(),0);
}
log.info("roomId={} 主播状态释放完毕",room.getRoomId());
}catch (Exception e){
log.info("roomId={} 主播状态释放失败",room.getRoomId(), e);
}
}