From 74a43b8dd5694edffa29c72ac4bd7af4c701e0ff Mon Sep 17 00:00:00 2001
From: zhuml4 <zhuml4@xiaopeng.com>
Date: 星期六, 07 三月 2026 21:02:13 +0800
Subject: [PATCH] 更新XPU报文解析逻辑
---
xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java | 80 +++++++++++++++++++++++++++------------
1 files changed, 55 insertions(+), 25 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 379ad0d..f06bcc2 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
@@ -130,11 +130,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;
}
}
@@ -220,31 +233,48 @@
}
//TODO 这里的逻辑要再确认下
//消息应答
- if(MessageType.MESSAGE_TYPE_RESPONSE == type){
- 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);
- }
- }
+ 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);
+// }
+// }
+// }
}
}
--
Gitblit v1.7.1