From c96fbe6b0aeb413a17c429133ee6d81d138144b9 Mon Sep 17 00:00:00 2001
From: liujie <1793218484@qq.com>
Date: 星期二, 09 六月 2026 10:58:52 +0800
Subject: [PATCH] Merge branch '1.0' of http://120.76.84.145:10101/gitblit/r/java/Robotaxi into 1.0
---
xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/util/XPUUtil.java | 154 ++++++++++++++++++++++++++++++++++++---------------
1 files changed, 109 insertions(+), 45 deletions(-)
diff --git a/xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/util/XPUUtil.java b/xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/util/XPUUtil.java
index 4eb8d5a..e90488d 100644
--- a/xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/util/XPUUtil.java
+++ b/xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/util/XPUUtil.java
@@ -6,8 +6,12 @@
import com.xiaopeng.api.dto.automaticPilot.MqttPublishDto;
import com.xiaopeng.api.feignClient.automaticPilot.VmpOpenServiceClient;
import com.xiaopeng.api.feignClient.automaticPilot.XPUClient;
+import com.xiaopeng.api.feignClient.vehicleConvoy.XpVehicleClient;
+import com.xiaopeng.api.model.vehicleConvoy.XpVehicle;
import com.xiaopeng.api.vo.automaticPilot.MqttPublishVo;
import com.xiaopeng.api.vo.automaticPilot.protobuf.*;
+import com.xiaopeng.common.core.domain.R;
+import com.xiaopeng.common.core.exception.ServiceException;
import com.xiaopeng.common.core.utils.StringUtils;
import com.xiaopeng.common.core.utils.uuid.UUID;
import lombok.extern.slf4j.Slf4j;
@@ -34,6 +38,8 @@
@Resource
private VmpOpenServiceClient vmpOpenServiceClient;
+ @Resource
+ private XpVehicleClient xpVehicleClient;
// 查询车管绑定类型，1：XPUID，2：待扩展
public static final String XPU_ID_TYPE = "1";
@@ -50,13 +56,12 @@
* @param taskList
* @return
*/
- //TODO 处理feign调用的异常
public MqttPublishDto cloudCreateOrder(RtTaskList taskList, String vin, Long sequence) {
MsgContent.Builder msgContentBuilder = MsgContent.newBuilder();
MsgHeader.Builder headerBuilder = MsgHeader.newBuilder();
headerBuilder.setMsgId(UUID.randomUUID().toString());
headerBuilder.setTimestamp(System.currentTimeMillis());
- headerBuilder.setCmd(RTCmdType.CLOUD_CREATE_ORDER.getValue());
+ headerBuilder.setCmd(RTCmdType.CLOUD_SEND_RT_TASK.getValue());
headerBuilder.setSequence(sequence);
headerBuilder.setType(MessageType.MESSAGE_TYPE_REQUEST);
headerBuilder.setSource(SourceType.CLOUD);
@@ -72,17 +77,26 @@
log.error("Protobuf转JSON失败", e);
throw new RuntimeException("Protobuf转JSON失败", e);
}
- log.info("云端-->XPU 创建订单：{}", vo);
- return xpuClient.publish(vo);
+ log.info("云端-->XPU 创建订单：{}", JSON.toJSONString(vo));
+ try {
+ MqttPublishDto dto = xpuClient.publish(vo);
+ log.info("XPU返回：{}", JSON.toJSONString(dto));
+ if (200 != dto.getCode()) {
+ dto = againPushMessage(vo);
+ }
+ return dto;
+ } catch (Exception e) {
+ throw new RuntimeException("云端-->XPU 创建订单失败：", e);
+ }
}
/**
* 取消订单、修改途经点、终点
+ *
* @param operation
* @return
*/
- //TODO 处理feign调用的异常
public MqttPublishDto cloudOrderOperation(OrderOperation operation, String vin, Long sequence) {
MsgContent.Builder msgContentBuilder = MsgContent.newBuilder();
MsgHeader.Builder headerBuilder = MsgHeader.newBuilder();
@@ -105,10 +119,56 @@
throw new RuntimeException("Protobuf转JSON失败", e);
}
log.info("云端-->XPU 修改订单：{}", vo);
- return xpuClient.publish(vo);
+ try {
+ MqttPublishDto dto = xpuClient.publish(vo);
+ log.info("XPU返回：{}", JSON.toJSONString(dto));
+ if (200 != dto.getCode()) {
+ dto = againPushMessage(vo);
+ }
+ return dto;
+ } catch (Exception e) {
+ throw new RuntimeException("云端-->XPU 修改订单失败：", e);
+ }
}
+ /**
+ * 下发导航地址到车端
+ *
+ * @param navigationData
+ * @return
+ */
+ public MqttPublishDto cloudPushNavigation(RestoreNavigationData navigationData, String xpuId, Long sequence) {
+ MsgContent.Builder msgContentBuilder = MsgContent.newBuilder();
+ MsgHeader.Builder headerBuilder = MsgHeader.newBuilder();
+ headerBuilder.setMsgId(UUID.randomUUID().toString());
+ headerBuilder.setTimestamp(System.currentTimeMillis());
+ headerBuilder.setCmd(RTCmdType.CLOUD_NAVIGATION_RESPONSE.getValue());
+ headerBuilder.setSequence(sequence);
+ headerBuilder.setType(MessageType.MESSAGE_TYPE_RESPONSE);
+ headerBuilder.setSource(SourceType.CLOUD);
+ msgContentBuilder.setHeader(headerBuilder.build());
+ msgContentBuilder.setRestoreNavigationData(navigationData);
+ MqttPublishVo vo = new MqttPublishVo();
+ vo.setTopic(buildTopicByXpuId(xpuId));
+ try {
+ vo.setPayload(jsonPrinter.print(msgContentBuilder.build()));
+ } catch (Exception e) {
+ log.error("Protobuf转JSON失败", e);
+ throw new RuntimeException("Protobuf转JSON失败", e);
+ }
+ log.info("云端-->XPU 下发导航信息：{}", vo);
+ try {
+ MqttPublishDto dto = xpuClient.publish(vo);
+ log.info("XPU返回：{}", JSON.toJSONString(dto));
+ if (200 != dto.getCode()) {
+ dto = againPushMessage(vo);
+ }
+ return dto;
+ } catch (Exception e) {
+ throw new RuntimeException("云端-->XPU 下发导航信息失败：", e);
+ }
+ }
/**
@@ -117,12 +177,12 @@
* @param responseMessage
* @return
*/
- public MqttPublishDto responseMessage(ResponseMessage responseMessage, String vin) {
+ public MqttPublishDto responseMessage(RTCmdType cmd, ResponseMessage responseMessage, String vin) {
MsgContent.Builder msgContentBuilder = MsgContent.newBuilder();
MsgHeader.Builder headerBuilder = MsgHeader.newBuilder();
headerBuilder.setMsgId(UUID.randomUUID().toString());
headerBuilder.setTimestamp(System.currentTimeMillis());
- headerBuilder.setCmd(RTCmdType.CLOUD_TASK_CONTROL.getValue());
+ headerBuilder.setCmd(cmd.getValue());
Snowflake snowflake = IdUtil.getSnowflake(0, 0);
headerBuilder.setSequence(snowflake.nextId());
headerBuilder.setType(MessageType.MESSAGE_TYPE_RESPONSE);
@@ -141,51 +201,32 @@
return xpuClient.publish(vo);
}
- /**
- * 下发导航地址到车端
- *
- * @param navigationData
- * @return
- */
- //TODO 处理feign调用的异常
- public MqttPublishDto cloudPushNavigation(RestoreNavigationData navigationData, String xpuId, Long sequence) {
- MsgContent.Builder msgContentBuilder = MsgContent.newBuilder();
- MsgHeader.Builder headerBuilder = MsgHeader.newBuilder();
- headerBuilder.setMsgId(UUID.randomUUID().toString());
- headerBuilder.setTimestamp(System.currentTimeMillis());
- headerBuilder.setCmd(RTCmdType.CLOUD_NAVIGATION_RESPONSE.getValue());
- headerBuilder.setSequence(sequence);
- headerBuilder.setType(MessageType.MESSAGE_TYPE_RESPONSE);
- headerBuilder.setSource(SourceType.CLOUD);
- msgContentBuilder.setHeader(headerBuilder.build());
-
- msgContentBuilder.setRestoreNavigationData(navigationData);
- MqttPublishVo vo = new MqttPublishVo();
- vo.setTopic(buildTopicByXpuId(xpuId));
- try {
- vo.setPayload(jsonPrinter.print(msgContentBuilder.build()));
- } catch (Exception e) {
- log.error("Protobuf转JSON失败", e);
- throw new RuntimeException("Protobuf转JSON失败", e);
- }
-
-
- log.info("云端-->XPU 下发导航信息：{}", vo);
- return xpuClient.publish(vo);
- }
-
/**
* @description: 根据vin查询xpuId
* @param: vin
* @return: xpuId
*/
- public String getXpuIdByVin(String vin) {
+ public String getXpuIdByVin(String vin) {
+ R<XpVehicle> xpVehicleR = xpVehicleClient.findByVehicleVin(vin);
+ if (xpVehicleR.getCode() != 200 || xpVehicleR.getData() == null) {
+ log.error("根据vin查询车辆信息失败,vin={},msg={}", vin, xpVehicleR.getMsg());
+ return "";
+ }
+ return xpVehicleR.getData().getXpuId();
+ }
+
+ /**
+ * @description: 根据vin查询xpuId
+ * @param: vin
+ * @return: xpuId
+ */
+ public String getXpuIdByVinForVehicleControl(String vin) {
Map<String, String> xpuIdMapping = vmpOpenServiceClient.getVehicleBindInfoByVin(vin, TYPE_LIST);
- log.info("getMqttPublishTopic vin,{},xpuIdMapping,{}", vin, JSON.toJSONString(xpuIdMapping));
- if (xpuIdMapping == null || StringUtils.isEmpty(xpuIdMapping.get(XPU_ID_TYPE)) ) {
+ log.info("getMqttPublishTopic vin,{},xpuIdMapping,{}", vin, JSON.toJSONString(xpuIdMapping));
+ if (xpuIdMapping == null || StringUtils.isEmpty(xpuIdMapping.get(XPU_ID_TYPE))) {
log.error("根据vin查询车辆xpuid信息为空,vin={}", vin);
- throw new RuntimeException("根据vin查询车辆xpuid信息为空");
+ throw new ServiceException("根据vin查询车辆xpuid信息为空");
}
return xpuIdMapping.get(XPU_ID_TYPE);
}
@@ -209,4 +250,27 @@
return prefix + xpuId;
}
+
+ /**
+ * 消息下发XPU重试机制
+ *
+ * @param message
+ * @return
+ */
+ public MqttPublishDto againPushMessage(MqttPublishVo message) {
+ log.info("云端-->XPU 消息重试：{}", message);
+ MqttPublishDto dto = null;
+ for (int i = 0; i < 3; i++) {
+ try {
+ dto = xpuClient.publish(message);
+ log.info("云端-->XPU 消息重试{}次：{}", (i + 1), message);
+ if (dto.getCode() == 200) {
+ break;
+ }
+ } catch (Exception e) {
+ log.error("云端-->XPU 推送失败：{}", message);
+ }
+ }
+ return dto;
+ }
}
--
Gitblit v1.7.1