123

ruoyi-cai/src/main/java/com/ruoyi/cai/ws/bean/Room.java

@@ -42,4 +42,12 @@ public class Room {
     public boolean isReleaseRes(){
         return roomData.isReleaseRes();
     }
+
+    public Long getCallId() {
+        return callUserData.getId();
+    }
+
+    public Long getReceiverId() {
+        return receiverUserData.getId();
+    }
 }

ruoyi-cai/src/main/java/com/ruoyi/cai/ws/handler/message/CancelMessageHandler.java

@@ -3,11 +3,9 @@ package com.ruoyi.cai.ws.handler.message;
 import com.alibaba.fastjson2.JSONObject;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
 import com.ruoyi.cai.domain.UserCall;
-import com.ruoyi.cai.executor.ExecutorConstant;
 import com.ruoyi.cai.service.UserCallService;
-import com.ruoyi.cai.trd.ImDataRes;
+import com.ruoyi.yunxin.data.ImDataRes;
-import com.ruoyi.cai.trd.ImMsgGen;
+import com.ruoyi.yunxin.data.ImMsgGen;
-import com.ruoyi.cai.ws.cache.RoomDataCache;
 import com.ruoyi.yunxin.Yunxin;
 import com.ruoyi.cai.ws.bean.FdCtxData;
 import com.ruoyi.cai.ws.bean.Room;
@@ -17,6 +15,7 @@ import com.ruoyi.cai.ws.handler.IMessageHandler;
 import com.ruoyi.cai.ws.service.RoomService;
 import com.ruoyi.cai.ws.dto.WsRMsgGen;
 import com.ruoyi.cai.ws.handler.AbstractMessageHandle;
+import com.ruoyi.yunxin.service.YunxinWsService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
@@ -49,13 +48,15 @@ public class CancelMessageHandler extends AbstractMessageHandle implements IMess
         sendToReceiver(roomId,WsRMsgGen.hangup("对方已取消",roomId, HangUpEnums.CANCEL.getCode()));
         roomService.closeAllFd(roomId);
         // IM
-        Long receiverId = room.getReceiverUserData().getId();
+        Long receiverId = room.getReceiverId();
-        Long callerId = room.getCallUserData().getId();
+        Long callerId = room.getCallId();
-        ImDataRes imDataRes = ImMsgGen.callNotice(1, callerId, receiverId, 0);
+        yunxinWsService.sendToCallNotifyAsync(callerId,receiverId,1,0);
-        yunxin.sendToSync(receiverId,callerId,imDataRes);
         // 更新房间状态
         userCallService.update(Wrappers.lambdaUpdate(UserCall.class)
                 .eq(UserCall::getId,roomId)
                 .set(UserCall::getStatus,RoomStatusEnums.STATUS_CALLER_CANCEL.getCode()));
     }
+
+    @Autowired
+    private YunxinWsService yunxinWsService;
 }

ruoyi-cai/src/main/java/com/ruoyi/cai/ws/handler/message/RefuseMessageHandler.java

@@ -4,18 +4,18 @@ import com.alibaba.fastjson2.JSONObject;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
 import com.ruoyi.cai.domain.UserCall;
 import com.ruoyi.cai.service.UserCallService;
-import com.ruoyi.cai.trd.ImDataRes;
+import com.ruoyi.yunxin.data.ImDataRes;
-import com.ruoyi.cai.trd.ImMsgGen;
+import com.ruoyi.yunxin.data.ImMsgGen;
 import com.ruoyi.yunxin.Yunxin;
 import com.ruoyi.cai.ws.bean.FdCtxData;
 import com.ruoyi.cai.ws.bean.Room;
-import com.ruoyi.cai.ws.cache.RoomDataCache;
 import com.ruoyi.cai.ws.constant.HangUpEnums;
 import com.ruoyi.cai.ws.constant.RoomStatusEnums;
 import com.ruoyi.cai.ws.handler.IMessageHandler;
 import com.ruoyi.cai.ws.service.RoomService;
 import com.ruoyi.cai.ws.dto.WsRMsgGen;
 import com.ruoyi.cai.ws.handler.AbstractMessageHandle;
