优化：websocket接收消息

liusuyi

2024-08-12 91b70d5ad2ada85cf00b25f7b9ecd9cf980bf138

优化：websocket接收消息

 ard-work/src/main/java/com/ruoyi/app/position/service/impl/AppPositionPushService.java

@@ -88,12 +88,12 @@
            IArdAppPositionService ardAppPositionService = SpringUtils.getBean(IArdAppPositionService.class);
            ArdAppPosition ardAppPosition = ardAppPositionService.selectLastArdAppPositionByUserId(sysUser.getUserId());
            if (StringUtils.isNotNull(ardAppPosition)) {
                Map<String, Object> params=new HashMap<>();
                params.put("longitude",ardAppPosition.getLongitude());
                params.put("latitude",ardAppPosition.getLatitude());
                params.put("speed",ardAppPosition.getSpeed());
                params.put("bearing",ardAppPosition.getBearing());
                params.put("name",sysUser.getUserName());
                Map<String, Object> params = new HashMap<>();
                params.put("longitude", ardAppPosition.getLongitude());
                params.put("latitude", ardAppPosition.getLatitude());
                params.put("speed", ardAppPosition.getSpeed());
                params.put("bearing", ardAppPosition.getBearing());
                params.put("name", sysUser.getUserName());
                ardAppPosition.setParams(params);
                ArdAppPositions.add(ardAppPosition);
            }
