java/Robotaxi.git

			@@ -1,188 +1,188 @@
			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.ChannelHandlerContext;
			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 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());
			}
			}
			}
			//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.ChannelHandlerContext;
			//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 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());
			// }
			// }
			//}