+import com.ruoyi.yunxin.service.YunxinWsService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
@@ -44,14 +44,16 @@ public class RefuseMessageHandler extends AbstractMessageHandle implements IMess
         roomService.closeAllFd(room.getRoomId());
 
         //发送IM通知
-        Long receiverId = room.getReceiverUserData().getId();
+        Long receiverId = room.getReceiverId();
-        Long callerId = room.getCallUserData().getId();
+        Long callerId = room.getCallId();
-        ImDataRes imDataRes = ImMsgGen.callNotice(1, callerId, receiverId, 0);
+        yunxinWsService.sendToCallNotifyAsync(callerId,receiverId,1,0);
-        yunxin.sendToSync(receiverId,callerId,imDataRes);
 
         // 更新房间状态
         userCallService.update(Wrappers.lambdaUpdate(UserCall.class)
                 .eq(UserCall::getId,room.getRoomId())
                 .set(UserCall::getStatus,RoomStatusEnums.STATUS_REFUSE.getCode()));
     }
+
+    @Autowired
+    private YunxinWsService yunxinWsService;
 }

ruoyi-cai/src/main/java/com/ruoyi/cai/ws/job/RoomCheckJobService.java

@@ -4,8 +4,8 @@ import cn.hutool.core.date.DateUtil;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
 import com.ruoyi.cai.domain.UserCall;
 import com.ruoyi.cai.service.UserCallService;
-import com.ruoyi.cai.trd.ImDataRes;
+import com.ruoyi.yunxin.data.ImDataRes;
-import com.ruoyi.cai.trd.ImMsgGen;
+import com.ruoyi.yunxin.data.ImMsgGen;
 import com.ruoyi.cai.ws.bean.Room;
 import com.ruoyi.cai.ws.bean.UserData;
 import com.ruoyi.cai.ws.cache.RoomCtxCache;
@@ -18,6 +18,7 @@ import com.ruoyi.cai.ws.service.SettleResp;
 import com.ruoyi.cai.ws.service.SettleService;
 import com.ruoyi.cai.ws.util.RoomWebSocketUtil;
 import com.ruoyi.yunxin.Yunxin;
+import com.ruoyi.yunxin.service.YunxinWsService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
@@ -128,13 +129,15 @@ public class RoomCheckJobService {
                 nextCreateJob = settleResp.isNextRun();
             }
             roomService.closeAllFd(roomId);
-            ImDataRes imDataRes = ImMsgGen.callNotice(3, callUserData.getId(), receiverUserData.getId(), 0);
+            yunxinWsService.sendToCallNotifyAsync(callUserData.getId(),receiverUserData.getId(),3,0);
-            yunxin.sendToSync(receiverUserData.getId(),callUserData.getId(),imDataRes);
             return JobResp.builder().nextCreateJob(nextCreateJob).build();
         }
         return JobResp.builder().nextCreateJob(true).build();
     }
 
