大疆不同任务类型执行逻辑，上云API源码分析
大疆司空2中有不同的任务类型：立即任务、定时任务、条件任务。
最初我们实现时，选择的是用Quartz创建定时任务，调用API中executeFlightTask接口实现任务下发。
在功能实现之后，随着对API的深入了解，发现大疆API中有相关的任务下发逻辑。
当时只看了实现逻辑，因为活比较多，加上懒，就一直拖到现在。
本文是对源码的学习总结。
所有任务下发的逻辑都从publishFlightTask()方法开始。
com.dji.sample.wayline.service.impl.FlightTaskServiceImpl#publishFlightTask
立即任务
fillImmediateTime(param)方法，如果是 “立即任务（IMMEDIATE）”，就把任务的执行日期taskDays与执行时间段taskPeriods强制设置为当前服务器时间。
由于taskDays = [now]，taskPeriods = [ [now] ]，所以只会进入循环一次，并且beginTime = 当前毫秒时间点，endTime = beginTime。
创建一条飞行任务记录存到数据库中---- waylineJobOpt。
调用publishOneFlight()方法发布任务
[1]publishFlightTask
/**
* 发布飞行任务的核心入口方法。
* 功能：
* 1. 处理立即任务（IMMEDIATE），强制以服务器当前时间作为任务执行时间；
* 2. 根据用户提交的任务日期（taskDays）与时间段（taskPeriods），
* 生成多个 beginTime/endTime 的任务实例；
* 3. 为每个时间段创建一条 WaylineJob（航线任务记录）；
* 4. 若为条件任务，写入对应的触发条件；
* 5. 立即调用 publishOneFlightTask 将任务下发给飞行设备（Dock）；
* 6. 如果任何一次下发失败，则返回失败并终止流程；
* 7. 所有任务下发成功后返回成功响应。
*/
@Override
public HttpResultResponse publishFlightTask(CreateJobParam param, CustomClaim customClaim) throws SQLException {
fillImmediateTime(param);
//立即任务只会进入循环一次
for (Long taskDay : param.getTaskDays()) {
LocalDate date = LocalDate.ofInstant(Instant.ofEpochSecond(taskDay), ZoneId.systemDefault());
for (List<Long> taskPeriod : param.getTaskPeriods()) {
//立即任务的beginTime = endTime = 当前毫秒时间
long beginTime = LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(taskPeriod.get(0)), ZoneId.systemDefault()))
.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
long endTime = taskPeriod.size() > 1 ?
LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(taskPeriod.get(1)), ZoneId.systemDefault()))
.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() : beginTime;
//立即任务直接跳过这个判断
if (TaskTypeEnum.IMMEDIATE != param.getTaskType() && endTime < System.currentTimeMillis()) {
continue;
}
//创建一条飞行任务记录
Optional<WaylineJobDTO> waylineJobOpt = waylineJobService.createWaylineJob(param, customClaim.getWorkspaceId(), customClaim.getUsername(), beginTime, endTime);
if (waylineJobOpt.isEmpty()) {
throw new SQLException("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);
//发布任务
HttpResultResponse response = this.publishOneFlightTask(waylineJob);
if (HttpResultResponse.CODE_SUCCESS != response.getCode()) {
return response;
}
}
}
return HttpResultResponse.success();
}
/**
* 如果任务类型为 IMMEDIATE（立即任务），则忽略用户传入的任何日期和时间段，
* 将任务的执行日期（taskDays）与执行时间段（taskPeriods）强制设置为当前服务器时间。
* 立即任务必须由服务器当前时间触发，不允许由客户端自定义时间。
*/
private void fillImmediateTime(CreateJobParam param) {
if (TaskTypeEnum.IMMEDIATE != param.getTaskType()) {
return;
}
long now = System.currentTimeMillis() / 1000;
param.setTaskDays(List.of(now));
param.setTaskPeriods(List.of(List.of(now)));
}
deviceRedisService.checkDeviceOnline判断机场是否在线
调用prepareFlightTask方法，下发飞行准备指令
调用executeFlightTask方法，下发飞行任务执行指令
[2]publishOneFlight
public HttpResultResponse 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 HttpResultResponse.error("Failed to prepare job.");
}
//下发立即任务执行指令
// Issue an immediate task execution command.
if (TaskTypeEnum.IMMEDIATE == waylineJob.getTaskType()) {
if (!executeFlightTask(waylineJob.getWorkspaceId(), waylineJob.getJobId())) {
return HttpResultResponse.error("Failed to execute job.");
}
}
..............(省略部分代码)
return HttpResultResponse.success();
}
定时任务
遍历定时任务设置中每一天的每一个时间段
忽略已经过期的时间段
创建waylineJob并存到数据库中。
调用publishOneFlightTask()发布任务
[1]publishFlightTask
@Override
public HttpResultResponse publishFlightTask(CreateJobParam param, CustomClaim customClaim) throws SQLException {
//定时任务在这个方法中直接返回，不会进行任何处理
fillImmediateTime(param);
//遍历每一天每一个时间段
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 ?
LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(taskPeriod.get(1)), ZoneId.systemDefault()))
.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() : beginTime;
//忽略已经过期的时间段
if (TaskTypeEnum.IMMEDIATE != param.getTaskType() && endTime < System.currentTimeMillis()) {
continue;
}
//创建waylineJob
Optional<WaylineJobDTO> waylineJobOpt = waylineJobService.createWaylineJob(param, customClaim.getWorkspaceId(), customClaim.getUsername(), beginTime, endTime);
if (waylineJobOpt.isEmpty()) {
throw new SQLException("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);
//发布任务
HttpResultResponse response = this.publishOneFlightTask(waylineJob);
if (HttpResultResponse.CODE_SUCCESS != response.getCode()) {
return response;
}
}
}
return HttpResultResponse.success();
}
deviceRedisService.checkDeviceOnline检查机场是否在线。
调用prepareFlightTask方法，下发飞行准备指令.
把定时任务添加到Redis的一个Sorted Set（有序集合）里，用任务的开始时间作为排序依据。
[2]publishOneFlightTask
public HttpResultResponse 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 HttpResultResponse.error("Failed to prepare job.");
}
..........(省略部分代码)
//把定时任务添加到Redis有序集合里，用开始时间排序
if (TaskTypeEnum.TIMED == 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 HttpResultResponse.error("Failed to create scheduled job.");
}
}
return HttpResultResponse.success();
}
checkScheduledJob定时任务，每5s对 Redis 中的定时任务集合扫描一次。
获取最早的定时任务，拆解任务信息。
如果任务信息开始比现在早30s (offset)以上，认为任务过期，删除 Redis 队列中的任务，更新任务状态为失败。
如果任务执行时间在now + offset这个时间区间内，调用executeFlightTask下发执行任务命令，且无论成功或失败，都从 Redis 队列中删除任务信息。
[3]checkScheduledJob定时任务
@Scheduled(initialDelay = 10, fixedRate = 5, timeUnit = TimeUnit.SECONDS)
public 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;
//任务信息开始比现在早30s以上，认为任务过期，删除Redis队列中的任务，更新任务状态为失败。
// 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;
}
//判断任务执行时间在now + offset这个时间区间内，下发执行任务命令。
if (now <= time && time <= now + offset) {
try {
this.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);
}
}
}
条件任务
条件任务和定时任务的 1-3步相同
同上
同上
调用addConditions()方法,设置readyConditions(预备条件)、executableConditions(执行条件)，然后将任务存入Redis，
wayline_job_condition_prepare（zset） → 按开始时间排序
wayline_job_condition:{jobId} → 任务详情。
调用publishOneFlightTask()发布任务
[1]publishFlightTask
@Override
public HttpResultResponse publishFlightTask(CreateJobParam param, CustomClaim customClaim) throws SQLException {
fillImmediateTime(param);
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 ?
LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(taskPeriod.get(1)), ZoneId.systemDefault()))
.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() : beginTime;
if (TaskTypeEnum.IMMEDIATE != param.getTaskType() && endTime < System.currentTimeMillis()) {
continue;
}
Optional<WaylineJobDTO> waylineJobOpt = waylineJobService.createWaylineJob(param, customClaim.getWorkspaceId(), customClaim.getUsername(), beginTime, endTime);
if (waylineJobOpt.isEmpty()) {
throw new SQLException("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);
HttpResultResponse response = this.publishOneFlightTask(waylineJob);
if (HttpResultResponse.CODE_SUCCESS != response.getCode()) {
return response;
}
}
}
return HttpResultResponse.success();
}
private void addConditions(WaylineJobDTO waylineJob, CreateJobParam param, Long beginTime, Long endTime) {
if (TaskTypeEnum.CONDITIONAL != param.getTaskType()) {
return;
}
//填入readyConditions(预备条件)、executableConditions(执行条件)
waylineJob.setConditions(
WaylineTaskConditionDTO.builder()
.executableConditions(Objects.nonNull(param.getMinStorageCapacity()) ?
new ExecutableConditions().setStorageCapacity(param.getMinStorageCapacity()) : null)
.readyConditions(new ReadyConditions()
.setBatteryCapacity(param.getMinBatteryCapacity())
.setBeginTime(beginTime)
.setEndTime(endTime))
.build());
waylineRedisService.setConditionalWaylineJob(waylineJob);
// key: wayline_job_condition, value: {workspace_id}:{dock_sn}:{job_id}
boolean isAdd = waylineRedisService.addPrepareConditionalWaylineJob(waylineJob);
if (!isAdd) {
throw new RuntimeException("Failed to create conditional job.");
}
}
@Override
public Boolean addPrepareConditionalWaylineJob(WaylineJobDTO waylineJob) {
if (Objects.isNull(waylineJob.getBeginTime())) {
return false;
}
// value: {workspace_id}:{dock_sn}:{job_id}
return RedisOpsUtils.zAdd(RedisConst.WAYLINE_JOB_CONDITION_PREPARE,
waylineJob.getWorkspaceId() + RedisConst.DELIMITER + waylineJob.getDockSn() + RedisConst.DELIMITER + waylineJob.getJobId(),
waylineJob.getBeginTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
}
deviceRedisService.checkDeviceOnline检查机场是否在线。
调用prepareFlightTask方法，下发飞行准备指令。
[2]publishOneFlightTask
public HttpResultResponse 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 HttpResultResponse.error("Failed to prepare job.");
}
..........(省略部分代码)
return HttpResultResponse.success();
}
private Boolean prepareFlightTask(WaylineJobDTO waylineJob) throws SQLException {
// get wayline file
Optional<GetWaylineListResponse> 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().getId());
FlighttaskPrepareRequest flightTask = new FlighttaskPrepareRequest()
.setFlightId(waylineJob.getJobId())
.setExecuteTime(waylineJob.getBeginTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli())
.setTaskType(waylineJob.getTaskType())
.setWaylineType(waylineJob.getWaylineType())
.setRthAltitude(waylineJob.getRthAltitude())
.setOutOfControlAction(waylineJob.getOutOfControlAction())
.setExitWaylineWhenRcLost(ExitWaylineWhenRcLostEnum.EXECUTE_RC_LOST_ACTION)
.setFile(new FlighttaskFile()
.setUrl(url.toString())
.setFingerprint(waylineFile.get().getSign()));
if (TaskTypeEnum.CONDITIONAL == waylineJob.getTaskType()) {
if (Objects.isNull(waylineJob.getConditions())) {
throw new IllegalArgumentException();
}
flightTask.setReadyConditions(waylineJob.getConditions().getReadyConditions());
flightTask.setExecutableConditions(waylineJob.getConditions().getExecutableConditions());
}
TopicServicesResponse<ServicesReplyData> serviceReply = abstractWaylineService.flighttaskPrepare(
SDKManager.getDeviceSDK(waylineJob.getDockSn()), flightTask);
if (!serviceReply.getData().getResult().isSuccess()) {
log.info("Prepare task ====> Error code: {}", serviceReply.getData().getResult());
waylineJobService.updateJob(WaylineJobDTO.builder()
.workspaceId(waylineJob.getWorkspaceId())
.jobId(waylineJob.getJobId())
.executeTime(LocalDateTime.now())
.status(WaylineJobStatusEnum.FAILED.getVal())
.completedTime(LocalDateTime.now())
.code(serviceReply.getData().getResult().getCode()).build());
return false;
}
return true;
}
prepareConditionJob定时任务，每5s对 Redis 中的条件任务集合扫描一次。
获取最近的条件任务，以及任务的开始时间，判断如果当前时间 + 1天 < 任务开始时间，直接返回。【判断任务开始时间是否达到提前准备阈值,默认提前一天】
从Redis中获取任务完整信息，再次调用publishOneFlightTask()发布任务,移除Redis中的条件任务，成功就直接返回
【两次调用publishOneFlightTask()方法，个人认为是为了冗余安全设计，避免条件任务因定时方法间隔没收到任务而触发失败】
失败，删除Redis中WAYLINE_JOB_CONDITION_PREFIX+jobId对应的条件任务，判断任务是否已经过期，过期就直接放弃，未过期就调用 retryPrepareJob()生成子任务并重新加入准备队列。
[3]prepareConditionJob定时方法
@Scheduled(initialDelay = 10, fixedRate = 5, timeUnit = TimeUnit.SECONDS)
public void prepareConditionJob() {
//获取最近的条件任务，如果队列为空，不做任何事。
Optional<ConditionalWaylineJobKey> jobKeyOpt = waylineRedisService.getNearestConditionalWaylineJob();
if (jobKeyOpt.isEmpty()) {
return;
}
//获取任务开始时间，判断如果当前时间 + 1天 < 任务开始时间，直接返回。
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;
}
//初始化失败任务对象，用于后续异常处理或Redis数据丢失时任务状态的更新
WaylineJobDTO job = WaylineJobDTO.builder()
.jobId(jobKey.getJobId())
.status(WaylineJobStatusEnum.FAILED.getVal())
.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.getCode());
waylineJobService.updateJob(job);
waylineRedisService.removePrepareConditionalWaylineJob(jobKey);
return;
}
WaylineJobDTO waylineJob = waylineJobOpt.get();
//调用publishOneFlightTask()发布任务
HttpResultResponse result = this.publishOneFlightTask(waylineJob);
waylineRedisService.removePrepareConditionalWaylineJob(jobKey);
if (HttpResultResponse.CODE_SUCCESS == result.getCode()) {
return;
}
//删除Redis中对应的条件任务
// If the end time is exceeded, no more retries will be made.
waylineRedisService.delConditionalWaylineJob(jobKey.getJobId());
//判断任务是否已经过期，过期就直接返回，未过期就调用 retryPrepareJob()重试。
if (waylineJob.getEndTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() - RedisConst.WAYLINE_JOB_BLOCK_TIME * 1000 < now) {
return;
}
// Retry if the end time has not been exceeded.
this.retryPrepareJob(jobKey, waylineJob);
} catch (Exception e) {
log.info("Failed to prepare the conditional task.");
waylineJobService.updateJob(job);
}
}
下发条件任务后，设备会定频检查下发任务 API 中的 ready_conditions 是否全部满足，若全部满足则会有任务就绪通知（flighttask_ready）事件通知。
读取 flightId，判断任务是否被阻塞
调用 executeFlightTask() 执行飞行任务
判断执行结果：
成功：任务开始执行
失败但为阻塞错误：写入 blocked 状态
失败且可重试：创建子任务 → retryPrepareJob
[4]flighttaskReady (mqtt事件通知接收)
@Override
public TopicEventsResponse<MqttReply> flighttaskReady(TopicEventsRequest<FlighttaskReady> response, MessageHeaders headers) {
List<String> flightIds = response.getData().getFlightIds();
log.info("ready task list：{}", Arrays.toString(flightIds.toArray()) );
// Check conditional task blocking status.
String blockedId = waylineRedisService.getBlockedWaylineJobId(response.getGateway());
//这段代码根据论坛回复来看，存在bug，详情见参考资料【3】
//if (!StringUtils.hasText(blockedId)) {
// return null;
//}
Optional<DeviceDTO> deviceOpt = deviceRedisService.getDeviceOnline(response.getGateway());
if (deviceOpt.isEmpty()) {
return null;
}
DeviceDTO device = deviceOpt.get();
try {
for (String jobId : flightIds) {
boolean isExecute = this.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 new TopicEventsResponse<>();
}
WaylineJobDTO waylineJob = waylineJobOpt.get();
this.retryPrepareJob(new ConditionalWaylineJobKey(device.getWorkspaceId(), response.getGateway(), jobId), waylineJob);
return new TopicEventsResponse<>();
}
} catch (Exception e) {
log.error("Failed to execute conditional task.");
e.printStackTrace();
}
return new TopicEventsResponse<>();
}
@Override
public void retryPrepareJob(ConditionalWaylineJobKey jobKey, WaylineJobDTO waylineJob) {
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.addPrepareConditionalWaylineJob(newJob);
if (!isAdd) {
log.error("Failed to create wayline job. {}", newJob.getJobId());
return;
}
waylineJob.setJobId(newJob.getJobId());
waylineRedisService.setConditionalWaylineJob(waylineJob);
}
@Override
public Boolean executeFlightTask(String workspaceId, String jobId) {
// get job
Optional<WaylineJobDTO> waylineJob = waylineJobService.getJobByJobId(workspaceId, jobId);
if (waylineJob.isEmpty()) {
throw new IllegalArgumentException("Job doesn't exist.");
}
boolean isOnline = deviceRedisService.checkDeviceOnline(waylineJob.get().getDockSn());
if (!isOnline) {
throw new RuntimeException("Dock is offline.");
}
WaylineJobDTO job = waylineJob.get();
TopicServicesResponse<ServicesReplyData> serviceReply = abstractWaylineService.flighttaskExecute(
SDKManager.getDeviceSDK(job.getDockSn()), new FlighttaskExecuteRequest().setFlightId(jobId));
if (!serviceReply.getData().getResult().isSuccess()) {
log.info("Execute job ====> Error: {}", serviceReply.getData().getResult());
waylineJobService.updateJob(WaylineJobDTO.builder()
.jobId(jobId)
.executeTime(LocalDateTime.now())
.status(WaylineJobStatusEnum.FAILED.getVal())
.completedTime(LocalDateTime.now())
.code(serviceReply.getData().getResult().getCode()).build());
// The conditional task fails and enters the blocking status.
if (TaskTypeEnum.CONDITIONAL == job.getTaskType()
&& WaylineErrorCodeEnum.find(serviceReply.getData().getResult().getCode()).isBlock()) {
waylineRedisService.setBlockedWaylineJob(job.getDockSn(), jobId);
}
return false;
}
waylineJobService.updateJob(WaylineJobDTO.builder()
.jobId(jobId)
.executeTime(LocalDateTime.now())
.status(WaylineJobStatusEnum.IN_PROGRESS.getVal())
.build());
waylineRedisService.setRunningWaylineJob(job.getDockSn(), EventsReceiver.<FlighttaskProgress>builder().bid(jobId).sn(job.getDockSn()).build());
return true;
}
参考资料：
