package com.xiaopeng.automaticPilot.service;
|
|
import cn.hutool.core.lang.Snowflake;
|
import com.alibaba.fastjson2.JSON;
|
import com.alibaba.fastjson2.JSONObject;
|
import com.google.protobuf.InvalidProtocolBufferException;
|
import com.google.protobuf.util.JsonFormat;
|
import com.xiaopeng.api.dto.automaticPilot.MqttPublishDto;
|
import com.xiaopeng.api.dto.map.OrderRouteAndFareDto;
|
import com.xiaopeng.api.dto.operation.CruisePointDto;
|
import com.xiaopeng.api.dto.operation.LonLatDto;
|
import com.xiaopeng.api.dto.operation.ReassignOrderDTO;
|
import com.xiaopeng.api.dto.order.AddCruiseMissionDto;
|
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.maintenance.XpTaskClient;
|
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.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.automaticPilot.VehiclePosition;
|
import com.xiaopeng.api.vo.automaticPilot.protobuf.*;
|
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 com.xiaopeng.common.redis.service.RedisService;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.beans.factory.annotation.Value;
|
import org.springframework.data.geo.Point;
|
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.mongodb.core.query.Update;
|
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 RedisService redisService;
|
|
@Resource
|
private XpCruisePlanClient xpCruisePlanClient;
|
|
@Resource
|
private Snowflake snowflake;
|
|
@Value("${robotaxi.appId:xp_robotaxi_api}")
|
private String appid;
|
|
@Resource
|
private XpTaskClient xpTaskClient;
|
|
private static final JsonFormat.Printer jsonPrinter = JsonFormat.printer()
|
.printingEnumsAsInts()
|
.preservingProtoFieldNames();
|
|
|
|
/**
|
* 消费消息 - XPU上报Robotaxi云端的消息topic（高优先级）
|
*
|
*/
|
public void topicProdRobotaxiXpuHigh(String message) {
|
MsgContent msgContent = parseMsgContent(message);
|
if (msgContent != null) {
|
handleXPUMessage(msgContent, message);
|
}
|
|
}
|
|
/**
|
* 消费消息 - XPU上报Robotaxi云端的消息topic（低优先级）
|
*/
|
public void topicProdRobotaxiXpuNormal(String message) {
|
MsgContent msgContent = parseMsgContent(message);
|
if (msgContent != null) {
|
handleXPUMessage(msgContent, message);
|
}
|
|
}
|
|
private MsgContent parseMsgContent(String message) {
|
try {
|
// 首先解析外层JSON，提取payload字段
|
JSONObject outerJson = JSON.parseObject(message);
|
String payload = outerJson.getString("payload");
|
|
if (StringUtils.isEmpty(payload)) {
|
log.error("Payload is empty in message: {}", message);
|
return null;
|
}
|
|
// 使用payload构建MsgContent
|
MsgContent.Builder builder = MsgContent.newBuilder();
|
JsonFormat.parser().ignoringUnknownFields().merge(payload, builder);
|
return builder.build();
|
} catch (InvalidProtocolBufferException e) {
|
log.error("Failed to parse message: {}", message, e);
|
return null;
|
} catch (Exception e) {
|
log.error("Failed to extract payload from message: {}", message, e);
|
return null;
|
}
|
}
|
|
|
/**
|
* 处理XPU消息
|
* @param msgContent
|
* @param message
|
*/
|
public void handleXPUMessage(MsgContent msgContent, String message){
|
MsgHeader header = msgContent.getHeader();
|
MessageType type = header.getType();
|
//处理请求消息
|
if (MessageType.MESSAGE_TYPE_REQUEST == type) {
|
//区分业务类型
|
int cmdValue = header.getCmd();
|
RTCmdType cmd = RTCmdType.getByValue(cmdValue);
|
if (cmd == null) {
|
log.warn("Unknown cmd value: {}", cmdValue);
|
return;
|
}
|
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) {
|
//区分业务类型
|
int cmdValue = header.getCmd();
|
RTCmdType cmd = RTCmdType.getByValue(cmdValue);
|
if (cmd == null) {
|
log.warn("Unknown cmd value: {}", cmdValue);
|
return;
|
}
|
//云端下发订单的响应报文
|
if (cmd == RTCmdType.CLOUD_SEND_RT_TASK) {
|
RtTaskListResponse response = msgContent.getRtTaskListResponse();
|
try {
|
log.info("云端下发订单操作的响应消息 (XPU→云端)：{}", jsonPrinter.print(response));
|
Object sequence = redisService.getCacheObject("CREATE_ORDER:" + response.getResults().getOrderId());
|
if(sequence!=null){
|
JSONObject jsonObject = JSON.parseObject(sequence.toString());
|
String orderId = jsonObject.getString("orderId");
|
String vin = jsonObject.getString("vin");
|
String orderType = jsonObject.getString("orderType");
|
PopOrderDto dto = new PopOrderDto();
|
dto.setOrderId(Long.valueOf(orderId.contains("XU") ? orderId.substring(2) : orderId));
|
dto.setXpVehicleVin(vin);
|
dto.setOrderType(orderType);
|
xpOrderClient.popOrder(dto);
|
log.info("订单状态更新：{}", response.getResults().getOrderId());
|
}
|
} catch (InvalidProtocolBufferException e) {
|
log.error("Failed to print RtTaskListResponse: {}", response, e);
|
}
|
} else {
|
log.info("响应消息不在处理范围内，忽略：{}", message);
|
}
|
}
|
}
|
|
|
|
|
|
/**
|
* XPU发布系统运行状态---车辆实时位置
|
*
|
* @param msgContent
|
*/
|
public void systemStatusXpu(MsgContent msgContent) {
|
//开始处理消息对应的业务逻辑
|
SystemRuntimeStatus systemRuntimeStatus = msgContent.getSystemRuntimeStatus();
|
//获取实时位置
|
RealtimePose pose = systemRuntimeStatus.getPose();
|
Wgs naviPose = pose.getNaviPose();
|
double latitude = naviPose.getLatitude();
|
double longitude = naviPose.getLongitude();
|
Double course = naviPose.getCourse();
|
Float alt = naviPose.getAlt();
|
//获取车辆静态信息
|
VehicleStaticInfo staticInfo = systemRuntimeStatus.getVehicleStatus().getStaticInfo();
|
String xpuId = staticInfo.getXpuId();
|
XpVehicle xpVehicle = xpVehicleClient.findByVehicleXpuId(xpuId).getData();
|
String vin = xpVehicle.getVehicleVin();
|
//先响应消息
|
long sequence = msgContent.getHeader().getSequence();
|
responseMessage(sequence, vin);
|
//保存车辆实时位置到redis
|
if(0 != longitude && 0 != latitude){
|
redisTemplate.opsForGeo().add("vehiclePosition:" + vin, new Point(longitude, latitude), vin);
|
}
|
|
VehicleStatus vehicleStatus = systemRuntimeStatus.getVehicleStatus();
|
OrderStatusInfo runningOrderStatusInfo = systemRuntimeStatus.getRunningOrderStatusInfo();
|
String order_id = runningOrderStatusInfo.getOrderId();
|
//车辆动态信息
|
VehicleDynamicStatus dynamicStatus = vehicleStatus.getDynamicStatus();
|
Long orderId = null;
|
Integer hasOrder = 0;
|
if (StringUtils.isNotEmpty(order_id)) {
|
if (order_id.contains("XU")) {
|
hasOrder = 1;
|
orderId = Long.parseLong(order_id.substring(2));
|
} else {
|
orderId = Long.parseLong(order_id);
|
}
|
}
|
//保存车辆实时信息到mongodb
|
if(0 != longitude && 0 != latitude){
|
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);
|
vehiclePosition.setSpeed(dynamicStatus.getSpeed());
|
vehiclePosition.setCurrentElectricity(dynamicStatus.getBmsSocDisp());
|
vehiclePosition.setMileageRange(dynamicStatus.getDstBatDisp());
|
vehiclePosition.setOrderId(orderId);
|
mongoTemplate.save(vehiclePosition);
|
}else{
|
vehiclePosition.setOrderId(orderId);
|
Update update = new Update()
|
.set("point", new GeoJsonPoint(longitude, latitude))
|
.set("course", course)
|
.set("alt", alt)
|
.set("speed", dynamicStatus.getSpeed())
|
.set("currentElectricity", dynamicStatus.getBmsSocDisp())
|
.set("mileageRange", dynamicStatus.getDstBatDisp())
|
.set("hasOrder", hasOrder)
|
.set("orderId", orderId);
|
mongoTemplate.updateFirst(new Query(Criteria.where("vin").is(vin)), update, VehiclePosition.class);
|
}
|
//调用推送服务接口
|
if (null != orderId) {
|
RtTripInfo rtTripInfo = systemRuntimeStatus.getRtTripInfo();
|
PushOrderRouteDTO dto = new PushOrderRouteDTO();
|
dto.setOrderId(orderId);
|
dto.setRouteType(order_id.contains("XU") ? 2 : 1);
|
dto.setLongitude(longitude);
|
dto.setLatitude(latitude);
|
//TODO 这两个字段现在拿不到
|
// dto.setDuration(rtTripInfo.getRemaining_time());
|
// dto.setDistance(rtTripInfo.getRemaining_distance().intValue());
|
// dto.setDirection();
|
xpPushMessageClient.pushOrderRoute(dto);
|
}
|
}
|
}
|
|
|
|
|
/**
|
* XPU发布系统运行状态---车辆实时位置
|
*
|
* @param msgContent
|
*/
|
public void systemStatusXpuAndReassign(MsgContent msgContent) {
|
//开始处理消息对应的业务逻辑
|
SystemRuntimeStatus systemRuntimeStatus = msgContent.getSystemRuntimeStatus();
|
AbnormalReason abnormalReason = systemRuntimeStatus.getSystemStatus().getAbnormalReason();
|
// 存储车辆挡位
|
GearLevel gearLevel = systemRuntimeStatus.getVehicleStatus().getDynamicStatus().getGearLevel();
|
int value = gearLevel.getNumber();
|
redisService.setCacheObject(CacheConstants.GEAR_LEVEL + systemRuntimeStatus.getVehicleStatus().getStaticInfo().getXpuId(), value);
|
// 判断是否为故障状态
|
XpVehicle vehicle = null;
|
if (abnormalReason == AbnormalReason.FAULT) {
|
log.info("车辆故障，开始处理故障");
|
String orderId = systemRuntimeStatus.getRunningOrderStatusInfo().getOrderId();
|
if(!orderId.contains("XU")){
|
//巡游订单，需要关闭订单
|
xpCruiseOrderClient.cancelCruiseOrderById(Long.parseLong(orderId.substring(2)));
|
XpCruiseOrder cruiseOrder = xpCruiseOrderClient.getCruiseOrder(Long.parseLong(orderId.substring(2))).getData();
|
vehicle = xpVehicleClient.queryById(cruiseOrder.getVehicleId()).getData();
|
if (vehicle != null) {
|
// 修改车辆状态
|
vehicle.setOperatingStatus(4);
|
xpVehicleClient.updateVehicleById(vehicle);
|
}
|
}else{
|
//正常订单
|
XpOrder xpOrder = xpOrderClient.queryOrderById(Long.parseLong(orderId)).getData();
|
vehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData();
|
if (vehicle != null) {
|
// 修改车辆状态
|
vehicle.setState(0);
|
vehicle.setOperatingStatus(4);
|
xpVehicleClient.updateVehicleById(vehicle);
|
// 解绑用户与车辆关系
|
CancelTaxiPassengerVo cancelTaxiPassengerVo = new CancelTaxiPassengerVo();
|
cancelTaxiPassengerVo.setVin(vehicle.getVehicleVin());
|
cancelTaxiPassengerVo.setUid(xpOrder.getAppUserId().toString());
|
cancelTaxiPassengerVo.setOrderNo(xpOrder.getOrderNumber());
|
cancelTaxiPassengerVo.setAppId(appid);
|
userVehicleClient.cancelTaxiPassenger(cancelTaxiPassengerVo);
|
}
|
// TODO 重新派单
|
ReassignOrderDTO reassignOrderDTO = new ReassignOrderDTO();
|
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);
|
reassignOrderClient.reassignOrder(reassignOrderDTO);
|
}
|
//添加车辆异常任务和发送消息
|
xpTaskClient.saveVehicleAccidentError(vehicle.getId());
|
}
|
}
|
|
/**
|
* XPU向云端请求导航
|
*
|
* @param msgContent
|
*/
|
public void xpuNavigationRequest(MsgContent msgContent) {
|
//开始处理消息对应的业务逻辑
|
NavigationRequest naviRequest = msgContent.getNavigationRequset();
|
String requestId = naviRequest.getRequestId();
|
String orderId = naviRequest.getOrderId();
|
String taskId = naviRequest.getTaskId();
|
OrderPhase orderPhase = naviRequest.getOrderPhase();
|
log.info("订单ID：{}", orderId);
|
log.info("任务ID：{}", taskId);
|
switch (orderPhase) {
|
case IDLE:
|
log.info("订单的任务未开始执行");
|
break;
|
case ROUTE_TO_PICKUP:
|
log.info("驶往上车点的行程 {}", orderId);
|
buildNaviRequest(requestId, orderId, orderPhase, msgContent.getHeader().getSequence());
|
break;
|
case ROUTE_TO_DESTINATION:
|
log.info("驶往下车点的行程 {}", orderId);
|
buildNaviRequest(requestId, orderId, orderPhase, msgContent.getHeader().getSequence());
|
break;
|
case ROUTE_TO_CRUISE:
|
log.info("巡游阶段路线 {}", orderId);
|
buildNaviRequest(requestId, orderId, orderPhase, msgContent.getHeader().getSequence());
|
break;
|
}
|
}
|
|
|
/**
|
* 车辆偏航后重新规划线路
|
*
|
* @param msgContent
|
*/
|
public void navigationDeviationEvent(MsgContent msgContent) {
|
//开始处理消息对应的业务逻辑
|
NavigationDeviationEvent deviationEvent = msgContent.getNaviDeviationEvent();
|
String requestId = deviationEvent.getNavigationDeviationMessage().getNaviRecalcRequestSeq() + "";
|
//导航id
|
String routeID = "";
|
Long vehicleId = null;
|
String orderType = "";
|
//偏航事件类型 1：请求导航，2：取消导航
|
NavigationDeviationMessage navigationDeviationMessage = deviationEvent.getNavigationDeviationMessage();
|
String orderId = navigationDeviationMessage.getOrderId();
|
String taskId = navigationDeviationMessage.getTaskId();
|
Wgs wgs = navigationDeviationMessage.getPoint();
|
Double longitude = wgs.getLongitude();
|
Double latitude = wgs.getLatitude();
|
//订单类型 1：乘客订单，2=巡游订单
|
LonLatDto destination = new LonLatDto();
|
LonLatDto origin = new LonLatDto();
|
|
if (orderId.contains("XU")) {
|
orderType = OrderType.CRUISE_ORDER;
|
|
R<XpCruiseOrder> cruiseOrder = xpCruiseOrderClient.getCruiseOrder(Long.parseLong(orderId.substring(2)));
|
XpCruiseOrder xpCruiseOrder = cruiseOrder.getData();
|
|
//构建导航数据给到车端
|
origin.setLongitude(longitude);
|
origin.setLatitude(latitude);
|
destination.setLongitude(xpCruiseOrder.getEndLon());
|
destination.setLatitude(xpCruiseOrder.getEndLat());
|
|
OrderRouteAndFareDto dto = new OrderRouteAndFareDto();
|
dto.setOrderId(Long.parseLong(orderId.substring(2)));
|
dto.setOrigin(origin);
|
dto.setDestination(destination);
|
dto.setRouteType(3);
|
R<RouteVo> routeVoR = xpMapClient.getOrderDistanceAndDuration(dto);
|
if (200 != routeVoR.getCode()) {
|
throw new RuntimeException("获取订单线路和运费失败");
|
}
|
RouteVo routeVo = routeVoR.getData();
|
routeID = routeVo.getRouteID();
|
|
|
vehicleId = xpCruiseOrder.getVehicleId();
|
} else {
|
orderType = OrderType.USER_ORDER;
|
//正常单
|
R<XpOrder> xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId));
|
if (200 != xpOrderR.getCode()) {
|
throw new RuntimeException("订单不存在");
|
}
|
XpOrder xpOrder = xpOrderR.getData();
|
XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData();
|
//驶往上车点的行程
|
if (3 == 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);
|
dto.setRouteType(1);
|
R<RouteVo> 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(), JSON.toJSONString(routeVo), 24L, TimeUnit.HOURS);
|
}
|
//驶往下车点的行程
|
if (6 == 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);
|
dto.setRouteType(2);
|
R<RouteVo> 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);
|
}
|
vehicleId = xpOrder.getVehicleId();
|
}
|
//直接响应回最新导航数据
|
if(StringUtils.isNotEmpty(routeID)){
|
XpVehicle xpVehicle = xpVehicleClient.queryById(vehicleId).getData();
|
R<String> 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")) {
|
orderType = OrderType.CRUISE_ORDER;
|
//巡游单
|
if (orderPhase.equals(OrderPhase.ROUTE_TO_CRUISE)) {
|
/* 构建导航数据给到车端 */
|
R<XpCruiseOrder> cruiseOrder = xpCruiseOrderClient.getCruiseOrder(Long.parseLong(orderId.substring(2)));
|
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.substring(2)));
|
dto.setOrigin(origin);
|
dto.setDestination(destination);
|
dto.setRouteType(3);
|
R<RouteVo> routeVoR = xpMapClient.getOrderDistanceAndDuration(dto);
|
if (200 != routeVoR.getCode()) {
|
throw new RuntimeException("获取订单线路和运费失败");
|
}
|
RouteVo routeVo = routeVoR.getData();
|
routeID = routeVo.getRouteID();
|
|
|
vehicleId = xpCruiseOrder.getVehicleId();
|
}
|
} else {
|
orderType = OrderType.USER_ORDER;
|
//正常单
|
R<XpOrder> xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId));
|
if (200 != xpOrderR.getCode()) {
|
throw new RuntimeException("订单不存在 " + orderId);
|
}
|
XpOrder xpOrder = xpOrderR.getData();
|
XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData();
|
//驶往上车点的行程
|
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);
|
dto.setRouteType(1);
|
R<RouteVo> 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(), JSON.toJSONString(routeVo), 24L, TimeUnit.HOURS);
|
}
|
//驶往下车点的行程
|
if (orderPhase.equals(OrderPhase.ROUTE_TO_DESTINATION)) {
|
//直接取缓存中的导航数据推送给车端和乘客
|
Object cacheObject = redisService.getCacheObject("ORDER_NAVIGATION_END_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);
|
dto.setRouteType(2);
|
R<RouteVo> 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)){
|
log.info("开始推送导航信息 {}", orderId);
|
XpVehicle xpVehicle = xpVehicleClient.queryById(vehicleId).getData();
|
R<String> stringR = xpMapClient.queryRouted(routeID);
|
if (200 != stringR.getCode()) {
|
throw new RuntimeException("根据导航ID获取车端需要的导航数据失败");
|
}
|
log.info("根据导航ID获取车端需要的导航数据 订单 {} 导航数据 {}", orderId, stringR.getData());
|
//构建导航下发请求对象
|
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.Builder orderOperationBuilder = OrderOperation.newBuilder();
|
orderOperationBuilder.setOrderId(orderId);
|
orderOperationBuilder.setTaskId(taskId);
|
orderOperationBuilder.setOperationType(OrderOperationType.CLOUD_UPDATE_WAYPOINTS);
|
|
|
//构建航点
|
List<Waypoint> waypoints = new ArrayList<>();
|
List<LonLatDto> 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();
|
// 处理数据
|
redisService.setCacheObject(sequence + "","pushOrder_"+orderId+"_"+vin,30L, 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.getOrderStatusInfoEvent();
|
//只处理请求消息
|
if (orderStatusInfoEvent.getEventType().equals(RtEventType.kEventOrderStateUpdate)) {
|
OrderStatusInfo orderStatusInfo = orderStatusInfoEvent.getOrderStatusInfo();
|
String orderId = orderStatusInfo.getOrderId();
|
//普通单
|
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.contains("XU") ? orderId.substring(2) : 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.contains("XU") ? orderId.substring(2) : 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.contains("XU") ? orderId.substring(2) : orderId), OrderStatusEnum.PICK_UP_GUESTS.getCode());
|
break;
|
case ORDER_PASSENGER_ONBOARD:
|
//乘客已上车
|
buildPushOrderStatus(orderId, OrderStatusEnum.PASSENGER_BOARDED, sequence);
|
xpOrderClient.updateOrderStatus(Long.parseLong(orderId.contains("XU") ? orderId.substring(2) : orderId), OrderStatusEnum.PASSENGER_BOARDED.getCode());
|
break;
|
case ORDER_EN_ROUTE_TO_DESTINATION:
|
//前往目的地
|
buildPushOrderStatus(orderId, OrderStatusEnum.IN_SERVICE, sequence);
|
xpOrderClient.updateOrderStatus(Long.parseLong(orderId.contains("XU") ? orderId.substring(2) : 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.contains("XU") ? orderId.substring(2) : orderId), OrderStatusEnum.SEE_OFF_GUESTS.getCode());
|
break;
|
case ORDER_DROPOFF_COMPLETED:
|
//下车完成
|
buildPushOrderStatus(orderId, OrderStatusEnum.APPROVAL_REJECTION, sequence);
|
xpOrderClient.updateOrderStatus(Long.parseLong(orderId.contains("XU") ? orderId.substring(2) : orderId), OrderStatusEnum.APPROVAL_REJECTION.getCode());
|
// 下发巡游任务
|
R<XpOrder> xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId));
|
if(xpOrderR.getCode()==200){
|
XpOrder orderData = xpOrderR.getData();
|
Long vehicleId =orderData.getVehicleId();
|
R<XpVehicle> xpVehicleR = xpVehicleClient.queryById(vehicleId);
|
if(xpVehicleR.getData()!=null){
|
XpVehicle xpVehicle = xpVehicleR.getData();
|
|
AddCruiseMissionDto addCruiseMissionDto = new AddCruiseMissionDto();
|
addCruiseMissionDto.setOrder(orderData);
|
addCruiseMissionDto.setVehicle(xpVehicle);
|
xpCruiseOrderClient.addCruiseMission(addCruiseMissionDto);
|
}
|
}
|
break;
|
case ORDER_COMPLETED:
|
//已完成
|
log.info("已完成，不做任何处理：{}", orderId);
|
break;
|
case ORDER_CANCELLED:
|
//已取消
|
buildPushOrderStatus(orderId, OrderStatusEnum.CANCELED, sequence);
|
xpOrderClient.updateOrderStatus(Long.parseLong(orderId.contains("XU") ? orderId.substring(2) : 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.COMPLETED.equals(status)){
|
// 返程找到目前的订单 -》之前的订单 查询生成一个 正常巡游 找到下一个点位 若到达最后点位 则反向找点位 每次判断是否收车状态 是则不进行处理了
|
// 找到这个订单
|
R<XpCruiseOrder> cruiseOrder = xpCruiseOrderClient.getCruiseOrder(Long.parseLong(orderId.substring(2)));
|
if(cruiseOrder!=null){
|
XpCruiseOrder xpCruiseOrder = cruiseOrder.getData();
|
Long vehicleId = xpCruiseOrder.getVehicleId();
|
// 完成当前巡游单子 已完成
|
xpCruiseOrderClient.completeCruiseOrder(Long.parseLong(orderId.substring(2)));
|
R<XpVehicle> xpVehicleR = xpVehicleClient.queryById(vehicleId);
|
XpVehicle xpVehicle = xpVehicleR.getData();
|
// 运营中 继续执行
|
if(xpVehicle.getOperatingStatus()==1 && xpVehicle.getState()==0){
|
// 创建新的巡游单子 判断是否为返程单子
|
if(xpCruiseOrder.getOrderType()==2){
|
R<XpCruiseOrder> lastOrder = xpCruiseOrderClient.getCruiseOrder(xpCruiseOrder.getLastOrderId());
|
XpCruiseOrder lastXpCruiseOrder = lastOrder.getData();
|
// 修改一下id 加一个一样的巡游单子
|
lastXpCruiseOrder.setId(null);
|
lastXpCruiseOrder.setStatus(1);
|
xpCruiseOrderClient.addCruiseOrder(lastXpCruiseOrder);
|
}else if(xpCruiseOrder.getOrderType()==1){
|
// 正常巡游 找到下一个点位 若到达最后点位 则反向找点位
|
// 起点
|
Double endLon = xpCruiseOrder.getEndLon();
|
Double endLat = xpCruiseOrder.getEndLat();
|
XpCruiseOrder newXpCruiseOrder = new XpCruiseOrder();
|
newXpCruiseOrder.setVehicleId(xpCruiseOrder.getVehicleId());
|
newXpCruiseOrder.setVehicleVin(xpCruiseOrder.getVehicleVin());
|
newXpCruiseOrder.setFleetId(xpCruiseOrder.getFleetId());
|
newXpCruiseOrder.setOrderType(1);
|
newXpCruiseOrder.setStartLat(endLat);
|
newXpCruiseOrder.setStartLon(endLon);
|
newXpCruiseOrder.setStartPointId(xpCruiseOrder.getEndPointId());
|
newXpCruiseOrder.setStartPointName(xpCruiseOrder.getEndPointName());
|
newXpCruiseOrder.setStatus(1);
|
newXpCruiseOrder.setLastOrderId(xpCruiseOrder.getId());
|
|
// 通过车找终点id
|
CruisePointDto cruisePointDto = new CruisePointDto();
|
cruisePointDto.setNewXpCruiseOrder(newXpCruiseOrder);
|
cruisePointDto.setVehicleId(xpCruiseOrder.getVehicleId());
|
cruisePointDto.setPointId(newXpCruiseOrder.getStartPointId());
|
cruisePointDto.setOrderReverse(xpCruiseOrder.getOrderReverse());
|
R<XpCruiseOrder> cruisePoint = xpCruisePlanClient.getCruisePoint(cruisePointDto);
|
if(cruisePoint!=null){
|
xpCruiseOrderClient.addCruiseOrder(cruisePoint.getData());
|
}
|
}
|
}
|
}
|
}
|
}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<>();
|
content.put("orderId", orderId.toString());
|
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<String> 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.Builder navigationDataBuilder = RestoreNavigationData.newBuilder();
|
navigationDataBuilder.setRouteId(routeId);
|
navigationDataBuilder.setRequestId(requestId);
|
navigationDataBuilder.setContent(routeDataContent);
|
navigationDataBuilder.setTimestamp(System.currentTimeMillis());
|
long sequence = snowflake.nextId();
|
log.info("开始推送导航数据 {}", orderId);
|
MqttPublishDto mqttPublishDto = xpuUtil.cloudPushNavigation(navigationDataBuilder.build(), xpuId, sequence);
|
log.info("推送导航的响应消息，{}", JSON.toJSONString(mqttPublishDto));
|
}
|
|
}
|
