From 39ffdf5754b4ebca682bd3c925f7a500e8a1bbda Mon Sep 17 00:00:00 2001
From: aix <vip_xiaobin810@163.com>
Date: Thu, 08 Aug 2024 16:56:36 +0800
Subject: [PATCH] 国土调查云身份凭证token
---
src/main/java/com/dji/sample/wayline/service/impl/FlightTaskServiceImpl.java | 326 ++++++++++++++++++++++++++++++++++++++++-------------
1 files changed, 245 insertions(+), 81 deletions(-)
diff --git a/src/main/java/com/dji/sample/wayline/service/impl/FlightTaskServiceImpl.java b/src/main/java/com/dji/sample/wayline/service/impl/FlightTaskServiceImpl.java
index 83b3247..ab4c9fd 100644
--- a/src/main/java/com/dji/sample/wayline/service/impl/FlightTaskServiceImpl.java
+++ b/src/main/java/com/dji/sample/wayline/service/impl/FlightTaskServiceImpl.java
@@ -1,34 +1,42 @@
package com.dji.sample.wayline.service.impl;
+import com.dji.sample.common.error.CommonErrorEnum;
import com.dji.sample.common.model.ResponseResult;
import com.dji.sample.component.mqtt.model.*;
import com.dji.sample.component.mqtt.service.IMessageSenderService;
import com.dji.sample.component.redis.RedisConst;
import com.dji.sample.component.redis.RedisOpsUtils;
import com.dji.sample.component.websocket.model.BizCodeEnum;
-import com.dji.sample.component.websocket.model.CustomWebSocketMessage;
import com.dji.sample.component.websocket.service.ISendMessageService;
-import com.dji.sample.component.websocket.service.IWebSocketManageService;
import com.dji.sample.manage.model.dto.DeviceDTO;
import com.dji.sample.manage.model.enums.UserTypeEnum;
-import com.dji.sample.wayline.model.dto.FlightTaskProgressReceiver;
+import com.dji.sample.manage.service.IDeviceRedisService;
+import com.dji.sample.media.model.MediaFileCountDTO;
import com.dji.sample.wayline.model.dto.WaylineJobDTO;
+import com.dji.sample.wayline.model.dto.WaylineJobKey;
+import com.dji.sample.wayline.model.dto.WaylineTaskProgressExtBreakPoint;
+import com.dji.sample.wayline.model.dto.WaylineTaskProgressReceiver;
+import com.dji.sample.wayline.model.entity.WaylineJobBreakPointEntity;
import com.dji.sample.wayline.model.enums.WaylineJobStatusEnum;
+import com.dji.sample.wayline.model.enums.WaylineTaskTypeEnum;
import com.dji.sample.wayline.service.IFlightTaskService;
+import com.dji.sample.wayline.service.IWaylineJobBreakPointService;
import com.dji.sample.wayline.service.IWaylineJobService;
+import com.dji.sample.wayline.service.IWaylineRedisService;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.annotation.ServiceActivator;
-import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.MessageHeaders;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
import java.time.LocalDateTime;
-import java.util.Objects;
+import java.time.ZoneId;
+import java.util.*;
import java.util.concurrent.TimeUnit;
/**
@@ -50,108 +58,264 @@
private ISendMessageService websocketMessageService;
@Autowired
- private IWebSocketManageService webSocketManageService;
-
- @Autowired
- private RedisOpsUtils redisOps;
-
- @Autowired
private IWaylineJobService waylineJobService;
- @Override
- @ServiceActivator(inputChannel = ChannelName.INBOUND_EVENTS_FLIGHT_TASK_PROGRESS, outputChannel = ChannelName.OUTBOUND)
- public void handleProgress(CommonTopicReceiver receiver, MessageHeaders headers) {
- EventsReceiver<FlightTaskProgressReceiver> eventsReceiver = mapper.convertValue(receiver.getData(),
- new TypeReference<EventsReceiver<FlightTaskProgressReceiver>>(){});
+ @Autowired
+ private IDeviceRedisService deviceRedisService;
+
+ @Autowired
+ private IWaylineRedisService waylineRedisService;
+
+ @Autowired
+ private IWaylineJobBreakPointService waylineJobBreakPointService;
+
+ /**
+ * Handle the progress messages of the flight tasks reported by the dock.
+ * 处理机场上报的飞行任务进度信息
+ * @param receiver
+ * @param headers
+ */
+ @ServiceActivator(inputChannel = ChannelName.INBOUND_EVENTS_FLIGHT_TASK_PROGRESS, outputChannel = ChannelName.OUTBOUND_EVENTS)
+ public CommonTopicReceiver handleProgress(CommonTopicReceiver receiver, MessageHeaders headers) {
+ log.info("上报航线任务进度: {}", receiver.toString());
+ EventsReceiver<WaylineTaskProgressReceiver> eventsReceiver = mapper.convertValue(receiver.getData(),
+ new TypeReference<EventsReceiver<WaylineTaskProgressReceiver>>(){});
eventsReceiver.setBid(receiver.getBid());
eventsReceiver.setSn(receiver.getGateway());
- FlightTaskProgressReceiver output = eventsReceiver.getOutput();
+ //用户记录飞行任务id
+ RedisOpsUtils.set(RedisConst.FLIGHT_LOG + "job_id",receiver.getBid());
- log.info("Task progress: {}", output.getProgress().toString());
+ WaylineTaskProgressReceiver output = eventsReceiver.getOutput();
+
+ log.info("任务进度: {}", output.getProgress().toString());
+
+ if (null != output.getExt().getBreakPoint()) {
+ WaylineTaskProgressExtBreakPoint breakPoint = output.getExt().getBreakPoint();
+ log.info("任务进度 ===> 断点信息：{}", breakPoint.toString());
+ // 保存断点信息
+ try {
+ Boolean isAddBp = waylineJobBreakPointService.addWaylineJobBreakPoint(WaylineJobBreakPointEntity.builder()
+ .jobId(receiver.getBid())
+ .bpIndex(breakPoint.getIndex())
+ .state(breakPoint.getState())
+ .progress(breakPoint.getProgress())
+ .waylineId(breakPoint.getWaylineId())
+ .breakReason(breakPoint.getBreakReason())
+ .latitude(breakPoint.getLatitude())
+ .longitude(breakPoint.getLongitude())
+ .height(breakPoint.getHeight())
+ .attitudeHead(breakPoint.getAttitudeHead())
+ .build());
+ if (isAddBp) {
+ log.info("任务进度 ===> 断点信息 ===> 保存成功：{}", breakPoint.toString());
+ } else {
+ log.info("任务进度 ===> 断点信息 ===> 保存失败：{}", breakPoint.toString());
+ }
+ }catch (Exception e) {
+ log.info("任务进度 ===> 断点信息 ===> 保存失败：{},\n {}", breakPoint.toString(),e.getMessage());
+ }
+
+ }
if (eventsReceiver.getResult() != ResponseResult.CODE_SUCCESS) {
- log.error("Task progress ===> Error code: " + eventsReceiver.getResult());
+ log.error("任务进度 ===> 错误编码: " + eventsReceiver.getResult());
}
EventsResultStatusEnum statusEnum = EventsResultStatusEnum.find(output.getStatus());
- if (statusEnum.getEnd()) {
- WaylineJobDTO job = WaylineJobDTO.builder()
- .jobId(receiver.getBid())
- .status(WaylineJobStatusEnum.SUCCESS.getVal())
- .endTime(LocalDateTime.now())
- .mediaCount(output.getExt().getMediaCount())
- .build();
+ waylineRedisService.setRunningWaylineJob(receiver.getGateway(), eventsReceiver);
- if (EventsResultStatusEnum.OK != statusEnum) {
- job.setCode(eventsReceiver.getResult());
- job.setStatus(WaylineJobStatusEnum.FAILED.getVal());
- }
-
- waylineJobService.updateJob(job);
- redisOps.del(receiver.getBid());
+ Optional<DeviceDTO> deviceOpt = deviceRedisService.getDeviceOnline(receiver.getGateway());
+ if (deviceOpt.isEmpty()) {
+ return null;
}
- redisOps.setWithExpire(receiver.getBid(), eventsReceiver, RedisConst.DEVICE_ALIVE_SECOND * RedisConst.DEVICE_ALIVE_SECOND);
- DeviceDTO device = (DeviceDTO) redisOps.get(RedisConst.DEVICE_ONLINE_PREFIX + receiver.getGateway());
- websocketMessageService.sendBatch(
- webSocketManageService.getValueWithWorkspaceAndUserType(
- device.getWorkspaceId(), UserTypeEnum.WEB.getVal()),
- CustomWebSocketMessage.builder()
- .data(eventsReceiver)
- .timestamp(System.currentTimeMillis())
- .bizCode(BizCodeEnum.FLIGHT_TASK_PROGRESS.getCode())
- .build());
+ if (statusEnum.getEnd()) {
+ handleEndStatus(receiver, statusEnum, output.getExt().getMediaCount(), eventsReceiver.getResult(), deviceOpt.get());
+ }
- if (receiver.getNeedReply() == 1) {
- String topic = headers.get(MqttHeaders.RECEIVED_TOPIC) + TopicConst._REPLY_SUF;
- messageSender.publish(topic,
- CommonTopicResponse.builder()
- .tid(receiver.getTid())
- .bid(receiver.getBid())
- .method(EventsMethodEnum.FLIGHT_TASK_PROGRESS.getMethod())
- .timestamp(System.currentTimeMillis())
- .data(RequestsReply.success())
- .build());
+ websocketMessageService.sendBatch(deviceOpt.get().getWorkspaceId(), UserTypeEnum.WEB.getVal(),
+ BizCodeEnum.FLIGHT_TASK_PROGRESS.getCode(), eventsReceiver);
+
+ return receiver;
+ }
+
+ private void handleEndStatus(CommonTopicReceiver receiver, EventsResultStatusEnum statusEnum, int mediaCount, int code, DeviceDTO dock) {
+
+ WaylineJobDTO job = WaylineJobDTO.builder()
+ .jobId(receiver.getBid())
+ .status(WaylineJobStatusEnum.SUCCESS.getVal())
+ .completedTime(LocalDateTime.now())
+ .mediaCount(mediaCount)
+ .build();
+
+ //记录媒体计数的更新
+ if (Objects.nonNull(job.getMediaCount()) && job.getMediaCount() != 0) {
+ RedisOpsUtils.hashSet(RedisConst.MEDIA_FILE_PREFIX + receiver.getGateway(), job.getJobId(),
+ MediaFileCountDTO.builder().jobId(receiver.getBid()).mediaCount(job.getMediaCount()).uploadedCount(0).build());
+ }
+
+ if (EventsResultStatusEnum.OK != statusEnum) {
+ job.setCode(code);
+ job.setStatus(WaylineJobStatusEnum.FAILED.getVal());
+ }
+
+ waylineRedisService.getConditionalWaylineJob(receiver.getBid()).ifPresent(waylineJob ->
+ retryPrepareConditionJob(new WaylineJobKey(dock.getWorkspaceId(), dock.getDeviceSn(), receiver.getBid()), waylineJob));
+ waylineJobService.updateJob(job);
+ waylineRedisService.delRunningWaylineJob(receiver.getGateway());
+ waylineRedisService.delPausedWaylineJob(receiver.getBid());
+ waylineRedisService.delBlockedWaylineJobId(receiver.getGateway());
+
+ }
+
+ /**
+ * Notifications will be received through this interface when tasks are ready on the device.
+ * @param receiver
+ * @param headers
+ */
+ @ServiceActivator(inputChannel = ChannelName.INBOUND_EVENTS_FLIGHT_TASK_READY, outputChannel = ChannelName.OUTBOUND_EVENTS)
+ public CommonTopicReceiver handleTaskNotifications(CommonTopicReceiver receiver, MessageHeaders headers) {
+ String dockSn = receiver.getGateway();
+ Set<String> flightIds = mapper.convertValue(receiver.getData(),
+ new TypeReference<Map<String, Set<String>>>(){}).get(MapKeyConst.FLIGHT_IDS);
+
+ log.info("ready task list：{}", Arrays.toString(flightIds.toArray()));
+ // Check conditional task blocking status.
+ String blockedId = waylineRedisService.getBlockedWaylineJobId(dockSn);
+ if (StringUtils.hasText(blockedId)) {
+ log.info("The dock is in a state of wayline congestion, and the task will not be executed.");
+ return null;
+ }
+
+ Optional<DeviceDTO> deviceOpt = deviceRedisService.getDeviceOnline(dockSn);
+ if (deviceOpt.isEmpty()) {
+ log.info("The dock is offline.");
+ return null;
+ }
+ DeviceDTO device = deviceOpt.get();
+ Optional<WaylineJobDTO> jobOpt = waylineJobService.getJobsByConditions(device.getWorkspaceId(), flightIds, WaylineJobStatusEnum.PENDING)
+ .stream().filter(job -> flightIds.contains(job.getJobId()))
+ .sorted(Comparator.comparingInt(a -> a.getTaskType().getVal()))
+ .min(Comparator.comparing(WaylineJobDTO::getBeginTime));
+ if (jobOpt.isEmpty()) {
+ return receiver;
+ }
+ executeReadyTask(jobOpt.get());
+
+ return receiver;
+ }
+
+ private void executeReadyTask(WaylineJobDTO waylineJob) {
+ try {
+ boolean isExecute = waylineJobService.executeFlightTask(waylineJob.getWorkspaceId(), waylineJob.getJobId());
+ if (isExecute || WaylineTaskTypeEnum.CONDITION != waylineJob.getTaskType()) {
+ return;
+ }
+ Optional<WaylineJobDTO> waylineJobOpt = waylineRedisService.getConditionalWaylineJob(waylineJob.getJobId());
+ if (waylineJobOpt.isEmpty()) {
+ log.info("The conditional job has expired and will no longer be executed.");
+ return;
+ }
+ waylineJob = waylineJobOpt.get();
+ this.retryPrepareConditionJob(new WaylineJobKey(waylineJob.getWorkspaceId(), waylineJob.getDockSn(), waylineJob.getJobId()), waylineJob);
+ } catch (Exception e) {
+ log.error("Failed to execute task. ID: {}, Name:{}", waylineJob.getJobId(), waylineJob.getJobName());
+ this.retryPrepareConditionJob(new WaylineJobKey(waylineJob.getWorkspaceId(), waylineJob.getDockSn(), waylineJob.getJobId()), waylineJob);
+ e.printStackTrace();
}
}
@Scheduled(initialDelay = 10, fixedRate = 5, timeUnit = TimeUnit.SECONDS)
- private void checkScheduledJob() {
- Object jobIdValue = redisOps.zGetMin(RedisConst.WAYLINE_JOB);
- log.info("Check the timed jobs of the wayline. {}", jobIdValue);
- if (Objects.isNull(jobIdValue)) {
+ private void prepareWaylineJob() {
+ Optional<WaylineJobKey> jobKeyOpt = waylineRedisService.getNearestPreparedWaylineJob();
+ if (jobKeyOpt.isEmpty()) {
return;
}
- String jobId = String.valueOf(jobIdValue);
- double time = redisOps.zScore(RedisConst.WAYLINE_JOB, jobIdValue);
- long now = System.currentTimeMillis();
- int offset = 30_000;
+ // format: {workspace_id}:{dock_sn}:{job_id}
+ WaylineJobKey jobKey = jobKeyOpt.get();
+ log.info("Check the prepared tasks of the wayline. {}", jobKey.toString());
- // Expired tasks are deleted directly.
- if (time < now - offset) {
- redisOps.zRemove(RedisConst.WAYLINE_JOB, jobId);
- waylineJobService.updateJob(WaylineJobDTO.builder()
- .jobId(jobId)
- .status(WaylineJobStatusEnum.FAILED.getVal())
- .endTime(LocalDateTime.now())
- .code(HttpStatus.SC_REQUEST_TIMEOUT).build());
+ WaylineJobDTO job = WaylineJobDTO.builder()
+ .jobId(jobKey.getJobId())
+ .status(WaylineJobStatusEnum.FAILED.getVal())
+ .executeTime(LocalDateTime.now())
+ .completedTime(LocalDateTime.now())
+ .code(HttpStatus.SC_INTERNAL_SERVER_ERROR).build();
+ Optional<WaylineJobDTO> waylineJobOpt = getPreparedJob(jobKey, job);
+ if (waylineJobOpt.isEmpty()) {
return;
}
- if (now <= time && time <= now + offset) {
- try {
- waylineJobService.executeFlightTask(jobId);
- } catch (Exception e) {
- log.info("The scheduled task delivery failed.");
- waylineJobService.updateJob(WaylineJobDTO.builder()
- .jobId(jobId)
- .status(WaylineJobStatusEnum.FAILED.getVal())
- .endTime(LocalDateTime.now())
- .code(HttpStatus.SC_INTERNAL_SERVER_ERROR).build());
- } finally {
- redisOps.zRemove(RedisConst.WAYLINE_JOB, jobId);
+ WaylineJobDTO waylineJob = waylineJobOpt.get();
+ try {
+ ResponseResult result = waylineJobService.publishOneFlightTask(waylineJob);
+ if (ResponseResult.CODE_SUCCESS == result.getCode()) {
+ return;
+ }
+ log.info("Failed to prepare the task. {}", result.getMessage());
+ job.setCode(result.getCode());
+ waylineJobService.updateJob(job);
+ // Retry if the end time has not been exceeded.
+ this.retryPrepareConditionJob(jobKey, waylineJob);
+ } catch (Exception e) {
+ log.info("Failed to prepare the task. {}", e.getLocalizedMessage());
+ waylineJobService.updateJob(job);
+ this.retryPrepareConditionJob(jobKey, waylineJob);
+ }
+ }
+
+ private boolean checkTime(long time) {
+ // prepare the task one day in advance.
+ int offset = 86_400_000;
+ return System.currentTimeMillis() + offset >= time;
+ }
+
+ private Optional<WaylineJobDTO> getPreparedJob(WaylineJobKey jobKey, WaylineJobDTO job) {
+ long time = waylineRedisService.getPreparedWaylineJobTime(jobKey).longValue();
+ if (!checkTime(time)) {
+ return Optional.empty();
+ }
+
+ Optional<WaylineJobDTO> waylineJobOpt = waylineRedisService.getConditionalWaylineJob(jobKey.getJobId());
+ // Determine whether the conditional task or the scheduled task has expired.
+ if (waylineJobOpt.isEmpty()) {
+ waylineJobOpt = waylineJobService.getJobByJobId(jobKey.getWorkspaceId(), jobKey.getJobId());
+ if (waylineJobOpt.isEmpty() || waylineJobOpt.get().getEndTime().isBefore(LocalDateTime.now())) {
+ job.setCode(CommonErrorEnum.REDIS_DATA_NOT_FOUND.getErrorCode());
+ waylineJobService.updateJob(job);
+ return Optional.empty();
}
}
+ waylineRedisService.removePreparedWaylineJob(jobKey);
+ return waylineJobOpt;
+ }
+
+ private void retryPrepareConditionJob(WaylineJobKey jobKey, WaylineJobDTO waylineJob) {
+ if (WaylineTaskTypeEnum.CONDITION != waylineJob.getTaskType()) {
+ return;
+ }
+ // If the end time is exceeded, no more retries will be made.
+ waylineRedisService.delConditionalWaylineJob(jobKey.getJobId());
+ if (waylineJob.getEndTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() < System.currentTimeMillis()) {
+ return;
+ }
+
+ Optional<WaylineJobDTO> childJobOpt = waylineJobService.createWaylineJobByParent(jobKey.getWorkspaceId(), jobKey.getJobId());
+ if (childJobOpt.isEmpty()) {
+ log.error("Failed to create wayline job.");
+ return;
+ }
+
+ WaylineJobDTO newJob = childJobOpt.get();
+ newJob.setBeginTime(LocalDateTime.now().plusSeconds(RedisConst.WAYLINE_JOB_BLOCK_TIME));
+ boolean isAdd = waylineRedisService.addPreparedWaylineJob(newJob);
+ if (!isAdd) {
+ log.error("Failed to create wayline job. {}", newJob.getJobId());
+ return;
+ }
+
+ waylineJob.setJobId(newJob.getJobId());
+ waylineRedisService.setConditionalWaylineJob(waylineJob);
}
}
--
Gitblit v1.9.3