package com.xiaopeng.automaticPilot.service; import cn.hutool.core.lang.Snowflake; import cn.hutool.core.util.IdUtil; 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.XpCruiseOrderClient; 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.XpCruiseOrder; import com.xiaopeng.api.model.order.XpOrder; import com.xiaopeng.api.model.vehicleConvoy.XpVehicle; import com.xiaopeng.api.vo.OrderType; import com.xiaopeng.api.vo.SystemEventType; import com.xiaopeng.api.vo.automaticPilot.VehiclePosition; import com.xiaopeng.api.vo.automaticPilot.proto.down.GeoPoint; 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.cloudCreateOrder.RTTask; import com.xiaopeng.api.vo.automaticPilot.proto.down.cloudCreateOrder.TaskStatus; import com.xiaopeng.api.vo.automaticPilot.proto.down.cloudCreateOrder.TaskType; 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; 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.SnowFlake; import com.xiaopeng.common.core.utils.StringUtils; import com.xiaopeng.common.core.utils.uuid.UUID; import com.xiaopeng.common.redis.service.RedisService; 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 XpCruiseOrderClient xpCruiseOrderClient; @Resource private Snowflake snowflake; @Autowired private RedisService redisService; /** * 消费消息 - 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){ 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); } } } } } /** * 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.getBms_soc_disp()); vehiclePosition.setMileageRange(dynamicStatus.getDst_bat_disp()); //订单信息 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(); //导航id String routeID = ""; Long vehicleId = null; String orderType = ""; //偏航事件类型 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")) { orderType = OrderType.CRUISE_ORDER; // todo 下发巡游单的导航数据 //先响应消息 responseMessage(sequence, ""); //构建导航数据给到车端 origin.setLongitude(116.40528); origin.setLatitude(39.90498); destination.setLongitude(116.40528); destination.setLatitude(39.90498); 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(); routeID = routeVo.getRouteID(); // vehicleId = xpOrder.getVehicleId(); } else { orderType = OrderType.USER_ORDER; //正常单 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()); 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(); routeID = routeVo.getRouteID(); //缓存数据，用于后续推给乘客导航线路 redisService.setCacheObject("ORDER_NAVIGATION_START_DATA:" + xpOrder.getAppUserId(), com.alibaba.fastjson.JSON.toJSONString(routeVo), 24L, TimeUnit.HOURS); } //驶往下车点的行程 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(); routeID = routeVo.getRouteID(); //缓存导航数据，用于后面的逻辑中给乘客端推送导航数据 redisService.setCacheObject("ORDER_NAVIGATION_END_DATA:" + xpOrder.getAppUserId(), com.alibaba.fastjson.JSON.toJSONString(routeVo), 24L, TimeUnit.HOURS); } } if(StringUtils.isNotEmpty(routeID)){ XpVehicle xpVehicle = xpVehicleClient.queryById(vehicleId).getData(); R stringR = xpMapClient.queryRouted(routeID); if (200 != stringR.getCode()) { throw new RuntimeException("根据导航ID获取车端需要的导航数据失败"); } //构建导航下发请求对象 cloudPushNavigationInfo(requestId, routeID, stringR.getData(), orderType, 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(); //导航id String routeID = ""; Long vehicleId = null; String orderType = ""; if (orderId.contains("XU")) { // todo 下发巡游单的导航数据 orderType = OrderType.CRUISE_ORDER; //巡游单 if (orderPhase.equals(OrderPhase.ROUTE_TO_CRUISE)) { //先响应消息 responseMessage(sequence, ""); /* 构建导航数据给到车端 */ R cruiseOrder = xpCruiseOrderClient.getCruiseOrder(orderId); XpCruiseOrder xpCruiseOrder = cruiseOrder.getData(); origin.setLongitude(xpCruiseOrder.getStartLon()); origin.setLatitude(xpCruiseOrder.getStartLat()); destination.setLongitude(xpCruiseOrder.getEndLon()); destination.setLatitude(xpCruiseOrder.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(); routeID = routeVo.getRouteID(); // vehicleId = xpOrder.getVehicleId(); } } else { orderType = OrderType.USER_ORDER; //正常单 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 (orderPhase.equals(OrderPhase.ROUTE_TO_PICKUP)) { //查询车辆位置 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()); 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(); routeID = routeVo.getRouteID(); //缓存数据，用于后续推给乘客导航线路 redisService.setCacheObject("ORDER_NAVIGATION_START_DATA:" + xpOrder.getAppUserId(), com.alibaba.fastjson.JSON.toJSONString(routeVo), 24L, TimeUnit.HOURS); } //驶往下车点的行程 if (orderPhase.equals(OrderPhase.ROUTE_TO_DESTINATION)) { //直接取缓存中的导航数据推送给车端和乘客 Object cacheObject = redisService.getCacheObject("ORDER_NAVIGATION_DATA:" + xpOrder.getAppUserId()); if(null == cacheObject){ //没有缓存数据，重新规划线路 //查询起点到终点的线路 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); R routeVoR = xpMapClient.getOrderDistanceAndDuration(dto); if (200 != routeVoR.getCode()) { throw new RuntimeException("获取订单线路和运费失败"); } RouteVo routeVo = routeVoR.getData(); routeID = routeVo.getRouteID(); //缓存导航数据，用于后面的逻辑中给乘客端推送导航数据 redisService.setCacheObject("ORDER_NAVIGATION_END_DATA:" + xpOrder.getAppUserId(), com.alibaba.fastjson.JSON.toJSONString(routeVo), 24L, TimeUnit.HOURS); }else{ RouteVo routeVo = JSON.parseObject(cacheObject.toString(), RouteVo.class); routeID = routeVo.getRouteID(); } } vehicleId = xpOrder.getVehicleId(); } if(StringUtils.isNotEmpty(routeID)){ XpVehicle xpVehicle = xpVehicleClient.queryById(vehicleId).getData(); R stringR = xpMapClient.queryRouted(routeID); if (200 != stringR.getCode()) { throw new RuntimeException("根据导航ID获取车端需要的导航数据失败"); } //构建导航下发请求对象 cloudPushNavigationInfo(requestId, routeID, stringR.getData(), orderType, orderId, xpVehicle.getVehicleVin(), xpVehicle.getXpuId()); } } /** * 云端订单操作 * 更新途经点/终点 * @param routeVo */ public void cloudOrderOperation(Integer orderType, String orderId, String taskId, RouteVo routeVo, String vin) { //构建消息体 OrderOperation orderOperation = new OrderOperation(); orderOperation.setOrder_id(orderId); orderOperation.setTask_id(taskId); orderOperation.setOperation_type(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 waypoint = new Waypoint(); waypoint.setWaypoint_id(UUID.randomUUID().toString()); if (0 == i) { waypoint.setPoint_type(PointType.START_POINT); //订单类型不同，点位的动作类型不同 waypoint.setAction_type(1 == orderType ? ActionType.PICK_UP : ActionType.ACTION_UNSPECIFIED); } else if (polyline.size() - 1 == i) { waypoint.setPoint_type(PointType.END_POINT); waypoint.setAction_type(1 == orderType ? ActionType.DROP_OFF : ActionType.ACTION_UNSPECIFIED); } else { waypoint.setPoint_type(PointType.WAY_POINT); //订单类型不同，点位的动作类型不同 waypoint.setAction_type(ActionType.ACTION_UNSPECIFIED); } //构建经纬度 GeoPoint geoPoint = new GeoPoint(); geoPoint.setLongitude(lonLatDto.getLongitude()); geoPoint.setLatitude(lonLatDto.getLatitude()); waypoint.setPoint(geoPoint); waypoint.setSequence(i + 1); waypoints.add(waypoint); } orderOperation.setWaypoints(waypoints); long sequence = snowflake.nextId(); // 处理数据 redisTemplate.opsForValue().set(sequence,"pushOrder_"+orderId+"_"+vin,30, TimeUnit.MINUTES); //开发发送请求到XPU xpuUtil.cloudOrderOperation(orderOperation, vin,sequence); } /** * 构建响应消息 */ 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, 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")){ //巡游单 //响应消息 responseMessage(sequence, ""); //前往目的地 if(OrderStatusEnum.IN_SERVICE.equals(status)){ } //到达目的地 if(OrderStatusEnum.COMPLETED.equals(status)){ // 返程找到目前的订单 -》之前的订单 查询生成一个 正常巡游 找到下一个点位 若到达最后点位 则反向找点位 每次判断是否收车状态 是则不进行处理了 // 找到这个订单 R cruiseOrder = xpCruiseOrderClient.getCruiseOrder(orderId); if(cruiseOrder!=null){ XpCruiseOrder xpCruiseOrder = cruiseOrder.getData(); Long vehicleId = xpCruiseOrder.getVehicleId(); // 完成当前巡游单子 已完成 xpCruiseOrderClient.closeCruiseOrder(xpCruiseOrder.getVehicleVin()); R xpVehicleR = xpVehicleClient.queryById(vehicleId); XpVehicle xpVehicle = xpVehicleR.getData(); // 运营中 继续执行 if(xpVehicle.getOperatingStatus()==1 && xpVehicle.getState()==0){ // 创建新的巡游单子 判断是否为返程单子 if(xpCruiseOrder.getOrderType()==2){ R lastOrder = xpCruiseOrderClient.getCruiseOrder(xpCruiseOrder.getLastOrderId()); XpCruiseOrder lastXpCruiseOrder = lastOrder.getData(); // 修改一下id 加一个一样的巡游单子 lastXpCruiseOrder.setId("XU"+ IdUtil.fastUUID()); lastXpCruiseOrder.setStatus(1); xpCruiseOrderClient.addCruiseOrder(lastXpCruiseOrder); RTTask task = new RTTask(); task.setOrder_id(lastXpCruiseOrder.getId()); task.setTask_id("MISSION_"+ IdUtil.fastUUID()); task.setTask_type(TaskType.CRUISE); task.setStatus(TaskStatus.PENDING); SnowFlake snowFlake = new SnowFlake(0, 0); snowFlake.withStartTimestamp(1420041600000L); long sequence1 = snowFlake.nextId(); xpuUtil.cloudCreateOrder(task, xpVehicle.getVehicleVin(),sequence1); }else if(xpCruiseOrder.getOrderType()==1){ // 正常巡游 找到下一个点位 若到达最后点位 则反向找点位 TODO v } } } } }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 orderType, String orderId, String vin, String xpuId) { //组装报文，将routeDateContent信息推送到XPU RestoreNavigationData navigationData = new RestoreNavigationData(); navigationData.setRoute_id(routeId); navigationData.setRequest_id(requestId); navigationData.setContent(routeDataContent); navigationData.setTimestamp(System.currentTimeMillis()); long sequence = snowflake.nextId(); // 处理数据 JSONObject jsonObject = new JSONObject(); jsonObject.put("event", SystemEventType.PUSH_NAVIGATION); jsonObject.put("orderType", orderType); jsonObject.put("orderId", orderId); jsonObject.put("vin", vin); redisTemplate.opsForValue().set(sequence, jsonObject.toJSONString(),30, TimeUnit.MINUTES); xpuUtil.cloudPushNavigation(navigationData, xpuId, sequence); } }