xp-robotaxi-api/src/main/java/com/xiaopeng/api/vo/automaticPilot/VehiclePosition.java
@@ -6,6 +6,8 @@ import org.springframework.data.mongodb.core.geo.GeoJsonPoint; import org.springframework.data.mongodb.core.index.CompoundIndex; import org.springframework.data.mongodb.core.index.CompoundIndexes; import org.springframework.data.mongodb.core.index.GeoSpatialIndexType; import org.springframework.data.mongodb.core.index.GeoSpatialIndexed; import org.springframework.data.mongodb.core.mapping.Document; import java.io.Serializable; @@ -32,6 +34,7 @@ /** * 车辆实时位置信息 */ @GeoSpatialIndexed(type = GeoSpatialIndexType.GEO_2DSPHERE) private GeoJsonPoint point; /** * 航向，单位：度，北零顺时针。范围：[0-360)。0 => 北, 90 => 东, 180 => 南, 270 => 西 xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java
@@ -6,9 +6,7 @@ import com.google.protobuf.InvalidProtocolBufferException; import com.google.protobuf.util.JsonFormat; import com.xiaopeng.api.dto.automaticPilot.MqttPublishDto; import com.xiaopeng.api.dto.automaticPilot.PushOrderDto; import com.xiaopeng.api.dto.automaticPilot.VehicleEventDTO; import com.xiaopeng.api.dto.map.HistoryPointDto; import com.xiaopeng.api.dto.map.OrderRouteAndFareDto; import com.xiaopeng.api.dto.operation.CruisePointDto; import com.xiaopeng.api.dto.operation.LonLatDto; @@ -50,7 +48,6 @@ import com.xiaopeng.common.redis.service.RedisService; import lombok.extern.slf4j.Slf4j; import org.redisson.api.RLock; import org.redisson.api.RTopic; import org.redisson.api.RedissonClient; import org.springframework.beans.factory.annotation.Value; import org.springframework.data.geo.Point; @@ -63,9 +60,10 @@ import org.springframework.stereotype.Service; import javax.annotation.Resource; import java.math.BigDecimal; import java.util.*; import java.util.concurrent.*; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; /** @@ -556,12 +554,12 @@ reassignOrderDTO.setUid(xpOrder.getAppUserId().toString()); reassignOrderDTO.setOrderNo(xpOrder.getOrderNumber()); reassignOrderDTO.setCount(10); RealtimePose pose = systemRuntimeStatus.getPose(); Wgs naviPose = pose.getNaviPose(); Double latitude = naviPose.getLatitude(); Double longitude = naviPose.getLongitude(); reassignOrderDTO.setLongitude(longitude); reassignOrderDTO.setLatitude(latitude); // RealtimePose pose = systemRuntimeStatus.getPose(); // Wgs naviPose = pose.getNaviPose(); // Double latitude = naviPose.getLatitude(); // Double longitude = naviPose.getLongitude(); reassignOrderDTO.setLongitude(xpOrder.getStartLon()); reassignOrderDTO.setLatitude(xpOrder.getStartLat()); reassignOrderClient.reassignOrder(reassignOrderDTO); } } @@ -1343,8 +1341,22 @@ buildPushOrderStatus(cmd, orderId, OrderStatusEnum.SEE_OFF_GUESTS, sequence); xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.SEE_OFF_GUESTS.getCode(), true); redisService.setCacheObject("ORDER_STATUS:" + orderId, OrderStatusEnum.SEE_OFF_GUESTS.getCode(), 24L, TimeUnit.HOURS); // todo 判断当前订单的车辆是否有连环派单，然后判断是都还满足连环派单的条件，不满足则需要重新派单 // 判断当前订单的车辆是否有连环派单，然后判断是都还满足连环派单的条件，不满足则需要重新派单 Long nextOrderId = redisService.getCacheObject(CacheConstants.nextOrder(orderId)); if(null != nextOrderId){ XpOrder nextOrder = xpOrderClient.queryOrderById(nextOrderId).getData(); if(null != nextOrder && nextOrder.getStatus() == OrderStatusEnum.PICK_UP.getCode()){ redisService.setCacheObject(CacheConstants.nextOrder(orderId),null); ReassignOrderDTO reassignOrderDTO = new ReassignOrderDTO(); reassignOrderDTO.setUid(xpOrder.getAppUserId().toString()); reassignOrderDTO.setOrderNo(xpOrder.getOrderNumber()); reassignOrderDTO.setCount(10); reassignOrderDTO.setLongitude(xpOrder.getStartLon()); reassignOrderDTO.setLatitude(xpOrder.getStartLon()); reassignOrderClient.reassignOrder(reassignOrderDTO); } } redisService.getCacheObject("ORDER_CONNECT_ORDER:" + orderId); }else{ buildPushOrderStatus(cmd, orderId, OrderStatusEnum.CANCELED, sequence); xp-robotaxi-itinerary-boot/src/main/java/com/xiaopeng/itinerary/controller/ActiveOrderController.java
@@ -2,6 +2,7 @@ import com.xiaopeng.api.dto.map.RouteAndFareDto; import com.xiaopeng.api.dto.order.OrderChangeDto; import com.xiaopeng.api.vo.automaticPilot.VehiclePosition; import com.xiaopeng.api.vo.operation.RouteAndFareVo; import com.xiaopeng.common.core.domain.R; import com.xiaopeng.itinerary.service.ActiveOrderService; @@ -10,11 +11,14 @@ import io.swagger.annotations.Api; import io.swagger.annotations.ApiOperation; import lombok.extern.slf4j.Slf4j; import org.springframework.data.geo.Distance; import org.springframework.data.geo.Metrics; import org.springframework.data.geo.Point; 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.validation.annotation.Validated; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import org.springframework.web.bind.annotation.*; import javax.annotation.Resource; import javax.servlet.http.HttpServletRequest; @@ -73,6 +77,26 @@ return R.ok(); } /** * 周围车辆点位查询 */ @ApiOperation(value = "周围车辆点位查询", notes = "", tags = {"小程序-周围车辆点位查询"}) @GetMapping("/api/v1/mini-program/nearby/cars") @XpClientCheck(checkToken = false, checkOtpToken = true) public R<List<VehiclePosition>> carAround(@RequestParam double lon,@RequestParam double lat,@RequestParam double distanceMeters,HttpServletRequest request) { String uid = request.getHeader("XP-Uid"); Point point = new Point(lon, lat); Distance distance = new Distance(distanceMeters / 1000.0, Metrics.KILOMETERS); Criteria criteria = Criteria.where("point") .nearSphere(point) .maxDistance(distance.getNormalizedValue()); Query query = new Query(criteria); return R.ok(mongoTemplate.find(query, VehiclePosition.class)); } @Resource private MongoTemplate mongoTemplate; } xp-robotaxi-itinerary-boot/src/main/java/com/xiaopeng/itinerary/service/impl/ActiveOrderServiceImpl.java
@@ -2,18 +2,23 @@ import com.alibaba.fastjson2.JSON; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.xiaopeng.api.dto.operation.ReassignOrderDTO; import com.xiaopeng.api.dto.order.OrderChangeDto; import com.xiaopeng.api.feignClient.map.XpMapClient; import com.xiaopeng.api.feignClient.operation.ReassignOrderClient; import com.xiaopeng.api.feignClient.operation.XpPointClient; import com.xiaopeng.api.feignClient.operation.XpSystemSettingClient; import com.xiaopeng.api.feignClient.order.XpOrderClient; import com.xiaopeng.api.model.operation.*; import com.xiaopeng.api.model.operation.XpElectronicFence; import com.xiaopeng.api.model.operation.XpElectronicFenceArea; import com.xiaopeng.api.model.operation.XpPoint; import com.xiaopeng.api.model.operation.XpSystemSetting; import com.xiaopeng.api.model.order.XpOrder; import com.xiaopeng.api.vo.operation.RouteOptionVo; import com.xiaopeng.api.vo.operation.RouteVo; import com.xiaopeng.common.core.constant.CacheConstants; import com.xiaopeng.common.core.domain.R; import com.xiaopeng.common.core.enums.OrderStatusEnum; import com.xiaopeng.common.core.exception.ServiceException; import com.xiaopeng.common.redis.service.RedisService; import com.xiaopeng.itinerary.constant.I18nKeyConstant; @@ -32,7 +37,7 @@ import javax.annotation.Resource; import java.time.LocalDateTime; import java.util.*; import java.util.StringJoiner; import java.util.concurrent.TimeUnit; @@ -183,10 +188,24 @@ xpOrder.setQueueName(key); xpOrderClient.updateOrder(xpOrder); // todo 判断当前订单的车辆是否有连环派单，然后判断是都还满足连环派单的条件，不满足则需要重新派单 // 判断当前订单的车辆是否有连环派单，然后判断是都还满足连环派单的条件，不满足则需要重新派单 Long nextOrderId = redisService.getCacheObject(CacheConstants.nextOrder(order.getId().toString())); if(null != nextOrderId){ XpOrder nextOrder = xpOrderClient.queryOrderById(nextOrderId).getData(); if(null != nextOrder && nextOrder.getStatus() == OrderStatusEnum.PICK_UP.getCode()){ redisService.setCacheObject(CacheConstants.nextOrder(order.getId().toString()),null); ReassignOrderDTO reassignOrderDTO = new ReassignOrderDTO(); reassignOrderDTO.setUid(xpOrder.getAppUserId().toString()); reassignOrderDTO.setOrderNo(xpOrder.getOrderNumber()); reassignOrderDTO.setCount(10); reassignOrderDTO.setLongitude(xpOrder.getStartLon()); reassignOrderDTO.setLatitude(xpOrder.getStartLon()); reassignOrderClient.reassignOrder(reassignOrderDTO); } } } @Resource private ReassignOrderClient reassignOrderClient; /** * 获取距离和时间 * @param uid xp-robotaxi-operation-boot/src/main/java/com/xiaopeng/operation/utils/TaskOrderUtil.java
@@ -1089,11 +1089,11 @@ log.info("没有找到空闲车辆 {}，尝试连环派单给即将服务结束的车辆", orderNo); // 查询状态为1的车辆vin R<List<String>> servingVehicles = xpVehicleClient.queryVehiclePositionByVin(new VehiclePositionByVinDto(xpElectronicFence.getId(), vinList, xpTradingStrategy.getSimultaneouslyOrderCount(),1)); if (vehiclePositionVos.getCode() != 200) { if (servingVehicles.getCode() != 200) { continue; } LonLatDto current = new LonLatDto(Double.parseDouble(split[0]), Double.parseDouble(split[1]));//当前点位 LonLatDto userPostion = new LonLatDto(Double.parseDouble(split[0]), Double.parseDouble(split[1])); XpTradingStrategyDetailVO strategyDetailVO = tradingStrategyClient.queryById().getData(); @@ -1114,21 +1114,15 @@ .map(R::getData) .filter(o -> { XpVehicle xpVehicle = xpVehicleClient.queryById(o.getVehicleId()).getData(); Object cacheObject = redisService.getCacheObject(CacheConstants.navigationEnd(o.getAppUserId().toString())); RouteVo routeRemain = JSON.parseObject(cacheObject.toString(), RouteVo.class); if (routeRemain.getDuration() / 60 > timeRegardAsEnding) { return false; } RouteAndFareDto remain = new RouteAndFareDto(); remain.setOrigin(current); LonLatDto ll = new LonLatDto(o.getStartLon(), o.getStartLat()); remain.setDestination(ll); RouteVo r = xpMapClient.getDistanceAndDuration(remain).getData(); if (r.getDistance() > distanceRegardAsEnding) { if (routeRemain.getDistance() > distanceRegardAsEnding) { return false; } if (r.getDuration() / 60 > timeRegardAsEnding) { if (routeRemain.getDuration() / 60 > timeRegardAsEnding) { return false; } RouteAndFareDto next = new RouteAndFareDto(); @@ -1137,7 +1131,7 @@ RouteVo route = xpMapClient.getDistanceAndDuration(next).getData(); if (route.getDistance() > backupPower + xpVehicle.getMileageRange()) return false; return r.getDistance() < destToPassenger return routeRemain.getDistance() < destToPassenger || now.plusSeconds(routeRemain.getDuration()) .plusMinutes(endingTimeToUseTime) .isBefore(rideTime); @@ -1150,7 +1144,7 @@ XpVehicle xpVehicle = xpVehicleClient.queryById(luckyOder.getVehicleId()).getData(); // 设置路线信息 RouteAndFareDto routeAndFareDto = new RouteAndFareDto(); routeAndFareDto.setDestination(current); routeAndFareDto.setDestination(userPostion); routeAndFareDto.setOrigin(new LonLatDto(luckyOder.getEndLon(),luckyOder.getEndLat())); RouteVo routeToOrigin = xpMapClient.getDistanceAndDuration(routeAndFareDto).getData(); Object cacheObject = redisService.getCacheObject(CacheConstants.navigationEnd(luckyOder.getAppUserId().toString())); @@ -1166,6 +1160,7 @@ redisService.setCacheObject(CacheConstants.navigationStart(xpOrder.getAppUserId().toString()), routeToOrigin); redisService.removeByPattern(quoteName,xpOrder.getOrderNumber()); log.info("将订单 {} 从队列 {} 弹出去", orderNo, quoteName); break; } } return false; @@ -1236,7 +1231,81 @@ // 获取车辆的区域信息 if (positionList.isEmpty()) { continue; log.info("没有找到空闲车辆 {}，尝试连环派单给即将服务结束的车辆", orderNo); // 查询状态为1的车辆vin Set<String> vinList = positionList.stream().map(VehiclePosition::getVin).collect(Collectors.toSet()); R<List<String>> servingVehicles = xpVehicleClient.queryVehiclePositionByVin(new VehiclePositionByVinDto(data.getFleetId(), vinList, xpTradingStrategy.getSimultaneouslyOrderCount(),1)); if (servingVehicles.getCode() != 200) { continue; } LonLatDto userPosition = new LonLatDto(data.getStartLon(), data.getStartLat());//人的位置 XpTradingStrategyDetailVO strategyDetailVO = tradingStrategyClient.queryById().getData(); Integer distanceRegardAsEnding = strategyDetailVO.getDistanceRegardAsEnding(); Integer timeRegardAsEnding = strategyDetailVO.getTimeRegardAsEnding(); Integer backupPower = strategyDetailVO.getBackupPower(); Integer destToPassenger = strategyDetailVO.getDestToPassenger(); Integer endingTimeToUseTime = strategyDetailVO.getEndingTimeToUseTime(); LocalDateTime rideTime = data.getRideTime(); LocalDateTime now = LocalDateTime.now(); Optional<XpOrder> luckyOrderOpt = servingVehicles.getData().stream() //获取位置 .map(vin -> mongoTemplate.findOne(new Query(Criteria.where("vin").is(vin)), VehiclePosition.class)) .map(VehiclePosition::getOrderId) .map(xpOrderClient::queryOrderById) .filter(R::isSuccess) .map(R::getData) .filter(o -> { XpVehicle xpVehicle = xpVehicleClient.queryById(o.getVehicleId()).getData(); Object cacheObject = redisService.getCacheObject(CacheConstants.navigationEnd(o.getAppUserId().toString())); RouteVo routeRemain = JSON.parseObject(cacheObject.toString(), RouteVo.class); if (routeRemain.getDuration() / 60 > timeRegardAsEnding) { return false; } if (routeRemain.getDistance() > distanceRegardAsEnding) { return false; } if (routeRemain.getDuration() / 60 > timeRegardAsEnding) { return false; } RouteAndFareDto next = new RouteAndFareDto(); next.setOrigin(new LonLatDto(o.getStartLon(),o.getStartLat())); next.setDestination(new LonLatDto(o.getEndLon(),o.getEndLat())); RouteVo route = xpMapClient.getDistanceAndDuration(next).getData(); if (route.getDistance() > backupPower + xpVehicle.getMileageRange()) return false; return routeRemain.getDistance() < destToPassenger || now.plusSeconds(routeRemain.getDuration()) .plusMinutes(endingTimeToUseTime) .isBefore(rideTime); }) .findAny(); if (luckyOrderOpt.isPresent()) { XpOrder luckyOder = luckyOrderOpt.get(); log.info("有空闲车辆 {} 订单 {}", luckyOder.getOrderNumber(), orderNo); XpVehicle xpVehicle = xpVehicleClient.queryById(luckyOder.getVehicleId()).getData(); // 设置路线信息 RouteAndFareDto pickUpRoute = new RouteAndFareDto(); pickUpRoute.setDestination(userPosition); pickUpRoute.setOrigin(new LonLatDto(luckyOder.getEndLon(),luckyOder.getEndLat())); RouteVo routeToOrigin = xpMapClient.getDistanceAndDuration(pickUpRoute).getData(); Object cacheObject = redisService.getCacheObject(CacheConstants.navigationEnd(luckyOder.getAppUserId().toString())); RouteVo routeRemain = JSON.parseObject(cacheObject.toString(), RouteVo.class); routeToOrigin.getPolyline().addAll(0, routeRemain.getPolyline()); routeToOrigin.setDuration(routeToOrigin.getDuration() + routeRemain.getDuration()); routeToOrigin.setDistance(routeToOrigin.getDistance() + routeRemain.getDistance()); PushOrderDto pushOrderDto = buildPushOrderDto(data); pushOrderDto.setVin(xpVehicle.getVehicleVin()); // 订单流转 orderMissionClient.pushOrder(pushOrderDto); redisService.setCacheObject(CacheConstants.nextOrder(luckyOder.getId().toString()), data.getId()); redisService.setCacheObject(CacheConstants.navigationStart(data.getAppUserId().toString()), routeToOrigin); break; } } // 直接派单 定义一个锁来锁定车辆 xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/webSocket/config/WebSocketConfig.java
@@ -1,27 +1,27 @@ package com.xiaopeng.pushMessage.webSocket.config; import com.xiaopeng.pushMessage.webSocket.handler.PushMessageWebSocketHandler; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Configuration; import org.springframework.web.socket.config.annotation.EnableWebSocket; import org.springframework.web.socket.config.annotation.WebSocketConfigurer; import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry; /** * Spring WebSocket 配置 * 注册 /websocket 端点，共用 Spring Boot HTTP 端口 */ @Configuration @EnableWebSocket public class WebSocketConfig implements WebSocketConfigurer { @Autowired private PushMessageWebSocketHandler pushMessageWebSocketHandler; @Override public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) { registry.addHandler(pushMessageWebSocketHandler, "/websocket") // 允许所有跨域来源，生产环境可按需收紧 .setAllowedOrigins("*"); } } //package com.xiaopeng.pushMessage.webSocket.config; // //import com.xiaopeng.pushMessage.webSocket.handler.PushMessageWebSocketHandler; //import org.springframework.beans.factory.annotation.Autowired; //import org.springframework.context.annotation.Configuration; //import org.springframework.web.socket.config.annotation.EnableWebSocket; //import org.springframework.web.socket.config.annotation.WebSocketConfigurer; //import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry; // ///** // * Spring WebSocket 配置 // * 注册 /websocket 端点，共用 Spring Boot HTTP 端口 // */ //@Configuration //@EnableWebSocket //public class WebSocketConfig implements WebSocketConfigurer { // // @Autowired // private PushMessageWebSocketHandler pushMessageWebSocketHandler; // // @Override // public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) { // registry.addHandler(pushMessageWebSocketHandler, "/websocket") // // 允许所有跨域来源，生产环境可按需收紧 // .setAllowedOrigins("*"); // } //} xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/webSocket/handler/PushMessageWebSocketHandler.java
@@ -1,186 +1,186 @@ 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()); } } } //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()); // } // } //}
