Merge branch '1.0' of http://120.76.84.145:10101/gitblit/r/java/Robotaxi in...

xuhy

2026-04-22 746517aca9d0b7ae334c70b3041a9b1de795addb

 xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/webSocket/handler/PushMessageWebSocketHandler.java

@@ -1,226 +1,186 @@
package com.xiaopeng.pushMessage.webSocket.handler;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.xiaopeng.api.dto.account.XpOperationsUserPositionDTO;
import com.xiaopeng.pushMessage.webSocket.service.WebSocketUserConnectionManager;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;

import javax.annotation.PostConstruct;
import java.time.LocalDateTime;
import java.util.UUID;

/**
 * Spring WebSocket 消息处理器
 * 路径: /websocket，共用 Spring Boot HTTP 端口
 * 替代原 Netty 独立端口方案
 */
@Component
@Slf4j
public class PushMessageWebSocketHandler extends TextWebSocketHandler {

    @Autowired
    private WebSocketUserConnectionManager userConnectionManager;

    @Autowired
    private WebSocketAuthHandler authHandler;

    private ObjectMapper om;

    @PostConstruct
    void setupObjectMapper() {
        // 允许对象忽略json中不存在的属性
        om = new ObjectMapper();
        om.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
        om.registerModule(new JavaTimeModule());
    }

