xp-robotaxi-api/src/main/java/com/xiaopeng/api/dto/automaticPilot/PushOrderDto.java
New file @@ -0,0 +1,16 @@ package com.xiaopeng.api.dto.automaticPilot; import io.swagger.annotations.ApiModel; import io.swagger.annotations.ApiModelProperty; import lombok.Data; @Data @ApiModel("推送订单Dto") public class PushOrderDto { @ApiModelProperty("车辆vin") private String vin; // 订单id @ApiModelProperty("订单id") private Long orderId; } xp-robotaxi-api/src/main/java/com/xiaopeng/api/dto/order/PopOrderDto.java
@@ -8,13 +8,11 @@ @Data @ApiModel(value = "订单成功下发命令信息") public class PopOrderDto { @ApiModelProperty(value = "队列名称") private String queueName; @ApiModelProperty(value = "订单Id") private Long orderId; @ApiModelProperty(value = "车辆Id") private Long xpVehicleId; @ApiModelProperty(value = "车辆Vin") private String xpVehicleVin; } xp-robotaxi-api/src/main/java/com/xiaopeng/api/feignClient/automatic/OrderMissionClient.java
@@ -1,6 +1,7 @@ package com.xiaopeng.api.feignClient.automatic; import com.xiaopeng.api.dto.account.XpOperationsUserQueryDTO; import com.xiaopeng.api.dto.automaticPilot.PushOrderDto; import com.xiaopeng.api.factory.account.XpAppUserClientFallbackFactory; import com.xiaopeng.api.factory.automatic.OrderMissionClientFallbackFactory; import com.xiaopeng.api.model.account.XpOperationsUser; @@ -27,9 +28,9 @@ /** * 下发命令 * @param orderId * @param * @return */ @PostMapping("/inner/pushOrder") R<Boolean> pushOrder(@RequestParam Long orderId); R<Boolean> pushOrder(@RequestBody PushOrderDto dto); } xp-robotaxi-automaticPilot-boot/src/main/java/com/xiaopeng/automaticPilot/controller/OrderMissionController.java
@@ -1,5 +1,8 @@ package com.xiaopeng.automaticPilot.controller; import cn.hutool.core.util.IdUtil; import com.xiaopeng.api.dto.automaticPilot.MqttPublishDto; import com.xiaopeng.api.dto.automaticPilot.PushOrderDto; 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; @@ -8,12 +11,12 @@ import com.xiaopeng.automaticPilot.util.XPUUtil; import com.xiaopeng.common.core.domain.R; import com.xiaopeng.common.core.utils.SnowFlake; import com.xiaopeng.common.redis.service.RedisService; import io.swagger.annotations.Api; import lombok.extern.slf4j.Slf4j; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; import org.springframework.web.bind.annotation.*; import java.util.concurrent.TimeUnit; @Slf4j @RestController @@ -23,28 +26,28 @@ private final XPUUtil xpuUtil; private final RedisService redisService; public OrderMissionController(XPUUtil xpuUtil) { public OrderMissionController(XPUUtil xpuUtil, RedisService redisService) { this.xpuUtil = xpuUtil; this.redisService = redisService; } @PostMapping("/inner/pushOrder") public R<Boolean> pushOrder(@RequestParam Long orderId) { public R<Boolean> pushOrder(@RequestBody PushOrderDto dto) { RTTask task = new RTTask(); task.setOrder_id(orderId.toString()); // TODO id处理 已经数据是否存储 task.setTask_id(System.currentTimeMillis()+ ""); task.setOrder_id(dto.getOrderId().toString()); task.setTask_id("MISSION_"+ IdUtil.fastUUID()); task.setTask_type(TaskType.ROBOTAXI); task.setStatus(TaskStatus.PENDING); SnowFlake snowFlake = new SnowFlake(0, 0); snowFlake.withStartTimestamp(1420041600000L); long sequence = snowFlake.nextId(); ResponseMessage responseMessage = xpuUtil.cloudCreateOrder(task,sequence); if (responseMessage.getCode()== StatusCode.OK) { MqttPublishDto mqttPublishDto = xpuUtil.cloudCreateOrder(task, dto.getVin(),sequence); if (mqttPublishDto.getCode()== 200) { return R.ok(true); } return R.ok(false); xp-robotaxi-automaticPilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java
@@ -6,6 +6,7 @@ import com.alibaba.fastjson2.JSONObject; import com.xiaopeng.api.dto.map.OrderRouteAndFareDto; import com.xiaopeng.api.dto.operation.LonLatDto; 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; @@ -35,6 +36,7 @@ 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; @@ -49,6 +51,7 @@ import javax.annotation.Resource; import java.util.*; import java.util.concurrent.TimeUnit; /** * Kafka 消费者服务 @@ -77,6 +80,7 @@ @Resource private XpPushMessageClient xpPushMessageClient; /** @@ -159,6 +163,17 @@ 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); } } } } @@ -527,8 +542,13 @@ waypoints.add(waypoint); } orderOperation.setWaypoints(waypoints); SnowFlake snowFlake = new SnowFlake(0, 0); snowFlake.withStartTimestamp(1420041600000L); long sequence = snowFlake.nextId(); // 处理数据 redisTemplate.opsForValue().set(sequence,"pushOrder_"+orderId+"_"+vin,30, TimeUnit.MINUTES); //开发发送请求到XPU xpuUtil.cloudOrderOperation(orderOperation, vin); xpuUtil.cloudOrderOperation(orderOperation, vin,sequence); }
@@ -36,14 +36,14 @@ * @param task * @return */ public MqttPublishDto cloudCreateOrder(RTTask task, String vin) { public MqttPublishDto cloudCreateOrder(RTTask task, String vin, Long sequence) { MsgContent msgContent = new MsgContent(); // 消息头 RtMsgHeader header = new RtMsgHeader(); header.setMsg_id(UUID.randomUUID().toString()); header.setTimestamp(System.currentTimeMillis()); header.setCmd(RTCmdType.CLOUD_CREATE_ORDER); header.setSequence(1L); header.setSequence(sequence); header.setType(MessageType.MESSAGE_TYPE_RESPONSE); msgContent.setHeader(header); // 消息内容 @@ -67,15 +67,14 @@ * @param operation * @return */ public MqttPublishDto cloudOrderOperation(OrderOperation operation, String vin) { public MqttPublishDto cloudOrderOperation(OrderOperation operation, String vin,Long sequence) { MsgContent msgContent = new MsgContent(); // 消息头 RtMsgHeader header = new RtMsgHeader(); header.setMsg_id(UUID.randomUUID().toString()); header.setTimestamp(System.currentTimeMillis()); header.setCmd(RTCmdType.CLOUD_ORDER_OPERATION); Snowflake snowflake = IdUtil.getSnowflake(0, 0); header.setSequence(snowflake.nextId()); header.setSequence(sequence); header.setType(MessageType.MESSAGE_TYPE_REQUEST); msgContent.setHeader(header); // 消息内容 @@ -87,6 +86,10 @@ MqttPublishVo vo = new MqttPublishVo(); vo.setTopic("topic-prod-xpu-robotaxi-high/" + vin); vo.setPayload(JSON.toJSONString(msgContent)); return xpuClient.publish(vo); }
