init

ruoyi-cai/src/main/java/com/ruoyi/cai/agora/Agora.java

@@ -0,0 +1,65 @@
+package com.ruoyi.cai.agora;
+
+import com.alibaba.fastjson2.JSONArray;
+import com.alibaba.fastjson2.JSONObject;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.stereotype.Component;
+import org.springframework.web.client.RestTemplate;
+
+import java.util.*;
+
+@Component
+public class Agora {
+
+    private RestTemplate restTemplate = new RestTemplate();
+
+    private final String APP_ID = "appId";
+
+    public String getAuthorizationHeader(){
+        // 客户 ID
+        final String customerKey = "Your customer key";
+        // 客户密钥
+        final String customerSecret = "Your customer secret";
+
+        // 拼接客户 ID 和客户密钥并使用 base64 编码
+        String plainCredentials = customerKey + ":" + customerSecret;
+        String base64Credentials = new String(Base64.getEncoder().encode(plainCredentials.getBytes()));
+        // 创建 authorization header
+        String authorizationHeader = "Basic " + base64Credentials;
+        return authorizationHeader;
+    }
+
+
+    public List<String> getChannelUsers(String roomId){
+        String url = "http://api.sd-rtn.com/dev/v1/channel/user/%s/%s";
+        url = String.format(url, APP_ID, roomId);
+        HttpHeaders headers = new HttpHeaders();
+        headers.setContentType(MediaType.APPLICATION_JSON);
+        headers.set("Authorization", getAuthorizationHeader());
+        HttpEntity httpEntity = new HttpEntity<>(headers);
+        JSONObject jsonobject = restTemplate.getForObject(url,JSONObject.class,httpEntity);
+        // "success":true,"data":{"channel_exist":true,"mode":2,"broadcasters":[1001,1025],"audience":[],"audience_total":0}}
+        if(jsonobject == null){
+            return Collections.emptyList();
+        }
+        JSONArray jsonArray = jsonobject.getJSONObject("data").getJSONArray("broadcasters");
+        return jsonArray.toJavaList(String.class);
+    }
+
+    public void closeChannel(String roomId){
+        String url = "http://api.sd-rtn.com/dev/v1/kicking-rule";
+        Map<String,Object> param = new HashMap<>();
+        param.put("appid",APP_ID);
+        param.put("cname",roomId);
+        param.put("time",0);
+        param.put("privileges",Collections.emptyList());
+        HttpHeaders headers = new HttpHeaders();
+        headers.setContentType(MediaType.APPLICATION_JSON);
+        headers.set("Authorization", getAuthorizationHeader());
+        HttpEntity httpEntity = new HttpEntity<>(param, headers);
+        restTemplate.postForObject(url, httpEntity, JSONObject.class);
+    }
+
+}

ruoyi-cai/src/main/java/com/ruoyi/cai/executor/ExecutorConstant.java