    // ==================== 连接生命周期 ====================

    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        log.info("WebSocket 连接建立，sessionId: {}", session.getId());
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        log.info("WebSocket 连接关闭，sessionId: {}, status: {}", session.getId(), status);
        userConnectionManager.removeUserConnection(session);
    }

    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        log.error("WebSocket 传输异常，sessionId: {}, error: {}", session.getId(), exception.getMessage(), exception);
        if (session.isOpen()) {
            session.close(CloseStatus.SERVER_ERROR);
        }
    }

    // ==================== 消息处理 ====================

    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        String payload = message.getPayload();
        log.info("[服务器] 收到消息 -> {}", payload);

        try {
            if (payload.startsWith("pos:")) {
                if (log.isDebugEnabled())
                    log.debug("收到运维人员位置上报消息：{}",payload);
                XpOperationsUserPositionDTO xpOperationsUserPositionDTO = om.readValue(payload.substring(4), XpOperationsUserPositionDTO.class);
                saveToMongo(xpOperationsUserPositionDTO);
            }
            JSONObject messageObj = JSON.parseObject(payload);
            String messageType = messageObj.getString("type");

            if ("auth".equals(messageType)) {
                handleUserAuth(session, messageObj);
            } else if ("ping".equals(messageType)) {
                handlePing(session);
            } else {
                handleBusinessMessage(session, messageObj);
            }
        } catch (Exception e) {
            log.debug("消息非 JSON 格式，按普通文本处理: {}", payload);
            sendText(session, "服务器时间: " + LocalDateTime.now() + "->" + payload);
        }
    }

    @Autowired
    private MongoTemplate mongoTemplate;
    private void saveToMongo(XpOperationsUserPositionDTO dto) {
        XpOperationsUserPositionDTO vehiclePosition = mongoTemplate.findOne(new Query(Criteria.where("id").is(dto.getId())), XpOperationsUserPositionDTO.class);
        if (null == vehiclePosition) {
            mongoTemplate.save(dto);
        }else{
            Update update = new Update()
                    .set("id",dto.getId())
                    .set("latitude", dto.getLatitude())
                    .set("latitude", dto.getLongitude());
            mongoTemplate.updateFirst(new Query(Criteria.where("id").is(dto.getId())), update, XpOperationsUserPositionDTO.class);
        }
    }

    // ==================== 业务处理 ====================

    /**
     * 处理用户认证消息
     */
    private void handleUserAuth(WebSocketSession session, JSONObject messageObj) {
        try {
            WebSocketAuthHandler.UserAuthInfo authInfo = authHandler.extractAuthInfo(messageObj);
            String appUserId = authInfo.getAppUserId();
            String token = authInfo.getToken();

            if (!StringUtils.hasText(appUserId)) {
                sendAuthFailure(session, "用户ID不能为空");
                return;
            }

            WebSocketAuthHandler.AuthResult authResult = authHandler.authenticateUser(appUserId, token);
            if (authResult.isSuccess()) {
                userConnectionManager.addUserConnection(appUserId, session);

                // TODO
                String uuid = UUID.randomUUID().toString();
                JSONObject response = new JSONObject();
                response.put("type", "auth_success");
                response.put("appUserId", appUserId);
                response.put("nickName", uuid);
                response.put("message", "认证成功");
                response.put("timestamp", LocalDateTime.now().toString());

                sendText(session, response.toJSONString());
                log.info("用户 {} ({}) 认证成功，sessionId: {}", appUserId, uuid, session.getId());
            } else {
                sendAuthFailure(session, authResult.getMessage());
                log.warn("用户 {} 认证失败: {}, sessionId: {}", appUserId, authResult.getMessage(), session.getId());
            }
        } catch (Exception e) {
            log.error("处理用户认证时发生错误", e);
            sendAuthFailure(session, "认证过程中发生错误: " + e.getMessage());
        }
    }

    private void sendAuthFailure(WebSocketSession session, String message) {
        JSONObject response = new JSONObject();
        response.put("type", "auth_failed");
        response.put("message", message);
        response.put("timestamp", LocalDateTime.now().toString());
        sendText(session, response.toJSONString());
    }

    /**
     * 处理心跳消息
     */
    private void handlePing(WebSocketSession session) {
        String appUserId = userConnectionManager.getUserIdBySessionId(session.getId());

        JSONObject response = new JSONObject();
        response.put("type", "pong");
        response.put("appUserId", appUserId);
        response.put("timestamp", LocalDateTime.now().toString());
        log.info("心跳用户id：{}", appUserId);

        sendText(session, response.toJSONString());
    }

    /**
     * 处理业务消息
     */
    private void handleBusinessMessage(WebSocketSession session, JSONObject messageObj) {
        String appUserId = userConnectionManager.getUserIdBySessionId(session.getId());

        if (appUserId != null) {
            log.info("收到用户 {} 的业务消息: {}", appUserId, messageObj.toJSONString());

            JSONObject response = new JSONObject();
            response.put("type", "message_response");
            response.put("appUserId", appUserId);
            response.put("originalMessage", messageObj);
            response.put("serverTime", LocalDateTime.now().toString());

            sendText(session, response.toJSONString());
        } else {
            log.warn("收到未认证连接的业务消息: {}", messageObj.toJSONString());

            JSONObject response = new JSONObject();
            response.put("type", "auth_required");
            response.put("message", "请先进行用户认证");
            response.put("timestamp", LocalDateTime.now().toString());

            sendText(session, response.toJSONString());
        }
    }

    // ==================== 工具方法 ====================

    /**
     * 发送文本消息（线程安全）
     */
    private void sendText(WebSocketSession session, String text) {
        try {
            if (session.isOpen()) {
                synchronized (session) {
                    session.sendMessage(new TextMessage(text));
                }
            }
        } catch (Exception e) {
            log.error("发送消息失败，sessionId: {}, error: {}", session.getId(), e.getMessage());
        }
    }
}
//package com.xiaopeng.pushMessage.webSocket.handler;
//
//import com.alibaba.fastjson.JSON;
//import com.alibaba.fastjson.JSONObject;
//import com.xiaopeng.pushMessage.webSocket.service.WebSocketUserConnectionManager;
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.stereotype.Component;
//import org.springframework.util.StringUtils;
//import org.springframework.web.socket.CloseStatus;
//import org.springframework.web.socket.TextMessage;
//import org.springframework.web.socket.WebSocketSession;
//import org.springframework.web.socket.handler.TextWebSocketHandler;
//
//import java.time.LocalDateTime;
//import java.util.UUID;
//
///**
// * Spring WebSocket 消息处理器
// * 路径: /websocket，共用 Spring Boot HTTP 端口
// * 替代原 Netty 独立端口方案
// */
//@Component
//@Slf4j
//public class PushMessageWebSocketHandler extends TextWebSocketHandler {
//
//    @Autowired
//    private WebSocketUserConnectionManager userConnectionManager;
//
//    @Autowired
//    private WebSocketAuthHandler authHandler;
//
//    // ==================== 连接生命周期 ====================
//
//    @Override
//    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
//        log.info("WebSocket 连接建立，sessionId: {}", session.getId());
//    }
//
//    @Override
//    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
//        log.info("WebSocket 连接关闭，sessionId: {}, status: {}", session.getId(), status);
//        userConnectionManager.removeUserConnection(session);
//    }
//
//    @Override
//    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
//        log.error("WebSocket 传输异常，sessionId: {}, error: {}", session.getId(), exception.getMessage(), exception);
//        if (session.isOpen()) {
//            session.close(CloseStatus.SERVER_ERROR);
//        }
//    }
//
//    // ==================== 消息处理 ====================
//
//    @Override
//    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
//        String payload = message.getPayload();
//        log.info("[服务器] 收到消息 -> {}", payload);
//
//        try {
//            JSONObject messageObj = JSON.parseObject(payload);
//            String messageType = messageObj.getString("type");
//
//            if ("auth".equals(messageType)) {
//                handleUserAuth(session, messageObj);
//            } else if ("ping".equals(messageType)) {
//                handlePing(session);
//            } else {
//                handleBusinessMessage(session, messageObj);
//            }
//        } catch (Exception e) {
//            log.debug("消息非 JSON 格式，按普通文本处理: {}", payload);
//            sendText(session, "服务器时间: " + LocalDateTime.now() + "->" + payload);
//        }
//    }
//
//    // ==================== 业务处理 ====================
//
//    /**
//     * 处理用户认证消息
//     */
//    private void handleUserAuth(WebSocketSession session, JSONObject messageObj) {
//        try {
//            WebSocketAuthHandler.UserAuthInfo authInfo = authHandler.extractAuthInfo(messageObj);
//            String appUserId = authInfo.getAppUserId();
//            String token = authInfo.getToken();
//
//            if (!StringUtils.hasText(appUserId)) {
//                sendAuthFailure(session, "用户ID不能为空");
//                return;
//            }
//
//            WebSocketAuthHandler.AuthResult authResult = authHandler.authenticateUser(appUserId, token);
//            if (authResult.isSuccess()) {
//                userConnectionManager.addUserConnection(appUserId, session);
//
//                // TODO
//                String uuid = UUID.randomUUID().toString();
//                JSONObject response = new JSONObject();
//                response.put("type", "auth_success");
//                response.put("appUserId", appUserId);
//                response.put("nickName", uuid);
//                response.put("message", "认证成功");
//                response.put("timestamp", LocalDateTime.now().toString());
//
//                sendText(session, response.toJSONString());
//                log.info("用户 {} ({}) 认证成功，sessionId: {}", appUserId, uuid, session.getId());
//            } else {
//                sendAuthFailure(session, authResult.getMessage());
//                log.warn("用户 {} 认证失败: {}, sessionId: {}", appUserId, authResult.getMessage(), session.getId());
//            }
//        } catch (Exception e) {
//            log.error("处理用户认证时发生错误", e);
//            sendAuthFailure(session, "认证过程中发生错误: " + e.getMessage());
//        }
//    }
//
//    private void sendAuthFailure(WebSocketSession session, String message) {
//        JSONObject response = new JSONObject();
//        response.put("type", "auth_failed");
//        response.put("message", message);
//        response.put("timestamp", LocalDateTime.now().toString());
//        sendText(session, response.toJSONString());
//    }
//
//    /**
//     * 处理心跳消息
//     */
//    private void handlePing(WebSocketSession session) {
//        String appUserId = userConnectionManager.getUserIdBySessionId(session.getId());
//
//        JSONObject response = new JSONObject();
//        response.put("type", "pong");
//        response.put("appUserId", appUserId);
//        response.put("timestamp", LocalDateTime.now().toString());
//        log.info("心跳用户id：{}", appUserId);
//
//        sendText(session, response.toJSONString());
//    }
//
//    /**
//     * 处理业务消息
//     */
//    private void handleBusinessMessage(WebSocketSession session, JSONObject messageObj) {
//        String appUserId = userConnectionManager.getUserIdBySessionId(session.getId());
//
//        if (appUserId != null) {
//            log.info("收到用户 {} 的业务消息: {}", appUserId, messageObj.toJSONString());
//
//            JSONObject response = new JSONObject();
//            response.put("type", "message_response");
//            response.put("appUserId", appUserId);
//            response.put("originalMessage", messageObj);
//            response.put("serverTime", LocalDateTime.now().toString());
//
//            sendText(session, response.toJSONString());
//        } else {
//            log.warn("收到未认证连接的业务消息: {}", messageObj.toJSONString());
//
//            JSONObject response = new JSONObject();
//            response.put("type", "auth_required");
//            response.put("message", "请先进行用户认证");
//            response.put("timestamp", LocalDateTime.now().toString());
//
//            sendText(session, response.toJSONString());
//        }
//    }
//
//    // ==================== 工具方法 ====================
//
//    /**
//     * 发送文本消息（线程安全）
//     */
//    private void sendText(WebSocketSession session, String text) {
//        try {
//            if (session.isOpen()) {
//                synchronized (session) {
//                    session.sendMessage(new TextMessage(text));
//                }
//            }
//        } catch (Exception e) {
//            log.error("发送消息失败，sessionId: {}, error: {}", session.getId(), e.getMessage());
//        }
//    }
//}