What's new?

1. Add new model: DJI Matrices 350 RTK.
2. Update file hms.json.
3. Fixed some issues.

pom.xml

@@ -11,7 +11,7 @@
 
 	<groupId>com.dji</groupId>
 	<artifactId>cloud-api-sample</artifactId>
-	<version>1.4.0</version>
+	<version>1.5.0</version>
 	<name>cloud-api-sample</name>
 
 	<properties>

sql/cloud_sample.sql

@@ -134,7 +134,8 @@ VALUES
 	(22,1,67,0,'Mavic 3T Camera',NULL),
 	(23,2,144,0,'DJI RC Pro','Remote control for Mavic 3E/T and Mavic 3M'),
 	(24,0,77,2,'Mavic 3M',NULL),
-	(25,1,68,0,'Mavic 3M Camera',NULL);
+	(25,1,68,0,'Mavic 3M Camera',NULL),
+	(26,0,89,0,'Matrice 350 RTK',NULL);
 
 /*!40000 ALTER TABLE `manage_device_dictionary` ENABLE KEYS */;
 UNLOCK TABLES;

src/main/java/com/dji/sample/component/redis/RedisConst.java

@@ -37,11 +37,9 @@ public final class RedisConst {
 
     public static final String LOGS_FILE_PREFIX = "logs_file" + DELIMITER;
 
-    public static final String WAYLINE_JOB_TIMED_EXECUTE = "wayline_job_timed_execute";
+    public static final String WAYLINE_JOB_PREPARED = "wayline_job_prepared";
 
-    public static final String WAYLINE_JOB_CONDITION_PREPARE = "wayline_job_condition_prepare";
-
-    public static final String WAYLINE_JOB_CONDITION_PREFIX = WAYLINE_JOB_CONDITION_PREPARE + DELIMITER;
+    public static final String WAYLINE_JOB_CONDITION_PREFIX = "wayline_job_condition" + DELIMITER;
 
     public static final String WAYLINE_JOB_BLOCK_PREFIX = "wayline_job_block" + DELIMITER;
 

src/main/java/com/dji/sample/component/websocket/service/impl/WebSocketManageServiceImpl.java

@@ -75,7 +75,7 @@ public class WebSocketManageServiceImpl implements IWebSocketManageService {
         return RedisOpsUtils.hashKeys(key)
                 .stream()
                 .map(SESSIONS::get)
-                .filter(Objects::nonNull)
+                .filter(this.getValueWithWorkspace(workspaceId)::contains)
                 .collect(Collectors.toSet());
     }
 

src/main/java/com/dji/sample/manage/service/impl/DeviceServiceImpl.java

@@ -518,11 +518,39 @@ public class DeviceServiceImpl implements IDeviceService {
             }
             entity.setId(deviceEntity.getId());
             mapper.updateById(entity);
-            return Optional.of(deviceEntity);
+            fillNullField(entity, deviceEntity);
+            return Optional.of(entity);
         }
         return mapper.insert(entity) > 0 ? Optional.of(entity) : Optional.empty();
     }
 
+    private void fillNullField(DeviceEntity entity, DeviceEntity oldEntity) {
+        if (Objects.isNull(entity) || Objects.isNull(oldEntity)) {
+            return;
+        }
+        if (Objects.isNull(entity.getWorkspaceId())) {
+            entity.setWorkspaceId(oldEntity.getWorkspaceId());
+        }
+        if (Objects.isNull(entity.getUserId())) {
+            entity.setUserId(oldEntity.getUserId());
+        }
+        if (Objects.isNull(entity.getChildSn())) {
+            entity.setChildSn(oldEntity.getChildSn());
+        }
+        if (Objects.isNull(entity.getBoundStatus())) {
+            entity.setBoundStatus(oldEntity.getBoundStatus());
+        }
+        if (Objects.isNull(entity.getBoundTime())) {
+            entity.setBoundTime(oldEntity.getBoundTime());
+        }
+        if (Objects.isNull(entity.getFirmwareVersion())) {
+            entity.setFirmwareVersion(oldEntity.getFirmwareVersion());
+        }
+        if (Objects.isNull(entity.getDeviceIndex())) {
+            entity.setDeviceIndex(oldEntity.getDeviceIndex());
+        }
+    }
+
     /**
      * Convert the received gateway device object into a database entity object.
      * @param gateway
@@ -650,7 +678,7 @@ public class DeviceServiceImpl implements IDeviceService {
             return;
         }
         if (entity.getFirmwareVersion().equals(firmwareReleaseNoteOpt.get().getProductVersion())) {
-            deviceDTO.setFirmwareStatus(entity.getCompatibleStatus() ?
+            deviceDTO.setFirmwareStatus(Objects.requireNonNullElse(entity.getCompatibleStatus(), true) ?
                     DeviceFirmwareStatusEnum.NOT_UPGRADE.getVal() :
                     DeviceFirmwareStatusEnum.CONSISTENT_UPGRADE.getVal());
             return;

src/main/java/com/dji/sample/wayline/model/dto/WaylineJobDTO.java

@@ -1,5 +1,7 @@
 package com.dji.sample.wayline.model.dto;
 
+import com.dji.sample.wayline.model.enums.WaylineTaskTypeEnum;
+import com.dji.sample.wayline.model.enums.WaylineTemplateTypeEnum;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
@@ -32,9 +34,9 @@ public class WaylineJobDTO {
 
     private String workspaceId;
 
-    private Integer waylineType;
+    private WaylineTemplateTypeEnum waylineType;
 
-    private Integer taskType;
+    private WaylineTaskTypeEnum taskType;
 
     private LocalDateTime executeTime;
 

src/main/java/com/dji/sample/wayline/model/dto/WaylineJobKey.java

@@ -3,15 +3,13 @@ package com.dji.sample.wayline.model.dto;
 import com.dji.sample.component.redis.RedisConst;
 import lombok.Data;
 
-import java.util.Objects;
-
 /**
  * @author sean
  * @version 1.4
  * @date 2023/3/28
  */
 @Data
