package com.xiaopeng.automaticPilot.service; import cn.hutool.core.lang.Snowflake; import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONObject; import com.xiaopeng.api.dto.map.OrderRouteAndFareDto; import com.xiaopeng.api.dto.operation.LonLatDto; import com.xiaopeng.api.dto.operation.ReassignOrderDTO; import com.xiaopeng.api.dto.order.PopOrderDto; import com.xiaopeng.api.dto.pushMessage.PushOrderRouteDTO; import com.xiaopeng.api.dto.pushMessage.SendMessageDTO; import com.xiaopeng.api.feignClient.map.XpMapClient; import com.xiaopeng.api.feignClient.operation.ReassignOrderClient; import com.xiaopeng.api.feignClient.order.XpOrderClient; import com.xiaopeng.api.feignClient.pushMessage.XpPushMessageClient; import com.xiaopeng.api.feignClient.vehicleConvoy.UserVehicleClient; import com.xiaopeng.api.feignClient.vehicleConvoy.XpVehicleClient; import com.xiaopeng.api.model.order.XpOrder; import com.xiaopeng.api.model.vehicleConvoy.XpVehicle; import com.xiaopeng.api.vo.automaticPilot.VehiclePosition; import com.xiaopeng.api.vo.automaticPilot.protobuf.ActionType; import com.xiaopeng.api.vo.automaticPilot.protobuf.GeoPoint; import com.xiaopeng.api.vo.automaticPilot.protobuf.OrderOperation; import com.xiaopeng.api.vo.automaticPilot.protobuf.OrderOperationType; import com.xiaopeng.api.vo.automaticPilot.protobuf.PointType; import com.xiaopeng.api.vo.automaticPilot.protobuf.ResponseMessage; import com.xiaopeng.api.vo.automaticPilot.protobuf.RestoreNavigationData; import com.xiaopeng.api.vo.automaticPilot.protobuf.StatusCode; import com.xiaopeng.api.vo.automaticPilot.protobuf.Waypoint; 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; import com.xiaopeng.api.vo.automaticPilot.proto.up.RtMsgHeader; import com.xiaopeng.api.vo.automaticPilot.proto.up.navigationDeviationEvent.NavigationDeviationEvent; import com.xiaopeng.api.vo.automaticPilot.proto.up.navigationDeviationEvent.NavigationDeviationMessage; import com.xiaopeng.api.vo.automaticPilot.proto.up.navigationDeviationEvent.RtEventType; import com.xiaopeng.api.vo.automaticPilot.proto.up.orderStateUpdateEvent.OrderStatusInfoEvent; import com.xiaopeng.api.vo.automaticPilot.proto.up.systemStatusXpu.*; import com.xiaopeng.api.vo.automaticPilot.proto.up.xpuNavigationRequest.NavigationRequest; import com.xiaopeng.api.vo.automaticPilot.proto.up.xpuNavigationRequest.OrderPhase; import com.xiaopeng.api.vo.operation.RouteVo; import com.xiaopeng.api.vo.vehicleConvoy.CancelTaxiPassengerVo; import com.xiaopeng.automaticPilot.util.XPUUtil; import com.xiaopeng.common.core.constant.CacheConstants; import com.xiaopeng.common.core.domain.R; import com.xiaopeng.common.core.enums.OrderStatusEnum; import com.xiaopeng.common.core.enums.PushMessageTypeEnum; import com.xiaopeng.common.core.utils.StringUtils; import com.xiaopeng.common.core.utils.uuid.UUID; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.mongodb.core.MongoTemplate; import org.springframework.data.mongodb.core.geo.GeoJsonPoint; import org.springframework.data.mongodb.core.query.Criteria; import org.springframework.data.mongodb.core.query.Query; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.stereotype.Service; import javax.annotation.Resource; import java.util.*; import java.util.concurrent.TimeUnit; /** * Kafka 消费者服务 */ @Slf4j @Service public class KafkaConsumerService { @Resource private XpMapClient xpMapClient; @Resource private XpOrderClient xpOrderClient; @Resource private MongoTemplate mongoTemplate; @Resource private XpVehicleClient xpVehicleClient; @Resource private XPUUtil xpuUtil; @Resource public RedisTemplate redisTemplate; @Resource private XpPushMessageClient xpPushMessageClient; @Resource private ReassignOrderClient reassignOrderClient; @Resource private UserVehicleClient userVehicleClient; @Resource private Snowflake snowflake; /** * 消费消息 - XPU上报Robotaxi云端的消息topic（高优先级） * */ public void topicProdRobotaxiXpuHigh(String message) { MsgContent msgContent = JSON.parseObject(message, MsgContent.class); handleXPUMessage(msgContent, message); } /** * 消费消息 - XPU上报Robotaxi云端的消息topic（低优先级） */ public void topicProdRobotaxiXpuNormal(String message) { MsgContent msgContent = JSON.parseObject(message, MsgContent.class); handleXPUMessage(msgContent, message); } /** * 处理XPU消息 * @param msgContent * @param message */ public void handleXPUMessage(MsgContent msgContent, String message){ RtMsgHeader header = msgContent.getHeader(); MessageType type = header.getType(); //处理请求消息 if (MessageType.MESSAGE_TYPE_REQUEST == type) { //区分业务类型 RTCmdType cmd = header.getCmd(); switch (cmd) { case SYSTEM_STATUS_XPU: log.info("XPU发布系统运行状态 (XPU→云端)：{}", message); systemStatusXpu(msgContent); systemStatusXpuAndReassign(msgContent); break; case XPU_NAVIGATION_REQUEST: log.info("XPU向云端请求导航 (XPU→云端)：{}", message); xpuNavigationRequest(msgContent); break; case TRIP_STATUS_UPDATE_EVENT: log.info("行程状态更新(XPU→云端)(XPU→座舱)：{}", message); break; case PULLOVER_STATUS_EVENT: log.info("靠边停车状态(XPU→云端)(XPU→座舱)：{}", message); break; case DRIVING_MODE_UPDATE_EVENT: log.info("驾驶模式变化(XPU→云端)(XPU→座舱)：{}", message); break; case SYSTEM_STATUS_UPDATE_EVENT: log.info("系统状态变化(XPU→云端)(XPU→座舱)：{}", message); break; case TASK_STATE_UPDATE_EVENT: log.info("任务状态变化(XPU→云端)(XPU→座舱)：{}", message); break; case ORDER_STATE_UPDATE_EVENT: log.info("订单状态变化(XPU→云端)(XPU→座舱)：{}", message); orderStateUpdateEvent(msgContent); break; case POINT_ARRIVAL_STATUS_EVENT: log.info("到达目标点状态(XPU→云端)(XPU→座舱)：{}", message); break; case PICKUP_CONFIRMATION_EVENT: log.info("用户上车确认(XPU→云端)(XPU→座舱)：{}", message); break; case DROPOFF_INFO_EVENT: log.info("用户下车信息(XPU→云端)：{}", message); break; case NAVIGATION_DEVIATION_EVENT: log.info("导航偏航(XPU→云端)：{}", message); navigationDeviationEvent(msgContent); break; case EMERGENCY_STOP_EVENT: log.info("急停(XPU→云端)(XPU→座舱)：{}", message); break; case NAVIGATION_UPDATE_EVENT: log.info("导航信息变化(XPU→云端)：{}", message); break; case TASK_QUEUE_UPDATE_EVENT: log.info("XPU待执行任务队列(XPU→云端)：{}", message); break; case SYSTEM_STATE_ANAPSHOT_SYNC_EVENT: log.info("每次MQTT连接之后触发 (XPU→云端)：{}", message); break; case ORDER_STATE_UPDATE_TO_CDU_EVENT: log.info("更新订单状态到CDU (XPU→云端)：{}", message); break; default: log.info("消息不在处理范围内，忽略：{}", message); } } //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); com.xiaopeng.api.vo.automaticPilot.proto.up.StatusCode code = resposne.getCode(); Long requestSeq = resposne.getRequest_seq(); if(code == com.xiaopeng.api.vo.automaticPilot.proto.up.StatusCode.OK){ Object sequence = redisTemplate.opsForValue().get(requestSeq); if(sequence!=null){ String[] split = sequence.toString().split("_"); if(split[0].equals("pushOrder")){ PopOrderDto dto = new PopOrderDto(); dto.setOrderId(Long.valueOf(split[1])); dto.setXpVehicleVin(split[2]); xpOrderClient.popOrder(dto); } } } } } /** * XPU发布系统运行状态---车辆实时位置 * * @param msgContent */ public void systemStatusXpu(MsgContent msgContent) { //开始处理消息对应的业务逻辑 SystemRuntimeStatus systemRuntimeStatus = msgContent.getContent_detail().getObject("systemRuntimeStatus", SystemRuntimeStatus.class); //获取实时位置 RealtimePose pose = systemRuntimeStatus.getPose(); Wgs naviPose = pose.getNavi_pose(); Double latitude = naviPose.getLatitude(); Double longitude = naviPose.getLongitude(); Double course = naviPose.getCourse(); Float alt = naviPose.getAlt(); //获取车辆静态信息 VehicleStaticInfo staticInfo = systemRuntimeStatus.getVehicle_status().getStatic_info(); String xpuId = staticInfo.getXpu_id(); XpVehicle xpVehicle = xpVehicleClient.findByVehicleXpuId(xpuId).getData(); String vin = xpVehicle.getVehicleVin(); //先响应消息 Long sequence = msgContent.getHeader().getSequence(); responseMessage(sequence, vin); //保存车辆实时位置到redis redisTemplate.opsForGeo().add("vehiclePosition:" + vin, new GeoJsonPoint(longitude, latitude), vin); //保存车辆实时信息到mongodb VehiclePosition vehiclePosition = mongoTemplate.findOne(new Query(Criteria.where("vin").is(vin)), VehiclePosition.class); if (null == vehiclePosition) { vehiclePosition = new VehiclePosition(); } vehiclePosition.setVin(vin); vehiclePosition.setPoint(new GeoJsonPoint(longitude, latitude)); vehiclePosition.setCourse(course); vehiclePosition.setAlt(alt); VehicleStatus vehicleStatus = systemRuntimeStatus.getVehicle_status(); //车辆动态信息 VehicleDynamicStatus dynamicStatus = vehicleStatus.getDynamic_status(); dynamicStatus.setSpeed(dynamicStatus.getSpeed()); vehiclePosition.setCurrentElectricity(dynamicStatus.getBattery_level()); //订单信息 OrderStatusInfo runningOrderStatusInfo = systemRuntimeStatus.getRunning_order_status_info(); Long orderId = null; String order_id = runningOrderStatusInfo.getOrder_id(); vehiclePosition.setHasOrder(0); if (StringUtils.isNotEmpty(order_id)) { if (order_id.contains("XU")) { vehiclePosition.setHasOrder(1); orderId = Long.parseLong(order_id.substring(2)); } else { orderId = Long.parseLong(order_id); } } vehiclePosition.setOrderId(orderId); mongoTemplate.save(vehiclePosition); //调用推送服务接口 if (null != orderId) { RtTripInfo rtTripInfo = systemRuntimeStatus.getRt_trip_info(); PushOrderRouteDTO dto = new PushOrderRouteDTO(); dto.setOrderId(orderId); dto.setRouteType(order_id.contains("XU") ? 2 : 1); dto.setLongitude(longitude); dto.setLatitude(latitude); dto.setDuration(rtTripInfo.getRemaining_time()); dto.setDistance(rtTripInfo.getRemaining_distance().intValue()); xpPushMessageClient.pushOrderRoute(dto); } } /** * XPU发布系统运行状态---车辆实时位置 * * @param msgContent */ public void systemStatusXpuAndReassign(MsgContent msgContent) { //开始处理消息对应的业务逻辑 SystemRuntimeStatus systemRuntimeStatus = msgContent.getContent_detail().getObject("systemRuntimeStatus", SystemRuntimeStatus.class); AbnormalReason abnormalReason = systemRuntimeStatus.getSystem_status().getAbnormal_reason(); // 存储车辆挡位 GearLevel gearLevel = systemRuntimeStatus.getVehicle_status().getDynamic_status().getGear_level(); int value = gearLevel.getValue(); redisTemplate.opsForValue().set(CacheConstants.GEAR_LEVEL + systemRuntimeStatus.getVehicle_status().getStatic_info().getVin(), value); // 判断是否为故障状态 if (abnormalReason == AbnormalReason.FAULT) { log.info("车辆故障，开始处理故障"); // TODO 添加异常任务 // 解绑用户与车辆关系 CancelTaxiPassengerVo cancelTaxiPassengerVo = new CancelTaxiPassengerVo(); String orderId = systemRuntimeStatus.getRunning_order_status_info().getOrder_id(); XpOrder xpOrder = xpOrderClient.queryOrderById(Long.parseLong(orderId)).getData(); XpVehicle vehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData(); if (vehicle != null) { // 修改车辆状态 vehicle.setOperatingStatus(4); xpVehicleClient.updateVehicleById(vehicle); cancelTaxiPassengerVo.setVin(vehicle.getVehicleVin()); cancelTaxiPassengerVo.setUid(xpOrder.getAppUserId().toString()); cancelTaxiPassengerVo.setOrderNo(xpOrder.getOrderNumber()); // TODO // cancelTaxiPassengerVo.setAppId(); userVehicleClient.cancelTaxiPassenger(cancelTaxiPassengerVo); } // 取消订单 // R r = xpOrderClient.cancelOrder(String.valueOf(xpOrder.getId())); // if (r.getCode() == R.SUCCESS){ log.info("取消订单成功"); // TODO 重新派单 ReassignOrderDTO reassignOrderDTO = new ReassignOrderDTO(); reassignOrderDTO.setUid(xpOrder.getAppUserId().toString()); reassignOrderDTO.setOrderNo(xpOrder.getOrderNumber()); reassignOrderDTO.setCount(10); RealtimePose pose = systemRuntimeStatus.getPose(); Wgs naviPose = pose.getNavi_pose(); Double latitude = naviPose.getLatitude(); Double longitude = naviPose.getLongitude(); reassignOrderDTO.setLongitude(longitude); reassignOrderDTO.setLatitude(latitude); reassignOrderClient.reassignOrder(reassignOrderDTO); // } } } /** * XPU向云端请求导航 * * @param msgContent */ 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(); log.info("订单ID：{}", orderId); log.info("任务ID：{}", taskId); switch (orderPhase) { case IDLE: log.info("订单的任务未开始执行"); break; case ROUTE_TO_PICKUP: log.info("驶往上车点的行程"); buildNaviRequest(requestId, orderId, orderPhase, msgContent.getHeader().getSequence()); break; case ROUTE_TO_DESTINATION: log.info("驶往下车点的行程"); buildNaviRequest(requestId, orderId, orderPhase, msgContent.getHeader().getSequence()); break; case ROUTE_TO_CRUISE: log.info("巡游阶段路线"); buildNaviRequest(requestId, orderId, orderPhase, msgContent.getHeader().getSequence()); break; } } /** * 车辆偏航后重新规划线路 * * @param msgContent */ public void navigationDeviationEvent(MsgContent msgContent) { //先响应消息 Long sequence = msgContent.getHeader().getSequence(); //开始处理消息对应的业务逻辑 NavigationDeviationEvent deviationEvent = msgContent.getContent_detail().getObject("deviationEvent", NavigationDeviationEvent.class); String requestId = deviationEvent.getRequest_id(); //偏航事件类型 1：请求导航，2：取消导航 if (deviationEvent.getEvent_type().equals(RtEventType.RT_EVENT_TYPE_REQUEST)) { NavigationDeviationMessage navigationDeviationMessage = deviationEvent.getNavigation_deviation_message(); String orderId = navigationDeviationMessage.getOrder_id(); String taskId = navigationDeviationMessage.getTask_id(); Double longitude = navigationDeviationMessage.getLongitude(); Double latitude = navigationDeviationMessage.getLatitude(); //订单类型 1：乘客订单，2=巡游订单 LonLatDto destination = new LonLatDto(); LonLatDto origin = new LonLatDto(); if (orderId.contains("XU")) { // todo 下发巡游单的导航数据 //巡游单 origin.setLongitude(longitude); origin.setLatitude(latitude); destination.setLongitude(116.40528); destination.setLatitude(39.90498); //响应消息 responseMessage(sequence, ""); } else { //正常单 R xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId)); if (200 != xpOrderR.getCode()) { 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())) { //查询司机当前位置到起点的线路 origin.setLongitude(longitude); origin.setLatitude(latitude); destination.setLongitude(xpOrder.getStartLon()); destination.setLatitude(xpOrder.getStartLat()); } //驶往下车点的行程 if (Arrays.asList(6).contains(xpOrder.getStatus())) { //查询起点到终点的线路 origin.setLongitude(longitude); origin.setLatitude(latitude); destination.setLongitude(xpOrder.getEndLon()); destination.setLatitude(xpOrder.getEndLat()); } OrderRouteAndFareDto dto = new OrderRouteAndFareDto(); dto.setOrderId(Long.parseLong(orderId)); dto.setOrigin(origin); dto.setDestination(destination); R routeVoR = xpMapClient.getOrderDistanceAndDuration(dto); if (200 != routeVoR.getCode()) { throw new RuntimeException("获取订单线路和运费失败"); } RouteVo routeVo = routeVoR.getData(); R stringR = xpMapClient.queryRouted(routeVo.getRouteID()); if (200 != stringR.getCode()) { throw new RuntimeException("根据导航ID获取车端需要的导航数据失败"); } //构建导航下发请求对象 cloudPushNavigationInfo(requestId, routeVo.getRouteID(), stringR.getData(), orderId, xpVehicle.getVehicleVin(), xpVehicle.getXpuId()); } } } /** * 构建导航请求 * * @param orderId * @param orderPhase */ public void buildNaviRequest(String requestId, String orderId, OrderPhase orderPhase, Long sequence) { //订单类型 1：乘客订单，2=巡游订单 LonLatDto destination = new LonLatDto(); LonLatDto origin = new LonLatDto(); if (orderId.contains("XU")) { // todo 下发巡游单的导航数据 //巡游单 if (orderPhase.equals(OrderPhase.ROUTE_TO_CRUISE)) { origin.setLongitude(116.40528); origin.setLatitude(39.90498); destination.setLongitude(116.40528); destination.setLatitude(39.90498); //先响应消息 responseMessage(sequence, ""); } } else { //正常单 R xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId)); if (200 != xpOrderR.getCode()) { throw new RuntimeException("订单不存在"); } XpOrder xpOrder = xpOrderR.getData(); //驶往上车点的行程 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("车辆位置信息不存在"); } //查询司机当前位置到起点的线路 origin.setLongitude(vehiclePosition.getPoint().getX()); origin.setLatitude(vehiclePosition.getPoint().getY()); destination.setLongitude(xpOrder.getStartLon()); destination.setLatitude(xpOrder.getStartLat()); } //驶往下车点的行程 if (orderPhase.equals(OrderPhase.ROUTE_TO_DESTINATION)) { //查询起点到终点的线路 origin.setLongitude(xpOrder.getStartLon()); origin.setLatitude(xpOrder.getStartLat()); destination.setLongitude(xpOrder.getEndLon()); destination.setLatitude(xpOrder.getEndLat()); } OrderRouteAndFareDto dto = new OrderRouteAndFareDto(); dto.setOrderId(Long.parseLong(orderId)); dto.setOrigin(origin); dto.setDestination(destination); //TODO 这里的routeID不应该每次都重新生成，应该有一个缓存，调用的接口有做缓存但是没有读取缓存 R routeVoR = xpMapClient.getOrderDistanceAndDuration(dto); if (200 != routeVoR.getCode()) { throw new RuntimeException("获取订单线路和运费失败"); } XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData(); RouteVo routeVo = routeVoR.getData(); R stringR = xpMapClient.queryRouted(routeVo.getRouteID()); if (200 != stringR.getCode()) { throw new RuntimeException("根据导航ID获取车端需要的导航数据失败"); } //构建导航下发请求对象 cloudPushNavigationInfo(requestId, routeVo.getRouteID(), stringR.getData(), orderId, xpVehicle.getVehicleVin(), xpVehicle.getXpuId()); } } /** * 云端订单操作 * 更新途经点/终点 * @param routeVo */ public void cloudOrderOperation(Integer orderType, String orderId, String taskId, RouteVo routeVo, String vin) { //构建消息体 OrderOperation.Builder orderOperationBuilder = OrderOperation.newBuilder(); orderOperationBuilder.setOrderId(orderId); orderOperationBuilder.setTaskId(taskId); orderOperationBuilder.setOperationType(OrderOperationType.CLOUD_UPDATE_WAYPOINTS); //构建航点 List waypoints = new ArrayList<>(); List polyline = routeVo.getPolyline(); for (int i = 0; i < polyline.size(); i++) { LonLatDto lonLatDto = polyline.get(i); Waypoint.Builder waypointBuilder = Waypoint.newBuilder(); waypointBuilder.setWaypointId(UUID.randomUUID().toString()); if (0 == i) { waypointBuilder.setPointType(PointType.START_POINT); //订单类型不同，点位的动作类型不同 waypointBuilder.setActionType(1 == orderType ? ActionType.PICK_UP : ActionType.ACTION_UNSPECIFIED); } else if (polyline.size() - 1 == i) { waypointBuilder.setPointType(PointType.END_POINT); waypointBuilder.setActionType(1 == orderType ? ActionType.DROP_OFF : ActionType.ACTION_UNSPECIFIED); } else { waypointBuilder.setPointType(PointType.WAY_POINT); //订单类型不同，点位的动作类型不同 waypointBuilder.setActionType(ActionType.ACTION_UNSPECIFIED); } //构建经纬度 GeoPoint.Builder geoPointBuilder = GeoPoint.newBuilder(); geoPointBuilder.setLongitude(lonLatDto.getLongitude()); geoPointBuilder.setLatitude(lonLatDto.getLatitude()); waypointBuilder.setPoint(geoPointBuilder.build()); waypointBuilder.setSequence(i + 1); waypoints.add(waypointBuilder.build()); } orderOperationBuilder.addAllWaypoints(waypoints); long sequence = snowflake.nextId(); // 处理数据 redisTemplate.opsForValue().set(sequence,"pushOrder_"+orderId+"_"+vin,30, TimeUnit.MINUTES); //开发发送请求到XPU xpuUtil.cloudOrderOperation(orderOperationBuilder.build(), vin,sequence); } /** * 构建响应消息 */ public void responseMessage(Long request_seq, String vin) { ResponseMessage.Builder responseMessageBuilder = ResponseMessage.newBuilder(); responseMessageBuilder.setRequestSeq(request_seq); responseMessageBuilder.setCode(StatusCode.OK); responseMessageBuilder.setTimestamp(System.currentTimeMillis()); xpuUtil.responseMessage(responseMessageBuilder.build(), vin); } /** * 订单状态更新事件 * @param msgContent */ public void orderStateUpdateEvent(MsgContent msgContent) { //先响应消息 Long sequence = msgContent.getHeader().getSequence(); //开始处理消息对应的业务逻辑 OrderStatusInfoEvent orderStatusInfoEvent = msgContent.getContent_detail().getObject("orderStatusInfoEvent", OrderStatusInfoEvent.class); //只处理请求消息 if (orderStatusInfoEvent.getEvent_type().equals(RtEventType.RT_EVENT_TYPE_REQUEST)) { com.xiaopeng.api.vo.automaticPilot.proto.up.orderStateUpdateEvent.OrderStatusInfo orderStatusInfo = orderStatusInfoEvent.getOrder_status_info(); String orderId = orderStatusInfo.getOrder_id(); //普通单 com.xiaopeng.api.vo.automaticPilot.proto.up.orderStateUpdateEvent.OrderStatus status = orderStatusInfo.getStatus(); switch (status) { case ORDER_STATUS_UNSPECIFIED: //订单状态未知 log.info("订单状态未知，不做任何处理：{}", orderId); break; case ORDER_RECEIVED: //收到订单 buildPushOrderStatus(orderId, OrderStatusEnum.READY_DEPART, sequence); xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.READY_DEPART.getCode()); break; case ORDER_STARTED: //开始执行订单 log.info("开始执行订单，不做任何处理：{}", orderId); break; case ORDER_EN_ROUTE_TO_PICKUP: //前往上车点 buildPushOrderStatus(orderId, OrderStatusEnum.PICK_UP, sequence); xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.PICK_UP.getCode()); break; case ORDER_ARRIVED_AT_PICKUP: //到达上车点 log.info("到达上车点，不做任何处理：{}", orderId); break; case ORDER_WAITING_FOR_PICKUP: //等待乘客上车 buildPushOrderStatus(orderId, OrderStatusEnum.PICK_UP_GUESTS, sequence); xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.PICK_UP_GUESTS.getCode()); break; case ORDER_PASSENGER_ONBOARD: //乘客已上车 buildPushOrderStatus(orderId, OrderStatusEnum.PASSENGER_BOARDED, sequence); xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.PASSENGER_BOARDED.getCode()); break; case ORDER_EN_ROUTE_TO_DESTINATION: //前往目的地 buildPushOrderStatus(orderId, OrderStatusEnum.IN_SERVICE, sequence); xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.IN_SERVICE.getCode()); break; case ORDER_ARRIVED_AT_DESTINATION: //到达目的地 log.info("到达目的地，不做任何处理：{}", orderId); break; case ORDER_WAITING_FOR_DROPOFF: //等待乘客下车 buildPushOrderStatus(orderId, OrderStatusEnum.SEE_OFF_GUESTS, sequence); xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.SEE_OFF_GUESTS.getCode()); break; case ORDER_DROPOFF_COMPLETED: //下车完成 buildPushOrderStatus(orderId, OrderStatusEnum.APPROVAL_REJECTION, sequence); xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.APPROVAL_REJECTION.getCode()); break; case ORDER_COMPLETED: //已完成 log.info("已完成，不做任何处理：{}", orderId); break; case ORDER_CANCELLED: //已取消 buildPushOrderStatus(orderId, OrderStatusEnum.CANCELED, sequence); xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.CANCELED.getCode()); break; case ORDER_ARRIVED_AT_CRUISE_DESTINATION: //到达巡游终点 buildPushOrderStatus(orderId, OrderStatusEnum.COMPLETED, sequence); break; case ORDER_PAUSED: //订单暂停 log.info("订单暂停，不做任何处理：{}", orderId); break; default: break; } } } /** * 构建订单状态推送消息 * @param orderId * @param status */ public void buildPushOrderStatus(String orderId, OrderStatusEnum status, Long sequence){ if(orderId.contains("XU")){ //巡游单 //响应消息 // TODO 1.处理结束当前巡游订单 2.创建新的巡游订单 返航 3.调用xpu车机接口 （判断是返航还是正常 返航需要存储之前的订单id） responseMessage(sequence, ""); //前往目的地 if(OrderStatusEnum.IN_SERVICE.equals(status)){ } //到达目的地 if(OrderStatusEnum.COMPLETED.equals(status)){ } }else{ //普通单 R xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId)); XpOrder xpOrder = xpOrderR.getData(); XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData(); //响应消息 responseMessage(sequence, xpVehicle.getVehicleVin()); //构建推送消息 Map params = new HashMap<>(); Map content = new HashMap<>(); content.put("orderId", orderId); content.put("status", status.getCode()); params.put("type", PushMessageTypeEnum.ORDER_STATUS.getDesc()); params.put("content", JSONObject.toJSONString(content)); params.put("timestamp", System.currentTimeMillis() + ""); String message = JSONObject.toJSONString(params); Long appUserId = xpOrder.getAppUserId(); SendMessageDTO dto = new SendMessageDTO(); dto.setAppUserId(appUserId + ""); dto.setMessage(message); R stringR = xpPushMessageClient.sendStatus(dto); if (R.isError(stringR)) { log.error("推送消息失败：{}", stringR.getMsg()); } } } /** * 云端下发导航到XPU * @param requestId XPU请求id * @param routeId 高德导航id * @param routeDataContent 高德导航原数据 * @param orderId 订单id * @param vin 车辆VIN * @param xpuId XPU设备id */ public void cloudPushNavigationInfo(String requestId, String routeId, String routeDataContent, String orderId, String vin, String xpuId) { //组装报文，将routeDateContent信息推送到XPU RestoreNavigationData.Builder navigationDataBuilder = RestoreNavigationData.newBuilder(); navigationDataBuilder.setRouteId(routeId); navigationDataBuilder.setRequestId(requestId); navigationDataBuilder.setContent(routeDataContent); navigationDataBuilder.setTimestamp(System.currentTimeMillis()); long sequence = snowflake.nextId(); // 处理数据 redisTemplate.opsForValue().set(sequence,"pushOrder_"+orderId+"_"+vin,30, TimeUnit.MINUTES); xpuUtil.cloudPushNavigation(navigationDataBuilder.build(), xpuId, sequence); } }