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-14 03:38:50 +08:00
parent 7a6f9eecd3
commit d5deaacfe8
32 changed files with 323 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -14,7 +14,7 @@ public class RoomData {
private BigDecimal videoDivide;
private Long payCoin = 0L;
private Long payIncome = 0L;
private Long hangUpTime; // 结束时间
private Long hangupTime; // 结束时间
private Long settleTime; // 结算时间

36
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/cache/CallerRoomCache.java vendored
View File

@@ -2,19 +2,23 @@ package com.ruoyi.cai.ws.cache;
import com.ruoyi.cai.ws.constant.RedisConstant;
import org.apache.commons.lang3.BooleanUtils;
import org.redisson.api.RMap;
import org.redisson.api.RScript;
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.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.TimeUnit;
@Component
public class CallerRoomCache {
@Autowired
private StringRedisTemplate redisTemplate;
@Autowired
private RedissonClient redissonClient;
public String getKey(Long fromUserId){
return String.format(RedisConstant.CALLER_ROOM_DATA,fromUserId);
@@ -22,36 +26,46 @@ public class CallerRoomCache {
public Long getRoomId(Long fromUserId, Long toUserId){
String key = getKey(fromUserId);
Object roomId = redisTemplate.opsForHash().get(key, toUserId);
RMap<String, Object> map = redissonClient.getMap(key);
Object roomId = map.get(toUserId+"");
return roomId == null ? null : Long.valueOf(roomId.toString());
}
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);
RMap<String, Object> map = redissonClient.getMap(key);
map.put(toUid+"",roomId+"");
map.expire(Duration.ofDays(7));
// map.expire(7,TimeUnit.DAYS);
// 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);
RMap<Object, Object> map = redissonClient.getMap(key);
return map.readAllMap();
// return redisTemplate.opsForHash().entries(key);
}
public void del(Long fromUid) {
String key = getKey(fromUid);
redisTemplate.delete(key);
redissonClient.getMap(key).delete();
}
private final static String DEL_ROOM_LUA = "return KEYS[1]\n" +
private final static String DEL_ROOM_LUA =
"local r = tonumber(redis.call('hget',KEYS[1],ARGV[1]))\n" +
"if r == tonumber(ARGV[2]) then\n" +
" return redis.call('hdel',KEYS[1],ARGV[1])\n" +
"end\n" +
"return 0";
public boolean delRoom(Long receiverId, Long roomId) {
DefaultRedisScript<Boolean> redisScript = new DefaultRedisScript<>(DEL_ROOM_LUA,Boolean.class);
Boolean execute = redisTemplate.execute(redisScript, Collections.singletonList(getKey(roomId)), receiverId,roomId);
public boolean delRoom(Long callerId, Long receiverId, Long roomId) {
RScript script = redissonClient.getScript();
Boolean execute = script.eval(RScript.Mode.READ_WRITE, DEL_ROOM_LUA, RScript.ReturnType.BOOLEAN,
Collections.singletonList(getKey(callerId)), receiverId+"", roomId+"");
// DefaultRedisScript<Boolean> redisScript = new DefaultRedisScript<>(DEL_ROOM_LUA,Boolean.class);
// Boolean execute = redisTemplate.execute(redisScript, Collections.singletonList(getKey(roomId)), receiverId,roomId);
return BooleanUtils.isTrue(execute);
}
}

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

@@ -1,12 +1,10 @@
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.apache.commons.lang3.StringUtils;
import org.redisson.api.RBucket;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.SearchStrategy;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;
@@ -21,24 +19,21 @@ import java.util.concurrent.TimeUnit;
@Component
public class FdCtxDataCache {
@Autowired
private StringRedisTemplate redisTemplate;
private RedissonClient redissonClient;
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);
RBucket<FdCtxData> bucket = redissonClient.getBucket(getKey(fdCtxData.getSessionKey()));
bucket.set(fdCtxData,5,TimeUnit.DAYS);
}
public FdCtxData getByRoomId(String sessionKey){
String key = getKey(sessionKey);
String s = redisTemplate.opsForValue().get(key);
if(StringUtils.isEmpty(s)){
return null;
}
return JSON.parseObject(s,FdCtxData.class);
RBucket<FdCtxData> bucket = redissonClient.getBucket(key);
return bucket.get();
}
}

