From a6d543bcef73e2a78d8da9b065e241643c166aa8 Mon Sep 17 00:00:00 2001
From: 无关风月 <443237572@qq.com>
Date: 星期五, 13 二月 2026 11:40:29 +0800
Subject: [PATCH] Merge remote-tracking branch 'origin/master'
---
xp-robotaxi-automaticPilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java | 673 +++++++++++++++++++++++++++++++++++++++++++++++++------
1 files changed, 599 insertions(+), 74 deletions(-)
diff --git a/xp-robotaxi-automaticPilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java b/xp-robotaxi-automaticPilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java
index 8eb081a..1f46d8e 100644
--- a/xp-robotaxi-automaticPilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java
+++ b/xp-robotaxi-automaticPilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java
@@ -1,8 +1,54 @@
package com.xiaopeng.automaticPilot.service;
+import cn.hutool.core.lang.Snowflake;
+import cn.hutool.core.util.IdUtil;
+import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.JSONObject;
+import com.xiaopeng.api.dto.map.OrderRouteAndFareDto;
+import com.xiaopeng.api.dto.operation.LonLatDto;
+import com.xiaopeng.api.dto.pushMessage.PushOrderRouteDTO;
+import com.xiaopeng.api.dto.pushMessage.SendMessageDTO;
+import com.xiaopeng.api.feignClient.map.XpMapClient;
+import com.xiaopeng.api.feignClient.order.XpOrderClient;
+import com.xiaopeng.api.feignClient.pushMessage.XpPushMessageClient;
+import com.xiaopeng.api.feignClient.vehicleConvoy.XpVehicleClient;
+import com.xiaopeng.api.model.order.XpOrder;
+import com.xiaopeng.api.model.vehicleConvoy.XpVehicle;
+import com.xiaopeng.api.vo.automaticPilot.VehiclePosition;
+import com.xiaopeng.api.vo.automaticPilot.proto.down.GeoPoint;
+import com.xiaopeng.api.vo.automaticPilot.proto.down.ResponseMessage;
+import com.xiaopeng.api.vo.automaticPilot.proto.down.StatusCode;
+import com.xiaopeng.api.vo.automaticPilot.proto.down.cloudOrderOperation.*;
+import com.xiaopeng.api.vo.automaticPilot.proto.up.MessageType;
+import com.xiaopeng.api.vo.automaticPilot.proto.up.MsgContent;
+import com.xiaopeng.api.vo.automaticPilot.proto.up.RTCmdType;
+import com.xiaopeng.api.vo.automaticPilot.proto.up.RtMsgHeader;
+import com.xiaopeng.api.vo.automaticPilot.proto.up.navigationDeviationEvent.NavigationDeviationEvent;
+import com.xiaopeng.api.vo.automaticPilot.proto.up.navigationDeviationEvent.NavigationDeviationMessage;
+import com.xiaopeng.api.vo.automaticPilot.proto.up.navigationDeviationEvent.RtEventType;
+import com.xiaopeng.api.vo.automaticPilot.proto.up.orderStateUpdateEvent.OrderStatusInfoEvent;
+import com.xiaopeng.api.vo.automaticPilot.proto.up.systemStatusXpu.*;
+import com.xiaopeng.api.vo.automaticPilot.proto.up.xpuNavigationRequest.NavigationRequest;
+import com.xiaopeng.api.vo.automaticPilot.proto.up.xpuNavigationRequest.OrderPhase;
+import com.xiaopeng.api.vo.operation.RouteVo;
+import com.xiaopeng.automaticPilot.util.XPUUtil;
+import com.xiaopeng.common.core.domain.R;
+import com.xiaopeng.common.core.enums.OrderStatusEnum;
+import com.xiaopeng.common.core.enums.PushMessageTypeEnum;
+import com.xiaopeng.common.core.utils.StringUtils;
+import com.xiaopeng.common.core.utils.uuid.UUID;
+import com.xiaopeng.common.redis.service.RedisService;
import lombok.extern.slf4j.Slf4j;
-import org.springframework.kafka.annotation.KafkaListener;
+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.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.util.*;
/**
* Kafka 消费者服务
@@ -11,111 +57,590 @@
@Service
public class KafkaConsumerService {
+ @Resource
+ private XpMapClient xpMapClient;
+
+ @Resource
+ private XpOrderClient xpOrderClient;
+
+ @Resource
+ private MongoTemplate mongoTemplate;
+
+ @Resource
+ private XpVehicleClient xpVehicleClient;
+
+ @Resource
+ private XPUUtil xpuUtil;
+
+ @Autowired
+ public RedisTemplate redisTemplate;
+
+ @Resource
+ private XpPushMessageClient xpPushMessageClient;
+
+
/**
- * 消费消息 - 自动驾驶主题
+ * 消费消息 - XPU上报Robotaxi云端的消息topic（高优先级）
*
- * @param message 消息内容
*/
- @KafkaListener(topics = "automatic-pilot-topic", groupId = "xp-automaticPilot-group")
- public void consumeAutomaticPilotMessage(String message) {
- try {
- log.info("收到自动驾驶主题消息: {}", message);
- // 处理消息逻辑
- processMessage(message);
- } catch (Exception e) {
- log.error("处理消息失败: {}", message, e);
- throw e;
+ public void topicProdRobotaxiXpuHigh(String message) {
+ MsgContent msgContent = JSON.parseObject(message, MsgContent.class);
+ RtMsgHeader header = msgContent.getHeader();
+ MessageType type = header.getType();
+ //处理请求消息
+ if (MessageType.MESSAGE_TYPE_REQUEST == type) {
+ //区分业务类型
+ RTCmdType cmd = header.getCmd();
+ switch (cmd) {
+ case SYSTEM_STATUS_XPU:
+ log.info("XPU发布系统运行状态 (XPU→云端)：{}", message);
+ systemStatusXpu(msgContent);
+ break;
+ case XPU_NAVIGATION_REQUEST:
+ log.info("XPU向云端请求导航 (XPU→云端)：{}", message);
+ xpuNavigationRequest(msgContent);
+ break;
+ case TRIP_STATUS_UPDATE_EVENT:
+ log.info("行程状态更新(XPU→云端)(XPU→座舱)：{}", message);
+ break;
+ case PULLOVER_STATUS_EVENT:
+ log.info("靠边停车状态(XPU→云端)(XPU→座舱)：{}", message);
+ break;
+ case DRIVING_MODE_UPDATE_EVENT:
+ log.info("驾驶模式变化(XPU→云端)(XPU→座舱)：{}", message);
+ break;
+ case SYSTEM_STATUS_UPDATE_EVENT:
+ log.info("系统状态变化(XPU→云端)(XPU→座舱)：{}", message);
+ break;
+ case TASK_STATE_UPDATE_EVENT:
+ log.info("任务状态变化(XPU→云端)(XPU→座舱)：{}", message);
+ break;
+ case ORDER_STATE_UPDATE_EVENT:
+ log.info("订单状态变化(XPU→云端)(XPU→座舱)：{}", message);
+ orderStateUpdateEvent(msgContent);
+ break;
+ case POINT_ARRIVAL_STATUS_EVENT:
+ log.info("到达目标点状态(XPU→云端)(XPU→座舱)：{}", message);
+ break;
+ case PICKUP_CONFIRMATION_EVENT:
+ log.info("用户上车确认(XPU→云端)(XPU→座舱)：{}", message);
+ break;
+ case DROPOFF_INFO_EVENT:
+ log.info("用户下车信息(XPU→云端)：{}", message);
+ break;
+ case NAVIGATION_DEVIATION_EVENT:
+ log.info("导航偏航(XPU→云端)：{}", message);
+ navigationDeviationEvent(msgContent);
+ break;
+ case EMERGENCY_STOP_EVENT:
+ log.info("急停(XPU→云端)(XPU→座舱)：{}", message);
+ break;
+ case NAVIGATION_UPDATE_EVENT:
+ log.info("导航信息变化(XPU→云端)：{}", message);
+ break;
+ case TASK_QUEUE_UPDATE_EVENT:
+ log.info("XPU待执行任务队列(XPU→云端)：{}", message);
+ break;
+ case SYSTEM_STATE_ANAPSHOT_SYNC_EVENT:
+ log.info("每次MQTT连接之后触发 (XPU→云端)：{}", message);
+ break;
+ case ORDER_STATE_UPDATE_TO_CDU_EVENT:
+ log.info("更新订单状态到CDU (XPU→云端)：{}", message);
+ break;
+ default:
+ log.info("消息不在处理范围内，忽略：{}", message);
+ }
+
+
+ }
+
+ }
+
+ /**
+ * 消费消息 - XPU上报Robotaxi云端的消息topic（低优先级）
+ */
+ public void topicProdRobotaxiXpuNormal(String message) {
+ MsgContent msgContent = JSON.parseObject(message, MsgContent.class);
+ RtMsgHeader header = msgContent.getHeader();
+ MessageType type = header.getType();
+ //处理请求消息
+ if (MessageType.MESSAGE_TYPE_REQUEST == type) {
+ //区分业务类型
+ RTCmdType cmd = header.getCmd();
+ switch (cmd) {
+ case SYSTEM_STATUS_XPU:
+ log.info("XPU发布系统运行状态 (XPU→云端)：{}", message);
+ systemStatusXpu(msgContent);
+ break;
+ case XPU_NAVIGATION_REQUEST:
+ log.info("XPU向云端请求导航 (XPU→云端)：{}", message);
+ xpuNavigationRequest(msgContent);
+ break;
+ case TRIP_STATUS_UPDATE_EVENT:
+ log.info("行程状态更新(XPU→云端)(XPU→座舱)：{}", message);
+ break;
+ case PULLOVER_STATUS_EVENT:
+ log.info("靠边停车状态(XPU→云端)(XPU→座舱)：{}", message);
+ break;
+ case DRIVING_MODE_UPDATE_EVENT:
+ log.info("驾驶模式变化(XPU→云端)(XPU→座舱)：{}", message);
+ break;
+ case SYSTEM_STATUS_UPDATE_EVENT:
+ log.info("系统状态变化(XPU→云端)(XPU→座舱)：{}", message);
+ break;
+ case TASK_STATE_UPDATE_EVENT:
+ log.info("任务状态变化(XPU→云端)(XPU→座舱)：{}", message);
+ break;
+ case ORDER_STATE_UPDATE_EVENT:
+ log.info("订单状态变化(XPU→云端)(XPU→座舱)：{}", message);
+ orderStateUpdateEvent(msgContent);
+ break;
+ case POINT_ARRIVAL_STATUS_EVENT:
+ log.info("到达目标点状态(XPU→云端)(XPU→座舱)：{}", message);
+ break;
+ case PICKUP_CONFIRMATION_EVENT:
+ log.info("用户上车确认(XPU→云端)(XPU→座舱)：{}", message);
+ break;
+ case DROPOFF_INFO_EVENT:
+ log.info("用户下车信息(XPU→云端)：{}", message);
+ break;
+ case NAVIGATION_DEVIATION_EVENT:
+ log.info("导航偏航(XPU→云端)：{}", message);
+ navigationDeviationEvent(msgContent);
+ break;
+ case EMERGENCY_STOP_EVENT:
+ log.info("急停(XPU→云端)(XPU→座舱)：{}", message);
+ break;
+ case NAVIGATION_UPDATE_EVENT:
+ log.info("导航信息变化(XPU→云端)：{}", message);
+ break;
+ case TASK_QUEUE_UPDATE_EVENT:
+ log.info("XPU待执行任务队列(XPU→云端)：{}", message);
+ break;
+ case SYSTEM_STATE_ANAPSHOT_SYNC_EVENT:
+ log.info("每次MQTT连接之后触发 (XPU→云端)：{}", message);
+ break;
+ case ORDER_STATE_UPDATE_TO_CDU_EVENT:
+ log.info("更新订单状态到CDU (XPU→云端)：{}", message);
+ break;
+ default:
+ log.info("消息不在处理范围内，忽略：{}", message);
+ }
+
+
+ }
+
+ }
+
+
+ /**
+ * XPU发布系统运行状态---车辆实时位置
+ *
+ * @param msgContent
+ */
+ public void systemStatusXpu(MsgContent msgContent) {
+ //先响应消息
+ Long sequence = msgContent.getHeader().getSequence();
+ responseMessage(sequence);
+ //开始处理消息对应的业务逻辑
+ SystemRuntimeStatus systemRuntimeStatus = msgContent.getContent_detail().getObject("systemRuntimeStatus", SystemRuntimeStatus.class);
+ //获取实时位姿
+ RealtimePose pose = systemRuntimeStatus.getPose();
+ Wgs naviPose = pose.getNavi_pose();
+ Double latitude = naviPose.getLatitude();
+ Double longitude = naviPose.getLongitude();
+ Double course = naviPose.getCourse();
+ Float alt = naviPose.getAlt();
+ //获取车辆静态信息
+ VehicleStaticInfo staticInfo = systemRuntimeStatus.getVehicle_status().getStatic_info();
+ String vin = staticInfo.getVin();
+ redisTemplate.opsForGeo().add("vehiclePosition:" + vin, new GeoJsonPoint(longitude, latitude), vin);
+
+ //保存车辆实时信息到mongodb
+ VehiclePosition vehiclePosition = mongoTemplate.findOne(new Query(Criteria.where("vin").is(vin)), VehiclePosition.class);
+ if (null == vehiclePosition) {
+ vehiclePosition = new VehiclePosition();
+ }
+ vehiclePosition.setVin(vin);
+ vehiclePosition.setPoint(new GeoJsonPoint(longitude, latitude));
+ vehiclePosition.setCourse(course);
+ vehiclePosition.setAlt(alt);
+ VehicleStatus vehicleStatus = systemRuntimeStatus.getVehicle_status();
+ //车辆动态信息
+ VehicleDynamicStatus dynamicStatus = vehicleStatus.getDynamic_status();
+ dynamicStatus.setSpeed(dynamicStatus.getSpeed());
+ vehiclePosition.setCurrentElectricity(dynamicStatus.getBattery_level());
+ //订单信息
+ OrderStatusInfo runningOrderStatusInfo = systemRuntimeStatus.getRunning_order_status_info();
+ Long orderId = null;
+ String order_id = runningOrderStatusInfo.getOrder_id();
+ vehiclePosition.setHasOrder(0);
+ if (StringUtils.isNotEmpty(order_id)) {
+ if (order_id.contains("XU")) {
+ vehiclePosition.setHasOrder(1);
+ orderId = Long.parseLong(order_id.substring(2));
+ } else {
+ orderId = Long.parseLong(order_id);
+ }
+ }
+ vehiclePosition.setOrderId(orderId);
+ mongoTemplate.save(vehiclePosition);
+ //调用推送服务接口
+ if (null != orderId) {
+ RtTripInfo rtTripInfo = systemRuntimeStatus.getRt_trip_info();
+ PushOrderRouteDTO dto = new PushOrderRouteDTO();
+ dto.setOrderId(orderId);
+ dto.setRouteType(order_id.contains("XU") ? 2 : 1);
+ dto.setLongitude(longitude);
+ dto.setLatitude(latitude);
+ dto.setDuration(rtTripInfo.getRemaining_time());
+ dto.setDistance(rtTripInfo.getRemaining_distance().intValue());
+ xpPushMessageClient.pushOrderRoute(dto);
}
}
+
/**
- * 消费消息 - 车辆状态主题
+ * XPU向云端请求导航
*
- * @param message 消息内容
+ * @param msgContent
*/
- @KafkaListener(topics = "vehicle-status-topic", groupId = "xp-automaticPilot-group")
- public void consumeVehicleStatusMessage(String message) {
- try {
- log.info("收到车辆状态主题消息: {}", message);
- // 处理车辆状态消息逻辑
- processVehicleStatus(message);
- } catch (Exception e) {
- log.error("处理车辆状态消息失败: {}", message, e);
- throw e;
+ public void xpuNavigationRequest(MsgContent msgContent) {
+ //先响应消息
+ Long sequence = msgContent.getHeader().getSequence();
+ responseMessage(sequence);
+ //开始处理消息对应的业务逻辑
+ NavigationRequest naviRequest = msgContent.getContent_detail().getObject("naviRequest", NavigationRequest.class);
+ String orderId = naviRequest.getOrder_id();
+ String taskId = naviRequest.getTask_id();
+ OrderPhase orderPhase = naviRequest.getOrder_phase();
+ log.info("订单ID：{}", orderId);
+ log.info("任务ID：{}", taskId);
+ switch (orderPhase) {
+ case IDLE:
+ log.info("订单的任务未开始执行");
+ break;
+ case ROUTE_TO_PICKUP:
+ log.info("驶往上车点的行程");
+ buildNaviRequest(orderId, taskId, orderPhase);
+ break;
+ case ROUTE_TO_DESTINATION:
+ log.info("驶往下车点的行程");
+ buildNaviRequest(orderId, taskId, orderPhase);
+ break;
+ case ROUTE_TO_CRUISE:
+ log.info("巡游阶段路线");
+ buildNaviRequest(orderId, taskId, orderPhase);
+ break;
}
}
+
/**
- * 消费消息 - 导航主题
+ * 车辆偏航后重新规划线路
*
- * @param message 消息内容
+ * @param msgContent
*/
- @KafkaListener(topics = "navigation-topic", groupId = "xp-automaticPilot-group")
- public void consumeNavigationMessage(String message) {
- try {
- log.info("收到导航主题消息: {}", message);
- // 处理导航消息逻辑
- processNavigation(message);
- } catch (Exception e) {
- log.error("处理导航消息失败: {}", message, e);
- throw e;
+ public void navigationDeviationEvent(MsgContent msgContent) {
+ //先响应消息
+ Long sequence = msgContent.getHeader().getSequence();
+ responseMessage(sequence);
+ //开始处理消息对应的业务逻辑
+ NavigationDeviationEvent deviationEvent = msgContent.getContent_detail().getObject("deviationEvent", NavigationDeviationEvent.class);
+ //偏航事件类型 1：请求导航，2：取消导航
+ if (deviationEvent.getEvent_type().equals(RtEventType.RT_EVENT_TYPE_REQUEST)) {
+ NavigationDeviationMessage navigationDeviationMessage = deviationEvent.getNavigation_deviation_message();
+ String orderId = navigationDeviationMessage.getOrder_id();
+ String taskId = navigationDeviationMessage.getTask_id();
+ Double longitude = navigationDeviationMessage.getLongitude();
+ Double latitude = navigationDeviationMessage.getLatitude();
+ //订单类型 1：乘客订单，2=巡游订单
+ Integer orderType = 1;
+ LonLatDto destination = new LonLatDto();
+ LonLatDto origin = new LonLatDto();
+ if (orderId.contains("XU")) {
+ orderType = 2;
+ // todo 下发巡游单的导航数据
+ //巡游单
+ origin.setLongitude(longitude);
+ origin.setLatitude(latitude);
+ destination.setLongitude(116.40528);
+ destination.setLatitude(39.90498);
+ } else {
+ //正常单
+ R<XpOrder> xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId));
+ if (200 != xpOrderR.getCode()) {
+ throw new RuntimeException("订单不存在");
+ }
+ XpOrder xpOrder = xpOrderR.getData();
+ //驶往上车点的行程
+ if (Arrays.asList(3).contains(xpOrder.getStatus())) {
+ //查询司机当前位置到起点的线路
+ origin.setLongitude(longitude);
+ origin.setLatitude(latitude);
+ destination.setLongitude(xpOrder.getStartLon());
+ destination.setLatitude(xpOrder.getStartLat());
+ }
+ //驶往下车点的行程
+ if (Arrays.asList(6).contains(xpOrder.getStatus())) {
+ //查询起点到终点的线路
+ origin.setLongitude(longitude);
+ origin.setLatitude(latitude);
+ destination.setLongitude(xpOrder.getEndLon());
+ destination.setLatitude(xpOrder.getEndLat());
+ }
+ OrderRouteAndFareDto dto = new OrderRouteAndFareDto();
+ dto.setOrderId(Long.parseLong(orderId));
+ dto.setOrigin(origin);
+ dto.setDestination(destination);
+ R<RouteVo> routeVoR = xpMapClient.getOrderDistanceAndDuration(dto);
+ if (200 != routeVoR.getCode()) {
+ throw new RuntimeException("获取订单线路和运费失败");
+ }
+ RouteVo routeVo = routeVoR.getData();
+ //构建导航下发请求对象
+ cloudOrderOperation(orderType, orderId, taskId, routeVo);
+ }
}
}
+
/**
- * 消费消息 - 订单主题
+ * 构建导航请求
*
- * @param message 消息内容
+ * @param orderId
+ * @param taskId
+ * @param orderPhase
*/
- @KafkaListener(topics = "order-topic", groupId = "xp-automaticPilot-group")
- public void consumeOrderMessage(String message) {
- try {
- log.info("收到订单主题消息: {}", message);
- // 处理订单消息逻辑
- processOrder(message);
- } catch (Exception e) {
- log.error("处理订单消息失败: {}", message, e);
- throw e;
+ public void buildNaviRequest(String orderId, String taskId, OrderPhase orderPhase) {
+ //订单类型 1：乘客订单，2=巡游订单
+ Integer orderType = 1;
+ LonLatDto destination = new LonLatDto();
+ LonLatDto origin = new LonLatDto();
+ if (orderId.contains("XU")) {
+ orderType = 2;
+ // todo 下发巡游单的导航数据
+ //巡游单
+ if (orderPhase.equals(OrderPhase.ROUTE_TO_CRUISE)) {
+ origin.setLongitude(116.40528);
+ origin.setLatitude(39.90498);
+ destination.setLongitude(116.40528);
+ destination.setLatitude(39.90498);
+ }
+ } else {
+ //正常单
+ R<XpOrder> xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId));
+ if (200 != xpOrderR.getCode()) {
+ throw new RuntimeException("订单不存在");
+ }
+ XpOrder xpOrder = xpOrderR.getData();
+ //驶往上车点的行程
+ if (orderPhase.equals(OrderPhase.ROUTE_TO_PICKUP)) {
+ XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData();
+ VehiclePosition vehiclePosition = mongoTemplate.findOne(new Query(Criteria.where("vin").is(xpVehicle.getVehicleVin())), VehiclePosition.class);
+ if (Objects.isNull(vehiclePosition)) {
+ throw new RuntimeException("车辆位置信息不存在");
+ }
+ //查询司机当前位置到起点的线路
+ origin.setLongitude(vehiclePosition.getPoint().getX());
+ origin.setLatitude(vehiclePosition.getPoint().getY());
+ destination.setLongitude(xpOrder.getStartLon());
+ destination.setLatitude(xpOrder.getStartLat());
+ }
+ //驶往下车点的行程
+ if (orderPhase.equals(OrderPhase.ROUTE_TO_DESTINATION)) {
+ //查询起点到终点的线路
+ origin.setLongitude(xpOrder.getStartLon());
+ origin.setLatitude(xpOrder.getStartLat());
+ destination.setLongitude(xpOrder.getEndLon());
+ destination.setLatitude(xpOrder.getEndLat());
+ }
+ OrderRouteAndFareDto dto = new OrderRouteAndFareDto();
+ dto.setOrderId(Long.parseLong(orderId));
+ dto.setOrigin(origin);
+ dto.setDestination(destination);
+ R<RouteVo> routeVoR = xpMapClient.getOrderDistanceAndDuration(dto);
+ if (200 != routeVoR.getCode()) {
+ throw new RuntimeException("获取订单线路和运费失败");
+ }
+ RouteVo routeVo = routeVoR.getData();
+ //构建导航下发请求对象
+ cloudOrderOperation(orderType, orderId, taskId, routeVo);
}
}
- /**
- * 处理消息逻辑（示例）
- *
- * @param message 消息内容
- */
- private void processMessage(String message) {
- // TODO: 实现具体的消息处理逻辑
- log.info("处理自动驾驶消息: {}", message);
- }
/**
- * 处理车辆状态消息（示例）
+ * 云端订单操作----下发导航数据
*
- * @param message 消息内容
+ * @param routeVo
*/
- private void processVehicleStatus(String message) {
- // TODO: 实现具体的车辆状态处理逻辑
- log.info("处理车辆状态消息: {}", message);
+ public void cloudOrderOperation(Integer orderType, String orderId, String taskId, RouteVo routeVo) {
+ //构建消息体
+ OrderOperation orderOperation = new OrderOperation();
+ orderOperation.setOrder_id(orderId);
+ orderOperation.setTask_id(taskId);
+ orderOperation.setOperation_type(OrderOperationType.CLOUD_UPDATE_WAYPOINTS);
+
+ //构建航点
+ List<Waypoint> waypoints = new ArrayList<>();
+ List<LonLatDto> polyline = routeVo.getPolyline();
+ for (int i = 0; i < polyline.size(); i++) {
+ LonLatDto lonLatDto = polyline.get(i);
+ Waypoint waypoint = new Waypoint();
+ waypoint.setWaypoint_id(UUID.randomUUID().toString());
+ if (0 == i) {
+ waypoint.setPoint_type(PointType.START_POINT);
+ //订单类型不同，点位的动作类型不同
+ waypoint.setAction_type(1 == orderType ? ActionType.PICK_UP : ActionType.ACTION_UNSPECIFIED);
+ } else if (polyline.size() - 1 == i) {
+ waypoint.setPoint_type(PointType.END_POINT);
+ waypoint.setAction_type(1 == orderType ? ActionType.DROP_OFF : ActionType.ACTION_UNSPECIFIED);
+ } else {
+ waypoint.setPoint_type(PointType.WAY_POINT);
+ //订单类型不同，点位的动作类型不同
+ waypoint.setAction_type(ActionType.ACTION_UNSPECIFIED);
+ }
+ //构建经纬度
+ GeoPoint geoPoint = new GeoPoint();
+ geoPoint.setLongitude(lonLatDto.getLongitude());
+ geoPoint.setLatitude(lonLatDto.getLatitude());
+ waypoint.setPoint(geoPoint);
+ waypoint.setSequence(i + 1);
+ waypoints.add(waypoint);
+ }
+ orderOperation.setWaypoints(waypoints);
+ //开发发送请求到XPU
+ xpuUtil.cloudOrderOperation(orderOperation);
}
- /**
- * 处理导航消息（示例）
- *
- * @param message 消息内容
- */
- private void processNavigation(String message) {
- // TODO: 实现具体的导航处理逻辑
- log.info("处理导航消息: {}", message);
- }
/**
- * 处理订单消息（示例）
- *
- * @param message 消息内容
+ * 构建响应消息
*/
- private void processOrder(String message) {
- // TODO: 实现具体的订单处理逻辑
- log.info("处理订单消息: {}", message);
+ public void responseMessage(Long request_seq) {
+ ResponseMessage responseMessage = new ResponseMessage();
+ responseMessage.setRequest_seq(request_seq);
+ responseMessage.setCode(StatusCode.OK);
+ responseMessage.setTimestamp(System.currentTimeMillis());
+ xpuUtil.responseMessage(responseMessage);
}
+
+
+ /**
+ * 订单状态更新事件
+ * @param msgContent
+ */
+ public void orderStateUpdateEvent(MsgContent msgContent) {
+ //先响应消息
+ Long sequence = msgContent.getHeader().getSequence();
+ responseMessage(sequence);
+ //开始处理消息对应的业务逻辑
+ OrderStatusInfoEvent orderStatusInfoEvent = msgContent.getContent_detail().getObject("orderStatusInfoEvent", OrderStatusInfoEvent.class);
+ //只处理请求消息
+ if (orderStatusInfoEvent.getEvent_type().equals(RtEventType.RT_EVENT_TYPE_REQUEST)) {
+ com.xiaopeng.api.vo.automaticPilot.proto.up.orderStateUpdateEvent.OrderStatusInfo orderStatusInfo = orderStatusInfoEvent.getOrder_status_info();
+ String orderId = orderStatusInfo.getOrder_id();
+ //普通单
+ com.xiaopeng.api.vo.automaticPilot.proto.up.orderStateUpdateEvent.OrderStatus status = orderStatusInfo.getStatus();
+ switch (status) {
+ case ORDER_STATUS_UNSPECIFIED:
+ //订单状态未知
+ log.info("订单状态未知，不做任何处理：{}", orderId);
+ break;
+ case ORDER_RECEIVED:
+ //收到订单
+ buildPushOrderStatus(orderId, OrderStatusEnum.READY_DEPART);
+ break;
+ case ORDER_STARTED:
+ //开始执行订单
+ log.info("开始执行订单，不做任何处理：{}", orderId);
+ break;
+ case ORDER_EN_ROUTE_TO_PICKUP:
+ //前往上车点
+ buildPushOrderStatus(orderId, OrderStatusEnum.PICK_UP);
+ break;
+ case ORDER_ARRIVED_AT_PICKUP:
+ //到达上车点
+ log.info("到达上车点，不做任何处理：{}", orderId);
+ break;
+ case ORDER_WAITING_FOR_PICKUP:
+ //等待乘客上车
+ buildPushOrderStatus(orderId, OrderStatusEnum.PICK_UP_GUESTS);
+ break;
+ case ORDER_PASSENGER_ONBOARD:
+ //乘客已上车
+ buildPushOrderStatus(orderId, OrderStatusEnum.PASSENGER_BOARDED);
+ break;
+ case ORDER_EN_ROUTE_TO_DESTINATION:
+ //前往目的地
+ buildPushOrderStatus(orderId, OrderStatusEnum.IN_SERVICE);
+ break;
+ case ORDER_ARRIVED_AT_DESTINATION:
+ //到达目的地
+ log.info("到达目的地，不做任何处理：{}", orderId);
+ break;
+ case ORDER_WAITING_FOR_DROPOFF:
+ //等待乘客下车
+ buildPushOrderStatus(orderId, OrderStatusEnum.SEE_OFF_GUESTS);
+ break;
+ case ORDER_DROPOFF_COMPLETED:
+ //下车完成
+ buildPushOrderStatus(orderId, OrderStatusEnum.APPROVAL_REJECTION);
+ break;
+ case ORDER_COMPLETED:
+ //已完成
+ log.info("已完成，不做任何处理：{}", orderId);
+ break;
+ case ORDER_CANCELLED:
+ //已取消
+ buildPushOrderStatus(orderId, OrderStatusEnum.CANCELED);
+ break;
+ case ORDER_ARRIVED_AT_CRUISE_DESTINATION:
+ //到达巡游终点
+ buildPushOrderStatus(orderId, null);
+ break;
+ case ORDER_PAUSED:
+ //订单暂停
+ log.info("订单暂停，不做任何处理：{}", orderId);
+ break;
+ default:
+ break;
+ }
+ }
+ }
+
+
+ /**
+ * 构建订单状态推送消息
+ * @param orderId
+ * @param status
+ */
+ public void buildPushOrderStatus(String orderId, OrderStatusEnum status){
+ if(orderId.contains("XU")){
+ //巡游单
+ }else{
+ //普通单
+ //构建推送消息
+ Map<String, Object> params = new HashMap<>();
+ Map<String, Object> content = new HashMap<>();
+ content.put("orderId", orderId);
+ content.put("status", status.getCode());
+ params.put("type", PushMessageTypeEnum.ORDER_STATUS.getDesc());
+ params.put("content", com.alibaba.fastjson2.JSONObject.toJSONString(content));
+ params.put("timestamp", System.currentTimeMillis() + "");
+ String message = JSONObject.toJSONString(params);
+ R<XpOrder> xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId));
+ XpOrder xpOrder = xpOrderR.getData();
+ Long appUserId = xpOrder.getAppUserId();
+ SendMessageDTO dto = new SendMessageDTO();
+ dto.setAppUserId(appUserId + "");
+ dto.setMessage(message);
+ R<String> stringR = xpPushMessageClient.sendStatus(dto);
+ if (R.isError(stringR)) {
+ log.error("推送消息失败：{}", stringR.getMsg());
+ }
+ }
+ }
+
}
--
Gitblit v1.7.1