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

liujie

2026-02-13 4472937d0cb9aaf3611cb6b6318659b49cf5ddf7

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

# Conflicts:
#	xp-robotaxi-automaticPilot-boot/src/main/java/com/xiaopeng/automaticPilot/util/XPUUtil.java

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/dto/automaticPilot/MqttPublishDto.java

New file
@@ -0,0 +1,19 @@
package com.xiaopeng.api.dto.automaticPilot;

import lombok.Data;

@Data
public class MqttPublishDto {
    /**
     * 状态码
     */
    private Integer code;
    /**
     * 状态信息
     */
    private String msg;
    /**
     * 数据
     */
    private Object data;
}

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/factory/automaticPilot/XPUClientFallbackFactory.java

New file
@@ -0,0 +1,22 @@
package com.xiaopeng.api.factory.automaticPilot;

import com.xiaopeng.api.dto.automaticPilot.MqttPublishDto;
import com.xiaopeng.api.feignClient.automaticPilot.XPUClient;
import com.xiaopeng.api.vo.automaticPilot.MqttPublishVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.openfeign.FallbackFactory;

@Slf4j
public class XPUClientFallbackFactory implements FallbackFactory<XPUClient> {
    @Override
    public XPUClient create(Throwable cause) {
        return new XPUClient(){

            @Override
            public MqttPublishDto publish(MqttPublishVo vo) {
                log.error("推送XPU-MQTT消息失败: {}", cause.getMessage());
                return null;
            }
        };
    }
}

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/feignClient/automaticPilot/XPUClient.java

New file
@@ -0,0 +1,25 @@
package com.xiaopeng.api.feignClient.automaticPilot;


import com.xiaopeng.api.dto.automaticPilot.MqttPublishDto;
import com.xiaopeng.api.factory.automaticPilot.XPUClientFallbackFactory;
import com.xiaopeng.api.vo.automaticPilot.MqttPublishVo;
import com.xiaopeng.common.core.dto.ServiceNameConstantsEnum;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;

@FeignClient(contextId = "XPUClient",
        value = ServiceNameConstantsEnum.XP_MQTT_ROUTER_THIRDPARTY_BOOT,
        fallbackFactory = XPUClientFallbackFactory.class)
public interface XPUClient {


    /**
     * 推送MQTT消息
     * @param vo
     */
    @PostMapping("/logan/mqtt/publish")
    MqttPublishDto publish(@RequestBody MqttPublishVo vo);

}

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/vo/automaticPilot/MqttPublishVo.java

New file
@@ -0,0 +1,15 @@
package com.xiaopeng.api.vo.automaticPilot;

import lombok.Data;

@Data
public class MqttPublishVo {
    /**
     * 消息内容
     */
    private String payload;
    /**
     * Mqtt topic
     */
    private String topic;
}

 xp-robotaxi-api/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -21,4 +21,5 @@
com.xiaopeng.api.factory.vehicleConvoy.XpFleetOperationsClientFallbackFactory
com.xiaopeng.api.factory.pushMessage.XpPushMessageClientFallbackFactory
com.xiaopeng.api.factory.account.UserResourcesClientFallbackFactory
com.xiaopeng.api.factory.vehicleConvoy.VehicleControlClientFallbackFactory
com.xiaopeng.api.factory.vehicleConvoy.VehicleControlClientFallbackFactory
com.xiaopeng.api.factory.automaticPilot.XPUClientFallbackFactory

 xp-robotaxi-automaticPilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java

