forked from uav/ard-uav-server
ard-sample/src/main/java/com/ard/sample/wayline/service/impl/FlightTaskServiceImpl.java
@@ -1,5 +1,6 @@ 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.*; @@ -16,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; @@ -32,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; /** @@ -66,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(); @@ -84,7 +89,7 @@ if (eventsReceiver.getResult() != ResponseResult.CODE_SUCCESS) { String errorMsg = CommonErrorEnum.getMsgByCode(eventsReceiver.getResult()); log.error("Task progress ===> Error code: {},Error Msg: {}", eventsReceiver.getResult(),errorMsg); log.error("Task progress ===> Error code: {},Error Msg: {}", eventsReceiver.getResult(), errorMsg); } EventsResultStatusEnum statusEnum = EventsResultStatusEnum.find(output.getStatus()); @@ -124,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()) @@ -135,17 +140,16 @@ .build()); } // 进度 if(eventsReceiver.getOutput().getProgress().getPercent()>=95){ if (eventsReceiver.getOutput().getProgress().getPercent() >= 95) { return; } String json=""; 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); } /** @@ -168,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; @@ -175,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())