+    @Autowired
+    private YunxinWsService yunxinWsService;
+
 
     /**
      * 是否心跳超时

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

@@ -21,8 +21,7 @@ import com.ruoyi.cai.ws.service.RoomService;
 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.service.YunxinWsService;
-import com.ruoyi.yunxin.data.SendAttachMsgData;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
@@ -55,7 +54,7 @@ public class OpenLogic {
     @Autowired
     private ChatManager chatManager;
     @Autowired
-    private Yunxin yunxin;
+    private YunxinWsService yunxinWsService;
 
     public void processOn(WebSocketSession session) {
         Map<String, Object> attributes = session.getAttributes();
@@ -151,13 +150,7 @@ public class OpenLogic {
             RoomWebSocketUtil.sendSendMessage(session, WsRMsgGen.response(room.getRoomId()));
         }
         if(isFirst){
-            ExecutorConstant.ROOM_EXECUTOR.execute(() -> {
+            yunxinWsService.sendCallAsync(room.getRoomId(),room.getCallId(),room.getReceiverId());
-                // 给对方发送呼叫页面
-                ;
-                SendAttachMsgData msg = SendAttachMsgData.init(room.getRoomId());
-                yunxin.sendAttachMsg(room.getCallUserData().getId(),
-                        room.getReceiverUserData().getId(), msg);
-            });
         }
     }
     public void receiverConnection(WebSocketSession session,Room room,Long userId){

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

@@ -2,29 +2,24 @@ package com.ruoyi.cai.ws.service;
 
 import cn.hutool.core.date.DateUtil;
 import cn.hutool.extra.spring.SpringUtil;
-import com.ruoyi.cai.domain.Account;
 import com.ruoyi.cai.dto.video.VideoSettleResp;
 import com.ruoyi.cai.dto.video.WithholdingFeeUserResp;
 import com.ruoyi.cai.manager.ConsumerManager;
 import com.ruoyi.cai.manager.LockManager;
 import com.ruoyi.cai.service.AccountService;
-import com.ruoyi.cai.trd.ImDataRes;
+import com.ruoyi.yunxin.data.ImDataRes;
-import com.ruoyi.cai.trd.ImMsgGen;
+import com.ruoyi.yunxin.data.ImMsgGen;
-import com.ruoyi.cai.ws.cache.CallerRoomCache;
+import com.ruoyi.cai.ws.bean.Room;
+import com.ruoyi.cai.ws.cache.OnlineDataCache;
 import com.ruoyi.cai.ws.cache.RoomCtxCache;
+import com.ruoyi.cai.ws.cache.RoomDataCache;
 import com.ruoyi.cai.ws.constant.HangUpEnums;
 import com.ruoyi.cai.ws.dto.WsR;
 import com.ruoyi.cai.ws.dto.WsRMsgGen;
 import com.ruoyi.cai.ws.util.RoomWebSocketUtil;
 import com.ruoyi.yunxin.Yunxin;
-import com.ruoyi.cai.ws.bean.Room;
+import com.ruoyi.yunxin.service.YunxinWsService;
-import com.ruoyi.cai.ws.bean.RoomData;
-import com.ruoyi.cai.ws.bean.UserData;
-import com.ruoyi.cai.ws.cache.OnlineDataCache;
-import com.ruoyi.cai.ws.cache.RoomDataCache;
-import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.poi.ss.formula.functions.Roman;
 import org.redisson.api.RLock;
 import org.redisson.api.RedissonClient;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -34,7 +29,6 @@ import org.springframework.transaction.annotation.Transactional;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.TimeUnit;
 
 @Component
 @Slf4j
@@ -164,6 +158,7 @@ public class SettleService {
         if(callTime > 0){
             ImDataRes imDataRes = ImMsgGen.callNotice(4, callId, receiveId, callTime);
             yunxin.sendToSync(callId,receiveId,imDataRes);
+            yunxinWsService.sendToCallNotifyAsync(callId,receiveId,4,callTime);
         }
         // 接收方通知
         Long anchorIncome = resp.getAnchorIncome();
@@ -177,4 +172,7 @@ public class SettleService {
         return SettleResp.builder().nextRun(false).build();
     }
 
+    @Autowired
+    private YunxinWsService yunxinWsService;
+
 }

ruoyi-yunxin/src/main/java/com/ruoyi/yunxin/YunExecutor.java

@@ -0,0 +1,40 @@
+package com.ruoyi.yunxin;
+
+import com.alibaba.ttl.threadpool.TtlExecutors;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+import java.util.concurrent.*;
+
+/**
+ * 线程变量定义
+ * <p>created on 2023/3/3 11:16</p>
+ * @author ZL
+ */
+public class YunExecutor {
+
+    private final static int CPU_NUM = Runtime.getRuntime().availableProcessors();
+
+    public static Executor YUN_EXECUTOR;
+
+    static {
+        ThreadPoolExecutor roomExecutor = new ThreadPoolExecutor(CPU_NUM,
+                CPU_NUM << 2,
+                5,
+                TimeUnit.SECONDS,
+                new ArrayBlockingQueue<>(5),
+                init("yunxinThreadPoll-%d"),
+                new ThreadPoolExecutor.CallerRunsPolicy());
+        YUN_EXECUTOR = TtlExecutors.getTtlExecutor(roomExecutor);
+
+    }
+
+    private static ThreadFactory init(String nameFormat){
+        return new ThreadFactoryBuilder().setNameFormat(nameFormat).build();
+    }
+
+    private static ThreadPoolExecutor initExecutor(int corePoolSize, int maxPoolSize, int keepAliveTime, TimeUnit timeUnit,
+                                         BlockingQueue<Runnable> workQueue, String nameFormat){
+        return new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, timeUnit, workQueue,
+                init(nameFormat));
+    }
+}

