From bbdd5f4f286fc593c5c30bd1bccd1f39af135dfa Mon Sep 17 00:00:00 2001
From: liujie <1793218484@qq.com>
Date: 星期一, 09 三月 2026 09:43:45 +0800
Subject: [PATCH] Merge branch 'master' of http://120.76.84.145:10101/gitblit/r/java/Robotaxi
---
xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java | 82 ++++++++++++++++++++++++++++++-----------
1 files changed, 60 insertions(+), 22 deletions(-)
diff --git a/xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java b/xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java
index d44fcd5..a2ebb22 100644
--- a/xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java
+++ b/xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java
@@ -100,6 +100,9 @@
@Resource
private XpCruisePlanClient xpCruisePlanClient;
+ private static final com.google.protobuf.util.JsonFormat.Printer jsonPrinter = com.google.protobuf.util.JsonFormat.printer()
+ .printingEnumsAsInts()
+ .preservingProtoFieldNames();
@@ -128,11 +131,24 @@
private MsgContent parseMsgContent(String message) {
try {
+ // 首先解析外层JSON，提取payload字段
+ JSONObject outerJson = JSON.parseObject(message);
+ String payload = outerJson.getString("payload");
+
+ if (StringUtils.isEmpty(payload)) {
+ log.error("Payload is empty in message: {}", message);
+ return null;
+ }
+
+ // 使用payload构建MsgContent
MsgContent.Builder builder = MsgContent.newBuilder();
- JsonFormat.parser().ignoringUnknownFields().merge(message, builder);
+ JsonFormat.parser().ignoringUnknownFields().merge(payload, builder);
return builder.build();
} catch (InvalidProtocolBufferException e) {
log.error("Failed to parse message: {}", message, e);
+ return null;
+ } catch (Exception e) {
+ log.error("Failed to extract payload from message: {}", message, e);
return null;
}
}
@@ -218,28 +234,48 @@
}
//TODO 这里的逻辑要再确认下
//消息应答
- if(MessageType.MESSAGE_TYPE_RESPONSE == type){
- ResponseMessage response = msgContent.getResponse();
- StatusCode code = response.getCode();
- long requestSeq = response.getRequestSeq();
- if(code == StatusCode.OK){
- Object sequence = redisTemplate.opsForValue().get(requestSeq);
- if(sequence!=null){
- JSONObject jsonObject = JSON.parseObject(sequence.toString());
- String event = jsonObject.getString("event");
- //推送导航的响应消息
- if(SystemEventType.PUSH_NAVIGATION.equals(event)){
- String orderId = jsonObject.getString("orderId");
- String vin = jsonObject.getString("vin");
- String orderType = jsonObject.getString("orderType");
- PopOrderDto dto = new PopOrderDto();
- dto.setOrderId(Long.valueOf(orderId));
- dto.setXpVehicleVin(vin);
- dto.setOrderType(orderType);
- xpOrderClient.popOrder(dto);
- }
- }
+ if(MessageType.MESSAGE_TYPE_RESPONSE == type) {
+ //区分业务类型
+ int cmdValue = header.getCmd();
+ RTCmdType cmd = RTCmdType.getByValue(cmdValue);
+ if (cmd == null) {
+ log.warn("Unknown cmd value: {}", cmdValue);
+ return;
}
+ if (cmd == RTCmdType.CLOUD_SEND_RT_TASK) {
+ RtTaskListResponse response = msgContent.getRtTaskListResponse();
+ try {
+ log.info("云端下发订单操作的响应消息 (XPU→云端)：{}", jsonPrinter.print(response));
+ } catch (InvalidProtocolBufferException e) {
+ log.error("Failed to print RtTaskListResponse: {}", response, e);
+ }
+ } else {
+ log.info("响应消息不在处理范围内，忽略：{}", message);
+ }
+// ResponseMessage response = msgContent.getResponse();
+// StatusCode code = response.getCode();
+// long requestSeq = response.getRequestSeq();
+// if(code == StatusCode.OK){
+// log.info("收到XPU消息的响应，code: {}, message: {}", code.getNumber(), message);
+// Object sequence = redisTemplate.opsForValue().get(requestSeq);
+// if(sequence!=null){
+// log.info("收到XPU消息的响应，sequence: {}", sequence);
+// JSONObject jsonObject = JSON.parseObject(sequence.toString());
+// String event = jsonObject.getString("event");
+// //推送导航的响应消息
+// if(SystemEventType.PUSH_NAVIGATION.equals(event)){
+// log.info("推送导航的响应消息，{}", jsonObject.toJSONString());
+// String orderId = jsonObject.getString("orderId");
+// String vin = jsonObject.getString("vin");
+// String orderType = jsonObject.getString("orderType");
+// PopOrderDto dto = new PopOrderDto();
+// dto.setOrderId(Long.valueOf(orderId));
+// dto.setXpVehicleVin(vin);
+// dto.setOrderType(orderType);
+// xpOrderClient.popOrder(dto);
+// }
+// }
+// }
}
}
@@ -518,6 +554,7 @@
//缓存导航数据，用于后面的逻辑中给乘客端推送导航数据
redisService.setCacheObject("ORDER_NAVIGATION_END_DATA:" + xpOrder.getAppUserId(), com.alibaba.fastjson.JSON.toJSONString(routeVo), 24L, TimeUnit.HOURS);
}
+ vehicleId = xpOrder.getVehicleId();
}
if(StringUtils.isNotEmpty(routeID)){
@@ -843,6 +880,7 @@
if(orderId.contains("XU")){
//巡游单
//响应消息
+ // TODO 1.处理结束当前巡游订单 2.创建新的巡游订单 返航 3.调用xpu车机接口 （判断是返航还是正常 返航需要存储之前的订单id）
responseMessage(sequence, "");
//前往目的地
if(OrderStatusEnum.IN_SERVICE.equals(status)){
--
Gitblit v1.7.1