xp-robotaxi-api/src/main/java/com/xiaopeng/api/vo/automaticPilot/proto/down/RTCmdType.java
@@ -6,6 +6,9 @@ CLOUD_ORDER_OPERATION(10001), // 云端操作订单 (云端→XPU) CLOUD_TASK_CONTROL(10003), // 云端任务操作 (云端→XPU) CLOUD_CREATE_ORDER(10004), // 云端新增订单 (云端→XPU) CLOUD_CANCEL_ORDER(10005), // 云端取消订单 (云端→XPU) CLOUD_SEND_RT_TASK(10006), // 云端发送新RT任务 (云端→XPU) CLOUD_NAVIGATION_RESPONSE(10007), // 云端下放导航信息 (云端→XPU) // RT CMD 座舱发起CMD范围：12001~14000 VEHICLE_ORDER_OPERATION(12001), // 车端操作订单 (座舱→XPU) xp-robotaxi-api/src/main/java/com/xiaopeng/api/vo/automaticPilot/proto/down/RtMsgHeader.java
@@ -27,4 +27,8 @@ * 请求或应答 */ private MessageType type; /** * 消息来源 */ private SourceType source; } xp-robotaxi-api/src/main/java/com/xiaopeng/api/vo/automaticPilot/proto/down/SourceType.java
New file @@ -0,0 +1,23 @@ package com.xiaopeng.api.vo.automaticPilot.proto.down; /** * 消息来源类型 */ public enum SourceType { SOURCE_UNKNOWN(0), CLOUD(1), CDU(2), XPU(3); private final int value; SourceType(int value) { this.value = value; } public int getValue() { return value; } }
@@ -1,5 +1,7 @@ package com.xiaopeng.api.vo.automaticPilot.proto.up; import com.xiaopeng.api.vo.automaticPilot.proto.down.SourceType; import lombok.Data; /** @@ -30,5 +32,5 @@ /** * 消息来源 {cloud,cdu,xpu} */ private String source; private SourceType source; } xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java
@@ -22,6 +22,7 @@ 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.down.cloudPushNavigation.RestoreNavigationData; 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; @@ -182,7 +183,7 @@ log.info("消息不在处理范围内，忽略：{}", message); } } //TODO 有点看不懂 //TODO 这里的逻辑要再确认下 //消息应答 if(MessageType.MESSAGE_TYPE_RESPONSE == type){ com.xiaopeng.api.vo.automaticPilot.proto.up.ResponseMessage resposne = msgContent.getContent_detail().getObject("resposne", com.xiaopeng.api.vo.automaticPilot.proto.up.ResponseMessage.class); @@ -331,7 +332,6 @@ } } /** * XPU向云端请求导航 * @@ -340,6 +340,7 @@ public void xpuNavigationRequest(MsgContent msgContent) { //开始处理消息对应的业务逻辑 NavigationRequest naviRequest = msgContent.getContent_detail().getObject("naviRequest", NavigationRequest.class); String requestId = naviRequest.getRequest_id(); String orderId = naviRequest.getOrder_id(); String taskId = naviRequest.getTask_id(); OrderPhase orderPhase = naviRequest.getOrder_phase(); @@ -351,15 +352,15 @@ break; case ROUTE_TO_PICKUP: log.info("驶往上车点的行程"); buildNaviRequest(orderId, taskId, orderPhase, msgContent.getHeader().getSequence()); buildNaviRequest(requestId, orderId, taskId, orderPhase, msgContent.getHeader().getSequence()); break; case ROUTE_TO_DESTINATION: log.info("驶往下车点的行程"); buildNaviRequest(orderId, taskId, orderPhase, msgContent.getHeader().getSequence()); buildNaviRequest(requestId, orderId, taskId, orderPhase, msgContent.getHeader().getSequence()); break; case ROUTE_TO_CRUISE: log.info("巡游阶段路线"); buildNaviRequest(orderId, taskId, orderPhase, msgContent.getHeader().getSequence()); buildNaviRequest(requestId, orderId, taskId, orderPhase, msgContent.getHeader().getSequence()); break; } } @@ -445,7 +446,7 @@ * @param taskId * @param orderPhase */ public void buildNaviRequest(String orderId, String taskId, OrderPhase orderPhase, Long sequence) { public void buildNaviRequest(String requestId, String orderId, String taskId, OrderPhase orderPhase, Long sequence) { //订单类型 1：乘客订单，2=巡游订单 Integer orderType = 1; LonLatDto destination = new LonLatDto(); @@ -506,7 +507,8 @@ RouteVo routeVo = routeVoR.getData(); //TODO 这里有问题，应该是下发导航信息，不是下发订单信息 //构建导航下发请求对象 cloudOrderOperation(orderType, orderId, taskId, routeVo, xpVehicle.getVehicleVin()); // cloudOrderOperation(orderType, orderId, taskId, routeVo, xpVehicle.getVehicleVin()); cloudPushNavigationInfo(requestId, orderId, taskId, xpVehicle.getVehicleVin(), xpVehicle.getXpuId()); } } @@ -555,7 +557,7 @@ // 处理数据 redisTemplate.opsForValue().set(sequence,"pushOrder_"+orderId+"_"+vin,30, TimeUnit.MINUTES); //开发发送请求到XPU xpuUtil.cloudOrderOperation(orderOperation, vin,sequence); xpuUtil.cloudOrderOperation(orderOperation, vin, sequence); } @@ -692,4 +694,20 @@ } } //TODO 完善逻辑 public void cloudPushNavigationInfo(String requestId, String orderId, String taskId, String vin, String xpuId) { //根据orderId查询派单时查询到的Route信息，获取其中的routeId信息，应该在小程序端下单的时候进行了缓存 String routeId = ""; //根据routeId，调用高德的https://tsapi.amap.com/v1/route/query/routeid?key=#接口，还原路线信息，提取其中的data字段信息 String routeDataContent = ""; //组装报文，将routeDateContent信息推送到XPU RestoreNavigationData navigationData = new RestoreNavigationData(); navigationData.setRoute_id(routeId); navigationData.setRequest_id(requestId); navigationData.setContent(routeDataContent); navigationData.setTimestamp(System.currentTimeMillis()); xpuUtil.cloudPushNavigation(navigationData, xpuId, snowflake.nextId()); } } xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/util/XPUUtil.java
@@ -11,6 +11,7 @@ import com.xiaopeng.api.vo.automaticPilot.proto.down.*; import com.xiaopeng.api.vo.automaticPilot.proto.down.cloudCreateOrder.RTTask; import com.xiaopeng.api.vo.automaticPilot.proto.down.cloudOrderOperation.OrderOperation; import com.xiaopeng.api.vo.automaticPilot.proto.down.cloudPushNavigation.RestoreNavigationData; import com.xiaopeng.common.core.utils.StringUtils; import com.xiaopeng.common.core.utils.uuid.UUID; import lombok.extern.slf4j.Slf4j; @@ -59,13 +60,14 @@ header.setCmd(RTCmdType.CLOUD_CREATE_ORDER); header.setSequence(sequence); header.setType(MessageType.MESSAGE_TYPE_RESPONSE); header.setSource(SourceType.CLOUD); msgContent.setHeader(header); // 消息内容 JSONObject jsonObject = new JSONObject(); jsonObject.put("task", task); msgContent.setContent_detail(jsonObject); MqttPublishVo vo = new MqttPublishVo(); vo.setTopic(buildTopic(vin)); vo.setTopic(buildTopicByVin(vin)); vo.setPayload(JSON.toJSONString(msgContent)); log.info("云端-->XPU 创建订单：{}", vo); @@ -81,7 +83,7 @@ * @return */ //TODO 处理feign调用的异常 public MqttPublishDto cloudOrderOperation(OrderOperation operation, String vin,Long sequence) { public MqttPublishDto cloudOrderOperation(OrderOperation operation, String vin, Long sequence) { MsgContent msgContent = new MsgContent(); // 消息头 RtMsgHeader header = new RtMsgHeader(); @@ -90,21 +92,19 @@ header.setCmd(RTCmdType.CLOUD_ORDER_OPERATION); header.setSequence(sequence); header.setType(MessageType.MESSAGE_TYPE_REQUEST); header.setSource(SourceType.CLOUD); msgContent.setHeader(header); // 消息内容 JSONObject jsonObject = new JSONObject(); jsonObject.put("operation", operation); msgContent.setContent_detail(jsonObject); MqttPublishVo vo = new MqttPublishVo(); vo.setTopic(buildTopic(vin)); vo.setTopic(buildTopicByVin(vin)); vo.setPayload(JSON.toJSONString(msgContent)); log.info("云端-->XPU 修改订单：{}", vo); return xpuClient.publish(vo); } /** * 响应消息 @@ -122,20 +122,49 @@ Snowflake snowflake = IdUtil.getSnowflake(0, 0); header.setSequence(snowflake.nextId()); header.setType(MessageType.MESSAGE_TYPE_RESPONSE); header.setSource(SourceType.CLOUD); msgContent.setHeader(header); // 消息内容 JSONObject jsonObject = new JSONObject(); jsonObject.put("resposne", responseMessage); msgContent.setContent_detail(jsonObject); MqttPublishVo vo = new MqttPublishVo(); vo.setTopic(buildTopic(vin)); vo.setTopic(buildTopicByVin(vin)); vo.setPayload(JSON.toJSONString(msgContent)); log.info("云端-->XPU 响应消息：{}", vo); return xpuClient.publish(vo); } /** * 下发导航地址到车端 * * @param operation * @return */ //TODO 处理feign调用的异常 public MqttPublishDto cloudPushNavigation(RestoreNavigationData navigationData, String xpuId, Long sequence) { MsgContent msgContent = new MsgContent(); // 消息头 RtMsgHeader header = new RtMsgHeader(); header.setMsg_id(UUID.randomUUID().toString()); header.setTimestamp(System.currentTimeMillis()); header.setCmd(RTCmdType.CLOUD_NAVIGATION_RESPONSE); header.setSequence(sequence); header.setType(MessageType.MESSAGE_TYPE_RESPONSE); header.setSource(SourceType.CLOUD); msgContent.setHeader(header); // 消息内容 JSONObject jsonObject = new JSONObject(); jsonObject.put("navigationData", navigationData); msgContent.setContent_detail(jsonObject); MqttPublishVo vo = new MqttPublishVo(); vo.setTopic(buildTopicByXpuId(xpuId)); vo.setPayload(JSON.toJSONString(msgContent)); log.info("云端-->XPU 下发导航信息：{}", vo); return xpuClient.publish(vo); } /** @@ -153,12 +182,16 @@ return xpuIdMapping.get(XPU_ID_TYPE); } private String buildTopic(String vin) { private String buildTopicByVin(String vin) { String xpuId = getXpuIdByVin(vin); return this.buildTopicByXpuId(xpuId); } private String buildTopicByXpuId(String xpuId) { String prefix = topicPrefix == null ? "" : topicPrefix; if (!prefix.endsWith("/")) { prefix = prefix + "/"; } String xpuId = getXpuIdByVin(vin); String value = xpuId; if (xpuId.length() > 10) { int start = 10; xp-robotaxi-automatic-pilot-boot/src/test/java/com/xiaopeng/automaticPilot/util/XPUUtilTest.java
@@ -33,8 +33,8 @@ } @Test public void testBuildTopic() throws Exception { Method buildTopic = XPUUtil.class.getDeclaredMethod("buildTopic", String.class); public void testBuildTopicByVin() throws Exception { Method buildTopic = XPUUtil.class.getDeclaredMethod("buildTopicByVin", String.class); buildTopic.setAccessible(true); String result = (String) buildTopic.invoke(xpuUtil, "VIN123"); Assertions.assertEquals("Robotaxi/Msg/Xpu/ABCDEFGHIJKL00000000", result);
