From 08cc6d5cf7e93b75cfabe7dcc2aeaa9afa1117eb Mon Sep 17 00:00:00 2001
From: puzhibing <393733352@qq.com>
Date: 星期二, 12 五月 2026 15:01:47 +0800
Subject: [PATCH] Merge branch '1.0' of http://120.76.84.145:10101/gitblit/r/java/Robotaxi into 1.1
---
xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/webSocket/handler/WebSocketTextFrameHandler.java | 86 ++++++++++++++++++++++++++++---------------
1 files changed, 56 insertions(+), 30 deletions(-)
diff --git a/xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/webSocket/handler/WebSocketTextFrameHandler.java b/xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/webSocket/handler/WebSocketTextFrameHandler.java
index ebd7ee5..0541a2b 100644
--- a/xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/webSocket/handler/WebSocketTextFrameHandler.java
+++ b/xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/webSocket/handler/WebSocketTextFrameHandler.java
@@ -2,6 +2,8 @@
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
+import com.xiaopeng.api.vo.automaticPilot.AppUserPosition;
+import com.xiaopeng.common.redis.service.RedisService;
import com.xiaopeng.pushMessage.webSocket.service.WebSocketUserConnectionManager;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
@@ -9,12 +11,19 @@
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
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.geo.GeoJsonPoint;
+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 javax.annotation.Resource;
import java.time.LocalDateTime;
-import java.util.Objects;
-import java.util.UUID;
+import java.util.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
/**
* TextWebSocketFrame 表示一个文本桢
@@ -29,20 +38,33 @@
@Autowired
private WebSocketUserConnectionManager userConnectionManager;
-
+
@Autowired
private WebSocketAuthHandler authHandler;
+
+ @Resource
+ private MongoTemplate mongoTemplate;
+
+ Map<String, BusinessMsgHandler> businessMsgHandlers;
+
+ public WebSocketTextFrameHandler(List<BusinessMsgHandler> handlers) {
+ this.businessMsgHandlers = handlers.stream()
+ .collect(Collectors.toMap(
+ BusinessMsgHandler::getMessageType,
+ Function.identity()
+ ));
+ }
@Override
protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
String message = msg.text();
log.info("[服务器] : 收到消息 -> {}", message);
-
+
try {
// 尝试解析消息，检查是否包含用户认证信息
JSONObject messageObj = JSON.parseObject(message);
String messageType = messageObj.getString("type");
-
+
if ("auth".equals(messageType)) {
// 处理用户认证消息
handleUserAuth(ctx, messageObj);
@@ -54,12 +76,13 @@
handleBusinessMessage(ctx, messageObj);
}
} catch (Exception e) {
+ e.printStackTrace();
// 如果不是JSON格式，按普通文本处理
log.debug("消息不是JSON格式，按普通文本处理: {}", message);
ctx.channel().writeAndFlush(new TextWebSocketFrame("服务器时间: " + LocalDateTime.now() + "->" + message));
}
}
-
+
/**
* 处理用户认证消息
*/
@@ -69,18 +92,18 @@
WebSocketAuthHandler.UserAuthInfo authInfo = authHandler.extractAuthInfo(messageObj);
String appUserId = authInfo.getAppUserId();
String token = authInfo.getToken();
-
+
if (!StringUtils.hasText(appUserId)) {
sendAuthFailureResponse(ctx, "用户ID不能为空");
return;
}
-
+
// 进行用户认证
WebSocketAuthHandler.AuthResult authResult = authHandler.authenticateUser(appUserId, token);
if (authResult.isSuccess()) {
// 认证成功，将连接与用户关联
userConnectionManager.addUserConnection(appUserId, ctx.channel());
-
+
// 发送认证成功响应
JSONObject response = new JSONObject();
response.put("type", "auth_success");
@@ -91,14 +114,14 @@
// response.put("nickName", authResult.getAppUser().getNickName());
response.put("message", "认证成功");
response.put("timestamp", LocalDateTime.now().toString());
-
+
ctx.channel().writeAndFlush(new TextWebSocketFrame(response.toJSONString()));
- log.info("用户 {} ({}) 认证成功，ChannelId: {}",
+ log.info("用户 {} ({}) 认证成功，ChannelId: {}",
appUserId, uuid, ctx.channel().id().asLongText());
} else {
// 认证失败
sendAuthFailureResponse(ctx, authResult.getMessage());
- log.warn("用户 {} 认证失败: {}, ChannelId: {}",
+ log.warn("用户 {} 认证失败: {}, ChannelId: {}",
appUserId, authResult.getMessage(), ctx.channel().id().asLongText());
}
} catch (Exception e) {
@@ -106,7 +129,7 @@
sendAuthFailureResponse(ctx, "认证过程中发生错误: " + e.getMessage());
}
}
-
+
/**
* 发送认证失败响应
*/
@@ -115,51 +138,53 @@
response.put("type", "auth_failed");
response.put("message", message);
response.put("timestamp", LocalDateTime.now().toString());
-
+
ctx.channel().writeAndFlush(new TextWebSocketFrame(response.toJSONString()));
}
-
+
/**
* 处理心跳消息
*/
private void handlePing(ChannelHandlerContext ctx, JSONObject messageObj) {
String appUserId = userConnectionManager.getUserIdByChannelId(ctx.channel().id().asLongText());
-
+
JSONObject response = new JSONObject();
response.put("type", "pong");
log.info("心跳用户id：{}",appUserId);
response.put("appUserId", appUserId);
response.put("timestamp", LocalDateTime.now().toString());
-
+
ctx.channel().writeAndFlush(new TextWebSocketFrame(response.toJSONString()));
}
-
+
/**
* 处理业务消息
*/
private void handleBusinessMessage(ChannelHandlerContext ctx, JSONObject messageObj) {
String appUserId = userConnectionManager.getUserIdByChannelId(ctx.channel().id().asLongText());
-
+
if (appUserId != null) {
log.info("收到用户 {} 的业务消息: {}", appUserId, messageObj.toJSONString());
-
+
+ businessMsgHandlers.get(messageObj.getString("type")).handle(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());
-
- ctx.channel().writeAndFlush(new TextWebSocketFrame(response.toJSONString()));
+// JSONObject response = new JSONObject();
+// response.put("type", "message_response");
+// response.put("appUserId", appUserId);
+// response.put("originalMessage", messageObj);
+// response.put("serverTime", LocalDateTime.now().toString());
+//
+// ctx.channel().writeAndFlush(new TextWebSocketFrame(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());
-
+
ctx.channel().writeAndFlush(new TextWebSocketFrame(response.toJSONString()));
}
}
@@ -180,7 +205,7 @@
log.info("handlerRemoved 被调用: {}", ctx.channel().id().asLongText());
// shortText 可能会重复
log.info("handlerRemoved 被调用: {}", ctx.channel().id().asShortText());
-
+
// 从用户连接管理器中移除连接
if (userConnectionManager != null) {
userConnectionManager.removeUserConnection(ctx.channel());
@@ -192,4 +217,5 @@
log.error("WebSocket连接异常: {}", cause.getMessage(), cause);
ctx.channel().close();
}
+
}
\ No newline at end of file
--
Gitblit v1.7.1