@@ -17,6 +17,7 @@ public class ExecutorConstant {
     public static Executor SYNC_EXECUTOR;
 
     public static Executor COMMON_EXECUTOR;
+    public static Executor ROOM_EXECUTOR;
 
     static {
         ThreadPoolExecutor syncExecutor = initExecutor(CPU_NUM,
@@ -36,6 +37,15 @@ public class ExecutorConstant {
                 new ThreadPoolExecutor.CallerRunsPolicy());
         COMMON_EXECUTOR = TtlExecutors.getTtlExecutor(commonExecutor);
 
+        ThreadPoolExecutor roomExecutor = new ThreadPoolExecutor(CPU_NUM,
+                CPU_NUM << 2,
+                5,
+                TimeUnit.SECONDS,
+                new ArrayBlockingQueue<>(50),
+                init("roomThreadPoll-%d"),
+                new ThreadPoolExecutor.CallerRunsPolicy());
+        ROOM_EXECUTOR = TtlExecutors.getTtlExecutor(roomExecutor);
+
     }
 
     private static ThreadFactory init(String nameFormat){

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

@@ -1,6 +1,8 @@
 package com.ruoyi.cai.ws.processon;
 
 import cn.hutool.core.date.DateUtil;
+import com.ruoyi.cai.agora.Agora;
+import com.ruoyi.cai.executor.ExecutorConstant;
 import com.ruoyi.cai.service.CaiAnchorService;
 import com.ruoyi.cai.service.CaiUserService;
 import com.ruoyi.cai.ws.bean.FdCtxData;
@@ -20,6 +22,7 @@ import org.springframework.stereotype.Component;
 import org.springframework.web.socket.WebSocketSession;
 
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 @Component
@@ -40,6 +43,8 @@ public class OpenLogic implements IOpenLogic {
     private CaiUserService userService;
     @Autowired
     private CaiAnchorService anchorService;
+    @Autowired
+    private Agora agora;
 
     @Override
     public void processOn(WebSocketSession session) {
@@ -104,25 +109,20 @@ public class OpenLogic implements IOpenLogic {
         }
         // 已经接通
         if(room.isOnline()){
-            // 如果视频也掉线了，则重连的时候发送消息提示
+            ExecutorConstant.ROOM_EXECUTOR.execute(() -> {
-//            sgo(function() use ($rs){
+                // 如果视频也掉线了，则重连的时候发送消息提示
-//                $users = Agora::getChannelUsers($rs->room->id);
+                List<String> channelUsers = agora.getChannelUsers(room.getRoomId());
-//                if (in_array($rs->caller->id, $users)) {
+                if(channelUsers.contains(userId+"")){
-//                    return;
+                    return;
-//                }
+                }
-//
+                Long callTime = roomService.getCallTime(room);
-//                $rs->sendToCurrent(WsMsgGen::startVideo($rs->room->id, $rs->callTime()))
+                RoomWebSocketUtil.sendSendMessage(sessionKey,WsRMsgGen.startVideo(room.getRoomId(), callTime));
-//                    ->sendToCurrent(WsMsgGen::sysNotice('重连成功，房间已通话 '.gmdate('H:i:s', $rs->callTime())))
+                RoomWebSocketUtil.sendSendMessage(sessionKey,WsRMsgGen.sysNotice("重连成功，房间已通话(转换成时分秒) "));
-//                    ->sendToReceiver(WsMsgGen::sysNotice('对方已重连成功'));
+                String sessionKeyReceiver = roomCtxCache.getSessionKeyReceiverByRoomId(room.getRoomId());
-//            });
+                if(StringUtils.isNotEmpty(sessionKeyReceiver)){
-
+                    RoomWebSocketUtil.sendSendMessage(sessionKey,WsRMsgGen.sysNotice("对方已重连成功"));
-            Long callTime = roomService.getCallTime(room);
+                }
-            RoomWebSocketUtil.sendSendMessage(sessionKey,WsRMsgGen.startVideo(room.getRoomId(), callTime));
+            });
-            RoomWebSocketUtil.sendSendMessage(sessionKey,WsRMsgGen.sysNotice("重连成功，房间已通话(转换成时分秒) "));
-            String sessionKeyReceiver = roomCtxCache.getSessionKeyReceiverByRoomId(room.getRoomId());
-            if(StringUtils.isNotEmpty(sessionKeyReceiver)){
-                RoomWebSocketUtil.sendSendMessage(sessionKey,WsRMsgGen.sysNotice("对方已重连成功"));
-            }
         }
         if(RoomStatusEnums.STATUS_CREATE.getCode().equals(status) ||
                 RoomStatusEnums.STATUS_CALLER_CONNECT.getCode().equals(status) ||
@@ -131,7 +131,7 @@ public class OpenLogic implements IOpenLogic {
             RoomWebSocketUtil.sendSendMessage(sessionKey, WsRMsgGen.response(room.getRoomId()));
         }
         if(isFirst){
-            // 给对方发送呼叫页面
+            // 给对方发送呼叫页面 TODO
         }
     }
     public void receiverConnection(String sessionKey,Room room,Long userId){

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

@@ -1,16 +1,23 @@
 package com.ruoyi.cai.ws.util;
 
+import com.esotericsoftware.minlog.Log;
 import com.ruoyi.cai.ws.constant.HangUpEnums;
+import com.ruoyi.websocket.dto.WsRMsgGen;
+import com.ruoyi.websocket.util.RoomWebSocketUtil;
+import com.ruoyi.websocket.util.WebSocketUtils;
+import lombok.extern.slf4j.Slf4j;
 
+@Slf4j
 public class WsExceptionUtil {
     public static void throwException(String sessionKey, String message, HangUpEnums hangUpType, String 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 String handUp(String sessionKey,String message,String roomId,HangUpEnums hang){
-        return null;
-    }
 }

ruoyi-websocket/src/main/java/com/ruoyi/websocket/dto/WsRMsgGen.java

@@ -49,4 +49,14 @@ public class WsRMsgGen {
         ok.setMsg("提示！");
         return ok;
     }
+
+    public static WsR hangup(String message, String roomId, Integer hangUpType) {
+        Map<String,Object> map = new HashMap<>();
+        map.put("roomid","roomId");
+        map.put("type","hangUpType");
+        WsR<Map<String, Object>> ok = WsR.ok(map);
+        ok.setMethod("hangup");
+        ok.setMsg(message);
+        return ok;
+    }
 }

ruoyi-websocket/src/main/java/com/ruoyi/websocket/util/WebSocketUtils.java

@@ -1,6 +1,7 @@
 package com.ruoyi.websocket.util;
 
 import com.ruoyi.websocket.holder.WebSocketSessionHolder;
+import jdk.nashorn.internal.runtime.regexp.JoniRegExp;
 import lombok.AccessLevel;
 import lombok.NoArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -31,6 +32,19 @@ public class WebSocketUtils {
         sendMessage(session, message);
     }
 
+    public static boolean close(String sessionKey) {
+        WebSocketSession sessions = WebSocketSessionHolder.getSessions(sessionKey);
+        if(sessions != null){
+            try {
+                sessions.close();
+                return true;
+            } catch (IOException e) {
+                log.error("关闭ws失败，sessionKey={}",sessionKey,e);
+            }
+        }
+        return false;
+    }
+
     public static void sendPongMessage(WebSocketSession session) {
         sendMessage(session, new PongMessage());
     }