@@ -247,9 +247,6 @@
     * @param msgContent
     */
    public void systemStatusXpu(MsgContent msgContent) {
        //先响应消息
        Long sequence = msgContent.getHeader().getSequence();
        responseMessage(sequence);
        //开始处理消息对应的业务逻辑
        SystemRuntimeStatus systemRuntimeStatus = msgContent.getContent_detail().getObject("systemRuntimeStatus", SystemRuntimeStatus.class);
        //获取实时位姿
@@ -262,6 +259,10 @@
        //获取车辆静态信息
        VehicleStaticInfo staticInfo = systemRuntimeStatus.getVehicle_status().getStatic_info();
        String vin = staticInfo.getVin();
        //先响应消息
        Long sequence = msgContent.getHeader().getSequence();
        responseMessage(sequence, vin);
        //保存车辆实时位置到redis
        redisTemplate.opsForGeo().add("vehiclePosition:" + vin, new GeoJsonPoint(longitude, latitude), vin);

        //保存车辆实时信息到mongodb
@@ -314,9 +315,6 @@
     * @param msgContent
     */
    public void xpuNavigationRequest(MsgContent msgContent) {
        //先响应消息
        Long sequence = msgContent.getHeader().getSequence();
        responseMessage(sequence);
        //开始处理消息对应的业务逻辑
        NavigationRequest naviRequest = msgContent.getContent_detail().getObject("naviRequest", NavigationRequest.class);
        String orderId = naviRequest.getOrder_id();
@@ -330,15 +328,15 @@
                break;
            case ROUTE_TO_PICKUP:
                log.info("驶往上车点的行程");
                buildNaviRequest(orderId, taskId, orderPhase);
                buildNaviRequest(orderId, taskId, orderPhase, msgContent.getHeader().getSequence());
                break;
            case ROUTE_TO_DESTINATION:
                log.info("驶往下车点的行程");
                buildNaviRequest(orderId, taskId, orderPhase);
                buildNaviRequest(orderId, taskId, orderPhase, msgContent.getHeader().getSequence());
                break;
            case ROUTE_TO_CRUISE:
                log.info("巡游阶段路线");
                buildNaviRequest(orderId, taskId, orderPhase);
                buildNaviRequest(orderId, taskId, orderPhase, msgContent.getHeader().getSequence());
                break;
        }
    }
@@ -352,7 +350,6 @@
    public void navigationDeviationEvent(MsgContent msgContent) {
        //先响应消息
        Long sequence = msgContent.getHeader().getSequence();
        responseMessage(sequence);
        //开始处理消息对应的业务逻辑
        NavigationDeviationEvent deviationEvent = msgContent.getContent_detail().getObject("deviationEvent", NavigationDeviationEvent.class);
        //偏航事件类型 1：请求导航，2：取消导航
@@ -374,6 +371,8 @@
                origin.setLatitude(latitude);
                destination.setLongitude(116.40528);
                destination.setLatitude(39.90498);
                //响应消息
                responseMessage(sequence, "");
            } else {
                //正常单
                R<XpOrder> xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId));
@@ -381,6 +380,9 @@
                    throw new RuntimeException("订单不存在");
                }
                XpOrder xpOrder = xpOrderR.getData();
                XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData();
                //响应消息
                responseMessage(sequence, xpVehicle.getVehicleVin());
                //驶往上车点的行程
                if (Arrays.asList(3).contains(xpOrder.getStatus())) {
                    //查询司机当前位置到起点的线路
@@ -407,7 +409,7 @@
                }
                RouteVo routeVo = routeVoR.getData();
                //构建导航下发请求对象
                cloudOrderOperation(orderType, orderId, taskId, routeVo);
                cloudOrderOperation(orderType, orderId, taskId, routeVo, xpVehicle.getVehicleVin());
            }
        }
    }