@@ -108,8 +108,8 @@
            for (ArdAppPosition ardAppPosition : data) {
                ISysUserService sysUserService = SpringUtils.getBean(ISysUserService.class);
                SysUser sysUser = sysUserService.selectUserById(ardAppPosition.getUserId());
                if(StringUtils.isNotNull(sysUser)) {
                    if(StringUtils.isNotNull(sysUser.getAppOnlineState())) {
                if (StringUtils.isNotNull(sysUser)) {
                    if (StringUtils.isNotNull(sysUser.getAppOnlineState())) {
                        if (sysUser.getAppOnlineState().equals("1")) {
                            onlineList.add(ardAppPosition);
                        }
@@ -126,8 +126,7 @@
            Map<String, Object> map = JSONObject.parseObject(message, Map.class);
            if (map.size() > 0) {
                Boolean enabled = (Boolean) map.get("enabled");
                if(enabled==null)
                {
                if (enabled == null) {
                    return;
                }
                if (enabled) {
@@ -177,10 +176,9 @@
                    AppPositionPushService.tempDataMap.remove(userId);
                }
            }
        }
        catch (Exception ex)
        {
        } catch (Exception ex) {
            log.error(ex.getMessage());
            WebSocketUtils.sendMessage(userId, "app位置推送解析消息【" + message + "】异常:" + ex.getMessage());
        }
    }


 ard-work/src/main/java/com/ruoyi/call/listener/WebsocketHandler.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,68 @@
package com.ruoyi.call.listener;

import com.alibaba.fastjson2.JSONObject;
import com.ruoyi.call.dto.CallMessage;
import com.ruoyi.utils.websocket.util.WebSocketUtils;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * å¤„理websocket消息处理视频通话消息
 *
 * @author liusuyi
 * @date 2024å¹´08月12日15:56
 **/
public class WebsocketHandler {
    public static void handleMessage(String userId, String message) {
        try {
            Map<String, Object> messageMap = JSONObject.parseObject(message, Map.class);
            if (messageMap.get("type") != null && messageMap.get("type").equals("callMessage")) {
                String msg = (String) messageMap.get("message");
                CallMessage callMessage = JSONObject.parseObject(msg, CallMessage.class);
                //单聊
                if (callMessage != null && callMessage.getType().equals(0)) {
                    WebSocketUtils.sendMessagePrefix(callMessage.getTargetId(), message);
                }
                //群聊
                if (callMessage != null && callMessage.getType().equals(1)) {
                    String step = callMessage.getStep();
                    switch (step) {
                        case "joinRoom":
                            Set<String> userSet = WebSocketUtils.ROOM_USER_SET.getOrDefault(callMessage.getRoomId(), new HashSet<>());
                            userSet.add(callMessage.getUserId());
                            WebSocketUtils.ROOM_USER_SET.put(callMessage.getRoomId(), userSet);
                            WebSocketUtils.sendRoomMessage(callMessage.getRoomId(), message);
                            break;
                        case "inviteRoom":
                            List<String> targetIds = callMessage.getTargetIds();
                            targetIds.stream().forEach(targetId -> {
                                WebSocketUtils.sendMessagePrefix(targetId, message);
                            });
                            break;
                        case "leaveRoom":
                            userSet = WebSocketUtils.ROOM_USER_SET.getOrDefault(callMessage.getRoomId(), new HashSet<>());
                            if (userSet.size() > 0) {
                                userSet.remove(callMessage.getUserId());
                                WebSocketUtils.ROOM_USER_SET.put(callMessage.getRoomId(), userSet);
                                WebSocketUtils.sendRoomMessage(callMessage.getRoomId(), message);
                            }
                            break;
                        default:
                            WebSocketUtils.sendMessagePrefix(callMessage.getTargetId(), message);
                            break;
                    }


                }
            }
            else
            {
                WebSocketUtils.sendMessage(userId, "视频通话无法解析消息【" + message + "】");
            }
        } catch (Exception ex) {
            WebSocketUtils.sendMessage(userId, "视频通话解析消息【" + message + "】异常:" + ex.getMessage());
        }
    }
}

 ard-work/src/main/java/com/ruoyi/utils/websocket/service/ChatServerEndpoint.java

@@ -1,8 +1,12 @@
package com.ruoyi.utils.websocket.service;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONException;
import com.alibaba.fastjson2.JSONObject;
import com.ruoyi.app.position.service.impl.AppPositionPushService;
import com.ruoyi.call.dto.CallMessage;
import com.ruoyi.call.listener.WebsocketHandler;
import com.ruoyi.common.exception.ServiceException;
import com.ruoyi.utils.websocket.util.WebSocketUtils;
import lombok.extern.slf4j.Slf4j;
@@ -44,49 +48,19 @@
    @OnMessage
    public void onMessage(@PathParam("userId") String userId, String message) {
        log.debug("收到消息：" + message);
        Map<String, Object> messageMap = JSONObject.parseObject(message, Map.class);
        if (messageMap.get("type")!=null &&messageMap.get("type").equals("callMessage")) {
            String msg = (String) messageMap.get("message");
            CallMessage callMessage = JSONObject.parseObject(msg, CallMessage.class);
            //单聊
            if (callMessage != null && callMessage.getType().equals(0)) {
                WebSocketUtils.sendMessagePrefix(callMessage.getTargetId(), message);
        try {
            if(isValidJson(message)) {
                // å¤„理视频通话消息
                WebsocketHandler.handleMessage(userId, message);
                // æ ¹æ®ç”¨æˆ·æ–°çš„频率重新调整定时任务
                AppPositionPushService.messageHandler(userId, message);
            } else {
                // å¤„理其他消息
                WebSocketUtils.sendMessage(userId,"Received an invalid JSON message："+message);
            }
            //群聊
            if (callMessage != null && callMessage.getType().equals(1)) {
                String step = callMessage.getStep();
                switch (step) {
                    case "joinRoom":
                        Set<String> userSet = WebSocketUtils.ROOM_USER_SET.getOrDefault(callMessage.getRoomId(), new HashSet<>());
                        userSet.add(callMessage.getUserId());
                        WebSocketUtils.ROOM_USER_SET.put(callMessage.getRoomId(), userSet);
                        WebSocketUtils.sendRoomMessage(callMessage.getRoomId(), message);
                        break;
                    case "inviteRoom":
                        List<String> targetIds = callMessage.getTargetIds();
                        targetIds.stream().forEach(targetId -> {
                            WebSocketUtils.sendMessagePrefix(targetId, message);
                        });
                        break;
                    case "leaveRoom":
                        userSet = WebSocketUtils.ROOM_USER_SET.getOrDefault(callMessage.getRoomId(), new HashSet<>());
                        if (userSet.size() > 0) {
                            userSet.remove(callMessage.getUserId());
                            WebSocketUtils.ROOM_USER_SET.put(callMessage.getRoomId(), userSet);
                            WebSocketUtils.sendRoomMessage(callMessage.getRoomId(), message);
                        }
                        break;
                    default:
                        WebSocketUtils.sendMessagePrefix(callMessage.getTargetId(), message);
                        break;
                }


            }
        } catch (Exception ex) {
            WebSocketUtils.sendMessage(userId, "无法解析消息【" + message + "】:" + ex.getMessage());
        }
        // æ ¹æ®ç”¨æˆ·æ–°çš„频率重新调整定时任务
        AppPositionPushService.messageHandler(userId, message);

    }

    @OnClose
@@ -114,4 +88,20 @@
        }
        log.info("Throwable msg " + throwable.getMessage());
    }

    /**
     * æ£€æŸ¥ç»™å®šçš„字符串是否为有效的JSON格式。
     *
     * @param message è¦æ£€æŸ¥çš„字符串
     * @return å¦‚果是有效的JSON字符串，返回true；否则返回false。
     */
    public static boolean isValidJson(String message) {
        try {
            Object json = JSON.parse(message);
            // ç¡®ä¿è§£æžç»“果是JSONObject或JSONArray类型
            return (json instanceof JSONObject || json instanceof JSONArray);
        } catch (JSONException e) {
            return false;  // è§£æžå¤±è´¥ï¼Œè¯´æ˜Žä¸æ˜¯åˆæ³•çš„JSON
        }
    }
}

 ard-work/src/main/java/com/ruoyi/utils/websocket/util/WebSocketUtils.java

@@ -95,7 +95,17 @@
    public static void sendMessageAll(String message) {
        ONLINE_USER_SESSIONS.forEach((sessionId, session) -> sendMessage(session, message));
    }

    /**
     * å‘送消息给指定用户
     *
     * @param  userId ç”¨æˆ·id
     * @param  message æ¶ˆæ¯å†…容
     * @author åˆ˜è‹ä¹‰
     * @date   2024/8/12 15:43
     */
    public static void sendMessage(String userId,String message) {
        WebSocketUtils.ONLINE_USER_SESSIONS.get(userId).getAsyncRemote().sendText(message);
    }
    /**
     * æŽ¨é€æ¶ˆæ¯åˆ°å…¶ä»–客户端
     *