forked from uav/ard-uav-server
ard-sample/src/main/java/com/ard/sample/wayline/service/impl/FlightTaskServiceImpl.java
@@ -1,10 +1,14 @@ package com.ard.sample.wayline.service.impl; import com.ard.common.utils.uuid.IdUtils; import com.ard.sample.common.error.CommonErrorEnum; import com.ard.sample.common.model.ResponseResult; import com.ard.sample.component.mqtt.model.*; import com.ard.sample.component.mqtt.service.IMessageSenderService; import com.ard.common.core.redis.RedisConst; import com.ard.common.core.redis.RedisOpsUtils; import com.ard.sample.component.websocket.enums.MessageType; import com.ard.sample.component.websocket.handler.DockWebSocketHandler; import com.ard.sample.component.websocket.model.BizCodeEnum; import com.ard.sample.component.websocket.model.CustomWebSocketMessage; import com.ard.sample.component.websocket.service.ISendMessageService; @@ -13,6 +17,7 @@ import com.ard.sample.manage.model.enums.UserTypeEnum; import com.ard.sample.media.model.MediaFileCountDTO; import com.ard.sample.wayline.model.dto.WaylineJobDTO; import com.ard.sample.wayline.model.dto.WaylineTaskProgressExt; import com.ard.sample.wayline.model.dto.WaylineTaskProgressReceiver; import com.ard.sample.wayline.model.enums.WaylineJobStatusEnum; import com.ard.sample.wayline.service.IFlightTaskService; @@ -29,7 +34,10 @@ import org.springframework.stereotype.Service; import java.time.LocalDateTime; import java.util.HashMap; import java.util.Map; import java.util.Objects; import java.util.Optional; import java.util.concurrent.TimeUnit; /** @@ -48,6 +56,9 @@ private ObjectMapper mapper; @Autowired private DockWebSocketHandler _dockWebSocketHandler; @Autowired private ISendMessageService websocketMessageService; @Autowired @@ -60,19 +71,25 @@ @ServiceActivator(inputChannel = ChannelName.INBOUND_EVENTS_FLIGHT_TASK_PROGRESS, outputChannel = ChannelName.OUTBOUND) public void handleProgress(CommonTopicReceiver receiver, MessageHeaders headers) { String receivedTopic = String.valueOf(headers.get(MqttHeaders.RECEIVED_TOPIC)); String dockSn = receivedTopic.substring((TopicConst.THING_MODEL_PRE + TopicConst.PRODUCT).length(), String dockSn = receivedTopic.substring((TopicConst.THING_MODEL_PRE + TopicConst.PRODUCT).length(), receivedTopic.indexOf(TopicConst.EVENTS_SUF)); EventsReceiver<WaylineTaskProgressReceiver> eventsReceiver = mapper.convertValue(receiver.getData(), new TypeReference<EventsReceiver<WaylineTaskProgressReceiver>>(){}); new TypeReference<EventsReceiver<WaylineTaskProgressReceiver>>() { }); eventsReceiver.setBid(receiver.getBid()); eventsReceiver.setSn(receiver.getGateway()); eventsReceiver.setGateway(receiver.getGateway()); eventsReceiver.setType(MessageType.STATUS.getType() + "-flighttask_progress"); eventsReceiver.setMethod(EventsMethodEnum.FLIGHT_TASK_PROGRESS.getMethod()); WaylineTaskProgressReceiver output = eventsReceiver.getOutput(); log.info("Task progress: {}", output.getProgress().toString()); if (eventsReceiver.getResult() != ResponseResult.CODE_SUCCESS) { log.error("Task progress ===> Error code: " + eventsReceiver.getResult()); String errorMsg = CommonErrorEnum.getMsgByCode(eventsReceiver.getResult()); log.error("Task progress ===> Error code: {},Error Msg: {}", eventsReceiver.getResult(), errorMsg); } EventsResultStatusEnum statusEnum = EventsResultStatusEnum.find(output.getStatus()); @@ -112,7 +129,7 @@ .bizCode(BizCodeEnum.FLIGHT_TASK_PROGRESS.getCode()) .build()); if (receiver.getNeedReply() == 1) { if (receiver.getNeedReply() != null && receiver.getNeedReply() == 1) { messageSender.publish(receivedTopic + TopicConst._REPLY_SUF, CommonTopicResponse.builder() .tid(receiver.getTid()) @@ -122,8 +139,29 @@ .data(RequestsReply.success()) .build()); } // 进度 if (eventsReceiver.getOutput().getProgress().getPercent() >= 95) { return; } String json = ""; try { json = mapper.writeValueAsString(eventsReceiver); } catch (Exception e) { e.printStackTrace(); } _dockWebSocketHandler.sendMessageToSubscribers(dockSn, MessageType.STATUS.getType(), json); } /** * 定时检查并处理航线定时任务 * <p> * 该方法每5秒执行一次，从Redis有序集合中获取最早需要执行的任务，检查其执行时间： * 1. 若任务已过期（超过当前时间30秒），则标记为失败并删除 * 2. 若任务在允许的时间窗口内（当前时间±30秒），则执行任务并删除 * </p> * * <p>任务ID格式：{workspace_id}:{dock_sn}:{job_id}</p> */ @Scheduled(initialDelay = 10, fixedRate = 5, timeUnit = TimeUnit.SECONDS) private void checkScheduledJob() { Object jobIdValue = RedisOpsUtils.zGetMin(RedisConst.WAYLINE_JOB_TIMED_EXECUTE); @@ -134,6 +172,31 @@ // format: {workspace_id}:{dock_sn}:{job_id} String[] jobArr = String.valueOf(jobIdValue).split(RedisConst.DELIMITER); double time = RedisOpsUtils.zScore(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, jobIdValue); // 取到待执行的定时任务ID，获取任务信息websocket发给前端 // region 取定时任务信息websocket发给前端 // pending status状态待执行 EventsReceiver<WaylineTaskProgressReceiver> eventsReceiver = new EventsReceiver<>(); WaylineTaskProgressReceiver waylineTaskProgressReceiver = new WaylineTaskProgressReceiver(); WaylineTaskProgressExt waylineTaskProgressExt = new WaylineTaskProgressExt(); waylineTaskProgressExt.setFlightId(jobArr[1]); waylineTaskProgressReceiver.setStatus("pending"); waylineTaskProgressReceiver.setExt(waylineTaskProgressExt); eventsReceiver.setOutput(waylineTaskProgressReceiver); eventsReceiver.setResult(ResponseResult.CODE_SUCCESS); eventsReceiver.setBid(IdUtils.simpleUUID()); eventsReceiver.setSn(jobArr[0]); eventsReceiver.setGateway(jobArr[0]); eventsReceiver.setType(MessageType.STATUS.getType() + "-flighttask_progress"); eventsReceiver.setMethod(EventsMethodEnum.FLIGHT_TASK_PROGRESS.getMethod()); String json = ""; try { json = mapper.writeValueAsString(eventsReceiver); } catch (Exception e) { e.printStackTrace(); } _dockWebSocketHandler.sendMessageToSubscribers(jobArr[0], MessageType.STATUS.getType(), json); log.info("定时任务自定义状态为待执行，json: {}", json); // endregion long now = System.currentTimeMillis(); int offset = 30_000; @@ -141,7 +204,7 @@ if (time < now - offset) { RedisOpsUtils.zRemove(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, jobIdValue); waylineJobService.updateJob(WaylineJobDTO.builder() .jobId(jobArr[2]) .jobId(jobArr[1]) .status(WaylineJobStatusEnum.FAILED.getVal()) .executeTime(LocalDateTime.now()) .completedTime(LocalDateTime.now())
