/*
|
* 【机场1】从 dock2 原类 com.ard.sample.wayline.service.impl.WaylineJobServiceImpl 复制/独立适配（DJI Dock 1 / M30）。
|
* 与 dock2 原类同服务共存；本文件内的适配差异以「// 机场1：」内联注释标出。
|
*/
|
package com.ard.sample.dock1.wayline.service.impl;
|
|
import com.ard.common.config.OssConfiguration;
|
import com.ard.oss.service.impl.OssServiceContext;
|
import com.ard.route.domain.route.ArdRouteInfo;
|
import com.ard.route.service.IArdRouteInfoService;
|
import com.ard.sample.config.DockConfiguration;
|
import com.ard.sample.utils.TimestampUtils;
|
import com.baomidou.mybatisplus.core.conditions.Wrapper;
|
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.ard.common.core.domain.model.LoginUser;
|
import com.ard.sample.common.error.CommonErrorEnum;
|
import com.ard.sample.common.model.Pagination;
|
import com.ard.sample.common.model.PaginationData;
|
import com.ard.sample.common.model.ResponseResult;
|
import com.ard.sample.component.mqtt.model.*;
|
import com.ard.sample.component.mqtt.service.IMessageSenderService;
|
import com.ard.common.core.redis.RedisConst;
|
import com.ard.common.core.redis.RedisOpsUtils;
|
import com.ard.sample.manage.model.dto.DeviceDTO;
|
import com.ard.sample.manage.model.enums.DeviceDomainEnum;
|
import com.ard.sample.manage.model.param.DeviceQueryParam;
|
import com.ard.sample.manage.service.IDeviceService;
|
import com.ard.sample.media.model.MediaFileCountDTO;
|
import com.ard.sample.media.model.MediaMethodEnum;
|
import com.ard.sample.media.service.IFileService;
|
import com.ard.sample.wayline.mapper.IWaylineJobMapper;
|
import com.ard.sample.wayline.model.dto.*;
|
import com.ard.sample.wayline.model.entity.WaylineJobEntity;
|
import com.ard.sample.wayline.model.enums.WaylineJobStatusEnum;
|
import com.ard.sample.wayline.model.enums.WaylineMethodEnum;
|
import com.ard.sample.wayline.model.enums.WaylineTaskTypeEnum;
|
import com.ard.sample.wayline.model.param.CreateJobParam;
|
import com.ard.sample.dock1.wayline.service.IWaylineJobService;
|
import com.fasterxml.jackson.core.type.TypeReference;
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
import lombok.extern.slf4j.Slf4j;
|
import org.apache.commons.codec.digest.DigestUtils;
|
import org.apache.ibatis.annotations.Param;
|
import org.springframework.beans.factory.annotation.Autowired;
|
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 org.springframework.util.StringUtils;
|
|
import javax.annotation.Resource;
|
import java.io.IOException;
|
import java.io.InputStream;
|
import java.net.URL;
|
import java.sql.SQLException;
|
import java.time.Instant;
|
import java.time.LocalDateTime;
|
import java.time.ZoneId;
|
import java.util.*;
|
import java.util.stream.Collectors;
|
|
import static com.ard.route.util.ToolUtils.kmzPath;
|
import static com.ard.route.util.ToolUtils.subStringMinioUrlName;
|
import com.ard.sample.dock1.wayline.model.dto.Dock1WaylineTaskCreateDTO;
|
import com.ard.sample.wayline.service.impl.WaylineJobServiceImpl;
|
|
/**
|
* @author sean
|
* @version 1.1
|
* @date 2022/6/1
|
*/
|
@Service
|
@Transactional
|
@Slf4j
|
@SuppressWarnings("all")
|
public class Dock1WaylineJobServiceImpl implements IWaylineJobService {
|
|
/**
|
* 机场1 设备类型（DJI Dock 1）。
|
*/
|
private static final int DOCK1_DEVICE_TYPE = 1;
|
|
@Autowired
|
private IWaylineJobMapper mapper;
|
|
// @Autowired
|
// private IWaylineFileService waylineFileService;
|
@Autowired
|
private TimestampUtils _timestampUtils;
|
|
@Autowired
|
private IArdRouteInfoService _ardRouteInfoService;
|
|
@Resource
|
private OssServiceContext _ossService;
|
|
@Autowired
|
private IDeviceService deviceService;
|
|
@Autowired
|
private IMessageSenderService messageSender;
|
|
@Autowired
|
private ObjectMapper objectMapper;
|
|
@Autowired
|
private IFileService fileService;
|
|
/**
|
* 插入航路作业
|
*
|
* @param jobEntity 航路作业实体
|
* @return 航路作业DTO的Optional包装对象，如果插入失败则返回Optional.empty()
|
*/
|
private Optional<WaylineJobDTO> insertWaylineJob(WaylineJobEntity jobEntity) {
|
int id = mapper.insert(jobEntity);
|
if (id <= 0) {
|
return Optional.empty();
|
}
|
return Optional.ofNullable(this.entity2Dto(jobEntity));
|
}
|
|
/**
|
* 创建航线任务
|
*
|
* @param param 任务参数
|
* @param username 用户名
|
* @param beginTime 开始时间
|
* @param endTime 结束时间
|
* @return Optional<WaylineJobDTO> 返回创建的航线任务对象，若创建失败则返回Optional.empty()
|
*/
|
@Override
|
public Optional<WaylineJobDTO> createWaylineJob(CreateJobParam param, String username, Long beginTime, Long endTime) {
|
if (Objects.isNull(param)) {
|
return Optional.empty();
|
}
|
// Immediate tasks, allocating time on the backend.
|
WaylineJobEntity jobEntity = WaylineJobEntity.builder()
|
.name(param.getName())
|
.dockSn(param.getDockSn())
|
.fileId(param.getFileId())
|
.username(username)
|
.jobId(UUID.randomUUID().toString())
|
.beginTime(beginTime)
|
.endTime(endTime)
|
.status(WaylineJobStatusEnum.PENDING.getVal())
|
.taskType(param.getTaskType())
|
.waylineType(param.getWaylineType())
|
.outOfControlAction(param.getOutOfControlAction())
|
.rthAltitude(param.getRthAltitude())
|
.rthMode(param.getRth_mode())
|
.mediaCount(0)
|
.parentId(param.getParentId())
|
.loopSum(param.getLoopSum())
|
.build();
|
|
return insertWaylineJob(jobEntity);
|
}
|
|
/**
|
* 根据父作业ID创建新的作业
|
*
|
* @param parentId 父作业ID
|
* @return 新创建的作业的DTO对象，如果父作业不存在，则返回Optional.empty()
|
*/
|
@Override
|
public Optional<WaylineJobDTO> createWaylineJobByParent(String parentId) {
|
Optional<WaylineJobDTO> parentJobOpt = this.getJobByJobId(parentId);
|
if (parentJobOpt.isEmpty()) {
|
return Optional.empty();
|
}
|
WaylineJobEntity jobEntity = this.dto2Entity(parentJobOpt.get());
|
jobEntity.setJobId(UUID.randomUUID().toString());
|
jobEntity.setErrorCode(null);
|
jobEntity.setCompletedTime(null);
|
jobEntity.setExecuteTime(null);
|
jobEntity.setStatus(WaylineJobStatusEnum.PENDING.getVal());
|
jobEntity.setParentId(parentId);
|
|
return this.insertWaylineJob(jobEntity);
|
}
|
|
/**
|
* 发布飞行任务
|
*
|
* @param param 任务参数
|
* @param loginUser 登录用户
|
* @return ResponseResult 返回结果
|
* @throws SQLException SQL异常
|
*/
|
@Override
|
public ResponseResult publishFlightTask(CreateJobParam param, LoginUser loginUser) throws SQLException {
|
if (WaylineTaskTypeEnum.IMMEDIATE.getVal() == param.getTaskType()) {
|
param.setExecuteTime(System.currentTimeMillis());
|
}
|
if (param.getExecuteTime() < _timestampUtils.plusMinutesByMillis(System.currentTimeMillis(), 10) && param.getTaskType() == WaylineTaskTypeEnum.TIMED.getVal()) {
|
return ResponseResult.error(1234,"定时任务执行时间不能小于当前时间的后10分钟！");
|
}
|
// 机场1：共享 WaylineLoopTaskServiceImpl 在生成循环子任务时会固定下发 rth_mode=1，
|
// 该字段 Dock1（M30）的 flighttask_prepare 不支持。在共享调度器按机型分流之前，
|
// 机场1 不创建循环任务，避免向设备下发非法参数。
|
if (param.getTaskType() == WaylineTaskTypeEnum.LOOP.getVal()) {
|
return ResponseResult.error(314013003, "机场1暂不支持循环任务，请选择立即执行或定时任务！");
|
}
|
// 机场1：/dock1 接口只允许给 Dock1（domain=3, type=1）创建任务，
|
// 避免误把机场2/3 的 SN 传进来，导致任务落到错误链路。
|
if (!isDock1Device(param.getDockSn())) {
|
return ResponseResult.error(314013004, "该设备不属于机场1，请使用对应的机场接口操作！");
|
}
|
param.setRth_mode(1);
|
Optional<WaylineJobDTO> waylineJobOpt = this.createWaylineJob(param, loginUser.getUsername(),
|
param.getExecuteTime(), param.getExecuteTime());
|
|
if (waylineJobOpt.isEmpty()) {
|
return ResponseResult.error(314013000, "创建机场任务失败！");
|
}
|
|
WaylineJobDTO waylineJob = waylineJobOpt.get();
|
|
boolean isOnline = deviceService.checkDeviceOnline(waylineJob.getDockSn());
|
if (!isOnline) {
|
return ResponseResult.error(CommonErrorEnum.DEVICE_OFFLINE.getErrorMsg());
|
}
|
|
// get wayline file
|
// Optional<WaylineFileDTO> waylineFile = waylineFileService.getWaylineByWaylineId(waylineJob.getFileId());
|
// if (waylineFile.isEmpty()) {
|
// return ResponseResult.error("机场任务航线不存在！");
|
// }
|
// 换航线表
|
ArdRouteInfo ardRouteOpt = _ardRouteInfoService.selectArdRouteInfoById(waylineJob.getFileId());
|
if (ardRouteOpt == null || ardRouteOpt.getPath() == null || ardRouteOpt.getPath().isEmpty()) {
|
return ResponseResult.error(314013001, "机场任务航线不存在！");
|
}
|
|
// get file url
|
// URL url = waylineFileService.getObjectUrl(waylineFile.get().getWaylineId());
|
URL url = null;
|
String minioUrl = subStringMinioUrlName(ardRouteOpt.getPath());
|
String md5 = "";
|
String httpUrl = "";
|
try (InputStream inputStream = _ossService.getObject(kmzPath, minioUrl)) {
|
md5 = DigestUtils.md5Hex(inputStream);
|
url = new URL(ardRouteOpt.getPath());
|
log.info("机场任务航线文件url地址转换前：{}", url);
|
httpUrl = url.toString().replace(OssConfiguration.endpoint, DockConfiguration.endpoint);
|
log.info("机场任务航线文件url地址转换后：{}", httpUrl);
|
log.info("机场任务航线文件md5值：{}", md5);
|
} catch (Exception e) {
|
log.error("机场任务航线文件地址错误！{}", e.getMessage());
|
return ResponseResult.error(314013002, "机场任务航线文件地址错误！" + e.getMessage());
|
}
|
|
log.info("机场任务-rth_mode的值：{}", waylineJob.getRthMode());
|
|
Dock1WaylineTaskCreateDTO flightTask = Dock1WaylineTaskCreateDTO.builder()
|
.flightId(waylineJob.getJobId())
|
.executeTime(waylineJob.getBeginTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli())
|
.taskType(waylineJob.getTaskType())
|
// 机场1：C4 dock1（M30）无 wayline_precision_type，不下发
|
.rthAltitude(waylineJob.getRthAltitude())
|
.outOfControlAction(waylineJob.getOutOfControlAction())
|
// 机场1：C4 dock1（M30）无 rth_mode，不下发
|
.file(WaylineTaskFileDTO.builder()
|
// .url(url.toString().replace("https://192.168.10.103:9001", "http://192.168.10.103:9011"))
|
// .fingerprint(waylineFile.get().getSign())
|
// .replace("https://", "http://")
|
// .replace("9001", "9011")
|
.url(httpUrl)
|
.fingerprint(md5)
|
.build())
|
// .simulateMission(WaylineSimulateMissionDTO.builder()
|
// .isEnable(1)
|
// .latitude(46.55819013988211)
|
// .longitude(125.14652370145113)
|
// .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();
|
|
// 下发参数打印
|
try {
|
String json = objectMapper.writeValueAsString(response);
|
log.info("机场下发航线任务的参数response：{}", json);
|
} catch (Exception e) {
|
log.error("机场下发航线任务的参数response异常：{}", e.getMessage());
|
}
|
|
ServiceReply serviceReply = messageSender.publishWithReply(topic, response);
|
|
log.info("下发任务结果： {}", serviceReply.getResult());
|
|
if (ResponseResult.CODE_SUCCESS != serviceReply.getResult()) {
|
String errorMsg = CommonErrorEnum.getMsgByCode(serviceReply.getResult());
|
log.info("Prepare task ====> Error code: {},Error Msg:{}", serviceReply.getResult(), errorMsg);
|
this.updateJob(WaylineJobDTO.builder()
|
.jobId(waylineJob.getJobId())
|
.executeTime(LocalDateTime.now())
|
.status(WaylineJobStatusEnum.FAILED.getVal())
|
.completedTime(LocalDateTime.now())
|
.code(serviceReply.getResult()).build());
|
//return ResponseResult.error("Prepare task ====> Error code: " + serviceReply.getResult());
|
return ResponseResult.error(serviceReply.getResult(), errorMsg);
|
}
|
|
// 飞行命令先注释掉
|
// Issue an immediate task execution command.
|
if (WaylineTaskTypeEnum.IMMEDIATE.getVal() == waylineJob.getTaskType()) {
|
if (!executeFlightTask(waylineJob.getJobId())) {
|
return ResponseResult.error("无法执行作业。");
|
}
|
}
|
|
if (WaylineTaskTypeEnum.TIMED.getVal() == waylineJob.getTaskType()) {
|
boolean isAdd = RedisOpsUtils.zAdd(RedisConst.WAYLINE_JOB_TIMED_EXECUTE,
|
waylineJob.getDockSn() + RedisConst.DELIMITER + waylineJob.getJobId(),
|
waylineJob.getBeginTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
|
if (!isAdd) {
|
return ResponseResult.error("无法创建计划作业。");
|
}
|
}
|
|
return ResponseResult.success();
|
}
|
|
public ResponseResult publishFlightTaskPrepare(Optional<WaylineJobDTO> waylineJobOpt) throws SQLException {
|
|
if (waylineJobOpt.isEmpty()) {
|
return ResponseResult.error(314013000, "创建机场任务失败！");
|
}
|
WaylineJobDTO waylineJob = waylineJobOpt.get();
|
boolean isOnline = deviceService.checkDeviceOnline(waylineJob.getDockSn());
|
if (!isOnline) {
|
return ResponseResult.error("机场不在线！");
|
}
|
// 换航线表
|
ArdRouteInfo ardRouteOpt = _ardRouteInfoService.selectArdRouteInfoById(waylineJob.getFileId());
|
if (ardRouteOpt == null || ardRouteOpt.getPath() == null || ardRouteOpt.getPath().isEmpty()) {
|
return ResponseResult.error(314013001, "机场任务航线不存在！");
|
}
|
// get file url
|
URL url = null;
|
String minioUrl = subStringMinioUrlName(ardRouteOpt.getPath());
|
String md5 = "";
|
String httpUrl = "";
|
try (InputStream inputStream = _ossService.getObject(kmzPath, minioUrl)) {
|
md5 = DigestUtils.md5Hex(inputStream);
|
url = new URL(ardRouteOpt.getPath());
|
log.info("机场任务航线文件url地址转换前：{}", url);
|
httpUrl = url.toString().replace(OssConfiguration.endpoint, DockConfiguration.endpoint);
|
log.info("机场任务航线文件url地址转换后：{}", httpUrl);
|
log.info("机场任务航线文件md5值：{}", md5);
|
} catch (Exception e) {
|
log.error("机场任务航线文件地址错误！{}", e.getMessage());
|
return ResponseResult.error(314013002, "机场任务航线文件地址错误！" + e.getMessage());
|
}
|
|
Dock1WaylineTaskCreateDTO flightTask = Dock1WaylineTaskCreateDTO.builder()
|
.flightId(waylineJob.getJobId())
|
.executeTime(waylineJob.getBeginTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli())
|
.taskType(waylineJob.getTaskType())
|
// 机场1：C4 dock1（M30）无 wayline_precision_type，不下发
|
.rthAltitude(waylineJob.getRthAltitude())
|
.outOfControlAction(waylineJob.getOutOfControlAction())
|
// 机场1：C4 dock1（M30）无 rth_mode，不下发
|
.file(WaylineTaskFileDTO.builder()
|
.url(httpUrl)
|
// .fingerprint(waylineFile.get().getSign())
|
.fingerprint(md5)
|
.build())
|
// .simulateMission(WaylineSimulateMissionDTO.builder()
|
// .isEnable(1)
|
// .latitude(46.558831038)
|
// .longitude(125.148072319)
|
// .build())
|
.build();
|
|
// 下发参数打印
|
try {
|
String json = objectMapper.writeValueAsString(flightTask);
|
log.info("机场下发循环任务子任务-定时任务的航线任务的参数flightTask：{}", json);
|
} catch (Exception e) {
|
log.error("机场下发循环任务子任务-定时任务的航线任务的参数flightTask异常：{}", e.getMessage());
|
}
|
|
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);
|
|
log.info("下发循环任务子任务-定时任务的结果： {}", serviceReply.getResult());
|
log.info("机场循环任务子任务-定时任务的航线文件url地址：{}", httpUrl);
|
log.info("机场循环任务子任务-定时任务的航线文件md5值：{}", md5);
|
|
if (ResponseResult.CODE_SUCCESS != serviceReply.getResult()) {
|
String errorMsg = CommonErrorEnum.getMsgByCode(serviceReply.getResult());
|
log.info("Prepare task ====> Error code: {},Error Msg： {}", serviceReply.getResult(), errorMsg);
|
this.updateJob(WaylineJobDTO.builder()
|
.jobId(waylineJob.getJobId())
|
.executeTime(LocalDateTime.now())
|
.status(WaylineJobStatusEnum.FAILED.getVal())
|
.completedTime(LocalDateTime.now())
|
.code(serviceReply.getResult()).build());
|
// return ResponseResult.error("Prepare task ====> Error code: " + serviceReply.getResult());
|
return ResponseResult.error(serviceReply.getResult(), errorMsg);
|
}
|
|
// 飞行命令先注释掉
|
// Issue an immediate task execution command.
|
// if (WaylineTaskTypeEnum.IMMEDIATE.getVal() == waylineJob.getTaskType()) {
|
// if (!executeFlightTask(waylineJob.getJobId())) {
|
// return ResponseResult.error("无法执行作业。");
|
// }
|
// }
|
|
if (WaylineTaskTypeEnum.TIMED.getVal() == waylineJob.getTaskType()) {
|
boolean isAdd = RedisOpsUtils.zAdd(RedisConst.WAYLINE_JOB_TIMED_EXECUTE,
|
waylineJob.getDockSn() + RedisConst.DELIMITER + waylineJob.getJobId(),
|
waylineJob.getBeginTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
|
if (!isAdd) {
|
return ResponseResult.error("无法创建计划作业。");
|
}
|
}
|
|
return ResponseResult.success();
|
}
|
|
/**
|
* 执行飞行任务
|
*
|
* @param jobId 任务ID
|
* @return 如果任务执行成功返回true，否则返回false
|
* @throws IllegalArgumentException 如果任务不存在
|
* @throws RuntimeException 如果设备处于离线状态
|
*/
|
@Override
|
public Boolean executeFlightTask(String jobId) {
|
// get job
|
Optional<WaylineJobDTO> waylineJob = this.getJobByJobId(jobId);
|
if (waylineJob.isEmpty()) {
|
throw new IllegalArgumentException("计划任务不存在。");
|
}
|
|
boolean isOnline = deviceService.checkDeviceOnline(waylineJob.get().getDockSn());
|
if (!isOnline) {
|
// throw new RuntimeException("Dock is offline.");
|
throw new RuntimeException(CommonErrorEnum.DEVICE_OFFLINE.getErrorMsg());
|
}
|
|
WaylineJobDTO job = waylineJob.get();
|
Dock1WaylineTaskCreateDTO flightTask = Dock1WaylineTaskCreateDTO.builder().flightId(jobId).build();
|
// flightTask中，只有任务id是有值的，其他字段不返回
|
String topic = TopicConst.THING_MODEL_PRE + TopicConst.PRODUCT +
|
job.getDockSn() + TopicConst.SERVICES_SUF;
|
CommonTopicResponse<Object> response = CommonTopicResponse.builder()
|
.tid(UUID.randomUUID().toString())
|
.bid(jobId)
|
.timestamp(System.currentTimeMillis())
|
.data(flightTask)
|
.method(WaylineMethodEnum.FLIGHT_TASK_EXECUTE.getMethod())
|
.build();
|
|
// 下发参数打印
|
try {
|
String json = objectMapper.writeValueAsString(response);
|
log.info("机场下发-执行飞行任务-的参数response：{}", json);
|
} catch (Exception e) {
|
log.error("机场下发-执行飞行任务-的参数response异常：{}", e.getMessage());
|
}
|
|
ServiceReply serviceReply = messageSender.publishWithReply(topic, response);
|
|
log.info("下发-执行飞行任务-结果： {}", serviceReply.getResult());
|
|
if (ResponseResult.CODE_SUCCESS != serviceReply.getResult()) {
|
String errorMsg = CommonErrorEnum.getMsgByCode(serviceReply.getResult());
|
log.info("Execute job ====> Error code: {},Error Msg:{}", serviceReply.getResult(), errorMsg);
|
this.updateJob(WaylineJobDTO.builder()
|
.jobId(jobId)
|
.executeTime(LocalDateTime.now())
|
.status(WaylineJobStatusEnum.FAILED.getVal())
|
.completedTime(LocalDateTime.now())
|
.code(serviceReply.getResult()).build());
|
return false;
|
}
|
|
this.updateJob(WaylineJobDTO.builder()
|
.jobId(jobId)
|
.executeTime(LocalDateTime.now())
|
.status(WaylineJobStatusEnum.IN_PROGRESS.getVal())
|
.build());
|
RedisOpsUtils.setWithExpire(RedisConst.WAYLINE_JOB_RUNNING_PREFIX + job.getDockSn(),
|
EventsReceiver.<WaylineTaskProgressReceiver>builder().bid(jobId).sn(job.getDockSn()).build(),
|
RedisConst.DEVICE_ALIVE_SECOND * RedisConst.DEVICE_ALIVE_SECOND);
|
return true;
|
}
|
|
/**
|
* 取消飞行任务
|
*
|
* @param jobIds 需要取消的任务ID集合
|
* @throws IllegalArgumentException 如果任务状态不正确，抛出非法参数异常
|
*/
|
@Override
|
public ResponseResult cancelFlightTask(Collection<String> jobIds) {
|
if (CollectionUtils.isEmpty(jobIds)) {
|
return ResponseResult.error(98, "任务ID不能为空");
|
}
|
// 机场1：按机场分组，逐个下发取消，不能在第一组处理完后直接 return，
|
// 否则同一请求中包含多个机场（或多个 SN）时，后面的任务不会被取消。
|
Map<String, Set<String>> dockJobs = new LinkedHashMap<>();
|
for (String jobId : jobIds) {
|
Optional<WaylineJobDTO> jobOpt = this.getJobByJobId(jobId);
|
if (jobOpt.isEmpty()) {
|
return ResponseResult.error("任务不存在，jobId: " + jobId);
|
}
|
WaylineJobDTO job = jobOpt.get();
|
// 机场1：/dock1 接口只允许取消机场1（Dock1）的任务，避免通过 jobId 误取消机场2/3 的任务。
|
if (!isDock1Device(job.getDockSn())) {
|
return ResponseResult.error(314013004, "该任务不属于机场1，请使用对应的机场接口操作！");
|
}
|
// 循环主任务只是模板，真正的执行体是子任务：主任务本地置为已取消，执行中的子任务下发取消。
|
if (job.getTaskType() != null && job.getTaskType() == WaylineTaskTypeEnum.LOOP.getVal()) {
|
for (WaylineJobDTO child : this.getJobByJobParentId(jobId)) {
|
if (isJobEnded(child.getStatus())) {
|
continue;
|
}
|
if (Objects.equals(child.getStatus(), WaylineJobStatusEnum.IN_PROGRESS.getVal())) {
|
dockJobs.computeIfAbsent(child.getDockSn(), k -> new LinkedHashSet<>()).add(child.getJobId());
|
} else {
|
this.cancelJobLocally(child.getJobId(), child.getDockSn());
|
}
|
}
|
this.cancelJobLocally(jobId, job.getDockSn());
|
continue;
|
}
|
if (isJobEnded(job.getStatus())) {
|
continue;
|
}
|
dockJobs.computeIfAbsent(job.getDockSn(), k -> new LinkedHashSet<>()).add(jobId);
|
}
|
|
ResponseResult errorResult = null;
|
for (Map.Entry<String, Set<String>> entry : dockJobs.entrySet()) {
|
ResponseResult result = this.publishCancelTask(entry.getKey(), new ArrayList<>(entry.getValue()));
|
if (ResponseResult.CODE_SUCCESS != result.getCode() && errorResult == null) {
|
errorResult = result;
|
}
|
}
|
return errorResult == null ? ResponseResult.success() : errorResult;
|
}
|
|
/**
|
* 机场1：判断 SN 是否为 Dock1 机场设备（domain=3, type=1）。
|
*
|
* @param sn 设备序列号
|
* @return true 表示属于机场1
|
*/
|
private boolean isDock1Device(String sn) {
|
if (!StringUtils.hasText(sn)) {
|
return false;
|
}
|
Optional<DeviceDTO> deviceOpt = deviceService.getDeviceBySn(sn);
|
return deviceOpt.isPresent()
|
&& Objects.equals(DeviceDomainEnum.DOCK.getVal(), deviceOpt.get().getDomain())
|
&& Objects.equals(DOCK1_DEVICE_TYPE, deviceOpt.get().getType());
|
}
|
|
/**
|
* 机场1：任务是否已经处于终态（成功/取消/失败/未知）。
|
*/
|
private boolean isJobEnded(Integer status) {
|
if (status == null) {
|
return false;
|
}
|
return Boolean.TRUE.equals(WaylineJobStatusEnum.find(status).getEnd());
|
}
|
|
/**
|
* 机场1：只在本地把任务标记为已取消，并清理定时执行队列，不删除历史记录。
|
*/
|
private void cancelJobLocally(String jobId, String dockSn) {
|
this.updateJob(WaylineJobDTO.builder()
|
.jobId(jobId)
|
.status(WaylineJobStatusEnum.CANCEL.getVal())
|
.completedTime(LocalDateTime.now())
|
.build());
|
if (dockSn != null) {
|
RedisOpsUtils.zRemove(RedisConst.WAYLINE_JOB_TIMED_EXECUTE,
|
dockSn + RedisConst.DELIMITER + jobId);
|
}
|
}
|
|
// region 代码备份
|
// @Override
|
// public void cancelFlightTask(Collection<String> jobIds) {
|
// List<WaylineJobDTO> waylineJobs = getJobsByConditions(jobIds, WaylineJobStatusEnum.PENDING);
|
//
|
//// Set<String> waylineJobIds = waylineJobs.stream().map(WaylineJobDTO::getJobId).collect(Collectors.toSet());
|
//// // Check if the task status is correct.
|
//// boolean isErr = !jobIds.removeAll(waylineJobIds) || !jobIds.isEmpty();
|
//// if (isErr) {
|
//// throw new IllegalArgumentException("任务开始后，无法取消。" + Arrays.toString(jobIds.toArray()));
|
////// throw new IllegalArgumentException("These tasks have an incorrect status and cannot be canceled. "
|
////// + Arrays.toString(jobIds.toArray()));
|
//// }
|
//
|
// // Group job id by dock sn.
|
// Map<String, List<String>> dockJobs = waylineJobs.stream()
|
// .collect(Collectors.groupingBy(WaylineJobDTO::getDockSn,
|
// Collectors.mapping(WaylineJobDTO::getJobId, Collectors.toList())));
|
// dockJobs.forEach((dockSn, idList) -> this.publishCancelTask(dockSn, idList));
|
//
|
// }
|
|
//@Override
|
//public void publishCancelTask(String dockSn, List<String> jobIds) {
|
//
|
// boolean isOnline = deviceService.checkDeviceOnline(dockSn);
|
// if (!isOnline) {
|
// Integer errorCode = CommonErrorEnum.DEVICE_OFFLINE.getErrorCode();
|
//
|
// //return ResponseResult.error(errorCode, CommonErrorEnum.getMsgByCode(errorCode)));
|
// throw new RuntimeException(CommonErrorEnum.DEVICE_OFFLINE.getErrorMsg());
|
// }
|
// 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()) {
|
// String errorMsg = CommonErrorEnum.getMsgByCode(serviceReply.getResult());
|
// log.info("取消飞行任务失败 ====> 设备sn: {}, 错误码: {}， {}", dockSn, serviceReply.getResult(), errorMsg);
|
// //throw new RuntimeException("Failed to cancel the wayline job of " + dockSn);
|
// throw new RuntimeException("设备：" + dockSn + "取消飞行任务失败，错误码：" + serviceReply.getResult() + "，" + errorMsg);
|
// }
|
//
|
// for (String jobId : jobIds) {
|
// this.updateJob(WaylineJobDTO.builder()
|
// .jobId(jobId)
|
// .status(WaylineJobStatusEnum.CANCEL.getVal())
|
// .completedTime(LocalDateTime.now())
|
// .build());
|
// RedisOpsUtils.zRemove(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, RedisConst.DELIMITER + dockSn + RedisConst.DELIMITER + jobId);
|
// // 原流程是修改表中数据的status状态值，现在的精力是直接删除
|
// this.deleteTask(jobId);
|
// }
|
//
|
//}
|
// endregion
|
|
/**
|
* 发布取消任务
|
*
|
* @param dockSn 设备序列号
|
* @param jobIds 任务ID列表
|
* @throws RuntimeException 如果设备离线或取消任务失败，则抛出运行时异常
|
*/
|
@Override
|
public ResponseResult publishCancelTask(String dockSn, List<String> jobIds) {
|
if (CollectionUtils.isEmpty(jobIds)) {
|
return ResponseResult.success();
|
}
|
for (String jobId : jobIds) {
|
Optional<WaylineJobDTO> waylineJobDTO = this.getJobByJobId(jobId);
|
if (waylineJobDTO.isEmpty()) {
|
return ResponseResult.error("任务不存在，jobId: " + jobId);
|
}
|
}
|
boolean isOnline = deviceService.checkDeviceOnline(dockSn);
|
if (!isOnline) {
|
Integer errorCode = CommonErrorEnum.DEVICE_OFFLINE.getErrorCode();
|
return ResponseResult.error(errorCode, CommonErrorEnum.getMsgByCode(errorCode));
|
}
|
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()) {
|
String errorMsg = CommonErrorEnum.getMsgByCode(serviceReply.getResult());
|
log.info("取消飞行任务失败 ====> 设备sn: {}, 错误码: {}， {}", dockSn, serviceReply.getResult(), errorMsg);
|
return ResponseResult.error(serviceReply.getResult(), errorMsg);
|
}
|
|
// 机场1：取消成功后只更新状态并清理定时队列，保留历史记录，避免“取消”等同于“删除”。
|
for (String jobId : jobIds) {
|
this.cancelJobLocally(jobId, dockSn);
|
}
|
return ResponseResult.success();
|
}
|
|
@Override
|
public void deleteTask(String jobid) {
|
mapper.delete(new LambdaQueryWrapper<WaylineJobEntity>()
|
.eq(WaylineJobEntity::getJobId, jobid));
|
}
|
|
/**
|
* 根据条件查询作业列表
|
*
|
* @param jobIds 作业ID集合
|
* @param status 作业状态枚举
|
* @return 满足条件的作业DTO列表
|
*/
|
@Override
|
public List<WaylineJobDTO> getJobsByConditions(Collection<String> jobIds, WaylineJobStatusEnum status) {
|
return mapper.selectList(
|
new LambdaQueryWrapper<WaylineJobEntity>()
|
// 机场1：原实现误用 ne（取的是“非该状态”的任务），修正为 eq。
|
.eq(Objects.nonNull(status), WaylineJobEntity::getStatus, status.getVal())
|
.and(!CollectionUtils.isEmpty(jobIds),
|
wrapper -> jobIds.forEach(id -> wrapper.eq(WaylineJobEntity::getJobId, id).or())))
|
.stream()
|
.map(this::entity2Dto)
|
.collect(Collectors.toList());
|
}
|
|
/**
|
* 根据任务ID获取任务详情
|
*
|
* @param jobId 任务ID
|
* @return Optional<WaylineJobDTO> 包含任务详情的Optional对象，如果不存在则返回空的Optional
|
*/
|
@Override
|
public Optional<WaylineJobDTO> getJobByJobId(String jobId) {
|
WaylineJobEntity jobEntity = mapper.selectOne(
|
new LambdaQueryWrapper<WaylineJobEntity>()
|
.eq(WaylineJobEntity::getJobId, jobId));
|
return Optional.ofNullable(entity2Dto(jobEntity));
|
}
|
|
public List<WaylineJobDTO> getJobByJobParentId(String parentId) {
|
List<WaylineJobEntity> jobEntities = mapper.selectList(
|
new LambdaQueryWrapper<WaylineJobEntity>()
|
.eq(WaylineJobEntity::getParentId, parentId));
|
if (CollectionUtils.isEmpty(jobEntities)) {
|
return Collections.emptyList(); // 返回不可变空列表，更安全
|
}
|
return jobEntities.stream()
|
.map(this::entity2Dto) // 调用原有的单个转换方法
|
.collect(Collectors.toList());
|
}
|
|
/**
|
* 更新作业信息
|
*
|
* @param dto 作业信息传输对象
|
* @return 更新成功返回true，否则返回false
|
*/
|
@Override
|
public Boolean updateJob(WaylineJobDTO dto) {
|
return mapper.update(this.dto2Entity(dto),
|
new LambdaUpdateWrapper<WaylineJobEntity>()
|
.eq(WaylineJobEntity::getJobId, dto.getJobId())) > 0;
|
}
|
|
/**
|
* 根据工作区ID获取工作线任务分页数据
|
*
|
* @param page 页码，从1开始
|
* @param pageSize 每页显示条数
|
* @return PaginationData 包含工作线任务分页数据的对象
|
*/
|
@Override
|
public PaginationData<WaylineJobDTO> getJobsByWorkspaceId(long page, long pageSize) {
|
Page<WaylineJobEntity> pageData = mapper.selectPage(
|
new Page<WaylineJobEntity>(page, pageSize),
|
new LambdaQueryWrapper<WaylineJobEntity>()
|
.orderByDesc(WaylineJobEntity::getId));
|
List<WaylineJobDTO> records = pageData.getRecords()
|
.stream()
|
.map(this::entity2Dto)
|
.collect(Collectors.toList());
|
|
return new PaginationData<WaylineJobDTO>(records, new Pagination(pageData));
|
}
|
|
/**
|
* 根据条件筛选获取任务分页数据
|
*
|
* @param page 页码，从1开始
|
* @param pageSize 每页显示条数
|
* @return PaginationData 包含工作线任务分页数据的对象
|
*/
|
@Override
|
public PaginationData<WaylineJobDTO> getJobsList(long page, long pageSize, Integer status,
|
Long startTime, Long endTime, Integer taskType, String name, Boolean isComplete) {
|
// 机场1：只查询机场1（Dock1）设备下的任务，避免 /dock1 接口返回机场2/3 的任务。
|
List<String> dock1SnList = deviceService.getDevicesByParams(DeviceQueryParam.builder()
|
.deviceType(DOCK1_DEVICE_TYPE)
|
.domains(Collections.singletonList(DeviceDomainEnum.DOCK.getVal()))
|
.build())
|
.stream()
|
.map(DeviceDTO::getDeviceSn)
|
.collect(Collectors.toList());
|
if (CollectionUtils.isEmpty(dock1SnList)) {
|
Page<WaylineJobEntity> emptyPage = new Page<>(page, pageSize);
|
return new PaginationData<>(Collections.emptyList(), new Pagination(emptyPage));
|
}
|
Page<WaylineJobEntity> pageData = mapper.selectPage(
|
new Page<WaylineJobEntity>(page, pageSize),
|
new LambdaQueryWrapper<WaylineJobEntity>()
|
.in(WaylineJobEntity::getDockSn, dock1SnList)
|
.isNull(WaylineJobEntity::getParentId)
|
// 实际开始时间筛选，允许为空，为空则不进行筛选
|
.ge(startTime != null, WaylineJobEntity::getBeginTime, startTime)
|
.le(endTime != null, WaylineJobEntity::getBeginTime, endTime)
|
// 任务类型筛选，允许为空，为空则不进行筛选
|
.eq(taskType != null, WaylineJobEntity::getTaskType, taskType)
|
// 任务名称筛选，允许为空，为空则不进行筛选;....模糊查询
|
.like(name != null, WaylineJobEntity::getName, "%" + name + "%")
|
// 根据是否完成筛选，允许为空，为空则不进行筛选
|
.eq(isComplete != null && isComplete, WaylineJobEntity::getStatus, 3)
|
.ne(isComplete != null && !isComplete, WaylineJobEntity::getStatus, 3)
|
// 任务状态筛选，允许为空，为空则不进行筛选
|
.eq(status != null, WaylineJobEntity::getStatus, status)
|
.orderByDesc(WaylineJobEntity::getBeginTime));
|
List<WaylineJobDTO> records = pageData.getRecords()
|
.stream()
|
.map(this::entity2Dto)
|
.peek(job -> {
|
List<WaylineJobDTO> childOpt = this.getJobByJobParentId(job.getJobId());
|
job.setChildren(childOpt);
|
})
|
.collect(Collectors.toList());
|
|
return new PaginationData<WaylineJobDTO>(records, new Pagination(pageData));
|
}
|
|
public List<WaylineJobEntity> selectList(Long taskType, Long executeTime, String parentId) {
|
return mapper.selectList(new LambdaQueryWrapper<WaylineJobEntity>()
|
.eq(taskType != null, WaylineJobEntity::getTaskType, taskType)
|
.ne(WaylineJobEntity::getStatus, WaylineJobStatusEnum.SUCCESS.getVal())
|
.eq(executeTime != null, WaylineJobEntity::getBeginTime, executeTime)
|
.eq(parentId != null, WaylineJobEntity::getParentId, parentId)
|
.ne(WaylineJobEntity::getStatus, 3)
|
.orderByDesc(WaylineJobEntity::getId));
|
}
|
|
/**
|
* 获取所有待定状态的定时任务，未执行即将要执行的。
|
*/
|
public List<WaylineJobEntity> selectScheduledTaskList() {
|
// "1":"定时任务"
|
return mapper.selectList(new LambdaQueryWrapper<WaylineJobEntity>()
|
.eq(WaylineJobEntity::getTaskType, 1)
|
.eq(WaylineJobEntity::getStatus, WaylineJobStatusEnum.PENDING.getVal())
|
.gt(WaylineJobEntity::getBeginTime, System.currentTimeMillis())
|
.orderByDesc(WaylineJobEntity::getBeginTime));
|
}
|
|
|
/**
|
* 根据航班任务ID获取航班任务资源
|
*
|
* @param receiver 接收到的消息对象
|
* @param headers 消息头信息
|
* @Override 覆盖父类方法
|
* @ServiceActivator(inputChannel = ChannelName.INBOUND_REQUESTS_FLIGHT_TASK_RESOURCE_GET, outputChannel = ChannelName.OUTBOUND)
|
* 指定输入通道和输出通道
|
* @Transactional(isolation = Isolation.READ_UNCOMMITTED)
|
* 设置事务隔离级别为未提交读
|
*/
|
@Override
|
// 机场1：不复制原 ServiceActivator(inputChannel = ChannelName.INBOUND_REQUESTS_FLIGHT_TASK_RESOURCE_GET)，消费职责由共享层原类承担
|
@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);
|
|
log.info("机场通过flightTaskResourceGet获取航线文件地址，jobId: " + jobId);
|
|
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;
|
|
DeviceDTO device = (DeviceDTO) RedisOpsUtils.get(RedisConst.DEVICE_ONLINE_PREFIX + receiver.getGateway());
|
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.getFileId());
|
// if (waylineFile.isEmpty()) {
|
// builder.data(RequestsReply.error(CommonErrorEnum.ILLEGAL_ARGUMENT));
|
// messageSender.publish(topic, builder.build());
|
// return;
|
// }
|
// 修改航线文件名称，从ard_route_info航线表中取;
|
// 换航线表
|
ArdRouteInfo ardRouteOpt = _ardRouteInfoService.selectArdRouteInfoById(waylineJob.getFileId());
|
if (ardRouteOpt == null) {
|
builder.data(RequestsReply.error(CommonErrorEnum.ILLEGAL_ARGUMENT));
|
messageSender.publish(topic, builder.build());
|
return;
|
}
|
|
// get file url
|
URL url = null;
|
String minioUrl = subStringMinioUrlName(ardRouteOpt.getPath());
|
String md5 = "";
|
String httpUrl = "";
|
try (InputStream inputStream = _ossService.getObject(kmzPath, minioUrl)) {
|
md5 = DigestUtils.md5Hex(inputStream);
|
url = new URL(ardRouteOpt.getPath());
|
|
log.info("机场通过flightTaskResourceGet获取航线文件地址转换前：{}", url);
|
httpUrl = url.toString().replace(OssConfiguration.endpoint, DockConfiguration.endpoint);
|
log.info("机场通过flightTaskResourceGet获取航线文件地址转换后：{}", httpUrl);
|
|
log.info("机场通过flightTaskResourceGet获取航线文件地址，jobId: " + jobId + " 返回信息为url：" + httpUrl + " md5：" + md5);
|
// }
|
// try {
|
// url = waylineFileService.getObjectUrl(waylineFile.get().getWaylineId());
|
// url = new URL(ardRouteOpt.getUpPath());
|
|
// Map<String, Object> jsonResourceGet = new HashMap<>();
|
// jsonResourceGet.put("url", url.toString().replace("https//192.168.10.103:9001", "http//192.168.10.103:9002"));
|
// jsonResourceGet.put("fingerprint", md5);
|
// .replace("https://", "http://")
|
// .replace("9001", "9011")
|
|
builder.data(RequestsReply.success(Dock1WaylineTaskCreateDTO.builder()
|
.file(WaylineTaskFileDTO.builder()
|
.url(httpUrl)
|
.fingerprint(md5)
|
.build())
|
.build()));
|
|
} catch (NullPointerException | IOException e) {
|
e.printStackTrace();
|
builder.data(RequestsReply.error(CommonErrorEnum.ILLEGAL_ARGUMENT));
|
messageSender.publish(topic, builder.build());
|
return;
|
}
|
|
// 下发参数打印
|
try {
|
String json = objectMapper.writeValueAsString(builder.build());
|
log.info("机场通过flighttask_resource_get获取航线文件的参数flightTask：{}", json);
|
} catch (Exception e) {
|
log.error("机场通过flighttask_resource_get获取航线文件的参数flightTask异常：{}", e.getMessage());
|
}
|
|
messageSender.publish(topic, builder.build());
|
|
}
|
|
/**
|
* 上传媒体文件的最高优先级
|
*
|
* @param jobId 作业ID
|
* @throws RuntimeException 如果作业ID不合法或设置优先级失败
|
*/
|
@Override
|
public void uploadMediaHighestPriority(String jobId) {
|
Optional<WaylineJobDTO> jobOpt = getJobByJobId(jobId);
|
if (jobOpt.isEmpty()) {
|
throw new RuntimeException(CommonErrorEnum.ILLEGAL_ARGUMENT.getErrorMsg());
|
}
|
// 机场1：只允许对 Dock1 的任务设置媒体上传优先级。
|
if (!isDock1Device(jobOpt.get().getDockSn())) {
|
throw new RuntimeException("该任务不属于机场1，请使用对应的机场接口操作！");
|
}
|
|
String dockSn = jobOpt.get().getDockSn();
|
String key = RedisConst.MEDIA_HIGHEST_PRIORITY_PREFIX + dockSn;
|
if (RedisOpsUtils.checkExist(key) &&
|
jobId.equals(((MediaFileCountDTO) RedisOpsUtils.get(key)).getJobId())) {
|
return;
|
}
|
|
ServiceReply reply = messageSender.publishWithReply(TopicConst.THING_MODEL_PRE + TopicConst.PRODUCT + dockSn + TopicConst.SERVICES_SUF,
|
CommonTopicResponse.builder()
|
.tid(UUID.randomUUID().toString())
|
.bid(UUID.randomUUID().toString())
|
.timestamp(System.currentTimeMillis())
|
.method(MediaMethodEnum.UPLOAD_FLIGHT_TASK_MEDIA_PRIORITIZE.getMethod())
|
.data(Map.of(MapKeyConst.FLIGHT_ID, jobId))
|
.build());
|
if (ResponseResult.CODE_SUCCESS != reply.getResult()) {
|
String errorMsg = CommonErrorEnum.getMsgByCode(reply.getResult());
|
throw new RuntimeException("未能设置媒体作业上传优先级. Error Code: " + reply.getResult() + ", Error Msg: " + errorMsg);
|
}
|
}
|
|
/**
|
* 将WaylineJobDTO对象转换为WaylineJobEntity对象
|
*
|
* @param dto WaylineJobDTO对象
|
* @return WaylineJobEntity对象
|
*/
|
private WaylineJobEntity dto2Entity(WaylineJobDTO dto) {
|
WaylineJobEntity.WaylineJobEntityBuilder builder = WaylineJobEntity.builder();
|
if (dto == null) {
|
return builder.build();
|
}
|
if (Objects.nonNull(dto.getBeginTime())) {
|
builder.beginTime(dto.getBeginTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
|
}
|
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());
|
}
|
if (Objects.nonNull(dto.getCompletedTime())) {
|
builder.completedTime(dto.getCompletedTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
|
}
|
return builder.status(dto.getStatus())
|
.mediaCount(dto.getMediaCount())
|
.name(dto.getJobName())
|
.errorCode(dto.getCode())
|
.jobId(dto.getJobId())
|
.fileId(dto.getFileId())
|
.dockSn(dto.getDockSn())
|
.taskType(dto.getTaskType())
|
.waylineType(dto.getWaylineType())
|
.username(dto.getUsername())
|
.rthAltitude(dto.getRthAltitude())
|
.outOfControlAction(dto.getOutOfControlAction())
|
.parentId(dto.getParentId())
|
.loopSum(dto.getLoopSum())
|
.build();
|
}
|
|
/**
|
* 将实体对象转换为数据传输对象
|
*
|
* @param entity WaylineJobEntity 实体对象
|
* @return WaylineJobDTO 数据传输对象
|
*/
|
private WaylineJobDTO entity2Dto(WaylineJobEntity entity) {
|
if (entity == null) {
|
return null;
|
}
|
// 修改航线文件名称，从ard_route_info航线表中取;
|
String fileId = entity.getFileId();
|
String fileName = "";
|
try {
|
if (fileId != null) {
|
ArdRouteInfo ardRouteInfo = _ardRouteInfoService.selectArdRouteInfoById(fileId);
|
if (ardRouteInfo != null) {
|
fileName = Optional.ofNullable(ardRouteInfo.getName()).orElse("");
|
}
|
}
|
} catch (Exception e) {
|
log.info("获取航线名称失败，fileId:{},异常原因：{}", fileId, e.getMessage());
|
fileName = "";
|
}
|
WaylineJobDTO.WaylineJobDTOBuilder builder = WaylineJobDTO.builder()
|
.jobId(entity.getJobId())
|
.jobName(entity.getName())
|
.fileId(entity.getFileId())
|
// .fileName(waylineFileService.getWaylineByWaylineId(entity.getFileId())
|
// .orElse(WaylineFileDTO.builder().build()).getName())
|
.fileName(fileName)
|
.dockSn(entity.getDockSn())
|
.dockName(deviceService.getDeviceBySn(entity.getDockSn())
|
.orElse(DeviceDTO.builder().build()).getNickname())
|
.username(entity.getUsername())
|
.status(WaylineJobStatusEnum.IN_PROGRESS.getVal() == entity.getStatus() &&
|
RedisOpsUtils.checkExist(RedisConst.WAYLINE_JOB_PAUSED_PREFIX + entity.getJobId()) ?
|
WaylineJobStatusEnum.PAUSED.getVal() : entity.getStatus())
|
.code(entity.getErrorCode())
|
.beginTime(LocalDateTime.ofInstant(Instant.ofEpochMilli(entity.getBeginTime()), ZoneId.systemDefault()))
|
.endTime(Objects.nonNull(entity.getEndTime()) ?
|
LocalDateTime.ofInstant(Instant.ofEpochMilli(entity.getEndTime()), ZoneId.systemDefault()) : null)
|
.executeTime(Objects.nonNull(entity.getExecuteTime()) ?
|
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())
|
.rthAltitude(entity.getRthAltitude())
|
.rthMode(entity.getRthMode())
|
.outOfControlAction(entity.getOutOfControlAction())
|
.mediaCount(entity.getMediaCount())
|
.loopSum(entity.getLoopSum());
|
|
|
if (Objects.nonNull(entity.getEndTime())) {
|
builder.endTime(LocalDateTime.ofInstant(Instant.ofEpochMilli(entity.getEndTime()), ZoneId.systemDefault()));
|
}
|
if (WaylineJobStatusEnum.IN_PROGRESS.getVal() == entity.getStatus() && RedisOpsUtils.getExpire(entity.getJobId()) > 0) {
|
EventsReceiver<WaylineTaskProgressReceiver> taskProgress = (EventsReceiver<WaylineTaskProgressReceiver>) RedisOpsUtils.get(RedisConst.WAYLINE_JOB_RUNNING_PREFIX + entity.getDockSn());
|
if (Objects.nonNull(taskProgress.getOutput()) && Objects.nonNull(taskProgress.getOutput().getProgress())) {
|
builder.progress(taskProgress.getOutput().getProgress().getPercent());
|
}
|
}
|
|
if (entity.getMediaCount() == 0) {
|
return builder.build();
|
}
|
|
// sync the number of media files
|
String key = RedisConst.MEDIA_HIGHEST_PRIORITY_PREFIX + entity.getDockSn();
|
String countKey = RedisConst.MEDIA_FILE_PREFIX + entity.getDockSn();
|
Object mediaFileCount = RedisOpsUtils.hashGet(countKey, entity.getJobId());
|
if (Objects.nonNull(mediaFileCount)) {
|
builder.uploadedCount(((MediaFileCountDTO) mediaFileCount).getUploadedCount())
|
.uploading(RedisOpsUtils.checkExist(key) && entity.getJobId().equals(((MediaFileCountDTO) RedisOpsUtils.get(key)).getJobId()));
|
return builder.build();
|
}
|
|
int uploadedSize = fileService.getFilesByWorkspaceAndJobId(entity.getJobId()).size();
|
// All media for this job have been uploaded.
|
if (uploadedSize >= entity.getMediaCount()) {
|
return builder.uploadedCount(uploadedSize).build();
|
}
|
RedisOpsUtils.hashSet(countKey, entity.getJobId(),
|
MediaFileCountDTO.builder()
|
.jobId(entity.getJobId())
|
.mediaCount(entity.getMediaCount())
|
.uploadedCount(uploadedSize).build());
|
return builder.build();
|
}
|
}
|