16
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/cache/OnlineDataCache.java vendored
View File

@@ -1,6 +1,8 @@
package com.ruoyi.cai.ws.cache;
import com.ruoyi.cai.ws.constant.RedisConstant;
import org.redisson.api.RSet;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
@@ -12,20 +14,28 @@ import java.util.Set;
public class OnlineDataCache {
@Autowired
private StringRedisTemplate redisTemplate;
@Autowired
private RedissonClient redissonClient;
public String getKey(){
return RedisConstant.ONLINE_ROOM_DATA;
}
public Set<String> getAll(){
return redisTemplate.opsForSet().members(getKey());
RSet<String> set = redissonClient.getSet(getKey());
return set.readAll();
// return redisTemplate.opsForSet().members(getKey());
}
public void add(Long roomId){
redisTemplate.opsForSet().add(getKey(), String.valueOf(roomId));
RSet<String> set = redissonClient.getSet(getKey());
set.add(String.valueOf(roomId));
// redisTemplate.opsForSet().add(getKey(), String.valueOf(roomId));
}
public void remove(Long roomId) {
redisTemplate.opsForSet().remove(getKey(),roomId);
RSet<String> set = redissonClient.getSet(getKey());
set.delete();
// redisTemplate.opsForSet().remove(getKey(),roomId);
}
}

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

