forked from uav/ard-uav-server
ard-sample/src/main/java/com/ard/sample/wayline/service/impl/FlightTaskServiceImpl.java
@@ -1,6 +1,9 @@ package com.ard.sample.wayline.service.impl; import com.ard.common.utils.uuid.IdUtils; import com.ard.route.domain.route.ArdRouteInfo; import com.ard.route.service.IArdRouteInfoService; import com.ard.route.service.IArdRouteMissionService; import com.ard.sample.common.error.CommonErrorEnum; import com.ard.sample.common.model.ResponseResult; import com.ard.sample.component.mqtt.model.*; @@ -15,12 +18,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; @@ -66,6 +72,18 @@ @Autowired private IWaylineJobService waylineJobService; @Autowired private IWaylineTrajectoryService _waylineTrajectoryService; @Autowired private IWaylineJobHistoryService _waylineJobHistoryService; @Autowired private IArdRouteMissionService _ardRouteMissionService; @Autowired private IArdRouteInfoService _ardRouteInfoService; @Override @ServiceActivator(inputChannel = ChannelName.INBOUND_EVENTS_FLIGHT_TASK_PROGRESS, outputChannel = ChannelName.OUTBOUND) @@ -143,13 +161,42 @@ // 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); _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); } // 同步更新4.2的计划任务的状态；等待夏吉建的接口。1:已执行 0:未执行 // 在任务中查找航线文件id，取航线文件信息中的状态，是3的修改状态 ArdRouteInfo ardRouteInfo = _ardRouteInfoService.selectArdRouteInfoById(job.get().getFileId()); if(ardRouteInfo != null && ardRouteInfo.getType().equals("3")) { _ardRouteMissionService.updateArdRouteMissionStatus(jobId, "1"); } } } } /** @@ -168,40 +215,45 @@ 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); // 取到待执行的定时任务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(); // 定时任务执行前，判断有没有任务正在执行，在执行则不发送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); } _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[1]) @@ -214,11 +266,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())
