From d9e201f71882d7d6faac88f1f2f69b31f5fb29f1 Mon Sep 17 00:00:00 2001
From: xuhy <3313886187@qq.com>
Date: 星期四, 23 四月 2026 15:03:32 +0800
Subject: [PATCH] [feat] 车辆服务取消登录校验，流量接入的流程接口定义
---
xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/webSocket/handler/WebSocketTextFrameHandler.java | 397 ++++++++++++++++++++++++++++++--------------------------
1 files changed, 211 insertions(+), 186 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 4a9ef48..6043071 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
@@ -1,149 +1,173 @@
package com.xiaopeng.pushMessage.webSocket.handler;
-//import com.alibaba.fastjson.JSON;
-//import com.alibaba.fastjson.JSONObject;
-//import com.xiaopeng.pushMessage.webSocket.service.WebSocketUserConnectionManager;
-//import io.netty.channel.ChannelHandler;
-//import io.netty.channel.ChannelHandlerContext;
-//import io.netty.channel.SimpleChannelInboundHandler;
-//import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
-//import lombok.extern.slf4j.Slf4j;
-//import org.springframework.beans.factory.annotation.Autowired;
-//import org.springframework.stereotype.Component;
-//import org.springframework.util.StringUtils;
-//
-//import java.time.LocalDateTime;
-//import java.util.Objects;
-//import java.util.UUID;
-//
-///**
-// * TextWebSocketFrame 表示一个文本桢
-// * 使用@Sharable注解使其可以被多个连接共享
-// * 支持基于AppUser的连接管理
-// */
-//@Component
-//@Slf4j
-//@ChannelHandler.Sharable
-//public class WebSocketTextFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
-//
-//
-// @Autowired
-// private WebSocketUserConnectionManager userConnectionManager;
-//
-// @Autowired
-// private WebSocketAuthHandler authHandler;
-//
-// @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);
-// } else if ("ping".equals(messageType)) {
-// // 处理心跳消息
-// handlePing(ctx, messageObj);
-// } else {
-// // 处理普通业务消息
-// handleBusinessMessage(ctx, messageObj);
-// }
-// } catch (Exception e) {
-// // 如果不是JSON格式，按普通文本处理
-// log.debug("消息不是JSON格式，按普通文本处理: {}", message);
-// ctx.channel().writeAndFlush(new TextWebSocketFrame("服务器时间: " + LocalDateTime.now() + "->" + message));
-// }
-// }
-//
-// /**
-// * 处理用户认证消息
-// */
-// private void handleUserAuth(ChannelHandlerContext ctx, JSONObject messageObj) {
-// try {
-// // 提取认证信息
-// 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");
-// response.put("appUserId", appUserId);
-// // TODO
-// String uuid = UUID.randomUUID().toString();
-// response.put("nickName", uuid);
-//// 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: {}",
-// appUserId, uuid, ctx.channel().id().asLongText());
-// } else {
-// // 认证失败
-// sendAuthFailureResponse(ctx, authResult.getMessage());
-// log.warn("用户 {} 认证失败: {}, ChannelId: {}",
-// appUserId, authResult.getMessage(), ctx.channel().id().asLongText());
-// }
-// } catch (Exception e) {
-// log.error("处理用户认证时发生错误", e);
-// sendAuthFailureResponse(ctx, "认证过程中发生错误: " + e.getMessage());
-// }
-// }
-//
-// /**
-// * 发送认证失败响应
-// */
-// private void sendAuthFailureResponse(ChannelHandlerContext ctx, String message) {
-// JSONObject response = new JSONObject();
-// 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());
-//
-// // 回复消息
+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;
+import io.netty.channel.SimpleChannelInboundHandler;
+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.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * TextWebSocketFrame 表示一个文本桢
+ * 使用@Sharable注解使其可以被多个连接共享
+ * 支持基于AppUser的连接管理
+ */
+@Component
+@Slf4j
+@ChannelHandler.Sharable
+public class WebSocketTextFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
+
+
+ @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);
+ } else if ("ping".equals(messageType)) {
+ // 处理心跳消息
+ handlePing(ctx, messageObj);
+ } else {
+ // 处理普通业务消息
+ handleBusinessMessage(ctx, messageObj);
+ }
+ } catch (Exception e) {
+ // 如果不是JSON格式，按普通文本处理
+ log.debug("消息不是JSON格式，按普通文本处理: {}", message);
+ ctx.channel().writeAndFlush(new TextWebSocketFrame("服务器时间: " + LocalDateTime.now() + "->" + message));
+ }
+ }
+
+ /**
+ * 处理用户认证消息
+ */
+ private void handleUserAuth(ChannelHandlerContext ctx, JSONObject messageObj) {
+ try {
+ // 提取认证信息
+ 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");
+ response.put("appUserId", appUserId);
+ // TODO
+ String uuid = UUID.randomUUID().toString();
+ response.put("nickName", uuid);
+// 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: {}",
+ appUserId, uuid, ctx.channel().id().asLongText());
+ } else {
+ // 认证失败
+ sendAuthFailureResponse(ctx, authResult.getMessage());
+ log.warn("用户 {} 认证失败: {}, ChannelId: {}",
+ appUserId, authResult.getMessage(), ctx.channel().id().asLongText());
+ }
+ } catch (Exception e) {
+ log.error("处理用户认证时发生错误", e);
+ sendAuthFailureResponse(ctx, "认证过程中发生错误: " + e.getMessage());
+ }
+ }
+
+ /**
+ * 发送认证失败响应
+ */
+ private void sendAuthFailureResponse(ChannelHandlerContext ctx, String message) {
+ JSONObject response = new JSONObject();
+ 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);
@@ -151,45 +175,46 @@
// 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()));
-// }
-// }
-//
-// @Override
-// public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
-// // id 表示唯一的值 LongText是唯一的
-// log.info("handlerAdded 被调用: {}", ctx.channel().id().asLongText());
-// // shortText 可能会重复
-// log.info("handlerAdded 被调用: {}", ctx.channel().id().asShortText());
-//
-// // 注意：用户连接管理在认证时添加，这里不添加
-// }
-//
-// @Override
-// public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
-// // id 表示唯一的值 LongText是唯一的
-// log.info("handlerRemoved 被调用: {}", ctx.channel().id().asLongText());
-// // shortText 可能会重复
-// log.info("handlerRemoved 被调用: {}", ctx.channel().id().asShortText());
-//
-// // 从用户连接管理器中移除连接
-// if (userConnectionManager != null) {
-// userConnectionManager.removeUserConnection(ctx.channel());
-// }
-// }
-//
-// @Override
-// public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
-// log.error("WebSocket连接异常: {}", cause.getMessage(), cause);
-// ctx.channel().close();
-// }
-//}
\ No newline at end of file
+ } 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()));
+ }
+ }
+
+ @Override
+ public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
+ // id 表示唯一的值 LongText是唯一的
+ log.info("handlerAdded 被调用: {}", ctx.channel().id().asLongText());
+ // shortText 可能会重复
+ log.info("handlerAdded 被调用: {}", ctx.channel().id().asShortText());
+
+ // 注意：用户连接管理在认证时添加，这里不添加
+ }
+
+ @Override
+ public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
+ // id 表示唯一的值 LongText是唯一的
+ log.info("handlerRemoved 被调用: {}", ctx.channel().id().asLongText());
+ // shortText 可能会重复
+ log.info("handlerRemoved 被调用: {}", ctx.channel().id().asShortText());
+
+ // 从用户连接管理器中移除连接
+ if (userConnectionManager != null) {
+ userConnectionManager.removeUserConnection(ctx.channel());
+ }
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+ log.error("WebSocket连接异常: {}", cause.getMessage(), cause);
+ ctx.channel().close();
+ }
+
+}
\ No newline at end of file
--
Gitblit v1.7.1