From 56df98ce4952239fbf7d0e99dbeb0e5c71531d6f Mon Sep 17 00:00:00 2001
From: sean.zhou <sean.zhou@dji.com>
Date: Fri, 18 Nov 2022 18:29:06 +0800
Subject: [PATCH] initial v1.3.0
---
src/main/java/com/dji/sample/wayline/service/impl/WaylineJobServiceImpl.java | 347 +++++++++++++++++++++++++++++++++++++++++++++++----------
1 files changed, 282 insertions(+), 65 deletions(-)
diff --git a/src/main/java/com/dji/sample/wayline/service/impl/WaylineJobServiceImpl.java b/src/main/java/com/dji/sample/wayline/service/impl/WaylineJobServiceImpl.java
index 9a2336c..5313863 100644
--- a/src/main/java/com/dji/sample/wayline/service/impl/WaylineJobServiceImpl.java
+++ b/src/main/java/com/dji/sample/wayline/service/impl/WaylineJobServiceImpl.java
@@ -3,40 +3,44 @@
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.dji.sample.common.error.CommonErrorEnum;
import com.dji.sample.common.model.CustomClaim;
import com.dji.sample.common.model.Pagination;
import com.dji.sample.common.model.PaginationData;
-import com.dji.sample.component.mqtt.model.CommonTopicResponse;
-import com.dji.sample.component.mqtt.model.ServiceReply;
-import com.dji.sample.component.mqtt.model.ServicesMethodEnum;
-import com.dji.sample.component.mqtt.model.TopicConst;
+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.manage.model.dto.DeviceDTO;
import com.dji.sample.manage.service.IDeviceService;
import com.dji.sample.wayline.dao.IWaylineJobMapper;
-import com.dji.sample.wayline.model.dto.FlightTaskCreateDTO;
-import com.dji.sample.wayline.model.dto.FlightTaskFileDTO;
-import com.dji.sample.wayline.model.dto.WaylineFileDTO;
-import com.dji.sample.wayline.model.dto.WaylineJobDTO;
+import com.dji.sample.wayline.model.dto.*;
import com.dji.sample.wayline.model.entity.WaylineJobEntity;
+import com.dji.sample.wayline.model.enums.WaylineJobStatusEnum;
+import com.dji.sample.wayline.model.enums.WaylineMethodEnum;
+import com.dji.sample.wayline.model.enums.WaylineTaskTypeEnum;
import com.dji.sample.wayline.model.param.CreateJobParam;
import com.dji.sample.wayline.service.IWaylineFileService;
import com.dji.sample.wayline.service.IWaylineJobService;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
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.stereotype.Service;
+import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.annotation.Transactional;
+import org.springframework.util.CollectionUtils;
import java.net.URL;
import java.sql.SQLException;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
-import java.util.List;
-import java.util.Optional;
-import java.util.UUID;
+import java.util.*;
import java.util.stream.Collectors;
/**
@@ -64,10 +68,18 @@
@Autowired
private RedisOpsUtils redisOps;
+ @Autowired
+ private ObjectMapper objectMapper;
+
+
@Override
- public Boolean createJob(CreateJobParam param, CustomClaim customClaim) throws SQLException {
- if (param == null) {
- return false;
+ public Optional<WaylineJobDTO> createWaylineJob(CreateJobParam param, CustomClaim customClaim) {
+ if (Objects.isNull(param)) {
+ return Optional.empty();
+ }
+ // Immediate tasks, allocating time on the backend.
+ if (Objects.isNull(param.getExecuteTime())) {
+ param.setExecuteTime(System.currentTimeMillis());
}
WaylineJobEntity jobEntity = WaylineJobEntity.builder()
.name(param.getName())
@@ -76,75 +88,203 @@
.username(customClaim.getUsername())
.workspaceId(customClaim.getWorkspaceId())
.jobId(UUID.randomUUID().toString())
- .type(param.getType())
+ .executeTime(param.getExecuteTime())
+ .status(WaylineJobStatusEnum.PENDING.getVal())
+ .taskType(param.getTaskType())
+ .waylineType(param.getWaylineType())
+ .outOfControlAction(param.getOutOfControlAction())
+ .rthAltitude(param.getRthAltitude())
.build();
int id = mapper.insert(jobEntity);
if (id <= 0) {
- return false;
+ return Optional.empty();
}
- if (param.isImmediate()) {
- publishFlightTask(jobEntity.getWorkspaceId(), jobEntity.getJobId());
- }
- return true;
+ return Optional.ofNullable(this.entity2Dto(jobEntity));
}
@Override
- public void publishFlightTask(String workspaceId, String jobId) throws SQLException {
+ public ResponseResult publishFlightTask(CreateJobParam param, CustomClaim customClaim) throws SQLException {
+ Optional<WaylineJobDTO> waylineJobOpt = this.createWaylineJob(param, customClaim);
+ if (waylineJobOpt.isEmpty()) {
+ throw new SQLException("Failed to create wayline job.");
+ }
+ WaylineJobDTO waylineJob = waylineJobOpt.get();
+
+ boolean isOnline = deviceService.checkDeviceOnline(waylineJob.getDockSn());
+ if (!isOnline) {
+ throw new RuntimeException("Dock is offline.");
+ }
+
+ // get wayline file
+ Optional<WaylineFileDTO> waylineFile = waylineFileService.getWaylineByWaylineId(waylineJob.getWorkspaceId(), waylineJob.getFileId());
+ if (waylineFile.isEmpty()) {
+ throw new SQLException("Wayline file doesn't exist.");
+ }
+
+ // get file url
+ URL url = waylineFileService.getObjectUrl(waylineJob.getWorkspaceId(), waylineFile.get().getWaylineId());
+
+ FlightTaskCreateDTO flightTask = FlightTaskCreateDTO.builder()
+ .flightId(waylineJob.getJobId())
+ .executeTime(waylineJob.getExecuteTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli())
+ .taskType(waylineJob.getTaskType())
+ .waylineType(waylineJob.getWaylineType())
+ .rthAltitude(waylineJob.getRthAltitude())
+ .outOfControlAction(waylineJob.getOutOfControlAction())
+ .file(FlightTaskFileDTO.builder()
+ .url(url.toString())
+ .fingerprint(waylineFile.get().getSign())
+ .build())
+ .build();
+
+ String topic = TopicConst.THING_MODEL_PRE + TopicConst.PRODUCT +
+ waylineJob.getDockSn() + TopicConst.SERVICES_SUF;
+ CommonTopicResponse<Object> response = CommonTopicResponse.builder()
+ .tid(UUID.randomUUID().toString())
+ .bid(waylineJob.getJobId())
+ .timestamp(System.currentTimeMillis())
+ .data(flightTask)
+ .method(WaylineMethodEnum.FLIGHT_TASK_PREPARE.getMethod())
+ .build();
+
+ ServiceReply serviceReply = messageSender.publishWithReply(topic, response);
+ if (ResponseResult.CODE_SUCCESS != serviceReply.getResult()) {
+ log.info("Prepare task ====> Error code: {}", serviceReply.getResult());
+ this.updateJob(WaylineJobDTO.builder()
+ .workspaceId(waylineJob.getWorkspaceId())
+ .jobId(waylineJob.getJobId())
+ .status(WaylineJobStatusEnum.FAILED.getVal())
+ .endTime(LocalDateTime.now())
+ .code(serviceReply.getResult()).build());
+ return ResponseResult.error("Prepare task ====> Error code: " + serviceReply.getResult());
+ }
+
+ // Issue an immediate task execution command.
+ if (WaylineTaskTypeEnum.IMMEDIATE.getVal() == waylineJob.getTaskType()) {
+ if (!executeFlightTask(waylineJob.getJobId())) {
+ return ResponseResult.error("Failed to execute job.");
+ }
+ }
+
+ if (WaylineTaskTypeEnum.TIMED.getVal() == waylineJob.getTaskType()) {
+ boolean isAdd = redisOps.zAdd(RedisConst.WAYLINE_JOB, waylineJob.getJobId(),
+ waylineJob.getExecuteTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
+ if (!isAdd) {
+ return ResponseResult.error("Failed to create scheduled job.");
+ }
+ }
+
+ return ResponseResult.success();
+ }
+
+ @Override
+ public Boolean executeFlightTask(String jobId) {
// get job
Optional<WaylineJobDTO> waylineJob = this.getJobByJobId(jobId);
if (waylineJob.isEmpty()) {
throw new IllegalArgumentException("Job doesn't exist.");
}
- long expire = redisOps.getExpire(RedisConst.DEVICE_ONLINE_PREFIX + waylineJob.get().getDockSn());
- if (expire < 0) {
+ boolean isOnline = deviceService.checkDeviceOnline(waylineJob.get().getDockSn());
+ if (!isOnline) {
throw new RuntimeException("Dock is offline.");
}
- // get wayline file
- Optional<WaylineFileDTO> waylineFile = waylineFileService.getWaylineByWaylineId(workspaceId, waylineJob.get().getFileId());
- if (waylineFile.isEmpty()) {
- throw new IllegalArgumentException("Wayline file doesn't exist.");
- }
-
- // get file url
- URL url = waylineFileService.getObjectUrl(workspaceId, waylineFile.get().getWaylineId());
-
WaylineJobDTO job = waylineJob.get();
- FlightTaskCreateDTO flightTask = FlightTaskCreateDTO.builder()
- .flightId(jobId)
- .type(job.getType())
- .file(FlightTaskFileDTO.builder()
- .url(url.toString())
- .sign(waylineFile.get().getSign())
- .build())
- .build();
+ FlightTaskCreateDTO flightTask = FlightTaskCreateDTO.builder().flightId(jobId).build();
String topic = TopicConst.THING_MODEL_PRE + TopicConst.PRODUCT +
job.getDockSn() + TopicConst.SERVICES_SUF;
CommonTopicResponse<Object> response = CommonTopicResponse.builder()
.tid(UUID.randomUUID().toString())
- .bid(UUID.randomUUID().toString())
+ .bid(jobId)
.timestamp(System.currentTimeMillis())
.data(flightTask)
- .method(ServicesMethodEnum.FLIGHTTASK_CREATE.getMethod())
+ .method(WaylineMethodEnum.FLIGHT_TASK_EXECUTE.getMethod())
.build();
- Optional<ServiceReply> serviceReplyOpt = messageSender.publishWithReply(topic, response);
- if (serviceReplyOpt.isEmpty()) {
- log.info("Timeout to receive reply.");
- throw new RuntimeException("Timeout to receive reply.");
- }
- if (serviceReplyOpt.get().getResult() != 0) {
- log.info("Error code: {}", serviceReplyOpt.get().getResult());
- throw new RuntimeException("Error code: " + serviceReplyOpt.get().getResult());
+ ServiceReply serviceReply = messageSender.publishWithReply(topic, response);
+ if (ResponseResult.CODE_SUCCESS != serviceReply.getResult()) {
+ log.info("Execute job ====> Error code: {}", serviceReply.getResult());
+ this.updateJob(WaylineJobDTO.builder()
+ .jobId(jobId)
+ .status(WaylineJobStatusEnum.FAILED.getVal())
+ .endTime(LocalDateTime.now())
+ .code(serviceReply.getResult()).build());
+ return false;
}
- job.setBid(response.getBid());
- boolean isUpd = this.updateJob(job);
- if (!isUpd) {
- throw new SQLException("Failed to update data.");
+ this.updateJob(WaylineJobDTO.builder()
+ .jobId(jobId)
+ .status(WaylineJobStatusEnum.IN_PROGRESS.getVal())
+ .build());
+ redisOps.setWithExpire(jobId,
+ EventsReceiver.<FlightTaskProgressReceiver>builder().bid(jobId).sn(job.getDockSn()).build(),
+ RedisConst.DEVICE_ALIVE_SECOND * RedisConst.DEVICE_ALIVE_SECOND);
+ return true;
+ }
+
+ @Override
+ public void cancelFlightTask(String workspaceId, Collection<String> jobIds) {
+ List<WaylineJobEntity> waylineJobs = mapper.selectList(
+ new LambdaQueryWrapper<WaylineJobEntity>()
+ .or(wrapper -> jobIds.forEach(id -> wrapper.eq(WaylineJobEntity::getJobId, id))));
+
+ // Check if the job have ended.
+ List<String> endJobs = waylineJobs.stream()
+ .filter(job -> WaylineJobStatusEnum.find(job.getStatus()).getEnd())
+ .map(WaylineJobEntity::getName)
+ .collect(Collectors.toList());
+ if (!CollectionUtils.isEmpty(endJobs)) {
+ throw new IllegalArgumentException("There are jobs that have ended." + Arrays.toString(endJobs.toArray()));
}
+
+ Set<String> ids = waylineJobs.stream().map(WaylineJobEntity::getJobId).collect(Collectors.toSet());
+ for (String id : jobIds) {
+ if (!ids.contains(id)) {
+ throw new IllegalArgumentException("Job id " + id + " doesn't exist.");
+ }
+ }
+
+ // Group job id by dock sn.
+ Map<String, List<String>> dockJobs = waylineJobs.stream()
+ .collect(Collectors.groupingBy(WaylineJobEntity::getDockSn,
+ Collectors.mapping(WaylineJobEntity::getJobId, Collectors.toList())));
+ dockJobs.forEach((dockSn, idList) -> this.publishCancelTask(workspaceId, dockSn, idList));
+
+ }
+
+ private void publishCancelTask(String workspaceId, String dockSn, List<String> jobIds) {
+ boolean isOnline = deviceService.checkDeviceOnline(dockSn);
+ if (isOnline) {
+ throw new RuntimeException("Dock is offline.");
+ }
+ String topic = TopicConst.THING_MODEL_PRE + TopicConst.PRODUCT + dockSn + TopicConst.SERVICES_SUF;
+
+ CommonTopicResponse<Object> response = CommonTopicResponse.builder()
+ .tid(UUID.randomUUID().toString())
+ .bid(UUID.randomUUID().toString())
+ .timestamp(System.currentTimeMillis())
+ .data(Map.of(MapKeyConst.FLIGHT_IDS, jobIds))
+ .method(WaylineMethodEnum.FLIGHT_TASK_CANCEL.getMethod())
+ .build();
+
+ ServiceReply serviceReply = messageSender.publishWithReply(topic, response);
+ if (ResponseResult.CODE_SUCCESS != serviceReply.getResult()) {
+ log.info("Cancel job ====> Error code: {}", serviceReply.getResult());
+ throw new RuntimeException("Failed to cancel the wayline job of " + dockSn);
+ }
+
+ for (String jobId : jobIds) {
+ this.updateJob(WaylineJobDTO.builder()
+ .workspaceId(workspaceId)
+ .jobId(jobId)
+ .status(WaylineJobStatusEnum.CANCEL.getVal())
+ .endTime(LocalDateTime.now())
+ .build());
+ redisOps.zRemove(RedisConst.WAYLINE_JOB, jobId);
+ }
+
}
@Override
@@ -159,9 +299,7 @@
public Boolean updateJob(WaylineJobDTO dto) {
return mapper.update(this.dto2Entity(dto),
new LambdaUpdateWrapper<WaylineJobEntity>()
- .eq(WaylineJobEntity::getWorkspaceId, dto.getWorkspaceId())
- .eq(WaylineJobEntity::getJobId, dto.getJobId()))
- > 0;
+ .eq(WaylineJobEntity::getJobId, dto.getJobId())) > 0;
}
@Override
@@ -169,7 +307,8 @@
Page<WaylineJobEntity> pageData = mapper.selectPage(
new Page<WaylineJobEntity>(page, pageSize),
new LambdaQueryWrapper<WaylineJobEntity>()
- .eq(WaylineJobEntity::getWorkspaceId, workspaceId));
+ .eq(WaylineJobEntity::getWorkspaceId, workspaceId)
+ .orderByDesc(WaylineJobEntity::getId));
List<WaylineJobDTO> records = pageData.getRecords()
.stream()
.map(this::entity2Dto)
@@ -178,14 +317,76 @@
return new PaginationData<WaylineJobDTO>(records, new Pagination(pageData));
}
+
+ @Override
+ @ServiceActivator(inputChannel = ChannelName.INBOUND_REQUESTS_FLIGHT_TASK_RESOURCE_GET, outputChannel = ChannelName.OUTBOUND)
+ @Transactional(isolation = Isolation.READ_UNCOMMITTED)
+ public void flightTaskResourceGet(CommonTopicReceiver receiver, MessageHeaders headers) {
+ Map<String, String> jobIdMap = objectMapper.convertValue(receiver.getData(), new TypeReference<Map<String, String>>() {});
+ String jobId = jobIdMap.get(MapKeyConst.FLIGHT_ID);
+
+ CommonTopicResponse.CommonTopicResponseBuilder<RequestsReply> builder = CommonTopicResponse.<RequestsReply>builder()
+ .tid(receiver.getTid())
+ .bid(receiver.getBid())
+ .method(RequestsMethodEnum.FLIGHT_TASK_RESOURCE_GET.getMethod())
+ .timestamp(System.currentTimeMillis());
+
+ String topic = headers.get(MqttHeaders.RECEIVED_TOPIC).toString() + TopicConst._REPLY_SUF;
+
+ Optional<WaylineJobDTO> waylineJobOpt = this.getJobByJobId(jobId);
+ if (waylineJobOpt.isEmpty()) {
+ builder.data(RequestsReply.error(CommonErrorEnum.ILLEGAL_ARGUMENT));
+ messageSender.publish(topic, builder.build());
+ return;
+ }
+
+ WaylineJobDTO waylineJob = waylineJobOpt.get();
+
+ // get wayline file
+ Optional<WaylineFileDTO> waylineFile = waylineFileService.getWaylineByWaylineId(waylineJob.getWorkspaceId(), waylineJob.getFileId());
+ if (waylineFile.isEmpty()) {
+ builder.data(RequestsReply.error(CommonErrorEnum.ILLEGAL_ARGUMENT));
+ messageSender.publish(topic, builder.build());
+ return;
+ }
+
+ // get file url
+ URL url = null;
+ try {
+ url = waylineFileService.getObjectUrl(waylineJob.getWorkspaceId(), waylineFile.get().getWaylineId());
+ builder.data(RequestsReply.success(FlightTaskCreateDTO.builder()
+ .file(FlightTaskFileDTO.builder()
+ .url(url.toString())
+ .fingerprint(waylineFile.get().getSign())
+ .build())
+ .build()));
+
+ } catch (SQLException | NullPointerException e) {
+ e.printStackTrace();
+ builder.data(RequestsReply.error(CommonErrorEnum.ILLEGAL_ARGUMENT));
+ messageSender.publish(topic, builder.build());
+ return;
+ }
+
+ messageSender.publish(topic, builder.build());
+
+ }
+
private WaylineJobEntity dto2Entity(WaylineJobDTO dto) {
WaylineJobEntity.WaylineJobEntityBuilder builder = WaylineJobEntity.builder();
if (dto == null) {
return builder.build();
}
- return builder.type(dto.getType())
- .bid(dto.getBid())
+ if (Objects.nonNull(dto.getEndTime())) {
+ builder.endTime(dto.getEndTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
+ }
+ if (Objects.nonNull(dto.getExecuteTime())) {
+ builder.executeTime(dto.getExecuteTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
+ }
+ return builder.status(dto.getStatus())
+ .mediaCount(dto.getMediaCount())
.name(dto.getJobName())
+ .errorCode(dto.getCode())
.build();
}
@@ -193,10 +394,9 @@
if (entity == null) {
return null;
}
- return WaylineJobDTO.builder()
+
+ WaylineJobDTO.WaylineJobDTOBuilder builder = WaylineJobDTO.builder()
.jobId(entity.getJobId())
- .bid(entity.getBid())
- .updateTime(LocalDateTime.ofInstant(Instant.ofEpochMilli(entity.getUpdateTime()), ZoneId.systemDefault()))
.jobName(entity.getName())
.fileId(entity.getFileId())
.fileName(waylineFileService.getWaylineByWaylineId(entity.getWorkspaceId(), entity.getFileId())
@@ -206,7 +406,24 @@
.orElse(DeviceDTO.builder().build()).getNickname())
.username(entity.getUsername())
.workspaceId(entity.getWorkspaceId())
- .type(entity.getType())
- .build();
+ .status(entity.getStatus())
+ .code(entity.getErrorCode())
+ .executeTime(LocalDateTime.ofInstant(Instant.ofEpochMilli(entity.getExecuteTime()), ZoneId.systemDefault()))
+ .taskType(entity.getTaskType())
+ .waylineType(entity.getWaylineType())
+ .rthAltitude(entity.getRthAltitude())
+ .outOfControlAction(entity.getOutOfControlAction())
+ .mediaCount(entity.getMediaCount());
+
+ if (Objects.nonNull(entity.getEndTime())) {
+ builder.endTime(LocalDateTime.ofInstant(Instant.ofEpochMilli(entity.getEndTime()), ZoneId.systemDefault()));
+ }
+ if (WaylineJobStatusEnum.IN_PROGRESS.getVal() == entity.getStatus() && redisOps.getExpire(entity.getJobId()) > 0) {
+ EventsReceiver<FlightTaskProgressReceiver> taskProgress = (EventsReceiver<FlightTaskProgressReceiver>) redisOps.get(entity.getJobId());
+ if (Objects.nonNull(taskProgress.getOutput()) && Objects.nonNull(taskProgress.getOutput().getProgress())) {
+ builder.progress(taskProgress.getOutput().getProgress().getPercent());
+ }
+ }
+ return builder.build();
}
}
--
Gitblit v1.9.3