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

通话逻辑

Browse Source
This commit is contained in:
张良(004796)
2023-12-28 21:35:37 +08:00
parent f337f57df5
commit 49922904ec
34 changed files with 984 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
ruoyi-cai/pom.xml
View File

@@ -26,5 +26,15 @@
<groupId>com.ruoyi</groupId>
<artifactId>ruoyi-system</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>2.0.19</version>
</dependency>
<dependency>
<groupId>com.ruoyi</groupId>
<artifactId>ruoyi-websocket</artifactId>
<version>4.8.2</version>
</dependency>
</dependencies>
</project>

15
ruoyi-cai/src/main/java/com/ruoyi/cai/chat/ChatManager.java
View File

@@ -14,6 +14,7 @@ import com.ruoyi.cai.service.CaiAnchorService;
import com.ruoyi.cai.service.CaiGuardTotalService;
import com.ruoyi.cai.service.CaiUserCallService;
import com.ruoyi.cai.service.CaiUserService;
import com.ruoyi.cai.ws.bean.Room;
import com.ruoyi.cai.ws.manager.WebSocketManager;
import com.ruoyi.common.exception.ServiceException;
import com.ruoyi.common.helper.LoginHelper;
@@ -58,10 +59,11 @@ public class ChatManager {
if(anchor == null){
throw new ServiceException("主播技能不存在");
}
String roomId = webSocketManager.checkOnlineRoom(fromUser.getId(), toUser.getId());
if(StringUtils.isEmpty(roomId)){
String roomId = null;
Room room = webSocketManager.checkOnlineRoom(fromUser.getId(), toUser.getId());
if(room == null){
CaiUserCall call = userCallService.createCall(fromUser, toUser, anchor);
roomId = webSocketManager.createRoom(call.getId());
roomId = webSocketManager.createRoom(call.getId()+"");
}
String weSocketUrl = String.format(properties.getWebSocketUrl(),"token",roomId);
Long guardPrice = systemConfigManager.getSystemConfigOfLong(SystemConfigEnum.GUARD_PRICE);
@@ -96,11 +98,10 @@ public class ChatManager {
if(!userCall.getFromUid().equals(userId) && !userCall.getToUid().equals(userId)){
throw new ServiceException("无权限操作房间");
}
String roomIdNew = webSocketManager.checkOnlineRoom(userCall.getFromUid(), userCall.getToUid());
if (StringUtils.isNotEmpty(roomIdNew)) {
throw new ServiceException("'服务繁忙'");
Room room = webSocketManager.checkOnlineRoom(userCall.getFromUid(), userCall.getToUid());
if (room == null) {
throw new ServiceException("'对方已取消通话'");
}
// 判断房间号是否还存在 TODO
List<CaiUser> userList = userService.listByIds(Arrays.asList(userCall.getFromUid(), userCall.getToUid()));
Map<Long, CaiUser> userMap = userList.stream().collect(Collectors.toMap(CaiUser::getId, Function.identity()));
CaiUser fromUser = userMap.get(userCall.getFromUid());

3
ruoyi-cai/src/main/java/com/ruoyi/cai/domain/CaiUserCall.java
View File

@@ -45,6 +45,7 @@ public class CaiUserCall implements Serializable {
* 结束通话时间
*/
private LocalDateTime endTime;
private String skillName;
/**
* 通话时长
*/
@@ -85,6 +86,8 @@ public class CaiUserCall implements Serializable {
* 接收者-工会-视频比例
*/
private BigDecimal receiverUnionVideoDivide;
private Boolean receiverUnionGet;
private Boolean receiverInviteGet;
/**
* 接收者-邀请人
*/

2
ruoyi-cai/src/main/java/com/ruoyi/cai/service/CaiAnchorService.java
View File

@@ -21,4 +21,6 @@ public interface CaiAnchorService extends IService<CaiAnchor> {
Page<AnchorListVo> pageApp(PageQuery pageQuery, AnchorListQuery query);
CaiAnchor getByUserId(Long userId);
void updateVideoStatus(Long userId, int videoStatus);
}

2
ruoyi-cai/src/main/java/com/ruoyi/cai/service/CaiUserService.java
View File

@@ -29,4 +29,6 @@ public interface CaiUserService extends IService<CaiUser> {
UserInfoVo info(Long userId);
CaiUser getByUserCode(String userCode);
void updateVideoStatus(Long userId, int videoStatus);
}

8
ruoyi-cai/src/main/java/com/ruoyi/cai/service/impl/CaiAnchorServiceImpl.java
View File

@@ -7,6 +7,7 @@ import com.ruoyi.cai.domain.CaiAnchor;
import com.ruoyi.cai.dto.admin.vo.CaiAnchorAdminVo;
import com.ruoyi.cai.dto.app.query.AnchorListQuery;
import com.ruoyi.cai.dto.app.vo.AnchorListVo;
import com.ruoyi.cai.dto.app.vo.AnchorVo;
import com.ruoyi.cai.mapper.CaiAnchorMapper;
import com.ruoyi.cai.service.CaiAnchorService;
import com.ruoyi.common.core.domain.PageQuery;
@@ -40,4 +41,11 @@ public class CaiAnchorServiceImpl extends ServiceImpl<CaiAnchorMapper,CaiAnchor>
public CaiAnchor getByUserId(Long userId){
return this.getOne(Wrappers.lambdaQuery(CaiAnchor.class).eq(CaiAnchor::getUserId,userId));
}
@Override
public void updateVideoStatus(Long userId, int videoStatus) {
this.update(Wrappers.lambdaUpdate(CaiAnchor.class)
.eq(CaiAnchor::getUserId,userId)
.set(CaiAnchor::getVideoStatus,videoStatus));
}
}

7
ruoyi-cai/src/main/java/com/ruoyi/cai/service/impl/CaiUserServiceImpl.java
View File

@@ -95,4 +95,11 @@ public class CaiUserServiceImpl extends ServiceImpl<CaiUserMapper, CaiUser> impl
public CaiUser getByUserCode(String userCode){
return this.getOne(Wrappers.lambdaQuery(CaiUser.class).eq(CaiUser::getUsercode,userCode).last("limit 1"));
}
@Override
public void updateVideoStatus(Long userId, int videoStatus) {
this.update(Wrappers.lambdaUpdate(CaiUser.class)
.eq(CaiUser::getId,userId)
.set(CaiUser::getVideoStatus,videoStatus));
}
}

4
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/bean/CallerRoom.java Normal file
View File

@@ -0,0 +1,4 @@
package com.ruoyi.cai.ws.bean;
public class CallerRoom {
}

14
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/bean/FdCtxData.java Normal file
View File

@@ -0,0 +1,14 @@
package com.ruoyi.cai.ws.bean;
import lombok.Data;
@Data
public class FdCtxData {
private String sessionKey;
private String roomId;
private Long userId;
private Integer userType;
private Long tarUserId;
private Integer tarUserType;
}

34
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/bean/Room.java Normal file
View File

@@ -0,0 +1,34 @@
package com.ruoyi.cai.ws.bean;
import com.ruoyi.cai.ws.constant.HangUpEnums;
import com.ruoyi.cai.ws.constant.RoomStatusEnums;
import com.ruoyi.cai.ws.service.CheckConnectionDTO;
import lombok.Data;
import java.util.HashMap;
import java.util.Map;
@Data
public class Room {
private RoomData roomData;
private UserData callUserData;
private UserData receiverUserData;
public String getRoomId(){
return roomData.getRoomId();
}
public Integer getStatus(){
return roomData.getStatus();
}
public boolean isCanCall(){
return RoomStatusEnums.isCanCall(roomData.getStatus());
}
public boolean isOnline() {
return RoomStatusEnums.STATUS_AGREE.getCode().equals(roomData.getStatus());
}
}

15
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/bean/RoomData.java Normal file
View File

@@ -0,0 +1,15 @@
package com.ruoyi.cai.ws.bean;
import lombok.Data;
import java.math.BigDecimal;
@Data
public class RoomData {
private String roomId;
private Long callPrice;
private String skillName;
private Integer status;
private BigDecimal videoDivide;
private Long hangUpTime;
}

21
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/bean/UserData.java Normal file
View File

@@ -0,0 +1,21 @@
package com.ruoyi.cai.ws.bean;
import lombok.Data;
import java.math.BigDecimal;
@Data
public class UserData {
private Long id;
private String roomId;
private int userType;
private String nickname;
private String userCode;
private Long inviterId;
private BigDecimal inviterRate;
private Boolean inviterIsGet;
private Long unionUserId;
private BigDecimal unionUserRate;
private Boolean unionIsGet;
}

41
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/cache/CallerRoomCache.java vendored Normal file
View File

@@ -0,0 +1,41 @@
package com.ruoyi.cai.ws.cache;
import com.ruoyi.cai.ws.constant.RedisConstant;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.concurrent.TimeUnit;
@Component
public class CallerRoomCache {
@Autowired
private StringRedisTemplate redisTemplate;
public String getKey(Long fromUserId){
return String.format(RedisConstant.CALLER_ROOM_DATA,fromUserId);
}
public String getRoomId(Long fromUserId, Long toUserId){
String key = getKey(fromUserId);
Object roomId = redisTemplate.opsForHash().get(key, toUserId);
return roomId == null ? null : String.valueOf(roomId);
}
public void addRoom(Long fromUid, Long toUid, Long roomId) {
String key = getKey(fromUid);
redisTemplate.opsForHash().put(key,toUid,roomId);
redisTemplate.expire(key,7, TimeUnit.DAYS);
}
public Map<Object,Object> getAll(Long fromUid){
String key = getKey(fromUid);
return redisTemplate.opsForHash().entries(key);
}
public void del(Long fromUid) {
String key = getKey(fromUid);
redisTemplate.delete(key);
}
}

26
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/cache/FdCtxDataCache.java vendored Normal file
View File

@@ -0,0 +1,26 @@
package com.ruoyi.cai.ws.cache;
import com.alibaba.fastjson2.JSON;
import com.ruoyi.cai.ws.bean.FdCtxData;
import com.ruoyi.cai.ws.constant.RedisConstant;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;
@Component
public class FdCtxDataCache {
@Autowired
private StringRedisTemplate redisTemplate;
public String getKey(String sessionKey){
return String.format(RedisConstant.FDCTX_DATA,sessionKey);
}
public void save(FdCtxData fdCtxData) {
String str = JSON.toJSONString(fdCtxData);
redisTemplate.opsForValue().set(getKey(fdCtxData.getSessionKey()),str,5, TimeUnit.DAYS);
}
}

20
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/cache/OnlineDataCache.java vendored Normal file
View File

@@ -0,0 +1,20 @@
package com.ruoyi.cai.ws.cache;
import com.ruoyi.cai.ws.constant.RedisConstant;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
@Component
public class OnlineDataCache {
@Autowired
private StringRedisTemplate redisTemplate;
public String getKey(){
return RedisConstant.ONLINE_ROOM_DATA;
}
public void add(String roomId){
redisTemplate.opsForSet().add(getKey(),roomId);
}
}

24
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/cache/RoomCtxCache.java vendored Normal file
View File

@@ -0,0 +1,24 @@
package com.ruoyi.cai.ws.cache;
import com.ruoyi.cai.ws.constant.RedisConstant;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;
@Component
public class RoomCtxCache {
@Autowired
private StringRedisTemplate redisTemplate;
public String getKey(String roomId){
return String.format(RedisConstant.FDCTX_ROOM_DATA,roomId);
}
public void addFd(String sessionKey,String roomId,Integer userType){
String key = getKey(roomId);
redisTemplate.opsForHash().putIfAbsent(key,sessionKey,userType);
redisTemplate.expire(key,7, TimeUnit.DAYS);
}
}

90
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/cache/RoomDataCache.java vendored Normal file
View File

@@ -0,0 +1,90 @@
package com.ruoyi.cai.ws.cache;
import cn.hutool.json.JSONObject;
import com.alibaba.fastjson2.JSON;
import com.ruoyi.cai.ws.bean.RoomData;
import com.ruoyi.cai.ws.constant.RedisConstant;
import com.ruoyi.cai.ws.constant.RoomStatusEnums;
import org.apache.commons.lang3.BooleanUtils;
import org.apache.poi.ss.formula.functions.Roman;
import org.redisson.api.RBucket;
import org.redisson.api.RMap;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;
import java.util.Arrays;
import java.util.Collections;
import java.util.Map;
@Component
public class RoomDataCache {
// 房间状态
private final static int STATUS_CREATE = 0; // 刚创建
private final static int STATUS_CALLER_CONNECT = 1; // 呼叫方连接上
private final static int STATUS_CALLER_CANCEL = 2; // 呼叫方取消通话
private final static int STATUS_RECEIVER_CONNECT = 3; // 接收方连接上
private final static int STATUS_CONNECT_CANCEL = 4; // 收到通话未接听
private final static int STATUS_TIMEOUT_CANCEL = 5; // 超时未接听
private final static int STATUS_REFUSE = 6; // 接收方已拒绝
private final static int STATUS_AGREE = 7; // 已接听
private final static int STATUS_HANGUP = 8; // 通话结束
@Autowired
private RedissonClient redissonClient;
@Autowired
private StringRedisTemplate stringRedisTemplate;
public String getKey(String roomId){
return String.format(RedisConstant.ROOM_DATA,roomId);
}
public RoomData getByRoomId(String roomId){
Map<Object, Object> map = stringRedisTemplate.opsForHash().entries(getKey(roomId));
if(map.get("roomId") == null){
return null;
}
RoomData roomData = new RoomData();
roomData.setRoomId(map.get("roomId").toString());
roomData.setCallPrice(Long.valueOf(map.get("callPrice").toString()));
return roomData;
}
public void init(RoomData roomData) {
Map<String,Object> map = JSON.parseObject(JSON.toJSONString(roomData));
stringRedisTemplate.opsForHash().putAll(getKey(roomData.getRoomId()),map);
}
private final static String HANG_UP = "local hangupTime = tonumber(redis.call('hGet', KEYS[1], 'hangupTime'))\n" +
"if hangupTime > 0 then\n" +
" return 0\n" +
"end\n" +
"return redis.call('hMSet', KEYS[1], 'status', 8, 'hangupTime', ARGV[1])";
public void hangUp(String roomId) {
DefaultRedisScript<Boolean> redisScript = new DefaultRedisScript<>(HANG_UP,Boolean.class);
String currentTime = String.valueOf(System.currentTimeMillis() / 1000);
stringRedisTemplate.execute(redisScript, Collections.singletonList(getKey(roomId)), currentTime);
}
public void setStatus(String roomId, RoomStatusEnums status) {
String key = getKey(roomId);
stringRedisTemplate.opsForHash().put(key,"status",status.getCode());
}
private final static String SET_STATUS_RECEIVER_CONNECTION = "local status = tonumber(redis.call('hget', KEYS[1], 'status'))\n" +
"if status ~= 1 then\n" +
" return 0\n" +
"end\n" +
"return redis.call('hmset', KEYS[1], 'status', 3)";
public boolean setStatusReceiverConnection(String roomId) {
DefaultRedisScript<Boolean> redisScript = new DefaultRedisScript<>(SET_STATUS_RECEIVER_CONNECTION,Boolean.class);
Boolean execute = stringRedisTemplate.execute(redisScript, Collections.singletonList(getKey(roomId)));
return BooleanUtils.isTrue(execute);
}
}

72
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/cache/UserDataCache.java vendored Normal file
View File

@@ -0,0 +1,72 @@
package com.ruoyi.cai.ws.cache;
import com.alibaba.fastjson2.JSON;
import com.ruoyi.cai.ws.bean.UserData;
import com.ruoyi.cai.ws.constant.RedisConstant;
import com.ruoyi.cai.ws.util.MapGetUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import java.util.Map;
@Component
public class UserDataCache {
public final static int TYPE_CALLER = 1; // 视频发起者
public final static int TYPE_RECEIVER = 2; // 视频接收者
@Autowired
private StringRedisTemplate redisTemplate;
public String getKey(String roomId,int type){
return String.format(RedisConstant.USER_ROOM_DATA,roomId,type==TYPE_CALLER?"caller":"receiver");
}
public UserData getCallerUserDataByRoom(String roomId){
return getUserDataByRoom(roomId,TYPE_CALLER);
}
public UserData getReceiverUserDataByRoom(String roomId){
return getUserDataByRoom(roomId,TYPE_RECEIVER);
}
public UserData getUserDataByRoom(String roomId,int type){
String key = getKey(roomId, type);
Map<Object, Object> entries = redisTemplate.opsForHash().entries(key);
if(entries.get("roomId") == null){
return null;
}
UserData userData = new UserData();
userData.setId(MapGetUtil.getLong(entries.get("id")));
userData.setRoomId(MapGetUtil.getString(entries.get("roomId")));
userData.setUserType(MapGetUtil.getInt(entries.get("userType")));
userData.setNickname(MapGetUtil.getString(entries.get("nickname")));
userData.setUserCode(MapGetUtil.getString(entries.get("userCode")));
userData.setInviterId(MapGetUtil.getLong(entries.get("inviterId")));
userData.setInviterRate(MapGetUtil.getBigDecimal(entries.get("inviterRate")));
userData.setInviterIsGet(MapGetUtil.getBoolean(entries.get("inviterIsGet")));
userData.setUnionUserId(MapGetUtil.getLong(entries.get("unionUserId")));
userData.setUnionUserRate(MapGetUtil.getBigDecimal(entries.get("unionUserRate")));
userData.setUnionIsGet(MapGetUtil.getBoolean(entries.get("unionIsGet")));
return userData;
}
public void init(UserData userData,int type){
String key = getKey(userData.getRoomId(), type);
userData.setUserType(type);
Map<String,Object> map = JSON.parseObject(JSON.toJSONString(userData));
redisTemplate.opsForHash().putAll(key,map);
}
public void initCaller(UserData callerUserData) {
init(callerUserData,TYPE_CALLER);
}
public void initReceiver(UserData callerUserData) {
init(callerUserData,TYPE_RECEIVER);
}
public void hMSet(String roomId,Integer userType,Map<String,Object> data) {
String key = getKey(roomId, userType);
redisTemplate.opsForHash().putAll(key,data);
}
}

25
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/constant/HangUpEnums.java Normal file
View File

@@ -0,0 +1,25 @@
package com.ruoyi.cai.ws.constant;
import lombok.Getter;
@Getter
public enum HangUpEnums {
CANCEL(1,"主动取消"),
REFUSE(2,"对方拒绝"),
TIMEOUT(3,"超时"),
TO(4,"被叫者挂断"),
FROM(5,"主叫者挂断"),
NOTMONEY(6,"费用不足"),
OTHER(7,"其他问题"),
TYPE_SPECIAL(8,"特殊用户挂断"),
PORN_FROM(10,"男方色情违规"),
PORN_TO(11,"女方色情违规"),
;
private final Integer code;
private final String msg;
HangUpEnums(Integer code, String msg) {
this.code = code;
this.msg = msg;
}
}

12
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/constant/RedisConstant.java Normal file
View File

@@ -0,0 +1,12 @@
package com.ruoyi.cai.ws.constant;
public class RedisConstant {
public static final String REDIS_P = "caiws-";
public static final String ONLINE_ROOM_DATA = REDIS_P + "onlineRoomData";
public static final String ROOM_DATA = REDIS_P + "roomData:%s";
public static final String FDCTX_DATA = REDIS_P + "fdctx:%s";
public static final String FDCTX_ROOM_DATA = REDIS_P + "roomCtx:%s";
public static final String CALLER_ROOM_DATA = REDIS_P + "caller:%s";
public static final String USER_ROOM_DATA = REDIS_P + "room:%s:%s";
public static final String INIT_ROOM_LOCK = REDIS_P + "lock:initRoom:%s-%s";
}

64
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/constant/RoomStatusEnums.java Normal file
View File

@@ -0,0 +1,64 @@
package com.ruoyi.cai.ws.constant;
import lombok.Getter;
@Getter
public enum RoomStatusEnums {
/**
* 刚创建
*/
STATUS_CREATE(0,"刚创建"),
/**
* 呼叫方连接上
*/
STATUS_CALLER_CONNECT(1,"呼叫方连接上"),
/**
* 呼叫方取消通话
*/
STATUS_CALLER_CANCEL(2,"呼叫方取消通话"),
/**
* 接收方连接上
*/
STATUS_RECEIVER_CONNECT(3,"接收方连接上"),
/**
* 收到通话未接听
*/
STATUS_CONNECT_CANCEL(4,"收到通话未接听"),
/**
* 超时未接听
*/
STATUS_TIMEOUT_CANCEL(5,"超时未接听"),
/**
* 接收方已拒绝
*/
STATUS_REFUSE(6,"接收方已拒绝"),
/**
* 已接听
*/
STATUS_AGREE(7,"已接听"),
/**
* 通话结束
*/
STATUS_HANGUP(8,"通话结束")
;
private final Integer code;
private final String name;
RoomStatusEnums(Integer code, String name) {
this.code = code;
this.name = name;
}
public static boolean isCanCall(Integer status){
if(STATUS_CREATE.getCode().equals(status)
|| STATUS_CALLER_CONNECT.getCode().equals(status)
|| STATUS_RECEIVER_CONNECT.getCode().equals(status)
|| STATUS_AGREE.getCode().equals(status)){
return true;
}
return false;
}
}

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

@@ -1,14 +1,134 @@
package com.ruoyi.cai.ws.manager;
import com.ruoyi.cai.domain.CaiAnchor;
import com.ruoyi.cai.domain.CaiUser;
import com.ruoyi.cai.domain.CaiUserCall;
import com.ruoyi.cai.service.CaiAnchorService;
import com.ruoyi.cai.service.CaiUserCallService;
import com.ruoyi.cai.service.CaiUserInviteService;
import com.ruoyi.cai.service.CaiUserService;
import com.ruoyi.cai.ws.bean.Room;
import com.ruoyi.cai.ws.bean.RoomData;
import com.ruoyi.cai.ws.bean.UserData;
import com.ruoyi.cai.ws.cache.CallerRoomCache;
import com.ruoyi.cai.ws.cache.RoomDataCache;
import com.ruoyi.cai.ws.cache.UserDataCache;
import com.ruoyi.cai.ws.constant.RedisConstant;
import com.ruoyi.cai.ws.service.RoomService;
import com.ruoyi.common.exception.ServiceException;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class WebSocketManager {
public String checkOnlineRoom(Long formUserId,Long toUserId){
return "name";
@Autowired
private CallerRoomCache callerRoomCache;
@Autowired
private RoomService roomService;
@Autowired
private CaiUserCallService userCallService;
@Autowired
private CaiUserService userService;
@Autowired
private CaiAnchorService anchorService;
@Autowired
private CaiUserInviteService userInviteService;
@Autowired
private RoomDataCache roomDataCache;
@Autowired
private UserDataCache userDataCache;
@Autowired
private RedissonClient redissonClient;
public Room checkOnlineRoom(Long fromUserId,Long toUserId){
String roomId = callerRoomCache.getRoomId(fromUserId, toUserId);
if(StringUtils.isEmpty(roomId)){
return null;
}
Room room = roomService.load(roomId);
if(room == null){
return null;
}
return room;
}
public String createRoom(Long id) {
return "123";
public String createRoom(String roomId) {
CaiUserCall userCall = userCallService.getById(roomId);
if(userCall == null){
throw new ServiceException("无效房间");
}
Room room = checkOnlineRoom(userCall.getFromUid(), userCall.getToUid());
if(room != null && room.isCanCall()){ // 缓存中存在旧房间，直接返回
return room.getRoomId();
}
RLock lock = redissonClient.getLock(String.format(RedisConstant.INIT_ROOM_LOCK, userCall.getFromUid(), userCall.getToUid()));
boolean b = lock.tryLock();
if(!b){
throw new ServiceException("系统繁忙");
}
try {
roomId = initRoom(userCall);
}finally {
lock.unlock();
}
return roomId;
}
public String initRoom(CaiUserCall call){
call = userCallService.getById(call.getId());
CaiUser callUser = userService.getById(call.getFromUid());
if(callUser == null){
throw new ServiceException("无效的呼叫方");
}
CaiUser receiverUser = userService.getById(call.getToUid());
if(receiverUser == null){
throw new ServiceException("无效的接收方");
}
CaiAnchor anchor = anchorService.getByUserId(call.getToUid());
if(anchor == null){
throw new ServiceException("接收方不是女神");
}
// 检查视频状态
if(anchor.getVideoStatus() != 0){
throw new ServiceException("对方正在视频中");
}
// 检查女神在线状态 TODO
// 关闭发起的其他房间
roomService.closeAllRoom(call.getFromUid());
// 删除旧房间记录
roomService.delCallRoom(call.getFromUid());
// 初始化
RoomData roomData = new RoomData();
roomData.setRoomId(call.getId()+"");
roomData.setCallPrice(call.getCallPrice());
roomData.setSkillName(call.getSkillName());
roomData.setStatus(call.getStatus());
roomData.setVideoDivide(call.getReceiverVideoDivide());
roomDataCache.init(roomData);
UserData callerUserData = new UserData();
callerUserData.setId(call.getFromUid());
callerUserData.setRoomId(call.getId()+"");
callerUserData.setNickname(callUser.getNickname());
callerUserData.setUserCode(callUser.getUsercode());
userDataCache.initCaller(callerUserData);
UserData receiveUserData = new UserData();
receiveUserData.setId(call.getToUid());
receiveUserData.setRoomId(call.getId()+"");
receiveUserData.setNickname(receiverUser.getNickname());
receiveUserData.setUserCode(receiverUser.getUsercode());
receiveUserData.setInviterId(call.getReceiverInviteUserId());
receiveUserData.setInviterRate(call.getReceiverIncomeDivide());
receiveUserData.setInviterIsGet(call.getReceiverInviteGet());
receiveUserData.setUnionUserId(call.getReceiverUnionUserId());
receiveUserData.setUnionUserRate(call.getReceiverUnionVideoDivide());
receiveUserData.setUnionIsGet(call.getReceiverUnionGet());
userDataCache.initReceiver(callerUserData);
callerRoomCache.addRoom(call.getFromUid(),call.getToUid(),call.getId());
return call.getId()+"";
}
}

147
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/processon/OpenLogic.java Normal file
View File

@@ -0,0 +1,147 @@
package com.ruoyi.cai.ws.processon;
import cn.hutool.core.date.DateUtil;
import com.ruoyi.cai.service.CaiAnchorService;
import com.ruoyi.cai.service.CaiUserService;
import com.ruoyi.cai.ws.bean.FdCtxData;
import com.ruoyi.cai.ws.bean.Room;
import com.ruoyi.cai.ws.cache.*;
import com.ruoyi.cai.ws.constant.HangUpEnums;
import com.ruoyi.cai.ws.constant.RoomStatusEnums;
import com.ruoyi.cai.ws.service.CheckConnectionDTO;
import com.ruoyi.cai.ws.service.RoomService;
import com.ruoyi.cai.ws.util.WsExceptionUtil;
import com.ruoyi.websocket.handle.IOpenLogic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.WebSocketSession;
import java.util.HashMap;
import java.util.Map;
@Component
public class OpenLogic implements IOpenLogic {
@Autowired
private RoomService roomService;
@Autowired
private FdCtxDataCache fdCtxDataCache;
@Autowired
private RoomCtxCache roomCtxCache;
@Autowired
private UserDataCache userDataCache;
@Autowired
private RoomDataCache roomDataCache;
@Autowired
private OnlineDataCache onlineDataCache;
@Autowired
private CaiUserService userService;
@Autowired
private CaiAnchorService anchorService;
@Override
public void processOn(WebSocketSession session) {
Map<String, Object> map = session.getAttributes();
String token = map.get("token").toString();
// 校验token
process(token,"",0L);
}
public void process(String sessionKey,String roomId,Long userId){
Room room = roomService.load(roomId);
if(room == null || (room.getCallUserData().getId().equals(userId) && room.getReceiverUserData().getId().equals(userId))){
WsExceptionUtil.throwException(sessionKey,"房间不可用", HangUpEnums.OTHER,roomId);
return;
}
CheckConnectionDTO checkConnect = roomService.checkConnect(room);
if(checkConnect != null){
WsExceptionUtil.throwException(sessionKey,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.setRoomId(room.getRoomId());
fdCtxData.setUserId(userId);
if(userId.equals(room.getCallUserData().getId())){
fdCtxData.setUserType(UserDataCache.TYPE_CALLER);
fdCtxData.setTarUserId(room.getReceiverUserData().getId());
fdCtxData.setTarUserType(UserDataCache.TYPE_RECEIVER);
} else {
fdCtxData.setUserType(UserDataCache.TYPE_RECEIVER);
fdCtxData.setTarUserId(room.getCallUserData().getId());
fdCtxData.setTarUserType(UserDataCache.TYPE_CALLER);
}
fdCtxDataCache.save(fdCtxData);
roomCtxCache.addFd(sessionKey,roomId,fdCtxData.getUserType());
}
public void callerConnection(String sessionKey,Room room,Long userId){
boolean isFirst = false;
Integer status = room.getStatus();
// 首次链接
if(RoomStatusEnums.STATUS_CREATE.getCode().equals(status)){
// 更新缓存
Map<String,Object> map = new HashMap<>();
map.put("connectTime", DateUtil.currentSeconds());
map.put("heartTime",DateUtil.currentSeconds());
userDataCache.hMSet(room.getRoomId(), UserDataCache.TYPE_CALLER,map);
roomDataCache.setStatus(room.getRoomId(),RoomStatusEnums.STATUS_CALLER_CONNECT);
onlineDataCache.add(room.getRoomId());
userService.updateVideoStatus(userId,1);
isFirst = true;
}
// 已经接通
if(room.isOnline()){
// TODO
// 如果视频也掉线了，则重连的时候发送消息提示
// sgo(function() use ($rs){
// $users = Agora::getChannelUsers($rs->room->id);
// if (in_array($rs->caller->id, $users)) {
// return;
// }
//
// $rs->sendToCurrent(WsMsgGen::startVideo($rs->room->id, $rs->callTime()))
// ->sendToCurrent(WsMsgGen::sysNotice('重连成功，房间已通话 '.gmdate('H:i:s', $rs->callTime())))
// ->sendToReceiver(WsMsgGen::sysNotice('对方已重连成功'));
// });
}
if(RoomStatusEnums.STATUS_CREATE.getCode().equals(status) ||
RoomStatusEnums.STATUS_CALLER_CONNECT.getCode().equals(status) ||
RoomStatusEnums.STATUS_RECEIVER_CONNECT.getCode().equals(status)){
// 给当前会话发送消息 - 连线成功
}
if(isFirst){
// 给对方发送呼叫页面
}
}
public void receiverConnection(String sessionKey,Room room,Long userId){
Integer status = room.getStatus();
// 首次链接
if(RoomStatusEnums.STATUS_CALLER_CONNECT.getCode().equals(status)){
// 更新缓存
Map<String,Object> map = new HashMap<>();
map.put("connectTime", DateUtil.currentSeconds());
map.put("heartTime",DateUtil.currentSeconds());
userDataCache.hMSet(room.getRoomId(), UserDataCache.TYPE_RECEIVER,map);
boolean res = roomDataCache.setStatusReceiverConnection(room.getRoomId());
if(!res){
return;
}
userService.updateVideoStatus(userId,1);
anchorService.updateVideoStatus(userId,1);
}
// 连线成功
// $rs->sendToCurrent(WsMsgGen::response($this->roomId))
// ->sendToCurrent(WsMsgGen::updateTip());
}
}

18
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/service/CheckConnectionDTO.java Normal file
View File

@@ -0,0 +1,18 @@
package com.ruoyi.cai.ws.service;
import com.ruoyi.cai.ws.constant.HangUpEnums;
import lombok.Data;
@Data
public class CheckConnectionDTO {
private HangUpEnums hangUpEnums;
private String message;
public CheckConnectionDTO() {
}
public CheckConnectionDTO(HangUpEnums hangUpEnums, String message) {
this.hangUpEnums = hangUpEnums;
this.message = message;
}
}

89
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/service/RoomService.java Normal file
View File

@@ -0,0 +1,89 @@
package com.ruoyi.cai.ws.service;
import com.ruoyi.cai.ws.bean.Room;
import com.ruoyi.cai.ws.bean.RoomData;
import com.ruoyi.cai.ws.bean.UserData;
import com.ruoyi.cai.ws.cache.CallerRoomCache;
import com.ruoyi.cai.ws.cache.RoomDataCache;
import com.ruoyi.cai.ws.cache.UserDataCache;
import com.ruoyi.cai.ws.constant.HangUpEnums;
import com.ruoyi.cai.ws.constant.RoomStatusEnums;
import com.ruoyi.cai.ws.util.WsExceptionUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.HashMap;
import java.util.Map;
@Component
public class RoomService {
@Autowired
private RoomDataCache roomDataCache;
@Autowired
private UserDataCache userDataCache;
@Autowired
private CallerRoomCache callerRoomCache;
public Room load(String roomId){
Room room = new Room();
RoomData roomData = roomDataCache.getByRoomId(roomId);
if(roomData == null){
return null;
}
room.setRoomData(roomData);
UserData callerUserData = userDataCache.getCallerUserDataByRoom(roomId);
if(callerUserData == null){
return null;
}
room.setCallUserData(callerUserData);
UserData receiverUserData = userDataCache.getReceiverUserDataByRoom(roomId);
if(receiverUserData == null){
return null;
}
room.setReceiverUserData(receiverUserData);
return room;
}
public void closeAllRoom(Long fromUid){
Map<Object, Object> all = callerRoomCache.getAll(fromUid);
for (Map.Entry<Object, Object> entry : all.entrySet()) {
String roomId = String.valueOf(entry.getValue());
Room room = load(roomId);
if(room == null){
continue;
}
if(room.isCanCall()){
// 给接收方发送对应已取消
}else if(room.isOnline()){
// 对接收方发送对方已挂断
}
// 挂掉房间
roomDataCache.hangUp(roomId);
// 关闭所有的websocket进程
}
}
public void delCallRoom(Long fromUid) {
callerRoomCache.del(fromUid);
}
private static Map<Integer, CheckConnectionDTO> STATUS_TO_HANG_UP = new HashMap<>();
static {
STATUS_TO_HANG_UP.put(RoomStatusEnums.STATUS_CALLER_CANCEL.getCode(),new CheckConnectionDTO(HangUpEnums.CANCEL,"通话已取消"));
STATUS_TO_HANG_UP.put(RoomStatusEnums.STATUS_CONNECT_CANCEL.getCode(),new CheckConnectionDTO(HangUpEnums.TIMEOUT,"通话未接听"));
STATUS_TO_HANG_UP.put(RoomStatusEnums.STATUS_REFUSE.getCode(),new CheckConnectionDTO(HangUpEnums.REFUSE,"对方已拒绝"));
STATUS_TO_HANG_UP.put(RoomStatusEnums.STATUS_TIMEOUT_CANCEL.getCode(),new CheckConnectionDTO(HangUpEnums.REFUSE,"通话已拒绝"));
STATUS_TO_HANG_UP.put(RoomStatusEnums.STATUS_HANGUP.getCode(),new CheckConnectionDTO(HangUpEnums.OTHER,"通话已结束"));
}
public CheckConnectionDTO checkConnect(Room room) {
CheckConnectionDTO checkConnectionDTO = STATUS_TO_HANG_UP.get(room.getRoomData().getStatus());
if(checkConnectionDTO != null){
return checkConnectionDTO;
}
if(room.getRoomData().getStatus() > RoomStatusEnums.STATUS_AGREE.getCode()){
return new CheckConnectionDTO(HangUpEnums.OTHER,"通话已结束");
}
return null;
}
}

51
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/util/MapGetUtil.java Normal file
View File

@@ -0,0 +1,51 @@
package com.ruoyi.cai.ws.util;
import java.math.BigDecimal;
public class MapGetUtil {
public static Long getLong(Object obj){
return getLong(obj,null);
}
public static Long getLong(Object obj, Long defaultValue){
if(obj == null){
return defaultValue;
}
return Long.valueOf(obj.toString());
}
public static String getString(Object obj){
if(obj == null){
return null;
}
return String.valueOf(obj);
}
public static Integer getInt(Object obj){
if(obj == null){
return null;
}
return Integer.valueOf(obj.toString());
}
public static BigDecimal getBigDecimal(Object obj){
if(obj == null){
return null;
}
return new BigDecimal(obj.toString());
}
public static Boolean getBoolean(Object obj) {
if(obj == null){
return null;
}
if(obj instanceof Boolean){
return (Boolean)obj;
}
String str = String.valueOf(obj);
return str.equals("true");
}
}

16
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/util/WsExceptionUtil.java Normal file
View File

@@ -0,0 +1,16 @@
package com.ruoyi.cai.ws.util;
import com.ruoyi.cai.ws.constant.HangUpEnums;
public class WsExceptionUtil {
public static void throwException(String sessionKey, String message, HangUpEnums hangUpType, String roomId){
if(hangUpType == null){
hangUpType = HangUpEnums.OTHER;
}
}
public String handUp(String sessionKey,String message,String roomId,HangUpEnums hang){
return null;
}
}

23
ruoyi-websocket/src/main/java/com/ruoyi/websocket/config/WebSocketConfig.java
View File

@@ -1,13 +1,10 @@
package com.ruoyi.websocket.config;
import cn.hutool.core.util.StrUtil;
import com.ruoyi.websocket.handler.PlusWebSocketHandler;
import com.ruoyi.websocket.interceptor.PlusWebSocketInterceptor;
import com.ruoyi.websocket.listener.WebSocketTopicListener;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
@@ -18,7 +15,7 @@ import org.springframework.web.socket.server.HandshakeInterceptor;
*
* @author zendwang
*/
@AutoConfiguration
@Configuration
@ConditionalOnProperty(value = "websocket.enabled", havingValue = "true")
@EnableConfigurationProperties(WebSocketProperties.class)
@EnableWebSocket
@@ -29,7 +26,7 @@ public class WebSocketConfig {
WebSocketHandler webSocketHandler,
WebSocketProperties webSocketProperties) {
if (StrUtil.isBlank(webSocketProperties.getPath())) {
webSocketProperties.setPath("/websocket");
webSocketProperties.setPath("/ws");
}
if (StrUtil.isBlank(webSocketProperties.getAllowedOrigins())) {
@@ -42,18 +39,4 @@ public class WebSocketConfig {
.setAllowedOrigins(webSocketProperties.getAllowedOrigins());
}
@Bean
public HandshakeInterceptor handshakeInterceptor() {
return new PlusWebSocketInterceptor();
}
@Bean
public WebSocketHandler webSocketHandler() {
return new PlusWebSocketHandler();
}
@Bean
public WebSocketTopicListener topicListener() {
return new WebSocketTopicListener();
}
}

9
ruoyi-websocket/src/main/java/com/ruoyi/websocket/handle/IOpenLogic.java Normal file
View File

@@ -0,0 +1,9 @@
package com.ruoyi.websocket.handle;
import org.springframework.web.socket.WebSocketSession;
public interface IOpenLogic {
void processOn(WebSocketSession session);
}

16
ruoyi-websocket/src/main/java/com/ruoyi/websocket/handler/PlusWebSocketHandler.java
View File

@@ -2,14 +2,18 @@ package com.ruoyi.websocket.handler;
import com.ruoyi.common.core.domain.model.LoginUser;
import com.ruoyi.websocket.dto.WebSocketMessageDto;
import com.ruoyi.websocket.handle.IOpenLogic;
import com.ruoyi.websocket.holder.WebSocketSessionHolder;
import com.ruoyi.websocket.util.WebSocketUtils;
import lombok.extern.slf4j.Slf4j;
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 java.util.Collections;
import java.util.List;
import java.util.Map;
import static com.ruoyi.websocket.constant.WebSocketConstants.LOGIN_USER_KEY;
@@ -19,16 +23,22 @@ import static com.ruoyi.websocket.constant.WebSocketConstants.LOGIN_USER_KEY;
* @author zendwang
*/
@Slf4j
@Component
public class PlusWebSocketHandler extends AbstractWebSocketHandler {
@Autowired
private IOpenLogic openLogic;
/**
* 连接成功后
*/
@Override
public void afterConnectionEstablished(WebSocketSession session) {
LoginUser loginUser = (LoginUser) session.getAttributes().get(LOGIN_USER_KEY);
WebSocketSessionHolder.addSession(loginUser.getUserId(), session);
log.info("[connect] sessionId: {},userId:{},userType:{}", session.getId(), loginUser.getUserId(), loginUser.getUserType());
Map<String, Object> attributes = session.getAttributes();
if(attributes.get("token") != null){
WebSocketSessionHolder.addSession(attributes.get("token").toString(), session);
}
openLogic.processOn(session);
log.info("[connect] sessionId: {},userId:{}", session.getId(), session.getId());
}
/**

12
ruoyi-websocket/src/main/java/com/ruoyi/websocket/holder/WebSocketSessionHolder.java
View File

@@ -16,27 +16,27 @@ import java.util.concurrent.ConcurrentHashMap;
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class WebSocketSessionHolder {
private static final Map<Long, WebSocketSession> USER_SESSION_MAP = new ConcurrentHashMap<>();
private static final Map<String, WebSocketSession> USER_SESSION_MAP = new ConcurrentHashMap<>();
public static void addSession(Long sessionKey, WebSocketSession session) {
public static void addSession(String sessionKey, WebSocketSession session) {
USER_SESSION_MAP.put(sessionKey, session);
}
public static void removeSession(Long sessionKey) {
public static void removeSession(String sessionKey) {
if (USER_SESSION_MAP.containsKey(sessionKey)) {
USER_SESSION_MAP.remove(sessionKey);
}
}
public static WebSocketSession getSessions(Long sessionKey) {
public static WebSocketSession getSessions(String sessionKey) {
return USER_SESSION_MAP.get(sessionKey);
}
public static Set<Long> getSessionsAll() {
public static Set<String> getSessionsAll() {
return USER_SESSION_MAP.keySet();
}
public static Boolean existSession(Long sessionKey) {
public static Boolean existSession(String sessionKey) {
return USER_SESSION_MAP.containsKey(sessionKey);
}
}

2
ruoyi-websocket/src/main/java/com/ruoyi/websocket/interceptor/PlusWebSocketInterceptor.java
View File

@@ -5,6 +5,7 @@ import com.ruoyi.common.helper.LoginHelper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.server.ServerHttpRequest;
import org.springframework.http.server.ServerHttpResponse;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.server.HandshakeInterceptor;
@@ -18,6 +19,7 @@ import static com.ruoyi.websocket.constant.WebSocketConstants.LOGIN_USER_KEY;
* @author zendwang
*/
@Slf4j
@Component
public class PlusWebSocketInterceptor implements HandshakeInterceptor {
/**

2
ruoyi-websocket/src/main/java/com/ruoyi/websocket/listener/WebSocketTopicListener.java
View File

@@ -7,6 +7,7 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.Ordered;
import org.springframework.stereotype.Component;
/**
* WebSocket 主题订阅监听器
@@ -14,6 +15,7 @@ import org.springframework.core.Ordered;
* @author zendwang
*/
@Slf4j
@Component
public class WebSocketTopicListener implements ApplicationRunner, Ordered {
@Override

12
ruoyi-websocket/src/main/java/com/ruoyi/websocket/util/WebSocketUtils.java
View File

@@ -34,20 +34,11 @@ public class WebSocketUtils {
* @param sessionKey session主键 一般为用户id
* @param message 消息文本
*/
public static void sendMessage(Long sessionKey, String message) {
public static void sendMessage(String sessionKey, String message) {
WebSocketSession session = WebSocketSessionHolder.getSessions(sessionKey);
sendMessage(session, message);
}
/**
* 订阅消息
*
* @param consumer 自定义处理
*/
public static void subscribeMessage(Consumer<WebSocketMessageDto> consumer) {
RedisUtils.subscribe(WEB_SOCKET_TOPIC, WebSocketMessageDto.class, consumer);
}
/**
* 发布订阅的消息
*
@@ -107,4 +98,5 @@ public class WebSocketUtils {
}
}
}
}