@@ -420,7 +422,7 @@
     * @param taskId
     * @param orderPhase
     */
    public void buildNaviRequest(String orderId, String taskId, OrderPhase orderPhase) {
    public void buildNaviRequest(String orderId, String taskId, OrderPhase orderPhase, Long sequence) {
        //订单类型  1：乘客订单，2=巡游订单
        Integer orderType = 1;
        LonLatDto destination = new LonLatDto();
@@ -434,6 +436,9 @@
                origin.setLatitude(39.90498);
                destination.setLongitude(116.40528);
                destination.setLatitude(39.90498);

                //先响应消息
                responseMessage(sequence, "");
            }
        } else {
            //正常单
@@ -445,6 +450,9 @@
            //驶往上车点的行程
            if (orderPhase.equals(OrderPhase.ROUTE_TO_PICKUP)) {
                XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData();
                //先响应消息
                responseMessage(sequence, xpVehicle.getVehicleVin());
                //查询车辆位置
                VehiclePosition vehiclePosition = mongoTemplate.findOne(new Query(Criteria.where("vin").is(xpVehicle.getVehicleVin())), VehiclePosition.class);
                if (Objects.isNull(vehiclePosition)) {
                    throw new RuntimeException("车辆位置信息不存在");
@@ -471,9 +479,10 @@
            if (200 != routeVoR.getCode()) {
                throw new RuntimeException("获取订单线路和运费失败");
            }
            XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData();
            RouteVo routeVo = routeVoR.getData();
            //构建导航下发请求对象
            cloudOrderOperation(orderType, orderId, taskId, routeVo);
            cloudOrderOperation(orderType, orderId, taskId, routeVo, xpVehicle.getVehicleVin());
        }
    }

@@ -483,7 +492,7 @@
     *
     * @param routeVo
     */
    public void cloudOrderOperation(Integer orderType, String orderId, String taskId, RouteVo routeVo) {
    public void cloudOrderOperation(Integer orderType, String orderId, String taskId, RouteVo routeVo, String vin) {
        //构建消息体
        OrderOperation orderOperation = new OrderOperation();
        orderOperation.setOrder_id(orderId);
@@ -519,19 +528,19 @@
        }
        orderOperation.setWaypoints(waypoints);
        //开发发送请求到XPU
        xpuUtil.cloudOrderOperation(orderOperation);
        xpuUtil.cloudOrderOperation(orderOperation, vin);
    }


    /**
     * 构建响应消息
     */
    public void responseMessage(Long request_seq) {
    public void responseMessage(Long request_seq, String vin) {
        ResponseMessage responseMessage = new ResponseMessage();
        responseMessage.setRequest_seq(request_seq);
        responseMessage.setCode(StatusCode.OK);
        responseMessage.setTimestamp(System.currentTimeMillis());
        xpuUtil.responseMessage(responseMessage);
        xpuUtil.responseMessage(responseMessage, vin);
    }


@@ -542,7 +551,6 @@
    public void orderStateUpdateEvent(MsgContent msgContent) {
        //先响应消息
        Long sequence = msgContent.getHeader().getSequence();
        responseMessage(sequence);
        //开始处理消息对应的业务逻辑
        OrderStatusInfoEvent orderStatusInfoEvent = msgContent.getContent_detail().getObject("orderStatusInfoEvent", OrderStatusInfoEvent.class);
        //只处理请求消息
@@ -558,7 +566,7 @@
                    break;
                case ORDER_RECEIVED:
                    //收到订单
                    buildPushOrderStatus(orderId, OrderStatusEnum.READY_DEPART);
                    buildPushOrderStatus(orderId, OrderStatusEnum.READY_DEPART, sequence);
                    break;
                case ORDER_STARTED:
                    //开始执行订单
@@ -566,7 +574,7 @@
                    break;
                case ORDER_EN_ROUTE_TO_PICKUP:
                    //前往上车点
                    buildPushOrderStatus(orderId, OrderStatusEnum.PICK_UP);
                    buildPushOrderStatus(orderId, OrderStatusEnum.PICK_UP, sequence);
                    break;
                case ORDER_ARRIVED_AT_PICKUP:
                    //到达上车点
@@ -574,15 +582,15 @@
                    break;
                case ORDER_WAITING_FOR_PICKUP:
                    //等待乘客上车
                    buildPushOrderStatus(orderId, OrderStatusEnum.PICK_UP_GUESTS);
                    buildPushOrderStatus(orderId, OrderStatusEnum.PICK_UP_GUESTS, sequence);
                    break;
                case ORDER_PASSENGER_ONBOARD:
                    //乘客已上车
                    buildPushOrderStatus(orderId, OrderStatusEnum.PASSENGER_BOARDED);
                    buildPushOrderStatus(orderId, OrderStatusEnum.PASSENGER_BOARDED, sequence);
                    break;
                case ORDER_EN_ROUTE_TO_DESTINATION:
                    //前往目的地
                    buildPushOrderStatus(orderId, OrderStatusEnum.IN_SERVICE);
                    buildPushOrderStatus(orderId, OrderStatusEnum.IN_SERVICE, sequence);
                    break;
                case ORDER_ARRIVED_AT_DESTINATION:
                    //到达目的地
@@ -590,11 +598,11 @@
                    break;
                case ORDER_WAITING_FOR_DROPOFF:
                    //等待乘客下车
                    buildPushOrderStatus(orderId, OrderStatusEnum.SEE_OFF_GUESTS);
                    buildPushOrderStatus(orderId, OrderStatusEnum.SEE_OFF_GUESTS, sequence);
                    break;
                case ORDER_DROPOFF_COMPLETED:
                    //下车完成
                    buildPushOrderStatus(orderId, OrderStatusEnum.APPROVAL_REJECTION);
                    buildPushOrderStatus(orderId, OrderStatusEnum.APPROVAL_REJECTION, sequence);
                    break;
                case ORDER_COMPLETED:
                    //已完成
@@ -602,11 +610,11 @@
                    break;
                case ORDER_CANCELLED:
                    //已取消
                    buildPushOrderStatus(orderId, OrderStatusEnum.CANCELED);
                    buildPushOrderStatus(orderId, OrderStatusEnum.CANCELED, sequence);
                    break;
                case ORDER_ARRIVED_AT_CRUISE_DESTINATION:
                    //到达巡游终点
                    buildPushOrderStatus(orderId, null);
                    buildPushOrderStatus(orderId, null, sequence);
                    break;
                case ORDER_PAUSED:
                    //订单暂停
@@ -624,11 +632,19 @@
     * @param orderId
     * @param status
     */
    public void buildPushOrderStatus(String orderId, OrderStatusEnum status){
    public void buildPushOrderStatus(String orderId, OrderStatusEnum status, Long sequence){
        if(orderId.contains("XU")){
            //巡游单
            //响应消息
            responseMessage(sequence, "");
        }else{
            //普通单
            R<XpOrder> xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId));
            XpOrder xpOrder = xpOrderR.getData();
            XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData();
            //响应消息
            responseMessage(sequence, xpVehicle.getVehicleVin());

            //构建推送消息
            Map<String, Object> params = new HashMap<>();
            Map<String, Object> content = new HashMap<>();
@@ -638,8 +654,6 @@
            params.put("content", com.alibaba.fastjson2.JSONObject.toJSONString(content));
            params.put("timestamp", System.currentTimeMillis() + "");
            String message = JSONObject.toJSONString(params);
            R<XpOrder> xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId));
            XpOrder xpOrder = xpOrderR.getData();
            Long appUserId = xpOrder.getAppUserId();
            SendMessageDTO dto = new SendMessageDTO();
            dto.setAppUserId(appUserId + "");

 xp-robotaxi-automaticPilot-boot/src/main/java/com/xiaopeng/automaticPilot/util/XPUUtil.java

@@ -7,13 +7,17 @@
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson2.JSON;
import com.xiaopeng.api.dto.automaticPilot.MqttPublishDto;
import com.xiaopeng.api.feignClient.automaticPilot.XPUClient;
import com.xiaopeng.api.vo.automaticPilot.MqttPublishVo;
import com.xiaopeng.api.vo.automaticPilot.proto.down.*;
import com.xiaopeng.api.vo.automaticPilot.proto.down.cloudOrderOperation.OrderOperation;
import com.xiaopeng.api.vo.automaticPilot.proto.down.cloudCreateOrder.RTTask;
import com.xiaopeng.api.vo.automaticPilot.proto.down.cloudOrderOperation.OrderOperation;
import com.xiaopeng.common.core.utils.uuid.UUID;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * 自动驾驶接口对接文档
@@ -22,11 +26,8 @@
@Component
public class XPUUtil {

    /**
     * 接口地址
     */
    @Value("${xpu.url}")
    private String xpuUrl;
    @Resource
    private XPUClient xpuClient;


    /**
@@ -35,16 +36,14 @@
     * @param task
     * @return
     */
    public ResponseMessage cloudCreateOrder(RTTask task,Long sequence) {
    public MqttPublishDto cloudCreateOrder(RTTask task, String vin) {
        MsgContent msgContent = new MsgContent();
        // 消息头
        RtMsgHeader header = new RtMsgHeader();
        header.setMsg_id(UUID.randomUUID().toString());
        header.setTimestamp(System.currentTimeMillis());
        header.setCmd(RTCmdType.CLOUD_CREATE_ORDER);


        header.setSequence(sequence);
        header.setSequence(1L);
        header.setType(MessageType.MESSAGE_TYPE_RESPONSE);
        msgContent.setHeader(header);
        // 消息内容
@@ -52,22 +51,12 @@
        jsonObject.put("task", task);
        msgContent.setContent_detail(jsonObject);
        log.info("云端-->XPU 创建订单：{}", msgContent);
        HttpRequest post = HttpUtil.createPost("");
        post.body(JSON.toJSONString(msgContent));
        HttpResponse execute = post.execute();
        int status = execute.getStatus();
        if (200 != status) {
            throw new RuntimeException("XPU 接口异常：" + execute.getStatus());
        }
        MsgContent resMsgContent = JSON.parseObject(execute.body(), MsgContent.class);
        RtMsgHeader resHeader = resMsgContent.getHeader();
        int value = resHeader.getType().getValue();
        //只处理应答消息
        if (value == MessageType.MESSAGE_TYPE_RESPONSE.getValue()) {
            JSONObject contentDetail = resMsgContent.getContent_detail();
            return contentDetail.getObject("response", ResponseMessage.class);
        }
        return null;

        //组装请求消息体
        MqttPublishVo vo = new MqttPublishVo();
        vo.setTopic("topic-prod-xpu-robotaxi-high/" + vin);
        vo.setPayload(JSON.toJSONString(msgContent));
        return xpuClient.publish(vo);
    }


@@ -78,7 +67,7 @@
     * @param operation
     * @return
     */
    public ResponseMessage cloudOrderOperation(OrderOperation operation) {
    public MqttPublishDto cloudOrderOperation(OrderOperation operation, String vin) {
        MsgContent msgContent = new MsgContent();
        // 消息头
        RtMsgHeader header = new RtMsgHeader();
@@ -94,22 +83,11 @@
        jsonObject.put("operation", operation);
        msgContent.setContent_detail(jsonObject);
        log.info("云端-->XPU 操作订单：{}", msgContent);
        HttpRequest post = HttpUtil.createPost("");
        post.body(JSON.toJSONString(msgContent));
        HttpResponse execute = post.execute();
        int status = execute.getStatus();
        if (200 != status) {
            throw new RuntimeException("XPU 接口异常：" + execute.getStatus());
        }
        MsgContent resMsgContent = JSON.parseObject(execute.body(), MsgContent.class);
        RtMsgHeader resHeader = resMsgContent.getHeader();
        int value = resHeader.getType().getValue();
        //只处理应答消息
        if (value == MessageType.MESSAGE_TYPE_RESPONSE.getValue()) {
            JSONObject contentDetail = resMsgContent.getContent_detail();
            return contentDetail.getObject("response", ResponseMessage.class);
        }
        return null;
        //组装请求消息体
        MqttPublishVo vo = new MqttPublishVo();
        vo.setTopic("topic-prod-xpu-robotaxi-high/" + vin);
        vo.setPayload(JSON.toJSONString(msgContent));
        return xpuClient.publish(vo);
    }


@@ -121,7 +99,7 @@
     * @param responseMessage
     * @return
     */
    public ResponseMessage responseMessage(ResponseMessage responseMessage) {
    public MqttPublishDto responseMessage(ResponseMessage responseMessage, String vin) {
        MsgContent msgContent = new MsgContent();
        // 消息头
        RtMsgHeader header = new RtMsgHeader();
@@ -137,14 +115,11 @@
        jsonObject.put("resposne", responseMessage);
        msgContent.setContent_detail(jsonObject);
        log.info("云端-->XPU 响应消息：{}", msgContent);
        HttpRequest post = HttpUtil.createPost("");
        post.body(JSON.toJSONString(msgContent));
        HttpResponse execute = post.execute();
        int status = execute.getStatus();
        if (200 != status) {
            throw new RuntimeException("XPU 接口异常：" + execute.getStatus());
        }
        return null;
        //组装请求消息体
        MqttPublishVo vo = new MqttPublishVo();
        vo.setTopic("topic-prod-xpu-robotaxi-high/" + vin);
        vo.setPayload(JSON.toJSONString(msgContent));
        return xpuClient.publish(vo);
    }

}

 xp-robotaxi-common/xp-common-core/src/main/java/com/xiaopeng/common/core/dto/ServiceNameConstantsEnum.java

@@ -61,5 +61,9 @@
     * 小鹏用车控服务
     */
    String XP_IM_OPEN_BOOT_SERVICE = "xp-im-open-boot";
    /**
     * 小鹏用车控服务
     */
    String XP_MQTT_ROUTER_THIRDPARTY_BOOT = "xp-mqtt-router-thirdparty-boot";

}