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

init

Browse Source
This commit is contained in:
dute7liang
2024-01-11 23:38:56 +08:00
parent f7485b9066
commit 728192d4f0
21 changed files with 231 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/bean/RoomData.java
View File

@@ -13,11 +13,11 @@ public class RoomData {
private BigDecimal videoDivide;
private Long payCoin = 0L;
private Long payIncome = 0L;
private Long hangUpTime;
private Long hangUpTime; // 掉线时间
private Long settleTime;
private Long settleTime; // 结算时间
private Long beginTime;
private Long beginTime; // 开始时间
private boolean releaseRes = false;
}

1
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/bean/UserData.java
View File

@@ -12,4 +12,5 @@ public class UserData {
private String nickname;
private String userCode;
private Long connectTime;
private Long heartTime;
}

7
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/cache/FdCtxDataCache.java vendored
View File

@@ -11,6 +11,13 @@ import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;
/**
* key ：sessionKey
* value ： 当前用户信息
* 一个call 有两个redis
* <p>created on 2024/1/11 22:44</p>
* @author duet
*/
@Component
public class FdCtxDataCache {
@Autowired

8
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/cache/RoomCtxCache.java vendored
View File

@@ -12,6 +12,12 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
/**
* key : 房间号
* value： { sessionKey ： 用户类型 }
* <p>created on 2024/1/11 22:45</p>
* @author duet
*/
@Component
public class RoomCtxCache {
@Autowired
@@ -23,7 +29,7 @@ public class RoomCtxCache {
public void addFd(String sessionKey,Long roomId,Integer userType){
String key = getKey(roomId);
redisTemplate.opsForHash().putIfAbsent(key,sessionKey,userType);
redisTemplate.opsForHash().put(key,sessionKey,userType);
redisTemplate.expire(key,7, TimeUnit.DAYS);
}

6
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/cache/RoomDataCache.java vendored
View File

@@ -20,6 +20,12 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.Map;
/**
* 房间信息
* 在拨打电话的时候就已经提前初始化
* <p>created on 2024/1/11 22:49</p>
* @author duet
*/
@Component
public class RoomDataCache {
@Autowired

5
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/cache/UserDataCache.java vendored
View File

@@ -10,6 +10,11 @@ import org.springframework.stereotype.Component;
import java.util.Map;
/**
* 房间的用户信息 拨打电话前就已经创建好
* <p>created on 2024/1/11 22:50</p>
* @author duet
*/
@Component
public class UserDataCache {
@Autowired

2
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/config/WebSocketConfig.java
View File

@@ -28,11 +28,9 @@ public class WebSocketConfig {
if (StrUtil.isBlank(webSocketProperties.getPath())) {
webSocketProperties.setPath("/ws");
}
if (StrUtil.isBlank(webSocketProperties.getAllowedOrigins())) {
webSocketProperties.setAllowedOrigins("*");
}
return registry -> registry
.addHandler(webSocketHandler, webSocketProperties.getPath())
.addInterceptors(handshakeInterceptor)

9
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/dto/WsToken.java
View File

@@ -7,4 +7,13 @@ public class WsToken {
private Long roomId;
private Long fromUid;
private Long toUid;
private Long userId;
public boolean isCall(){
return userId.equals(fromUid);
}
public boolean isReceive(){
return userId.equals(toUid);
}
}

11
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/handler/RoomWebSocketHandler.java
View File

@@ -1,15 +1,18 @@
package com.ruoyi.cai.ws.handler;
import com.ruoyi.cai.chat.ChatManager;
import com.ruoyi.cai.ws.constant.WebSocketConstants;
import com.ruoyi.cai.ws.holder.WebSocketSessionHolder;
import com.ruoyi.cai.ws.processon.OpenLogic;
import com.ruoyi.cai.ws.util.WebSocketUtils;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.*;
import org.springframework.web.socket.handler.AbstractWebSocketHandler;
import javax.websocket.server.PathParam;
import java.util.Map;
/**
@@ -29,13 +32,9 @@ public class RoomWebSocketHandler extends AbstractWebSocketHandler {
* 连接成功后
*/
@Override
public void afterConnectionEstablished(WebSocketSession session) {
Map<String, Object> attributes = session.getAttributes();
if(attributes.get("token") != null){
WebSocketSessionHolder.addSession(attributes.get("token").toString(), session);
}
public void afterConnectionEstablished(@NotNull WebSocketSession session) {
openLogic.processOn(session);
log.info("[connect] sessionId: {},userId:{}", session.getId(), session.getId());
// log.info("[connect] sessionId: {},userId:{}", session.getId(), session.getId());
}
/**

8
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/handler/message/GiftMessageHandler.java
View File

@@ -3,7 +3,9 @@ package com.ruoyi.cai.ws.handler.message;
import com.alibaba.fastjson2.JSONObject;
import com.ruoyi.cai.domain.Account;
import com.ruoyi.cai.domain.Gift;
import com.ruoyi.cai.dto.ConsumeResp;
import com.ruoyi.cai.dto.app.query.GiveGiftRes;
import com.ruoyi.cai.manager.ConsumerManager;
import com.ruoyi.cai.service.AccountService;
import com.ruoyi.cai.service.GiftService;
import com.ruoyi.cai.service.UserGiftService;
@@ -23,6 +25,8 @@ public class GiftMessageHandler extends AbstractMessageHandle implements IMessag
private AccountService accountService;
@Autowired
private UserGiftService userGiftService;
@Autowired
private ConsumerManager consumerManager;
@Override
public void processOn(Room room, FdCtxData fdCtxData, JSONObject map) {
Long giftId = map.getLong("giftId");
@@ -50,8 +54,8 @@ public class GiftMessageHandler extends AbstractMessageHandle implements IMessag
giveGiftRes.setToUserId(fdCtxData.getTarUserId());
giveGiftRes.setGiftId(giftId);
giveGiftRes.setGiftCount(giftCount);
boolean b = userGiftService.giveGift(giveGiftRes);
if(!b){
ConsumeResp consumeResp = consumerManager.sendGift(giveGiftRes);
if(!consumeResp.isSuccess()){
sendToCurrent(fdCtxData,WsRMsgGen.sysNotice("赠送失败，请重试"));
return;
}

2
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/manager/WebSocketManager.java
View File

@@ -97,8 +97,6 @@ public class WebSocketManager {
// 检查女神在线状态 TODO
// 关闭发起的其他房间
roomService.closeAllRoom(call.getFromUid());
// 删除旧房间记录
roomService.delCallRoom(call.getFromUid());

88
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/processon/OpenLogic.java
View File

@@ -1,10 +1,11 @@
package com.ruoyi.cai.ws.processon;
import cn.hutool.core.date.DateUtil;
import com.ruoyi.cai.trd.Agora;
import com.ruoyi.cai.chat.ChatManager;
import com.ruoyi.cai.executor.ExecutorConstant;
import com.ruoyi.cai.service.AnchorService;
import com.ruoyi.cai.service.UserService;
import com.ruoyi.cai.trd.Agora;
import com.ruoyi.cai.ws.bean.FdCtxData;
import com.ruoyi.cai.ws.bean.Room;
import com.ruoyi.cai.ws.cache.*;
@@ -12,16 +13,21 @@ import com.ruoyi.cai.ws.constant.HangUpEnums;
import com.ruoyi.cai.ws.constant.RoomStatusEnums;
import com.ruoyi.cai.ws.constant.UserDataConstant;
import com.ruoyi.cai.ws.dto.WsRMsgGen;
import com.ruoyi.cai.ws.dto.WsToken;
import com.ruoyi.cai.ws.holder.WebSocketSessionHolder;
import com.ruoyi.cai.ws.service.CheckConnectionDTO;
import com.ruoyi.cai.ws.service.RoomService;
import com.ruoyi.cai.ws.util.MapGetUtil;
import com.ruoyi.cai.ws.util.RoomWebSocketUtil;
import com.ruoyi.cai.ws.util.TimeConverter;
import com.ruoyi.cai.ws.util.WsExceptionUtil;
import com.ruoyi.yunxin.Yunxin;
import com.ruoyi.yunxin.data.SendAttachMsgData;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.WebSocketSession;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -46,37 +52,39 @@ public class OpenLogic {
private AnchorService anchorService;
@Autowired
private Agora agora;
@Autowired
private ChatManager chatManager;
@Autowired
private Yunxin yunxin;
public void processOn(WebSocketSession session) {
Map<String, Object> map = session.getAttributes();
String token = MapGetUtil.getString(map.get("token"));
Long roomId = MapGetUtil.getLong(map.get("roomId"));
Long userId = MapGetUtil.getLong(map.get("userId"));
// 校验token
process(token,roomId,userId);
}
public void process(String sessionKey,Long roomId,Long userId){
public void processOn(WebSocketSession session) throws IOException {
Map<String, Object> attributes = session.getAttributes();
Object token = attributes.get("token");
if(token == null){
WsExceptionUtil.throwExceptionFast(session,"参数异常");
return;
}
WsToken wsToken = chatManager.getToken(String.valueOf(token));
if(wsToken == null){
WsExceptionUtil.throwExceptionFast(session,"无效token");
return;
}
WebSocketSessionHolder.addSession(session.getId(), session);
Long roomId = wsToken.getRoomId();
Long userId = wsToken.getUserId();
Room room = roomService.load(roomId);
if(room == null || (room.getCallUserData().getId().equals(userId) && room.getReceiverUserData().getId().equals(userId))){
WsExceptionUtil.throwException(sessionKey,"房间不可用", HangUpEnums.OTHER,roomId);
WsExceptionUtil.throwException(session,"房间不可用", HangUpEnums.OTHER,roomId);
return;
}
CheckConnectionDTO checkConnect = roomService.checkConnect(room);
if(checkConnect != null){
WsExceptionUtil.throwException(sessionKey,checkConnect.getMessage(),checkConnect.getHangUpEnums(),roomId);
WsExceptionUtil.throwException(session,checkConnect.getMessage(),checkConnect.getHangUpEnums(),roomId);
return;
}
if(userId.equals(room.getCallUserData().getId())){
// 走主叫方逻辑
callerConnection(sessionKey,room,userId);
}else {
// 走接收方逻辑
receiverConnection(sessionKey,room,userId);
}
// 保存上下文到redis
FdCtxData fdCtxData = new FdCtxData();
fdCtxData.setSessionKey(sessionKey);
fdCtxData.setSessionKey(session.getId());
fdCtxData.setRoomId(room.getRoomId());
fdCtxData.setUserId(userId);
if(userId.equals(room.getCallUserData().getId())){
@@ -89,10 +97,18 @@ public class OpenLogic {
fdCtxData.setTarUserType(UserDataConstant.TYPE_CALLER);
}
fdCtxDataCache.save(fdCtxData);
roomCtxCache.addFd(sessionKey,roomId,fdCtxData.getUserType());
roomCtxCache.addFd(session.getId(),roomId,fdCtxData.getUserType());
if(userId.equals(room.getCallUserData().getId())){
// 走主叫方逻辑
callerConnection(session,room,userId);
}else {
// 走接收方逻辑
receiverConnection(session,room,userId);
}
}
public void callerConnection(String sessionKey,Room room,Long userId){
public void callerConnection(WebSocketSession session,Room room,Long userId){
boolean isFirst = false;
Integer status = room.getStatus();
// 首次链接
@@ -110,17 +126,18 @@ public class OpenLogic {
// 已经接通
if(room.isOnline()){
ExecutorConstant.ROOM_EXECUTOR.execute(() -> {
// 已经接通，看一下是否掉线了，为重连
// 如果视频也掉线了，则重连的时候发送消息提示
List<String> channelUsers = agora.getChannelUsers(room.getRoomId());
if(channelUsers.contains(userId+"")){
return;
}
Long callTime = roomService.getCallTime(room);
RoomWebSocketUtil.sendSendMessage(sessionKey, WsRMsgGen.startVideo(room.getRoomId(), callTime));
RoomWebSocketUtil.sendSendMessage(sessionKey,WsRMsgGen.sysNotice("重连成功，房间已通话(转换成时分秒) "));
RoomWebSocketUtil.sendSendMessage(session, WsRMsgGen.startVideo(room.getRoomId(), callTime));
RoomWebSocketUtil.sendSendMessage(session,WsRMsgGen.sysNotice("重连成功，房间已通话 "+ TimeConverter.convertSecondsToHMS(callTime)));
String sessionKeyReceiver = roomCtxCache.getSessionKeyReceiverByRoomId(room.getRoomId());
if(StringUtils.isNotEmpty(sessionKeyReceiver)){
RoomWebSocketUtil.sendSendMessage(sessionKey,WsRMsgGen.sysNotice("对方已重连成功"));
RoomWebSocketUtil.sendSendMessage(session,WsRMsgGen.sysNotice("对方已重连成功"));
}
});
}
@@ -128,13 +145,19 @@ public class OpenLogic {
RoomStatusEnums.STATUS_CALLER_CONNECT.getCode().equals(status) ||
RoomStatusEnums.STATUS_RECEIVER_CONNECT.getCode().equals(status)){
// 给当前会话发送消息 - 连线成功
RoomWebSocketUtil.sendSendMessage(sessionKey, WsRMsgGen.response(room.getRoomId()));
RoomWebSocketUtil.sendSendMessage(session, WsRMsgGen.response(room.getRoomId()));
}
if(isFirst){
// 给对方发送呼叫页面 TODO
ExecutorConstant.ROOM_EXECUTOR.execute(() -> {
// 给对方发送呼叫页面
;
SendAttachMsgData msg = SendAttachMsgData.init(room.getRoomId());
yunxin.sendAttachMsg(room.getCallUserData().getId(),
room.getReceiverUserData().getId(), msg);
});
}
}
public void receiverConnection(String sessionKey,Room room,Long userId){
public void receiverConnection(WebSocketSession session,Room room,Long userId){
Integer status = room.getStatus();
// 首次链接
if(RoomStatusEnums.STATUS_CALLER_CONNECT.getCode().equals(status)){
@@ -143,6 +166,7 @@ public class OpenLogic {
map.put("connectTime", DateUtil.currentSeconds());
map.put("heartTime",DateUtil.currentSeconds());
userDataCache.hMSet(room.getRoomId(), UserDataConstant.TYPE_RECEIVER,map);
// 房间号状态设置为 接收方已连接
boolean res = roomDataCache.setStatusReceiverConnection(room.getRoomId());
if(!res){
return;
@@ -150,8 +174,8 @@ public class OpenLogic {
userService.updateVideoStatus(userId,1);
anchorService.updateVideoStatus(userId,1);
}
RoomWebSocketUtil.sendSendMessage(sessionKey, WsRMsgGen.response(room.getRoomId()));
RoomWebSocketUtil.sendSendMessage(sessionKey, WsRMsgGen.updateTip());
RoomWebSocketUtil.sendSendMessage(session, WsRMsgGen.response(room.getRoomId()));
RoomWebSocketUtil.sendSendMessage(session, WsRMsgGen.updateTip());
}

5
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/util/RoomWebSocketUtil.java
View File

@@ -2,6 +2,7 @@ package com.ruoyi.cai.ws.util;
import com.alibaba.fastjson2.JSON;
import com.ruoyi.cai.ws.dto.WsR;
import org.springframework.web.socket.WebSocketSession;
import java.util.List;
@@ -10,6 +11,10 @@ public class RoomWebSocketUtil {
WebSocketUtils.sendMessage(sessionKey, JSON.toJSONString(r));
}
public static void sendSendMessage(WebSocketSession sessionKey, WsR r){
WebSocketUtils.sendMessage(sessionKey, JSON.toJSONString(r));
}
public static void sendSendMessage(List<String> sessionKey, WsR r){
for (String s : sessionKey) {
WebSocketUtils.sendMessage(s, JSON.toJSONString(r));

10
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/util/TimeConverter.java Normal file
View File

@@ -0,0 +1,10 @@
package com.ruoyi.cai.ws.util;
public class TimeConverter {
public static String convertSecondsToHMS(Long seconds){
long hours = seconds / 3600;
long minutes = (seconds % 3600) / 60;
long secondsShow = seconds % 60;
return hours + ":" + minutes + ":" + secondsShow;
}
}

10
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/util/WebSocketUtils.java
View File

@@ -33,12 +33,16 @@ public class WebSocketUtils {
public static boolean close(String sessionKey) {
WebSocketSession sessions = WebSocketSessionHolder.getSessions(sessionKey);
if(sessions != null){
return close(sessions);
}
public static boolean close(WebSocketSession session) {
if(session != null){
try {
sessions.close();
session.close();
return true;
} catch (IOException e) {
log.error("关闭ws失败，sessionKey={}",sessionKey,e);
log.error("关闭ws失败，session={}",session,e);
}
}
return false;

22
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/util/WsExceptionUtil.java
View File

@@ -3,6 +3,9 @@ package com.ruoyi.cai.ws.util;
import com.ruoyi.cai.ws.constant.HangUpEnums;
import com.ruoyi.cai.ws.dto.WsRMsgGen;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.socket.WebSocketSession;
import java.io.IOException;
@Slf4j
public class WsExceptionUtil {
@@ -15,6 +18,25 @@ public class WsExceptionUtil {
if(!close){
log.warn("ws连接时发现异常：{}，sessionKey {} roomid {}", message, sessionKey, roomId);
}
}
public static void throwException(WebSocketSession sessionKey, String message, HangUpEnums hangUpType, Long roomId){
if(hangUpType == null){
hangUpType = HangUpEnums.OTHER;
}
RoomWebSocketUtil.sendSendMessage(sessionKey, WsRMsgGen.hangup(message,roomId,hangUpType.getCode()));
boolean close = WebSocketUtils.close(sessionKey);
if(!close){
log.warn("ws连接时发现异常：{}，sessionKey {} roomid {}", message, sessionKey, roomId);
}
}
public static void throwExceptionFast(WebSocketSession session,String message){
RoomWebSocketUtil.sendSendMessage(session, WsRMsgGen.hangup(message, null, HangUpEnums.OTHER.getCode()));
try {
session.close();
} catch (IOException e) {
log.error("快速关闭ws 失败",e);
}
}
}