From 71d2d3557012509c69442dfdc5398685423d4f86 Mon Sep 17 00:00:00 2001
From: xuhy <3313886187@qq.com>
Date: 星期六, 18 四月 2026 15:50:23 +0800
Subject: [PATCH] [fix] 取消订单情况的连环派单 [feat] 周围车辆点位查询
---
xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/webSocket/handler/PushMessageWebSocketHandler.java | 412 ++++++++++++++++++++++++++--------------------------------
1 files changed, 186 insertions(+), 226 deletions(-)
diff --git a/xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/webSocket/handler/PushMessageWebSocketHandler.java b/xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/webSocket/handler/PushMessageWebSocketHandler.java
index 863d1a1..b28663d 100644
--- a/xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/webSocket/handler/PushMessageWebSocketHandler.java
+++ b/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());
+// }
+// }
+//}
--
Gitblit v1.7.1