ruoyi-yunxin/src/main/java/com/ruoyi/yunxin/Yunxin.java

@@ -28,9 +28,7 @@ public class Yunxin {
     private ImMessageClient messageClient;
 
     public void sendToSync(Long toUid,Long fromUid,Object data){
-        YunxinExecutorConstant.COMMON_EXECUTOR.execute(() -> {
         this.sendTo(toUid,fromUid,data);
-        });
     }
 
     public YxDataR<SendMsgResp> sendTo(Long toUid,Long fromUid,Object data){
@@ -39,6 +37,7 @@ public class Yunxin {
         req.setTo(toUid+"");
         req.setBody(JSON.toJSONString(data));
         req.setOption(JSON.toJSONString(new Option()));
+        req.setType(100);
         return messageClient.sendMsg(req);
     }
 

ruoyi-yunxin/src/main/java/com/ruoyi/yunxin/client/ImMessageClient.java

@@ -15,7 +15,7 @@ import com.ruoyi.yunxin.resp.YxDataR;
 public interface ImMessageClient {
 
     /**
-     * 发送自定义消息
+     * 发送消息
      * @param req
      * @return
      */

ruoyi-yunxin/src/main/java/com/ruoyi/yunxin/data/ImDataRes.java

@@ -1,4 +1,4 @@
-package com.ruoyi.cai.trd;
+package com.ruoyi.yunxin.data;
 
 import lombok.Data;
 
@@ -9,11 +9,11 @@ public class ImDataRes {
 
     @Data
     public static class ImData {
+        private Integer calltype;
         // 通话状态，1 取消 2 拒绝 3 超时 4 已通话
-        private Integer callType;
         private Integer status;
-        private Long fromUid;
+        private Long from_uid;
-        private Long toUid;
+        private Long to_uid;
-        private Long callTime;
+        private Long call_time;
     }
 }

ruoyi-yunxin/src/main/java/com/ruoyi/yunxin/data/ImMsgGen.java

@@ -1,4 +1,4 @@
-package com.ruoyi.cai.trd;
+package com.ruoyi.yunxin.data;
 
 public class ImMsgGen {
 
@@ -13,11 +13,11 @@ public class ImMsgGen {
         ImDataRes imDataRes = new ImDataRes();
         imDataRes.setType(15);
         ImDataRes.ImData imData = new ImDataRes.ImData();
-        imData.setCallType(1); // 通话状态，1 取消 2 拒绝 3 超时 4 已通话
+        imData.setCalltype(1);
-        imData.setStatus(status);
+        imData.setStatus(status); // 通话状态，1 取消 2 拒绝 3 超时 4 已通话
-        imData.setFromUid(from);
+        imData.setFrom_uid(from);
-        imData.setToUid(to);
+        imData.setTo_uid(to);
-        imData.setCallTime(callTime);
+        imData.setCall_time(callTime);
         imDataRes.setData(imData);
         return imDataRes;
     }

ruoyi-yunxin/src/main/java/com/ruoyi/yunxin/data/SendAttachMsgData.java

@@ -1,33 +0,0 @@
-package com.ruoyi.yunxin.data;
-
-import cn.hutool.core.date.DateUtil;
-import lombok.Data;
-
-/**
- * 系统消息通知
- */
-@Data
-public class SendAttachMsgData {
-    private Long id = 11L;
-    private SendAttachMsgDataMsg data;
-
-    @Data
-    public static class SendAttachMsgDataMsg {
-        private Long amount;
-        private Long callTime = DateUtil.currentSeconds();
-        private Long giftTotal = 0L;
-        private Long id = 0L;
-        private Integer linkType = 0;
-        private Long time = 0L;
-        private Long toid = 0L;
-        private Long roomId;
-    }
-
-    public static SendAttachMsgData init(Long roomId){
-        SendAttachMsgData data = new SendAttachMsgData();
-        SendAttachMsgDataMsg msgDataMsg = new SendAttachMsgDataMsg();
-        msgDataMsg.setRoomId(roomId);
-        data.setData(msgDataMsg);
-        return data;
-    }
-}

ruoyi-yunxin/src/main/java/com/ruoyi/yunxin/data/WsSendAttachMsgData.java

@@ -0,0 +1,26 @@
+package com.ruoyi.yunxin.data;
+
+import cn.hutool.core.date.DateUtil;
+import lombok.Data;
+
+/**
+ * ws专用
+ */
+@Data
+public class WsSendAttachMsgData {
+    private Integer id = 11;
+    private SendAttachMsgDataMsg data;
+
+    @Data
+    public static class SendAttachMsgDataMsg {
+        private Integer amount = 0;
+        private Long callTime = DateUtil.currentSeconds();
+        private Long gifttotal = 0L;
+        private Integer id = 0;
+        private Integer link_type = 0;
+        private Long time = 0L;
+        private Long toid = 0L;
+        private Long roomid;
+    }
+
+}

ruoyi-yunxin/src/main/java/com/ruoyi/yunxin/service/YunxinWsService.java

@@ -0,0 +1,45 @@
+package com.ruoyi.yunxin.service;
+
+import com.alibaba.fastjson.JSON;
+import com.ruoyi.yunxin.YunExecutor;
+import com.ruoyi.yunxin.Yunxin;
+import com.ruoyi.yunxin.data.ImDataRes;
+import com.ruoyi.yunxin.data.ImMsgGen;
+import com.ruoyi.yunxin.data.WsSendAttachMsgData;
+import com.ruoyi.yunxin.resp.SendMsgResp;
+import com.ruoyi.yunxin.resp.YxCommonR;
+import com.ruoyi.yunxin.resp.YxDataR;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+@Component
+@Slf4j
+public class YunxinWsService {
+
+    @Autowired
+    private Yunxin yunxin;
+    public void sendCallAsync(Long roomId,Long callId,Long receiverId){
+        YunExecutor.YUN_EXECUTOR.execute(() -> {
+            WsSendAttachMsgData data = new WsSendAttachMsgData();
+            WsSendAttachMsgData.SendAttachMsgDataMsg msgDataMsg = new WsSendAttachMsgData.SendAttachMsgDataMsg();
+            msgDataMsg.setRoomid(roomId);
+            data.setData(msgDataMsg);
+            YxDataR<YxCommonR> r = yunxin.sendAttachMsg(callId, receiverId, data);
+            if(r == null || !r.isSuccess()){
+                log.error("云信发送失败【sendCallAsync】r={}", JSON.toJSONString(r));
+            }
+        });
+    }
+
+    public void sendToCallNotifyAsync(Long from,Long to,Integer status,Long calltime){
+        YunExecutor.YUN_EXECUTOR.execute(() -> {
+            ImDataRes imDataRes = ImMsgGen.callNotice(status, from, to, calltime);
+            YxDataR<SendMsgResp> r = yunxin.sendTo(from, to, imDataRes);
+            if(r == null || !r.isSuccess()){
+                log.error("云信发送失败【sendCallAsync】r={}", JSON.toJSONString(r));
+            }
+        });
+    }
+
+}