@@ -3,13 +3,18 @@ package com.ruoyi.cai.ws.cache;
import com.ruoyi.cai.ws.constant.RedisConstant;
import com.ruoyi.cai.ws.constant.UserDataConstant;
import com.ruoyi.cai.ws.holder.WebSocketSessionHolder;
import org.redisson.api.RMap;
import org.redisson.api.RSet;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
/**
@@ -22,6 +27,8 @@ import java.util.concurrent.TimeUnit;
public class RoomCtxCache {
@Autowired
private StringRedisTemplate redisTemplate;
@Autowired
private RedissonClient redissonClient;
public String getKey(Long roomId){
return String.format(RedisConstant.FDCTX_ROOM_DATA,roomId);
@@ -29,23 +36,28 @@ public class RoomCtxCache {
public void addFd(String sessionKey,Long roomId,Integer userType){
String key = getKey(roomId);
redisTemplate.opsForHash().put(key,sessionKey,userType);
redisTemplate.expire(key,7, TimeUnit.DAYS);
RMap<Object, Object> map = redissonClient.getMap(key);
map.put(sessionKey,userType);
map.expire(Duration.ofDays(7));
}
public List<String> getSessionKeysByRoomId(Long roomId){
String key = getKey(roomId);
Map<Object, Object> entries = redisTemplate.opsForHash().entries(key);
List<String> res = new ArrayList<>();
for (Object o : entries.keySet()) {
res.add(String.valueOf(o));
RMap<String, Object> entries = redissonClient.getMap(key);
Set<String> set = entries.readAllKeySet();
List<String> keys = new ArrayList<>();
for (String sessionKey : set) {
if(WebSocketSessionHolder.existSession(sessionKey)){
keys.add(sessionKey);
}
}
return res;
return keys;
}
public String getSessionKeyByRoomIdAndUserType(Long roomId,Integer userType){
String key = getKey(roomId);
Map<Object, Object> entries = redisTemplate.opsForHash().entries(key);
RMap<Object, Object> map = redissonClient.getMap(key);
Map<Object, Object> entries = map.readAllMap();
for (Map.Entry<Object, Object> entry : entries.entrySet()) {
String sessionKey = String.valueOf(entry.getKey());
Integer userTypeK = Integer.valueOf(entry.getValue().toString());

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

@@ -6,12 +6,16 @@ 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.redisson.api.RMap;
import org.redisson.api.RScript;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;
import java.time.Duration;
import java.util.Collections;
import java.util.Map;
@@ -26,6 +30,8 @@ public class RoomDataCache {
@Autowired
private RedissonClient redissonClient;
@Autowired
private RedisTemplate redisTemplate;
@Autowired
private StringRedisTemplate stringRedisTemplate;
public String getKey(Long roomId){
@@ -33,82 +39,105 @@ public class RoomDataCache {
}
public RoomData getByRoomId(Long roomId){
Map<Object, Object> map = stringRedisTemplate.opsForHash().entries(getKey(roomId));
if(map.get("roomId") == null){
RMap<String, Object> map = redissonClient.getMap(getKey(roomId));
Map<String, Object> allMap = map.readAllMap();
// Map<Object, Object> map = redisTemplate.opsForHash().entries(getKey(roomId));
if(allMap.get("roomId") == null){
return null;
}
return JSON.parseObject(JSON.toJSONString(map),RoomData.class);
return JSON.parseObject(JSON.toJSONString(allMap),RoomData.class);
}
public void init(RoomData roomData) {
Map<String,Object> map = JSON.parseObject(JSON.toJSONString(roomData));
stringRedisTemplate.opsForHash().putAll(getKey(roomData.getRoomId()),map);
// redisTemplate.opsForHash().putAll(getKey(roomData.getRoomId()),map);
RMap<String, Object> mapRedis = redissonClient.getMap(getKey(roomData.getRoomId()));
mapRedis.putAll(map);
mapRedis.expire(Duration.ofDays(15));
}
public void hMSet(Long roomId,Map<String,Object> data) {
String key = getKey(roomId);
stringRedisTemplate.opsForHash().putAll(key,data);
// redisTemplate.opsForHash().putAll(key,data);
RMap<Object, Object> map = redissonClient.getMap(key);
map.putAll(data);
}
public void hMSet(Long roomId,String mKey,Object mData) {
String key = getKey(roomId);
stringRedisTemplate.opsForHash().put(key,mKey,mData);
RMap<String, Object> map = redissonClient.getMap(key);
map.put(mKey,mData);
// redisTemplate.opsForHash().put(key,mKey,mData);
}
private final static String HANG_UP = "local hangupTime = tonumber(redis.call('hGet', KEYS[1], 'hangupTime'))\n" +
"if hangupTime > 0 then\n" +
private final static String HANG_UP = "local hangupTime = tonumber(redis.call('hGet', KEYS[1], '\"hangupTime\"'))\n" +
"if hangupTime and hangupTime > 0 then\n" +
" return 0\n" +
"end\n" +
"return redis.call('hMSet', KEYS[1], 'status', 8, 'hangupTime', ARGV[1])";
"return redis.call('hMSet', KEYS[1], '\"status\"', 8, '\"hangupTime\"', ARGV[1])";
public boolean hangUp(Long roomId) {
DefaultRedisScript<Boolean> redisScript = new DefaultRedisScript<>(HANG_UP,Boolean.class);
String currentTime = String.valueOf(System.currentTimeMillis() / 1000);
Boolean execute = stringRedisTemplate.execute(redisScript, Collections.singletonList(getKey(roomId)), currentTime);
RScript script = redissonClient.getScript();
Long currentTime = System.currentTimeMillis() / 1000;
Boolean execute = script.eval(RScript.Mode.READ_WRITE, HANG_UP, RScript.ReturnType.BOOLEAN,
Collections.singletonList(getKey(roomId)), currentTime);
// DefaultRedisScript<Boolean> redisScript = new DefaultRedisScript<>(HANG_UP,Boolean.class);
// Boolean execute = stringRedisTemplate.execute(redisScript, Collections.singletonList(getKey(roomId)), currentTime);
return BooleanUtils.isTrue(execute);
}
public void setStatus(Long roomId, RoomStatusEnums status) {
String key = getKey(roomId);
stringRedisTemplate.opsForHash().put(key,"status",status.getCode());
// redisTemplate.opsForHash().put(key,"status",status.getCode());
RMap<String, Object> map = redissonClient.getMap(key);
map.put("status",status.getCode());
}
private final static String SET_STATUS_RECEIVER_CONNECTION = "local status = tonumber(redis.call('hget', KEYS[1], 'status'))\n" +
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)";
"return redis.call('hmset', KEYS[1], '\"status\"', 3)";
public boolean setStatusReceiverConnection(Long roomId) {
DefaultRedisScript<Boolean> redisScript = new DefaultRedisScript<>(SET_STATUS_RECEIVER_CONNECTION,Boolean.class);
Boolean execute = stringRedisTemplate.execute(redisScript, Collections.singletonList(getKey(roomId)));
RScript script = redissonClient.getScript();
Boolean execute = script.eval(RScript.Mode.READ_WRITE, SET_STATUS_RECEIVER_CONNECTION, RScript.ReturnType.BOOLEAN,
Collections.singletonList(getKey(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);
}
private final static String SET_STATUS_AGREE =
"local status = tonumber(redis.call('hget', KEYS[1], 'status'))\n" +
"local beginTime = tonumber(redis.call('hget', KEYS[1], 'beginTime'))\n" +
"if status ~= 3 or beginTime > 0 then\n" +
"local status = tonumber(redis.call('hget', KEYS[1], '\"status\"'))\n" +
"local beginTime = tonumber(redis.call('hget', KEYS[1], '\"beginTime\"'))\n" +
"if status ~= 3 or (beginTime and beginTime > 0) then\n" +
" return 0\n" +
"end\n" +
"return redis.call('hmset', KEYS[1], 'status', 7, 'beginTime', ARGV[1])";
"return redis.call('hmset', KEYS[1], '\"status\"', 7, '\"beginTime\"', ARGV[1])";
public boolean setStatusAgree(Long roomId) {
DefaultRedisScript<Boolean> redisScript = new DefaultRedisScript<>(SET_STATUS_AGREE,Boolean.class);
String currentTime = String.valueOf(System.currentTimeMillis() / 1000);
Boolean execute = stringRedisTemplate.execute(redisScript, Collections.singletonList(getKey(roomId)),currentTime);
RScript script = redissonClient.getScript();
Boolean execute = script.eval(RScript.Mode.READ_WRITE, SET_STATUS_AGREE, RScript.ReturnType.BOOLEAN,
Collections.singletonList(getKey(roomId)),currentTime);
// DefaultRedisScript<Boolean> redisScript = new DefaultRedisScript<>(SET_STATUS_AGREE,Boolean.class);
// Boolean execute = stringRedisTemplate.execute(redisScript, Collections.singletonList(getKey(roomId)),currentTime);
return BooleanUtils.isTrue(execute);
}
private final static String INCS_BLACK_AMOUNT =
"redis.call('hIncrBy', KEYS[1], 'payCoin', ARGV[1])\n" +
"redis.call('hIncrBy', KEYS[1], 'payIncome', ARGV[2])";
"redis.call('hIncrBy', KEYS[1], '\"payCoin\"', ARGV[1])\n" +
"redis.call('hIncrBy', KEYS[1], '\"payIncome\"', ARGV[2])";
public boolean incsBlackAmount(Long roomId, Long decrCoin, Long decrIncomeCoin) {
DefaultRedisScript<Boolean> redisScript = new DefaultRedisScript<>(INCS_BLACK_AMOUNT,Boolean.class);
Boolean execute = stringRedisTemplate.execute(redisScript, Collections.singletonList(getKey(roomId)),
decrCoin,decrIncomeCoin);
RScript script = redissonClient.getScript();
Boolean execute = script.eval(RScript.Mode.READ_WRITE, INCS_BLACK_AMOUNT, RScript.ReturnType.BOOLEAN,
Collections.singletonList(getKey(roomId)),decrCoin.intValue(),decrIncomeCoin.intValue());
// DefaultRedisScript<Boolean> redisScript = new DefaultRedisScript<>(INCS_BLACK_AMOUNT,Boolean.class);
// Boolean execute = stringRedisTemplate.execute(redisScript, Collections.singletonList(getKey(roomId)),
// decrCoin,decrIncomeCoin);
return BooleanUtils.isTrue(execute);
}
}

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

@@ -4,10 +4,13 @@ import com.alibaba.fastjson2.JSON;
import com.ruoyi.cai.ws.bean.UserData;
import com.ruoyi.cai.ws.constant.RedisConstant;
import com.ruoyi.cai.ws.constant.UserDataConstant;
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.stereotype.Component;
import java.time.Duration;
import java.util.Map;
/**
@@ -19,6 +22,8 @@ import java.util.Map;
public class UserDataCache {
@Autowired
private StringRedisTemplate redisTemplate;
@Autowired
private RedissonClient redissonClient;
public String getKey(Long roomId,int type){
return String.format(RedisConstant.USER_ROOM_DATA,roomId,type== UserDataConstant.TYPE_CALLER?"caller":"receiver");
@@ -34,18 +39,23 @@ public class UserDataCache {
public UserData getUserDataByRoom(Long roomId,int type){
String key = getKey(roomId, type);
Map<Object, Object> entries = redisTemplate.opsForHash().entries(key);
if(entries.get("roomId") == null){
// Map<Object, Object> entries = redisTemplate.opsForHash().entries(key);
RMap<Object, Object> entries = redissonClient.getMap(key);
Map<Object, Object> map = entries.readAllMap();
if(map.get("roomId") == null){
return null;
}
return JSON.parseObject(JSON.toJSONString(entries),UserData.class);
return JSON.parseObject(JSON.toJSONString(map),UserData.class);
}
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);
// redisTemplate.opsForHash().putAll(key,map);
RMap<Object, Object> mapRedis = redissonClient.getMap(key);
mapRedis.putAll(map);
mapRedis.expire(Duration.ofDays(15));
}
public void initCaller(UserData callerUserData) {
@@ -58,6 +68,7 @@ public class UserDataCache {
public void hMSet(Long roomId,Integer userType,Map<String,Object> data) {
String key = getKey(roomId, userType);
redisTemplate.opsForHash().putAll(key,data);
// redisTemplate.opsForHash().putAll(key,data);
redissonClient.getMap(key).putAll(data);
}
}

4
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/dto/WsRMsgGen.java
View File

@@ -53,8 +53,8 @@ public class WsRMsgGen {
public static WsR hangup(String message, Long roomId, Integer hangUpType) {
Map<String,Object> map = new HashMap<>();
map.put("roomid","roomId");
map.put("type","hangUpType");
map.put("roomid",roomId);
map.put("type",hangUpType);
WsR<Map<String, Object>> ok = WsR.ok(map);
ok.setMethod("hangup");
ok.setMsg(message);

5
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/handler/MessageHandleApplication.java
View File

@@ -10,8 +10,8 @@ import com.ruoyi.cai.ws.constant.HangUpEnums;
import com.ruoyi.cai.ws.dto.WsToken;
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.WsExceptionUtil;
import jodd.util.StringUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.TextMessage;
@@ -36,6 +36,9 @@ public class MessageHandleApplication {
public void processOn(WebSocketSession session, TextMessage message) {
String payload = message.getPayload();
if(StringUtil.isEmpty(payload)){
return;
}
JSONObject jsonObject = JSON.parseObject(payload);
Object method = jsonObject.get("method");
if(method == null){

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

@@ -1,11 +1,17 @@
package com.ruoyi.cai.ws.handler;
import cn.hutool.core.util.URLUtil;
import com.google.common.base.Splitter;
import com.ruoyi.cai.chat.ChatManager;
import com.ruoyi.cai.ws.cache.FdCtxDataCache;
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.service.RoomService;
import com.ruoyi.cai.ws.util.WebSocketUtils;
import com.ruoyi.common.utils.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.client.utils.URIUtils;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@@ -14,6 +20,9 @@ import org.springframework.web.socket.handler.AbstractWebSocketHandler;
import javax.websocket.server.PathParam;
import java.io.IOException;
import java.net.URI;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
/**
@@ -34,10 +43,27 @@ public class RoomWebSocketHandler extends AbstractWebSocketHandler {
*/
@Override
public void afterConnectionEstablished(WebSocketSession session) {
URI uri = session.getUri();
Map<String, String> para = getPara(uri == null ? null : uri.getQuery());
Map<String, Object> attributes = session.getAttributes();
attributes.putAll(para);
openLogic.processOn(session);
log.info("[connect] sessionId: {},userId:{}", session.getId(), session.getId());
}
private Map<String,String> getPara(String uri) {
Map<String,String> map = new HashMap<>();
if(StringUtils.isEmpty(uri)){
return map;
}
String[] keys = uri.split("&");
for (String key : keys) {
String[] split = key.split("=");
map.put(split[0],split.length > 1 ? split[1] : "");
}
return map;
}
/**
* 处理发送来的文本消息
*
@@ -47,7 +73,11 @@ public class RoomWebSocketHandler extends AbstractWebSocketHandler {
*/
@Override
protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
messageHandleApplication.processOn(session,message);
try {
messageHandleApplication.processOn(session,message);
}catch (Exception e){
log.error("ws消息处理失败！需要开发检查问题!",e);
}
}
@Override
@@ -79,6 +109,10 @@ public class RoomWebSocketHandler extends AbstractWebSocketHandler {
log.error("[transport error] sessionId: {} , exception:{}", session.getId(), exception.getMessage());
}
@Autowired
private RoomService roomService;
@Autowired
private FdCtxDataCache fdCtxDataCache;
/**
* 连接关闭后
*
@@ -88,6 +122,7 @@ public class RoomWebSocketHandler extends AbstractWebSocketHandler {
@Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
String token = String.valueOf(session.getAttributes().get("token"));
// TODO 连接关闭 是否要删除fd的关系
WebSocketSessionHolder.removeSession(token);
log.info("[disconnect] sessionId: {},token:{}", session.getId(), token);
}

2
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/handler/message/HangupMessageHandler.java
View File

@@ -37,7 +37,7 @@ public class HangupMessageHandler extends AbstractMessageHandle implements IMess
return;
}
// 触发结算
settleService.processOn(room);
settleService.processOn(room.getRoomId());
Integer type = fdCtxData.isCaller() ? HangUpEnums.FROM.getCode() : HangUpEnums.TO.getCode();
sendToCurrent(fdCtxData,WsRMsgGen.hangup("您已挂断",room.getRoomId(), type));
sendToTar(fdCtxData,WsRMsgGen.hangup("对方已挂断",room.getRoomId(), type));

18
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/job/CheckTimeOutJob.java
View File

@@ -9,18 +9,18 @@ import com.ruoyi.cai.trd.ImMsgGen;
import com.ruoyi.cai.ws.bean.Room;
import com.ruoyi.cai.ws.bean.UserData;
import com.ruoyi.cai.ws.cache.OnlineDataCache;
import com.ruoyi.cai.ws.cache.RoomCtxCache;
import com.ruoyi.cai.ws.constant.RoomStatusEnums;
import com.ruoyi.cai.ws.service.RoomService;
import com.ruoyi.cai.ws.service.SettleService;
import com.ruoyi.yunxin.Yunxin;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.util.Set;
import java.util.concurrent.TimeUnit;
@Component
@Slf4j
public class CheckTimeOutJob {
@Autowired
private OnlineDataCache onlineDataCache;
@@ -29,13 +29,14 @@ public class CheckTimeOutJob {
@Autowired
private UserCallService userCallService;
@Autowired
private RoomCtxCache roomCtxCache;
private SettleService settleService;
@Autowired
private Yunxin yunxin;
/**
* 1 分钟执行一次
* 检查房间是不是三分钟没有接听， 需要自动挂断掉
*/
@Scheduled(fixedDelay = 60,timeUnit = TimeUnit.SECONDS)
// @Scheduled(fixedDelay = 60,timeUnit = TimeUnit.SECONDS)
public void run(){
Set<String> roomIdStr = onlineDataCache.getAll();
for (String roomIdS : roomIdStr) {
@@ -45,7 +46,11 @@ public class CheckTimeOutJob {
}
public void deal(Long roomId){
log.info("开始执行房间超时检测，是否一直不接 roomId={}",roomId);
Room room = roomService.load(roomId);
if(room == null){
return;
}
if(!RoomStatusEnums.STATUS_CALLER_CONNECT.getCode().equals(room.getStatus())
&& !RoomStatusEnums.STATUS_RECEIVER_CONNECT.getCode().equals(room.getStatus())){
return;
@@ -66,6 +71,7 @@ public class CheckTimeOutJob {
userCallService.update(Wrappers.lambdaUpdate(UserCall.class)
.eq(UserCall::getId,roomId)
.set(UserCall::getStatus, RoomStatusEnums.STATUS_TIMEOUT_CANCEL.getCode()));
// TODO 需要放开主播的接听状态
roomService.closeAllFd(roomId);
ImDataRes imDataRes = ImMsgGen.callNotice(3, callUserData.getId(), receiverUserData.getId(), 0);
yunxin.sendToSync(receiverUserData.getId(),callUserData.getId(),imDataRes);

7
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/job/HeartbeatJob.java
View File

@@ -41,7 +41,7 @@ public class HeartbeatJob {
/**
* 30秒执行一次
* 30秒执行一次 心跳只处理接听后的心跳
*/
@Scheduled(fixedDelay = 30,timeUnit = TimeUnit.SECONDS)
public void run(){
@@ -50,6 +50,9 @@ public class HeartbeatJob {
try {
Long roomId = Long.valueOf(roomIdS);
Room room = roomService.load(roomId);
if(room == null){
return;
}
if(!room.isCanCall()){
return;
}
@@ -69,7 +72,7 @@ public class HeartbeatJob {
userCallService.update(Wrappers.lambdaUpdate(UserCall.class)
.eq(UserCall::getId,roomId)
.set(UserCall::getStatus, RoomStatusEnums.STATUS_TIMEOUT_CANCEL.getCode()));
settleService.processOn(room);
settleService.processOn(roomId);
}
List<String> keys = roomCtxCache.getSessionKeysByRoomId(roomId);
RoomWebSocketUtil.sendSendMessage(keys, hangup);

8
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/job/SettleJob.java
View File

@@ -25,17 +25,13 @@ public class SettleJob {
/**
* 每 1 分钟执行一次
*/
@Scheduled(fixedDelay = 60,timeUnit = TimeUnit.SECONDS)
// @Scheduled(fixedDelay = 60,timeUnit = TimeUnit.SECONDS)
public void run(){
Set<String> all = onlineDataCache.getAll();
for (String roomIdS : all) {
try {
Long roomId = Long.valueOf(roomIdS);
Room room = roomService.load(roomId);
if(room == null){
return;
}
settleService.processOn(room);
settleService.processOn(roomId);
}catch (Exception e){
log.info("定时任务结算失败！",e);
}

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

@@ -48,7 +48,7 @@ public class WebSocketManager {
return null;
}
Room room = roomService.load(roomId);
if(room == null){
if(room == null || !room.isCanCall()){
return null;
}
return room;
@@ -104,7 +104,6 @@ public class WebSocketManager {
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);
@@ -119,7 +118,7 @@ public class WebSocketManager {
receiveUserData.setRoomId(call.getId());
receiveUserData.setNickname(receiverUser.getNickname());
receiveUserData.setUserCode(receiverUser.getUsercode());
userDataCache.initReceiver(callerUserData);
userDataCache.initReceiver(receiveUserData);
callerRoomCache.addRoom(call.getFromUid(),call.getToUid(),call.getId());
return call.getId();

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

@@ -73,7 +73,7 @@ public class OpenLogic {
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))){
if(room == null || (!room.getCallUserData().getId().equals(userId) && !room.getReceiverUserData().getId().equals(userId))){
WsExceptionUtil.throwException(session,"房间不可用", HangUpEnums.OTHER,roomId);
return;
}

22
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/service/RoomService.java
View File

@@ -138,8 +138,8 @@ public class RoomService {
if(beginTime == null){
return 0L;
}
if(roomData.getHangUpTime() != null){
return roomData.getHangUpTime() - roomData.getBeginTime();
if(roomData.getHangupTime() != null){
return roomData.getHangupTime() - roomData.getBeginTime();
}
return DateUtil.currentSeconds() - beginTime;
@@ -180,7 +180,7 @@ public class RoomService {
Long receiverId = receiverUserData.getId();
Long callerId = room.getCallUserData().getId();
// 呼叫方释放资源
callerRoomCache.delRoom(receiverId,roomId);
callerRoomCache.delRoom(callerId,receiverId,roomId);
userService.updateVideoStatus(callerId,0);
// 接收方释放资源（已连接的情况下）
if(receiverUserData.getConnectTime() != null && receiverUserData.getConnectTime() > 0){
@@ -195,11 +195,25 @@ public class RoomService {
}
// 修改释放状态
Map<String,Object> map = new HashMap<>();
map.put("releaseRes",1);
map.put("releaseRes",true);
roomDataCache.hMSet(roomId,map);
}catch (Exception e){
log.error("释放房间资源失败!房间号：{}",roomId,e);
}
}
/**
* 检查是否正在通话中
*/
public boolean checkRunningVideo(Long toUserId, Long fromUserId) {
Long roomId = callerRoomCache.getRoomId(toUserId, fromUserId);
if(roomId != null){
Room room = load(roomId);
if(room != null && room.isCanCall()){
return true;
}
}
return false;
}
}

13
ruoyi-cai/src/main/java/com/ruoyi/cai/ws/service/SettleService.java
View File

@@ -76,7 +76,7 @@ public class SettleService {
boolean b = roomService.hangUp(room.getRoomId());
if(b){
// 结算操作
settleService.processOn(room);
settleService.processOn(room.getRoomId());
}
// 向客户端发送挂断指令
WsR r = WsRMsgGen.hangup("拨打方余额不足", room.getRoomId(), HangUpEnums.NOTMONEY.getCode());
@@ -99,10 +99,14 @@ public class SettleService {
/**
* 结算处理
* @param room
*/
@SneakyThrows
public void processOn(Room room){
public void processOn(Long roomId){
Room room = roomService.load(roomId);
if(room == null){
log.warn("房间不存在，无需结算 roomId={}",roomId);
return;
}
String lock = LockManager.getVideoSettleLock(room.getRoomId());
RLock clientLock = redissonClient.getLock(lock);
boolean locked = clientLock.isLocked();
@@ -110,7 +114,7 @@ public class SettleService {
log.info("正在结算中，稍等！");
return;
}
boolean lockFlag = clientLock.tryLock(5, TimeUnit.SECONDS);
boolean lockFlag = clientLock.tryLock();
if(!lockFlag){
log.info("正在结算中，稍等！");
return;
@@ -133,7 +137,6 @@ public class SettleService {
}
if(!room.isReleaseRes()){ // 房间资源是否已经释放
roomService.releaseRes(room.getRoomId());
return;
}
Long roomId = room.getRoomId();
// 未通话，无需结算