java/Robotaxi.git

			@@ -3,12 +3,11 @@
			import cn.hutool.core.lang.Snowflake;
			import com.alibaba.fastjson2.JSON;
			import com.alibaba.fastjson2.JSONObject;
			import com.google.common.collect.ImmutableMap;
			import com.google.protobuf.InvalidProtocolBufferException;
			import com.google.protobuf.util.JsonFormat;
			import com.xiaopeng.api.dto.automaticPilot.MqttPublishDto;
			import com.xiaopeng.api.dto.automaticPilot.PushOrderDto;
			import com.xiaopeng.api.dto.automaticPilot.VehicleEventDTO;
			import com.xiaopeng.api.dto.map.HistoryPointDto;
			import com.xiaopeng.api.dto.map.OrderRouteAndFareDto;
			import com.xiaopeng.api.dto.operation.CruisePointDto;
			import com.xiaopeng.api.dto.operation.LonLatDto;
			@@ -22,12 +21,14 @@
			import com.xiaopeng.api.feignClient.map.XpMapClient;
			import com.xiaopeng.api.feignClient.operation.ReassignOrderClient;
			import com.xiaopeng.api.feignClient.operation.XpCruisePlanClient;
			import com.xiaopeng.api.feignClient.operation.XpSystemSettingClient;
			import com.xiaopeng.api.feignClient.order.XpCruiseOrderClient;
			import com.xiaopeng.api.feignClient.order.XpOrderClient;
			import com.xiaopeng.api.feignClient.order.XpTestOrderClient;
			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.operation.XpSystemSetting;
			import com.xiaopeng.api.model.order.XpCruiseOrder;
			import com.xiaopeng.api.model.order.XpOrder;
			import com.xiaopeng.api.model.order.XpTestOrder;
			@@ -38,6 +39,9 @@
			import com.xiaopeng.api.vo.operation.RouteVo;
			import com.xiaopeng.api.vo.order.OrderTrajectoryVo;
			import com.xiaopeng.api.vo.vehicleConvoy.CancelTaxiPassengerVo;
			import com.xiaopeng.automaticPilot.job.base.QuartzManager;
			import com.xiaopeng.automaticPilot.job.base.TimeJobType;
			import com.xiaopeng.automaticPilot.job.jobs.OrderExceptionJob;
			import com.xiaopeng.automaticPilot.util.XPUUtil;
			import com.xiaopeng.common.core.constant.CacheConstants;
			import com.xiaopeng.common.core.constant.Constants;
			@@ -45,12 +49,12 @@
			import com.xiaopeng.common.core.enums.CancelOrderSceneEnum;
			import com.xiaopeng.common.core.enums.OrderStatusEnum;
			import com.xiaopeng.common.core.enums.PushMessageTypeEnum;
			import com.xiaopeng.common.core.utils.DateUtils;
			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.redisson.api.RLock;
			import org.redisson.api.RTopic;
			import org.redisson.api.RedissonClient;
			import org.springframework.beans.factory.annotation.Value;
			import org.springframework.data.geo.Point;
			@@ -63,9 +67,11 @@
			import org.springframework.stereotype.Service;

			import javax.annotation.Resource;
			import java.math.BigDecimal;
			import java.time.LocalDateTime;
			import java.util.*;
			import java.util.concurrent.*;
			import java.util.concurrent.LinkedBlockingQueue;
			import java.util.concurrent.ThreadPoolExecutor;
			import java.util.concurrent.TimeUnit;


			/**
			@@ -75,6 +81,7 @@
			@Service
			public class KafkaConsumerService {
			private static final String REASSIGN_CANCEL_CONTEXT_PREFIX = "REASSIGN_CANCEL_CONTEXT:";
			private static final String CANCEL_ORDER_CONTEXT_PREFIX = "CANCEL_ORDER_CONTEXT:";


			@Resource
			@@ -135,6 +142,9 @@

			@Resource
			private RedissonClient redissonClient;

			@Resource
			private XpSystemSettingClient systemSettingClient;

			private static final JsonFormat.Printer jsonPrinter = JsonFormat.printer()
			.printingEnumsAsInts()
			@@ -222,6 +232,7 @@
			break;
			case PULLOVER_STATUS_EVENT:
			log.info("靠边停车状态(XPU→云端)(XPU→座舱)：{}", message);
			pulloverStatusEvent(msgContent);
			break;
			case DRIVING_MODE_UPDATE_EVENT:
			log.info("驾驶模式变化(XPU→云端)(XPU→座舱)：{}", message);
			@@ -430,10 +441,12 @@
			RtTripInfo rtTripInfo = systemRuntimeStatus.getRunningOrderStatusInfo().getRtTripInfo();
			if (null != orderId && !order_id.contains("XU")) {
			PushOrderRouteDTO dto = new PushOrderRouteDTO();
			dto.setMessageType(PushMessageTypeEnum.ORDER_ROUTE.getDesc());
			dto.setOrderId(orderId);
			dto.setRouteType(1);
			dto.setLongitude(longitude);
			dto.setLatitude(latitude);
			dto.setSpeed(dynamicStatus.getSpeed());
			if (rtTripInfo.hasCurrentTask() && rtTripInfo.getCurrentTask().hasNaviProgressInfo()) {
			dto.setDuration(rtTripInfo.getCurrentTask().getNaviProgressInfo().getRouteRemainTime());
			dto.setDistance(rtTripInfo.getCurrentTask().getNaviProgressInfo().getRouteRemainDist());
			@@ -451,9 +464,12 @@
			if(!order_id.contains("TEST")){
			xpPushMessageClient.pushOrderRoute(dto);//存储历史轨迹
			// 给连环派单的用户推送
			Long nextOrderId = redisService.getCacheObject(CacheConstants.nextOrder(order_id));
			if (null != nextOrderId){
			dto.setOrderId(nextOrderId);
			String json = redisService.getCacheObject(CacheConstants.nextOrder(order_id));
			if (StringUtils.isNotEmpty(json)){
			JSONObject jsonObject = JSON.parseObject(json);
			dto.setOrderId(jsonObject.getLong("orderId"));
			dto.setEndLongitude(jsonObject.getDouble("endLongitude"));
			dto.setEndLatitude(jsonObject.getDouble("endLatitude"));
			xpPushMessageClient.pushOrderRoute(dto);
			}
			xpOrderClient.saveVehicleLocation(orderTrajectory);
			@@ -556,12 +572,12 @@
			reassignOrderDTO.setUid(xpOrder.getAppUserId().toString());
			reassignOrderDTO.setOrderNo(xpOrder.getOrderNumber());
			reassignOrderDTO.setCount(10);
			RealtimePose pose = systemRuntimeStatus.getPose();
			Wgs naviPose = pose.getNaviPose();
			Double latitude = naviPose.getLatitude();
			Double longitude = naviPose.getLongitude();
			reassignOrderDTO.setLongitude(longitude);
			reassignOrderDTO.setLatitude(latitude);
			// RealtimePose pose = systemRuntimeStatus.getPose();
			// Wgs naviPose = pose.getNaviPose();
			// Double latitude = naviPose.getLatitude();
			// Double longitude = naviPose.getLongitude();
			reassignOrderDTO.setLongitude(xpOrder.getStartLon());
			reassignOrderDTO.setLatitude(xpOrder.getStartLat());
			reassignOrderClient.reassignOrder(reassignOrderDTO);
			}
			}
			@@ -647,7 +663,7 @@
			LonLatDto origin = new LonLatDto();
			RouteVo routeVo = null;
			String key = "";

			Float speed = 0f;

			//添加redis锁，用于在发生偏航后重新规划线路与正常推送剩余导航数据之间的冲突
			RLock lock = redissonClient.getLock("PUSH_ORDER_ROUTE_LOCK:" + orderId);
			@@ -686,6 +702,13 @@
			} else if(orderId.contains("TEST")){
			R<XpTestOrder> cruiseOrder = testOrderClient.queryOrderById(Long.parseLong(orderId.substring(4)));
			XpTestOrder testOrder = cruiseOrder.getData();

			XpVehicle xpVehicle = xpVehicleClient.queryById(testOrder.getVehicleId()).getData();
			VehiclePosition vehiclePosition = mongoTemplate.findOne(Query.query(Criteria.where("vin").is(xpVehicle.getVehicleVin())), VehiclePosition.class);
			if(null != vehiclePosition){
			speed = vehiclePosition.getSpeed();
			}

			//区分测试单和测试巡游单
			if(1 == testOrder.getOrderType()){
			orderType = OrderType.TEST_ORDER;
			@@ -764,6 +787,10 @@
			}
			XpOrder xpOrder = xpOrderR.getData();
			XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData();
			VehiclePosition vehiclePosition = mongoTemplate.findOne(Query.query(Criteria.where("vin").is(xpVehicle.getVehicleVin())), VehiclePosition.class);
			if(null != vehiclePosition){
			speed = vehiclePosition.getSpeed();
			}
			//驶往上车点的行程
			if (3 == xpOrder.getStatus()) {
			//查询司机当前位置到起点的线路
			@@ -819,6 +846,7 @@
			dto.setRouteType(1);
			dto.setLongitude(longitude);
			dto.setLatitude(latitude);
			dto.setSpeed(speed);
			if (null != routeVo) {
			dto.setDuration(null == routeVo.getDuration() ? 0 : routeVo.getDuration().doubleValue());
			dto.setDistance(null == routeVo.getDistance() ? 0 : routeVo.getDistance().doubleValue());
			@@ -1164,7 +1192,7 @@
			testOrderClient.updateOrderStatus(Long.parseLong(orderId.substring(4)), OrderStatusEnum.READY_DEPART.getCode());
			}else{
			buildPushOrderStatus(cmd, orderId, OrderStatusEnum.READY_DEPART, sequence);
			xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.READY_DEPART.getCode());
			xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.READY_DEPART.getCode(), false);
			}
			}
			break;
			@@ -1179,7 +1207,7 @@
			testOrderClient.updateOrderStatus(Long.parseLong(orderId.substring(4)), OrderStatusEnum.PICK_UP.getCode());
			}else{
			buildPushOrderStatus(cmd, orderId, OrderStatusEnum.PICK_UP, sequence);
			xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.PICK_UP.getCode());
			xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.PICK_UP.getCode(), false);
			}
			}
			break;
			@@ -1194,24 +1222,18 @@
			testOrderClient.updateOrderStatus(Long.parseLong(orderId.substring(4)), OrderStatusEnum.PICK_UP_GUESTS.getCode());
			}else{
			buildPushOrderStatus(cmd, orderId, OrderStatusEnum.PICK_UP_GUESTS, sequence);
			xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.PICK_UP_GUESTS.getCode());
			xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.PICK_UP_GUESTS.getCode(), false);
			}
			}
			break;
			case ORDER_PASSENGER_ONBOARD:
			//先手动暂停触发，因为现目前车机响应状态时机不对
			try {
			Thread.sleep(waittime);
			} catch (InterruptedException e) {
			throw new RuntimeException(e);
			}
			//乘客已上车
			if(!orderId.contains("XU")){
			if(orderId.contains("TEST")){
			testOrderClient.updateOrderStatus(Long.parseLong(orderId.substring(4)), OrderStatusEnum.PASSENGER_BOARDED.getCode());
			}else{
			buildPushOrderStatus(cmd, orderId, OrderStatusEnum.PASSENGER_BOARDED, sequence);
			xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.PASSENGER_BOARDED.getCode());
			xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.PASSENGER_BOARDED.getCode(), false);
			}
			}
			break;
			@@ -1222,7 +1244,7 @@
			testOrderClient.updateOrderStatus(Long.parseLong(orderId.substring(4)), OrderStatusEnum.IN_SERVICE.getCode());
			}else{
			buildPushOrderStatus(cmd, orderId, OrderStatusEnum.IN_SERVICE, sequence);
			xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.IN_SERVICE.getCode());
			xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.IN_SERVICE.getCode(), false);
			}
			}
			break;
			@@ -1237,7 +1259,7 @@
			testOrderClient.updateOrderStatus(Long.parseLong(orderId.substring(4)), OrderStatusEnum.SEE_OFF_GUESTS.getCode());
			}else{
			buildPushOrderStatus(cmd, orderId, OrderStatusEnum.SEE_OFF_GUESTS, sequence);
			xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.SEE_OFF_GUESTS.getCode());
			xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.SEE_OFF_GUESTS.getCode(), false);
			}
			}
			break;
			@@ -1291,7 +1313,7 @@
			log.info("改派旧车取消回调，跳过订单取消处理，orderId: {}, taskId: {}, cancelScene: {}, context: {}",
			orderId, taskId, reassignCancelContext.getString("cancelScene"), reassignCancelContext);
			//响应消息
			String responseVin = resolveReassignCancelResponseVin(reassignCancelContext);
			String responseVin = resolveReassignCancelResponseVin(orderId, reassignCancelContext);
			if (StringUtils.isNotEmpty(responseVin)) {
			responseMessage(cmd, sequence, responseVin);
			} else {
			@@ -1341,14 +1363,30 @@
			}};
			if(xczqx.contains(xpOrder.getStatus())){//行程中取消订单，按照行程结束处理
			buildPushOrderStatus(cmd, orderId, OrderStatusEnum.SEE_OFF_GUESTS, sequence);
			xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.SEE_OFF_GUESTS.getCode());
			xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.SEE_OFF_GUESTS.getCode(), true);
			redisService.setCacheObject("ORDER_STATUS:" + orderId, OrderStatusEnum.SEE_OFF_GUESTS.getCode(), 24L, TimeUnit.HOURS);
			// todo 判断当前订单的车辆是否有连环派单，然后判断是都还满足连环派单的条件，不满足则需要重新派单

			// 判断当前订单的车辆是否有连环派单，然后判断是都还满足连环派单的条件，不满足则需要重新派单
			String json = redisService.getCacheObject(CacheConstants.nextOrder(orderId));
			if (StringUtils.isNotEmpty(json)){
			JSONObject jsonObject = JSON.parseObject(json);
			Long nextOrderId = jsonObject.getLong("orderId");
			XpOrder nextOrder = xpOrderClient.queryOrderById(nextOrderId).getData();
			if(null != nextOrder && nextOrder.getStatus() == OrderStatusEnum.PICK_UP.getCode()){
			redisService.setCacheObject(CacheConstants.nextOrder(orderId),null);
			ReassignOrderDTO reassignOrderDTO = new ReassignOrderDTO();
			reassignOrderDTO.setUid(xpOrder.getAppUserId().toString());
			reassignOrderDTO.setOrderNo(xpOrder.getOrderNumber());
			reassignOrderDTO.setCount(10);
			reassignOrderDTO.setLongitude(xpOrder.getStartLon());
			reassignOrderDTO.setLatitude(xpOrder.getStartLon());
			reassignOrderClient.reassignOrder(reassignOrderDTO);
			}
			}
			redisService.getCacheObject("ORDER_CONNECT_ORDER:" + orderId);

			}else{
			buildPushOrderStatus(cmd, orderId, OrderStatusEnum.CANCELED, sequence);
			xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.CANCELED.getCode());
			xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.CANCELED.getCode(), true);
			redisService.setCacheObject("ORDER_STATUS:" + orderId, OrderStatusEnum.CANCELED.getCode(), 24L, TimeUnit.HOURS);
			}
			}
			@@ -1455,6 +1493,7 @@
			newXpCruiseOrder.setStartPointName(xpCruiseOrder.getEndPointName());
			newXpCruiseOrder.setStatus(1);
			newXpCruiseOrder.setLastOrderId(xpCruiseOrder.getId());
			xpCruiseOrder.setXpuTaskId(snowflake.nextId() + "");

			// 通过车找终点id
			CruisePointDto cruisePointDto = new CruisePointDto();
			@@ -1464,7 +1503,12 @@
			cruisePointDto.setOrderReverse(xpCruiseOrder.getOrderReverse());
			R<XpCruiseOrder> cruisePoint = xpCruisePlanClient.getCruisePoint(cruisePointDto);
			if(cruisePoint!=null){
			xpCruiseOrderClient.addCruiseOrder(cruisePoint.getData());
			XpCruiseOrder cruiseOrder1 = cruisePoint.getData();
			//不停车替换新任务
			cruiseOrder1.setLastOrderId1("XU" + xpCruiseOrder.getId());
			cruiseOrder1.setLastXpuTaskId(xpCruiseOrder.getXpuTaskId());
			cruiseOrder1.setTaskActionType(TaskActionType.kTerminateForNewTask);
			xpCruiseOrderClient.addCruiseOrder(cruiseOrder1);
			}
			}
			}
			@@ -1514,6 +1558,10 @@
			}
			}

			/**
			* 当前 cancelOrder 指令是按 taskId 记录取消上下文。
			* 改派旧车取消回调优先读取该上下文，避免依赖订单当前车/状态造成竞态误判。
			*/
			private JSONObject getReassignCancelContext(String orderId) {
			if (StringUtils.isEmpty(orderId)) {
			return null;
			@@ -1529,9 +1577,56 @@
			if (OrderStateChangeSource.ORDER_STATE_CHANGE_SOURCE_VEHICLE != orderStatusInfo.getChangeSource()) {
			return null;
			}
			JSONObject cancelContext = getReassignCancelContext(orderId);
			String taskId = orderStatusInfo.getTaskId();
			// 先按 taskId 读取当前生效的取消上下文，再兼容历史按 orderId 存储的改派上下文。
			JSONObject cancelContext = getCancelOrderContext(taskId);
			if (cancelContext == null) {
			cancelContext = getReassignCancelContext(orderId);
			}
			if (cancelContext == null) {
			log.info("改派取消回调未命中专用上下文，orderId: {}, taskId: {}, missReason: context_not_found",
			orderId, taskId);
			return null;
			}
			String cancelScene = cancelContext.getString("cancelScene");
			if (CancelOrderSceneEnum.REASSIGN_OLD_VEHICLE != CancelOrderSceneEnum.fromCode(cancelScene)) {
			log.info("改派取消回调未命中专用上下文，orderId: {}, taskId: {}, missReason: scene_not_match, cancelScene: {}, context: {}",
			orderId, taskId, cancelScene, cancelContext);
			return null;
			}
			String contextOrderId = cancelContext.getString("orderId");
			if (StringUtils.isNotEmpty(contextOrderId) && !Objects.equals(contextOrderId, orderId)) {
			log.info("改派取消回调未命中专用上下文，orderId: {}, taskId: {}, missReason: order_not_match, contextOrderId: {}, context: {}",
			orderId, taskId, contextOrderId, cancelContext);
			return null;
			}
			return cancelContext;
			}


			/**
			* 当前 cancelOrder 指令是按 taskId 记录取消上下文。
			* 改派旧车取消回调优先读取该上下文，避免依赖订单当前车/状态造成竞态误判。
			*/
			private JSONObject getCancelOrderContext(String taskId) {
			if (StringUtils.isEmpty(taskId)) {
			return null;
			}
			Object cacheObject = redisService.getCacheObject(CANCEL_ORDER_CONTEXT_PREFIX + taskId);
			if (cacheObject == null) {
			return null;
			}
			return JSON.parseObject(cacheObject.toString());
			}


			private String resolveReassignCancelResponseVin(String orderId, JSONObject cancelContext) {
			if (cancelContext == null) {
			return null;
			}
			String vin = cancelContext.getString("vin");
			if (StringUtils.isNotEmpty(vin)) {
			return vin;
			}
			if (CancelOrderSceneEnum.REASSIGN_OLD_VEHICLE != CancelOrderSceneEnum.fromCode(
			cancelContext.getString("cancelScene"))) {
			@@ -1548,16 +1643,7 @@
			boolean vehicleMatched = Objects.nonNull(newVehicleId) && Objects.equals(newVehicleId, xpOrder.getVehicleId());
			boolean statusMatched = Objects.nonNull(currentStatus) && currentStatus > 0 && currentStatus < 8;
			if (vehicleMatched && statusMatched) {
			return cancelContext;
			}
			log.info("改派取消回调未命中专用上下文，orderId: {}, currentVehicleId: {}, newVehicleId: {}, currentStatus: {}",
			orderId, xpOrder.getVehicleId(), newVehicleId, currentStatus);
			return null;
			}

			private String resolveReassignCancelResponseVin(JSONObject cancelContext) {
			if (cancelContext == null) {
			return null;
			return newVehicleId.toString();
			}
			String oldVehicleVin = cancelContext.getString("oldVehicleVin");
			if (StringUtils.isNotEmpty(oldVehicleVin)) {
			@@ -1569,6 +1655,8 @@
			}
			R<XpVehicle> vehicleResult = xpVehicleClient.queryById(oldVehicleId);
			XpVehicle xpVehicle = vehicleResult == null ? null : vehicleResult.getData();
			log.info("改派取消回调未命中专用上下文，orderId: {}, currentVehicleId: {}, newVehicleId: {}, currentStatus: {}",
			orderId, xpOrder.getVehicleId(), newVehicleId, currentStatus);
			return xpVehicle == null ? null : xpVehicle.getVehicleVin();
			}

			@@ -1596,4 +1684,36 @@
			log.info("推送导航的响应消息，{}", JSON.toJSONString(mqttPublishDto));
			}



			//靠边停车处理逻辑
			public void pulloverStatusEvent(MsgContent msgContent){
			// todo 等对接完成数据结构后再实现相关功能
			//开始处理消息对应的业务逻辑
			NavigationRequest naviRequest = msgContent.getNavigationRequest();
			String requestId = naviRequest.getRequestId();
			String orderId = naviRequest.getOrderId();
			String taskId = naviRequest.getTaskId();
			OrderPhase orderPhase = naviRequest.getOrderPhase();
			log.info("订单ID：{}", orderId);
			log.info("任务ID：{}", taskId);

			//获取系统异常订单超时配置，创建定时任务取消订单
			if(!orderId.contains("XU")){
			XpSystemSetting xpSystemSetting = systemSettingClient.getSystemSettingOne().getData();
			Map<String, ? extends Object> jobMaps =
			new ImmutableMap.Builder<String, String>()
			.put("id", orderId)
			.put("orderType", orderId.contains("TEST") ? OrderType.TEST_ORDER : OrderType.USER_ORDER)
			.build();
			QuartzManager.addJob(
			OrderExceptionJob.class,
			(OrderExceptionJob.name + orderId).toUpperCase(),
			TimeJobType.ERROR_ORDER_CANCEL,
			DateUtils.localDateTimeToDate(LocalDateTime.now().plusMinutes(xpSystemSetting.getOrderExceptionalTimeout())),
			jobMaps
			);
			}

			}
			}