-public class ConditionalWaylineJobKey {
+public class WaylineJobKey {
 
     private String workspaceId;
 
@@ -19,18 +17,17 @@ public class ConditionalWaylineJobKey {
 
     private String jobId;
 
-    public ConditionalWaylineJobKey(String workspaceId, String dockSn, String jobId) {
+    public WaylineJobKey(String workspaceId, String dockSn, String jobId) {
         this.workspaceId = workspaceId;
         this.dockSn = dockSn;
         this.jobId = jobId;
     }
+    private WaylineJobKey(String[] keyArr) {
+        this(keyArr[0], keyArr[1], keyArr[2]);
+    }
 
-    public ConditionalWaylineJobKey(String key) {
-        if (Objects.isNull(key)) {
-            return;
-        }
-        String[] keyArr = key.split(RedisConst.DELIMITER);
-        new ConditionalWaylineJobKey(keyArr[0], keyArr[1], keyArr[2]);
+    public WaylineJobKey(String key) {
+        this(key.split(RedisConst.DELIMITER));
     }
 
     public String getKey() {

src/main/java/com/dji/sample/wayline/model/dto/WaylineTaskCreateDTO.java

@@ -1,5 +1,7 @@
 package com.dji.sample.wayline.model.dto;
 
+import com.dji.sample.wayline.model.enums.WaylineTaskTypeEnum;
+import com.dji.sample.wayline.model.enums.WaylineTemplateTypeEnum;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
@@ -18,9 +20,9 @@ public class WaylineTaskCreateDTO {
 
     private String flightId;
 
-    private Integer taskType;
+    private WaylineTaskTypeEnum taskType;
 
-    private Integer waylineType;
+    private WaylineTemplateTypeEnum waylineType;
 
     private Long executeTime;
 

src/main/java/com/dji/sample/wayline/service/IWaylineRedisService.java

@@ -1,8 +1,8 @@
 package com.dji.sample.wayline.service;
 
 import com.dji.sample.component.mqtt.model.EventsReceiver;
-import com.dji.sample.wayline.model.dto.ConditionalWaylineJobKey;
 import com.dji.sample.wayline.model.dto.WaylineJobDTO;
+import com.dji.sample.wayline.model.dto.WaylineJobKey;
 import com.dji.sample.wayline.model.dto.WaylineTaskProgressReceiver;
 
 import java.util.Optional;
@@ -70,6 +70,13 @@ public interface IWaylineRedisService {
      */
     String getBlockedWaylineJobId(String dockSn);
 
+    /**
+     * Delete the wayline job id blocked by the dock in redis.
+     * @param dockSn
+     * @return
+     */
+    Boolean delBlockedWaylineJobId(String dockSn);
+
     /**
      * Save the conditional wayline job by the dock to redis.
      * @param waylineJob
@@ -90,11 +97,30 @@ public interface IWaylineRedisService {
      */
     Boolean delConditionalWaylineJob(String jobId);
 
-    Boolean addPrepareConditionalWaylineJob(WaylineJobDTO waylineJob);
+    /**
+     * Add the wayline job that needs to be issued.
+     * @param waylineJob
+     * @return
+     */
+    Boolean addPreparedWaylineJob(WaylineJobDTO waylineJob);
 
-    Optional<ConditionalWaylineJobKey> getNearestConditionalWaylineJob();
+    /**
+     * Get the latest wayline job that needs to be issued.
+     * @return
+     */
+    Optional<WaylineJobKey> getNearestPreparedWaylineJob();
 
-    Double getConditionalWaylineJobTime(ConditionalWaylineJobKey jobKey);
+    /**
+     * Get the time when the wayline job is issued.
+     * @param jobKey
+     * @return
+     */
+    Double getPreparedWaylineJobTime(WaylineJobKey jobKey);
 
-    Boolean removePrepareConditionalWaylineJob(ConditionalWaylineJobKey jobKey);
+    /**
+     * Delete the wayline job that needs to be issued in redis.
+     * @param jobKey
+     * @return
+     */
+    Boolean removePreparedWaylineJob(WaylineJobKey jobKey);
 }

src/main/java/com/dji/sample/wayline/service/impl/FlightTaskServiceImpl.java

@@ -12,10 +12,11 @@ import com.dji.sample.manage.model.dto.DeviceDTO;
 import com.dji.sample.manage.model.enums.UserTypeEnum;
 import com.dji.sample.manage.service.IDeviceRedisService;
 import com.dji.sample.media.model.MediaFileCountDTO;
-import com.dji.sample.wayline.model.dto.ConditionalWaylineJobKey;
 import com.dji.sample.wayline.model.dto.WaylineJobDTO;
+import com.dji.sample.wayline.model.dto.WaylineJobKey;
 import com.dji.sample.wayline.model.dto.WaylineTaskProgressReceiver;
 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.IWaylineJobService;
 import com.dji.sample.wayline.service.IWaylineRedisService;
@@ -25,7 +26,6 @@ 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;
@@ -70,9 +70,6 @@ public class FlightTaskServiceImpl implements IFlightTaskService {
      */
     @ServiceActivator(inputChannel = ChannelName.INBOUND_EVENTS_FLIGHT_TASK_PROGRESS, outputChannel = ChannelName.OUTBOUND_EVENTS)
     public CommonTopicReceiver 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(),
-                receivedTopic.indexOf(TopicConst.EVENTS_SUF));
         EventsReceiver<WaylineTaskProgressReceiver> eventsReceiver = mapper.convertValue(receiver.getData(),
                 new TypeReference<EventsReceiver<WaylineTaskProgressReceiver>>(){});
         eventsReceiver.setBid(receiver.getBid());
@@ -87,42 +84,52 @@ public class FlightTaskServiceImpl implements IFlightTaskService {
         }
 
         EventsResultStatusEnum statusEnum = EventsResultStatusEnum.find(output.getStatus());
-        waylineRedisService.setRunningWaylineJob(dockSn, eventsReceiver);
-
-        if (statusEnum.getEnd()) {
-            WaylineJobDTO job = WaylineJobDTO.builder()
-                    .jobId(receiver.getBid())
-                    .status(WaylineJobStatusEnum.SUCCESS.getVal())
-                    .completedTime(LocalDateTime.now())
-                    .mediaCount(output.getExt().getMediaCount())
-                    .build();
-
-            // record the update of the media count.
-            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(eventsReceiver.getResult());
-                job.setStatus(WaylineJobStatusEnum.FAILED.getVal());
-            }
-
-            waylineJobService.updateJob(job);
-            waylineRedisService.delRunningWaylineJob(dockSn);
-            waylineRedisService.delPausedWaylineJob(receiver.getBid());
-        }
+        waylineRedisService.setRunningWaylineJob(receiver.getGateway(), eventsReceiver);
 
         Optional<DeviceDTO> deviceOpt = deviceRedisService.getDeviceOnline(receiver.getGateway());
         if (deviceOpt.isEmpty()) {
             return null;
         }
+
+        if (statusEnum.getEnd()) {
+            handleEndStatus(receiver, statusEnum, output.getExt().getMediaCount(), eventsReceiver.getResult(), deviceOpt.get());
+        }
+
         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();
+
+        // record the update of the media count.
+        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
@@ -130,106 +137,65 @@ public class FlightTaskServiceImpl implements IFlightTaskService {
      */
     @ServiceActivator(inputChannel = ChannelName.INBOUND_EVENTS_FLIGHT_TASK_READY, outputChannel = ChannelName.OUTBOUND_EVENTS)
     public CommonTopicReceiver handleTaskNotifications(CommonTopicReceiver receiver, MessageHeaders headers) {
-        String receivedTopic = String.valueOf(headers.get(MqttHeaders.RECEIVED_TOPIC));
-        String dockSn  = receivedTopic.substring((TopicConst.THING_MODEL_PRE + TopicConst.PRODUCT).length(),
-                receivedTopic.indexOf(TopicConst.EVENTS_SUF));
-        List<String> flightIds = mapper.convertValue(receiver.getData(),
-                new TypeReference<Map<String, List<String>>>(){}).get(MapKeyConst.FLIGHT_IDS);
+        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()) );
+        log.info("ready task list：{}", Arrays.toString(flightIds.toArray()));
         // Check conditional task blocking status.
         String blockedId = waylineRedisService.getBlockedWaylineJobId(dockSn);
-        if (!StringUtils.hasText(blockedId)) {
+        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();
-
-        try {
-            for (String jobId : flightIds) {
-                boolean isExecute = waylineJobService.executeFlightTask(device.getWorkspaceId(), jobId);
-                if (!isExecute) {
-                    return null;
-                }
-                Optional<WaylineJobDTO> waylineJobOpt = waylineRedisService.getConditionalWaylineJob(jobId);
-                if (waylineJobOpt.isEmpty()) {
-                    log.info("The conditional job has expired and will no longer be executed.");
-                    return receiver;
-                }
-                WaylineJobDTO waylineJob = waylineJobOpt.get();
-                this.retryPrepareJob(new ConditionalWaylineJobKey(device.getWorkspaceId(), dockSn, jobId), waylineJob);
-                return receiver;
-            }
-        } catch (Exception e) {
-            log.error("Failed to execute conditional task.");
-            e.printStackTrace();
+        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;
     }
 
-    @Scheduled(initialDelay = 10, fixedRate = 5, timeUnit = TimeUnit.SECONDS)
-    private void checkScheduledJob() {
-        Object jobIdValue = RedisOpsUtils.zGetMin(RedisConst.WAYLINE_JOB_TIMED_EXECUTE);
-        if (Objects.isNull(jobIdValue)) {
-            return;
-        }
-        log.info("Check the timed tasks of the 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);
-        long now = System.currentTimeMillis();
-        int offset = 30_000;
-
-        // Expired tasks are deleted directly.
-        if (time < now - offset) {
-            RedisOpsUtils.zRemove(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, jobIdValue);
-            waylineJobService.updateJob(WaylineJobDTO.builder()
-                    .jobId(jobArr[2])
-                    .status(WaylineJobStatusEnum.FAILED.getVal())
-                    .executeTime(LocalDateTime.now())
-                    .completedTime(LocalDateTime.now())
-                    .code(HttpStatus.SC_REQUEST_TIMEOUT).build());
-            return;
-        }
-
-        if (now <= time && time <= now + offset) {
-            try {
-                waylineJobService.executeFlightTask(jobArr[0], jobArr[2]);
-            } catch (Exception e) {
-                log.info("The scheduled task delivery failed.");
-                waylineJobService.updateJob(WaylineJobDTO.builder()
-                        .jobId(jobArr[2])
-                        .status(WaylineJobStatusEnum.FAILED.getVal())
-                        .executeTime(LocalDateTime.now())
-                        .completedTime(LocalDateTime.now())
-                        .code(HttpStatus.SC_INTERNAL_SERVER_ERROR).build());
-            } finally {
-                RedisOpsUtils.zRemove(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, jobIdValue);
+    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 prepareConditionJob() {
-        Optional<ConditionalWaylineJobKey> jobKeyOpt = waylineRedisService.getNearestConditionalWaylineJob();
+    private void prepareWaylineJob() {
+        Optional<WaylineJobKey> jobKeyOpt = waylineRedisService.getNearestPreparedWaylineJob();
         if (jobKeyOpt.isEmpty()) {
             return;
         }
-        ConditionalWaylineJobKey jobKey = jobKeyOpt.get();
-        log.info("Check the conditional tasks of the wayline. {}", jobKey.toString());
         // format: {workspace_id}:{dock_sn}:{job_id}
-        double time = waylineRedisService.getConditionalWaylineJobTime(jobKey);
-        long now = System.currentTimeMillis();
-        // prepare the task one day in advance.
-        int offset = 86_400_000;
-
-        if (now + offset < time) {
-            return;
-        }
+        WaylineJobKey jobKey = jobKeyOpt.get();
+        log.info("Check the prepared tasks of the wayline. {}", jobKey.toString());
 
         WaylineJobDTO job = WaylineJobDTO.builder()
                 .jobId(jobKey.getJobId())
@@ -237,40 +203,65 @@ public class FlightTaskServiceImpl implements IFlightTaskService {
                 .executeTime(LocalDateTime.now())
                 .completedTime(LocalDateTime.now())
                 .code(HttpStatus.SC_INTERNAL_SERVER_ERROR).build();
-        try {
-            Optional<WaylineJobDTO> waylineJobOpt = waylineRedisService.getConditionalWaylineJob(jobKey.getJobId());
-            if (waylineJobOpt.isEmpty()) {
-                job.setCode(CommonErrorEnum.REDIS_DATA_NOT_FOUND.getErrorCode());
-                waylineJobService.updateJob(job);
-                waylineRedisService.removePrepareConditionalWaylineJob(jobKey);
-                return;
-            }
-            WaylineJobDTO waylineJob = waylineJobOpt.get();
+        Optional<WaylineJobDTO> waylineJobOpt = getPreparedJob(jobKey, job);
+        if (waylineJobOpt.isEmpty()) {
+            return;
+        }
 
+        WaylineJobDTO waylineJob = waylineJobOpt.get();
+        try {
             ResponseResult result = waylineJobService.publishOneFlightTask(waylineJob);
-            waylineRedisService.removePrepareConditionalWaylineJob(jobKey);
-
             if (ResponseResult.CODE_SUCCESS == result.getCode()) {
                 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() - RedisConst.WAYLINE_JOB_BLOCK_TIME * 1000 < now) {
-                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.retryPrepareJob(jobKey, waylineJob);
-
+            this.retryPrepareConditionJob(jobKey, waylineJob);
         } catch (Exception e) {
-            log.info("Failed to prepare the conditional task.");
+            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 retryPrepareJob(ConditionalWaylineJobKey jobKey, WaylineJobDTO waylineJob) {
+    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.");
@@ -279,7 +270,7 @@ public class FlightTaskServiceImpl implements IFlightTaskService {
 
         WaylineJobDTO newJob = childJobOpt.get();
         newJob.setBeginTime(LocalDateTime.now().plusSeconds(RedisConst.WAYLINE_JOB_BLOCK_TIME));
-        boolean isAdd = waylineRedisService.addPrepareConditionalWaylineJob(newJob);
+        boolean isAdd = waylineRedisService.addPreparedWaylineJob(newJob);
         if (!isAdd) {
             log.error("Failed to create wayline job. {}", newJob.getJobId());
             return;

src/main/java/com/dji/sample/wayline/service/impl/WaylineJobServiceImpl.java

@@ -27,10 +27,7 @@ import com.dji.sample.media.service.IFileService;
 import com.dji.sample.wayline.dao.IWaylineJobMapper;
 import com.dji.sample.wayline.model.dto.*;
 import com.dji.sample.wayline.model.entity.WaylineJobEntity;
-import com.dji.sample.wayline.model.enums.WaylineErrorCodeEnum;
-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.enums.*;
 import com.dji.sample.wayline.model.param.CreateJobParam;
 import com.dji.sample.wayline.model.param.UpdateJobParam;
 import com.dji.sample.wayline.service.IWaylineFileService;
@@ -152,98 +149,91 @@ public class WaylineJobServiceImpl implements IWaylineJobService {
             return;
         }
         long now = System.currentTimeMillis() / 1000;
-        if (CollectionUtils.isEmpty(param.getTaskDays())) {
-            param.setTaskDays(List.of(now));
-        }
-        if (CollectionUtils.isEmpty(param.getTaskPeriods())) {
-            param.setTaskPeriods(List.of(List.of(now)));
-        }
+        param.setTaskDays(Collections.singletonList(now));
+        param.setTaskPeriods(Collections.singletonList(Collections.singletonList(now)));
     }
 
     @Override
     public ResponseResult publishFlightTask(CreateJobParam param, CustomClaim customClaim) throws SQLException {
         fillImmediateTime(param);
 
+        param.getTaskDays().sort((a, b) -> (int) (a - b));
+        param.getTaskPeriods().sort((a, b) -> (int) (a.get(0) - b.get(0)));
         for (Long taskDay : param.getTaskDays()) {
             LocalDate date = LocalDate.ofInstant(Instant.ofEpochSecond(taskDay), ZoneId.systemDefault());
             for (List<Long> taskPeriod : param.getTaskPeriods()) {
                 long beginTime = LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(taskPeriod.get(0)), ZoneId.systemDefault()))
                         .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
-                long endTime = taskPeriod.size() > 1 ?
+                long endTime = taskPeriod.size() > 1 && Objects.nonNull(taskPeriod.get(1)) ?
                         LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(taskPeriod.get(1)), ZoneId.systemDefault()))
                                 .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() : beginTime;
+                if (WaylineTaskTypeEnum.IMMEDIATE != param.getTaskType() && endTime < System.currentTimeMillis()) {
+                    return ResponseResult.error("The task has expired.");
+                }
                 Optional<WaylineJobDTO> waylineJobOpt = this.createWaylineJob(param, customClaim.getWorkspaceId(), customClaim.getUsername(), beginTime, endTime);
                 if (waylineJobOpt.isEmpty()) {
-                    throw new SQLException("Failed to create wayline job.");
+                    return ResponseResult.error("Failed to create wayline job.");
                 }
 
                 WaylineJobDTO waylineJob = waylineJobOpt.get();
-                // If it is a conditional task type, add conditions to the job parameters.
-                addConditions(waylineJob, param, beginTime, endTime);
+                if (WaylineTaskTypeEnum.IMMEDIATE == param.getTaskType()) {
+                    return this.publishOneFlightTask(waylineJob);
+                }
 
-                return this.publishOneFlightTask(waylineJob);
+                // If it is a conditional task type, add conditions to the job parameters.
+                addPreparedJob(waylineJob, param, beginTime, endTime);
             }
         }
-        return ResponseResult.error();
+        return ResponseResult.success();
     }
 
-    private void addConditions(WaylineJobDTO waylineJob, CreateJobParam param, Long beginTime, Long endTime) {
-        if (WaylineTaskTypeEnum.CONDITION != param.getTaskType()) {
-            return;
-        }
-
-        waylineJob.setConditions(
-                WaylineTaskConditionDTO.builder()
-                        .executableConditions(Objects.nonNull(param.getMinStorageCapacity()) ?
-                                WaylineTaskExecutableConditionDTO.builder().storageCapacity(param.getMinStorageCapacity()).build() : null)
-                        .readyConditions(WaylineTaskReadyConditionDTO.builder()
-                                .batteryCapacity(param.getMinBatteryCapacity())
-                                .beginTime(beginTime)
-                                .endTime(endTime)
-                                .build())
-                        .build());
-
-        waylineRedisService.setConditionalWaylineJob(waylineJob);
-        // key: wayline_job_condition, value: {workspace_id}:{dock_sn}:{job_id}
-        boolean isAdd = waylineRedisService.addPrepareConditionalWaylineJob(waylineJob);
+    private void addPreparedJob(WaylineJobDTO waylineJob, CreateJobParam param, Long beginTime, Long endTime) {
+        if (WaylineTaskTypeEnum.CONDITION == param.getTaskType()) {
+            waylineJob.setConditions(
+                    WaylineTaskConditionDTO.builder()
+                            .executableConditions(Objects.nonNull(param.getMinStorageCapacity()) ?
+                                    WaylineTaskExecutableConditionDTO.builder().storageCapacity(param.getMinStorageCapacity()).build() : null)
+                            .readyConditions(WaylineTaskReadyConditionDTO.builder()
+                                    .batteryCapacity(param.getMinBatteryCapacity())
+                                    .beginTime(beginTime)
+                                    .endTime(endTime)
+                                    .build())
+                            .build());
+
+            waylineRedisService.setConditionalWaylineJob(waylineJob);
+        }
+        // value: {workspace_id}:{dock_sn}:{job_id}
+        boolean isAdd = waylineRedisService.addPreparedWaylineJob(waylineJob);
         if (!isAdd) {
-            throw new RuntimeException("Failed to create conditional job.");
+            throw new RuntimeException("Failed to create prepare job.");
         }
     }
 
     public ResponseResult publishOneFlightTask(WaylineJobDTO waylineJob) throws SQLException {
 
-        boolean isOnline = deviceRedisService.checkDeviceOnline(waylineJob.getDockSn());
-        if (!isOnline) {
-            throw new RuntimeException("Dock is offline.");
-        }
-
         boolean isSuccess = this.prepareFlightTask(waylineJob);
         if (!isSuccess) {
             return ResponseResult.error("Failed to prepare job.");
         }
 
         // Issue an immediate task execution command.
-        if (WaylineTaskTypeEnum.IMMEDIATE.getVal() == waylineJob.getTaskType()) {
-            if (!executeFlightTask(waylineJob.getWorkspaceId(), waylineJob.getJobId())) {
+        if (WaylineTaskTypeEnum.IMMEDIATE == waylineJob.getTaskType()) {
+            boolean isExecuted = executeFlightTask(waylineJob.getWorkspaceId(), waylineJob.getJobId());
+            if (!isExecuted) {
                 return ResponseResult.error("Failed to execute job.");
             }
         }
 
-        if (WaylineTaskTypeEnum.TIMED.getVal() == waylineJob.getTaskType()) {
-            // key: wayline_job_timed, value: {workspace_id}:{dock_sn}:{job_id}
-            boolean isAdd = RedisOpsUtils.zAdd(RedisConst.WAYLINE_JOB_TIMED_EXECUTE,
-                    waylineJob.getWorkspaceId() + RedisConst.DELIMITER + waylineJob.getDockSn() + RedisConst.DELIMITER + waylineJob.getJobId(),
-                    waylineJob.getBeginTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
-            if (!isAdd) {
-                return ResponseResult.error("Failed to create scheduled job.");
-            }
-        }
-
         return ResponseResult.success();
     }
 
     private Boolean prepareFlightTask(WaylineJobDTO waylineJob) throws SQLException {
+
+        boolean isOnline = deviceRedisService.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()) {
@@ -266,7 +256,7 @@ public class WaylineJobServiceImpl implements IWaylineJobService {
                         .build())
                 .build();
 
-        if (WaylineTaskTypeEnum.CONDITION.getVal() == waylineJob.getTaskType()) {
+        if (WaylineTaskTypeEnum.CONDITION == waylineJob.getTaskType()) {
             if (Objects.isNull(waylineJob.getConditions())) {
                 throw new IllegalArgumentException();
             }
@@ -317,7 +307,7 @@ public class WaylineJobServiceImpl implements IWaylineJobService {
                     .completedTime(LocalDateTime.now())
                     .code(serviceReply.getResult()).build());
             // The conditional task fails and enters the blocking status.
-            if (WaylineTaskTypeEnum.CONDITION.getVal() == job.getTaskType()
+            if (WaylineTaskTypeEnum.CONDITION == job.getTaskType()
                     && WaylineErrorCodeEnum.find(serviceReply.getResult()).isBlock()) {
                 waylineRedisService.setBlockedWaylineJob(job.getDockSn(), jobId);
             }
@@ -372,7 +362,6 @@ public class WaylineJobServiceImpl implements IWaylineJobService {
                     .status(WaylineJobStatusEnum.CANCEL.getVal())
                     .completedTime(LocalDateTime.now())
                     .build());
-            RedisOpsUtils.zRemove(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, workspaceId + RedisConst.DELIMITER + dockSn + RedisConst.DELIMITER + jobId);
         }
 
     }
@@ -382,8 +371,7 @@ public class WaylineJobServiceImpl implements IWaylineJobService {
                 new LambdaQueryWrapper<WaylineJobEntity>()
                         .eq(WaylineJobEntity::getWorkspaceId, workspaceId)
                         .eq(Objects.nonNull(status), WaylineJobEntity::getStatus, status.getVal())
-                        .and(!CollectionUtils.isEmpty(jobIds),
-                                wrapper -> jobIds.forEach(id -> wrapper.eq(WaylineJobEntity::getJobId, id).or())))
+                        .in(!CollectionUtils.isEmpty(jobIds), WaylineJobEntity::getJobId, jobIds))
                 .stream()
                 .map(this::entity2Dto)
                 .collect(Collectors.toList());
@@ -524,8 +512,8 @@ public class WaylineJobServiceImpl implements IWaylineJobService {
                 .fileId(dto.getFileId())
                 .dockSn(dto.getDockSn())
                 .workspaceId(dto.getWorkspaceId())
-                .taskType(dto.getTaskType())
-                .waylineType(dto.getWaylineType())
+                .taskType(Optional.ofNullable(dto.getTaskType()).map(WaylineTaskTypeEnum::getVal).orElse(null))
+                .waylineType(Optional.ofNullable(dto.getWaylineType()).map(WaylineTemplateTypeEnum::getVal).orElse(null))
                 .username(dto.getUsername())
                 .rthAltitude(dto.getRthAltitude())
                 .outOfControlAction(dto.getOutOfControlAction())
@@ -645,8 +633,8 @@ public class WaylineJobServiceImpl implements IWaylineJobService {
                         LocalDateTime.ofInstant(Instant.ofEpochMilli(entity.getExecuteTime()), ZoneId.systemDefault()) : null)
                 .completedTime(WaylineJobStatusEnum.find(entity.getStatus()).getEnd() ?
                         LocalDateTime.ofInstant(Instant.ofEpochMilli(entity.getUpdateTime()), ZoneId.systemDefault()) : null)
-                .taskType(entity.getTaskType())
-                .waylineType(entity.getWaylineType())
+                .taskType(WaylineTaskTypeEnum.find(entity.getTaskType()))
+                .waylineType(WaylineTemplateTypeEnum.find(entity.getWaylineType()))
                 .rthAltitude(entity.getRthAltitude())
                 .outOfControlAction(entity.getOutOfControlAction())
                 .mediaCount(entity.getMediaCount());

src/main/java/com/dji/sample/wayline/service/impl/WaylineRedisServiceImpl.java

@@ -3,8 +3,8 @@ package com.dji.sample.wayline.service.impl;
 import com.dji.sample.component.mqtt.model.EventsReceiver;
 import com.dji.sample.component.redis.RedisConst;
 import com.dji.sample.component.redis.RedisOpsUtils;
-import com.dji.sample.wayline.model.dto.ConditionalWaylineJobKey;
 import com.dji.sample.wayline.model.dto.WaylineJobDTO;
+import com.dji.sample.wayline.model.dto.WaylineJobKey;
 import com.dji.sample.wayline.model.dto.WaylineTaskProgressReceiver;
 import com.dji.sample.wayline.service.IWaylineRedisService;
 import org.springframework.stereotype.Service;
@@ -64,13 +64,18 @@ public class WaylineRedisServiceImpl implements IWaylineRedisService {
         return (String) RedisOpsUtils.get(RedisConst.WAYLINE_JOB_BLOCK_PREFIX + dockSn);
     }
 
+    @Override
+    public Boolean delBlockedWaylineJobId(String dockSn) {
+        return RedisOpsUtils.del(RedisConst.WAYLINE_JOB_BLOCK_PREFIX + dockSn);
+    }
+
     @Override
     public void setConditionalWaylineJob(WaylineJobDTO waylineJob) {
         if (!StringUtils.hasText(waylineJob.getJobId())) {
             throw new RuntimeException("Job id can't be null.");
         }
         RedisOpsUtils.setWithExpire(RedisConst.WAYLINE_JOB_CONDITION_PREFIX + waylineJob.getJobId(), waylineJob,
-                (Duration.between(waylineJob.getEndTime(), LocalDateTime.now()).getSeconds()));
+                Math.abs(Duration.between(waylineJob.getEndTime(), LocalDateTime.now()).getSeconds()));
     }
 
     @Override
@@ -84,29 +89,29 @@ public class WaylineRedisServiceImpl implements IWaylineRedisService {
     }
 
     @Override
-    public Boolean addPrepareConditionalWaylineJob(WaylineJobDTO waylineJob) {
+    public Boolean addPreparedWaylineJob(WaylineJobDTO waylineJob) {
         if (Objects.isNull(waylineJob.getBeginTime())) {
             return false;
         }
         // value: {workspace_id}:{dock_sn}:{job_id}
-        return RedisOpsUtils.zAdd(RedisConst.WAYLINE_JOB_CONDITION_PREPARE,
+        return RedisOpsUtils.zAdd(RedisConst.WAYLINE_JOB_PREPARED,
                 waylineJob.getWorkspaceId() + RedisConst.DELIMITER + waylineJob.getDockSn() + RedisConst.DELIMITER + waylineJob.getJobId(),
                 waylineJob.getBeginTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
     }
 
     @Override
-    public Optional<ConditionalWaylineJobKey> getNearestConditionalWaylineJob() {
-        return Optional.ofNullable(RedisOpsUtils.zGetMin(RedisConst.WAYLINE_JOB_CONDITION_PREPARE))
-                .map(Object::toString).map(ConditionalWaylineJobKey::new);
+    public Optional<WaylineJobKey> getNearestPreparedWaylineJob() {
+        return Optional.ofNullable(RedisOpsUtils.zGetMin(RedisConst.WAYLINE_JOB_PREPARED))
+                .map(Object::toString).map(WaylineJobKey::new);
     }
 
     @Override
-    public Double getConditionalWaylineJobTime(ConditionalWaylineJobKey jobKey) {
-        return RedisOpsUtils.zScore(RedisConst.WAYLINE_JOB_CONDITION_PREPARE, jobKey.getKey());
+    public Double getPreparedWaylineJobTime(WaylineJobKey jobKey) {
+        return RedisOpsUtils.zScore(RedisConst.WAYLINE_JOB_PREPARED, jobKey.getKey());
     }
 
     @Override
-    public Boolean removePrepareConditionalWaylineJob(ConditionalWaylineJobKey jobKey) {
-        return RedisOpsUtils.zRemove(RedisConst.WAYLINE_JOB_CONDITION_PREPARE, jobKey.getKey());
+    public Boolean removePreparedWaylineJob(WaylineJobKey jobKey) {
+        return RedisOpsUtils.zRemove(RedisConst.WAYLINE_JOB_PREPARED, jobKey.getKey());
     }
 }