Merge branch 'master' of http://120.76.84.145:10101/gitblit/r/java/Robotaxi

liujie

2026-03-09 bbdd5f4f286fc593c5c30bd1bccd1f39af135dfa

 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)){