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; @@ -15,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; @@ -31,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; /** @@ -65,16 +71,16 @@ @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.setType(MessageType.STATUS.getType() + "-flighttask_progress"); eventsReceiver.setMethod(EventsMethodEnum.FLIGHT_TASK_PROGRESS.getMethod()); WaylineTaskProgressReceiver output = eventsReceiver.getOutput(); @@ -82,7 +88,8 @@ 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()); @@ -122,7 +129,7 @@ .bizCode(BizCodeEnum.FLIGHT_TASK_PROGRESS.getCode()) .build()); 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()) @@ -133,17 +140,16 @@ .build()); } // 进度 if(eventsReceiver.getOutput().getProgress().getPercent()>=95){ return; } String json=""; // if (eventsReceiver.getOutput().getProgress().getPercent() >= 95) { // return; // } String json = ""; try { json = mapper.writeValueAsString(eventsReceiver); } catch (Exception e){ } catch (Exception e) { e.printStackTrace(); } _dockWebSocketHandler.sendMessageToSubscribers(dockSn, MessageType.STATUS.getType(),json); _dockWebSocketHandler.sendMessageToSubscribers(dockSn, MessageType.STATUS.getType(), json, ""); } /** @@ -162,18 +168,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()) @@ -183,11 +219,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())
