forked from uav/ard-uav-server
ard-sample/src/main/java/com/ard/sample/wayline/service/impl/FlightTaskServiceImpl.java
@@ -1,5 +1,7 @@ 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; @@ -13,11 +15,15 @@ import com.ard.sample.component.websocket.service.IWebSocketManageService; import com.ard.sample.manage.model.dto.DeviceDTO; import com.ard.sample.manage.model.enums.UserTypeEnum; import com.ard.sample.manage.service.IWaylineTrajectoryService; 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.entity.WaylineJobHistoryEntity; import com.ard.sample.wayline.model.enums.WaylineJobStatusEnum; import com.ard.sample.wayline.service.IFlightTaskService; import com.ard.sample.wayline.service.IWaylineJobHistoryService; import com.ard.sample.wayline.service.IWaylineJobService; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; @@ -31,7 +37,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; /** @@ -61,25 +70,35 @@ @Autowired private IWaylineJobService waylineJobService; @Autowired private IWaylineTrajectoryService _waylineTrajectoryService; @Autowired private IWaylineJobHistoryService _waylineJobHistoryService; @Override @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.setType(MessageType.STATUS.getType()+"-flighttask_progress"); 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()); @@ -119,16 +138,7 @@ .bizCode(BizCodeEnum.FLIGHT_TASK_PROGRESS.getCode()) .build()); String json=""; try { json = mapper.writeValueAsString(eventsReceiver); } catch (Exception e){ e.printStackTrace(); } _dockWebSocketHandler.sendMessageToSubscribers(dockSn, MessageType.STATUS.getType() + "-flighttask_progress",json); if (receiver.getNeedReply()!=null &&receiver.getNeedReply() == 1) { if (receiver.getNeedReply() != null && receiver.getNeedReply() == 1) { messageSender.publish(receivedTopic + TopicConst._REPLY_SUF, CommonTopicResponse.builder() .tid(receiver.getTid()) @@ -138,7 +148,41 @@ .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, ""); _waylineTrajectoryService.setList(dockSn,"1",json); // 计划任务成功后，保存计划任务到历史记录 String status = eventsReceiver.getOutput().getStatus(); String jobId = eventsReceiver.getOutput().getExt().getFlightId(); if(status.equals("ok") && jobId != null) { Optional<WaylineJobDTO> job = waylineJobService.getJobByJobId(jobId); if(job.isPresent()) { if (_waylineJobHistoryService.getJobByJobId(jobId).isPresent()) { return; } WaylineJobHistoryEntity waylineJobHistoryEntity = _waylineJobHistoryService.dto2Entity(job.get()); Optional<WaylineJobDTO> jobHistory = _waylineJobHistoryService.createWaylineJob(waylineJobHistoryEntity); if(jobHistory.isPresent()) { log.info("任务执行成功，保存到历史记录。 {}", jobHistory); } } } } /** @@ -157,18 +201,48 @@ if (Objects.isNull(jobIdValue)) { return; } log.info("Check the timed tasks of the wayline. {}", jobIdValue); log.info("检查 wayline 的定时任务。 {}", jobIdValue); // 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); // region 取定时任务信息websocket发给前端 // 定时任务执行前，判断有没有任务正在执行，在执行则不发送websocket消息 String key = RedisConst.WAYLINE_JOB_RUNNING_PREFIX + jobArr[0]; if (!RedisOpsUtils.checkExist(key)) { // 取到待执行的定时任务ID，获取任务信息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; // Expired tasks are deleted directly. // 过期任务会直接删除。 if (time < now - offset) { log.info("定时任务超时，jobId: {}", jobArr[1]); 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()) @@ -178,11 +252,11 @@ if (now <= time && time <= now + offset) { try { waylineJobService.executeFlightTask(jobArr[2]); waylineJobService.executeFlightTask(jobArr[1]); } catch (Exception e) { log.info("The scheduled task delivery failed."); log.info("计划中的任务交付失败了。jobId: {}，异常信息：{}", jobArr[1], e.getMessage()); waylineJobService.updateJob(WaylineJobDTO.builder() .jobId(jobArr[2]) .jobId(jobArr[1]) .status(WaylineJobStatusEnum.FAILED.getVal()) .executeTime(LocalDateTime.now()) .completedTime(LocalDateTime.now())
