package com.ard.sample.wayline.service.impl;
|
|
import com.ard.common.utils.uuid.IdUtils;
|
import com.ard.sample.common.error.CommonErrorEnum;
|
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.component.websocket.enums.MessageType;
|
import com.ard.sample.component.websocket.handler.DockWebSocketHandler;
|
import com.ard.sample.component.websocket.model.BizCodeEnum;
|
import com.ard.sample.component.websocket.model.CustomWebSocketMessage;
|
import com.ard.sample.component.websocket.service.ISendMessageService;
|
import com.ard.sample.component.websocket.service.IWebSocketManageService;
|
import com.ard.sample.manage.model.dto.DeviceDTO;
|
import com.ard.sample.manage.model.enums.UserTypeEnum;
|
import com.ard.sample.media.model.MediaFileCountDTO;
|
import com.ard.sample.wayline.model.dto.WaylineJobDTO;
|
import com.ard.sample.wayline.model.dto.WaylineTaskProgressExt;
|
import com.ard.sample.wayline.model.dto.WaylineTaskProgressReceiver;
|
import com.ard.sample.wayline.model.enums.WaylineJobStatusEnum;
|
import com.ard.sample.wayline.service.IFlightTaskService;
|
import com.ard.sample.wayline.service.IWaylineJobService;
|
import com.fasterxml.jackson.core.type.TypeReference;
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
import lombok.extern.slf4j.Slf4j;
|
import org.apache.http.HttpStatus;
|
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.integration.annotation.ServiceActivator;
|
import org.springframework.integration.mqtt.support.MqttHeaders;
|
import org.springframework.messaging.MessageHeaders;
|
import org.springframework.scheduling.annotation.Scheduled;
|
import org.springframework.stereotype.Service;
|
|
import java.time.LocalDateTime;
|
import java.util.HashMap;
|
import java.util.Map;
|
import java.util.Objects;
|
import java.util.Optional;
|
import java.util.concurrent.TimeUnit;
|
|
/**
|
* @author sean
|
* @version 1.1
|
* @date 2022/6/9
|
*/
|
@Service
|
@Slf4j
|
public class FlightTaskServiceImpl implements IFlightTaskService {
|
|
@Autowired
|
private IMessageSenderService messageSender;
|
|
@Autowired
|
private ObjectMapper mapper;
|
|
@Autowired
|
private DockWebSocketHandler _dockWebSocketHandler;
|
|
@Autowired
|
private ISendMessageService websocketMessageService;
|
|
@Autowired
|
private IWebSocketManageService webSocketManageService;
|
|
@Autowired
|
private IWaylineJobService waylineJobService;
|
|
@Override
|
@ServiceActivator(inputChannel = ChannelName.INBOUND_EVENTS_FLIGHT_TASK_PROGRESS, outputChannel = ChannelName.OUTBOUND)
|
public void 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());
|
eventsReceiver.setSn(receiver.getGateway());
|
eventsReceiver.setGateway(receiver.getGateway());
|
eventsReceiver.setType(MessageType.STATUS.getType() + "-flighttask_progress");
|
eventsReceiver.setMethod(EventsMethodEnum.FLIGHT_TASK_PROGRESS.getMethod());
|
|
|
WaylineTaskProgressReceiver output = eventsReceiver.getOutput();
|
|
log.info("Task progress: {}", output.getProgress().toString());
|
|
if (eventsReceiver.getResult() != ResponseResult.CODE_SUCCESS) {
|
String errorMsg = CommonErrorEnum.getMsgByCode(eventsReceiver.getResult());
|
log.error("Task progress ===> Error code: {},Error Msg: {}", eventsReceiver.getResult(), errorMsg);
|
}
|
|
EventsResultStatusEnum statusEnum = EventsResultStatusEnum.find(output.getStatus());
|
String key = RedisConst.WAYLINE_JOB_RUNNING_PREFIX + dockSn;
|
RedisOpsUtils.setWithExpire(key, eventsReceiver, RedisConst.DEVICE_ALIVE_SECOND * RedisConst.DEVICE_ALIVE_SECOND);
|
|
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);
|
RedisOpsUtils.del(key);
|
RedisOpsUtils.del(RedisConst.WAYLINE_JOB_PAUSED_PREFIX + receiver.getBid());
|
}
|
|
DeviceDTO device = (DeviceDTO) RedisOpsUtils.get(RedisConst.DEVICE_ONLINE_PREFIX + receiver.getGateway());
|
websocketMessageService.sendBatch(
|
webSocketManageService.getValueAndUserType(UserTypeEnum.WEB.getVal()),
|
CustomWebSocketMessage.builder()
|
.data(eventsReceiver)
|
.timestamp(System.currentTimeMillis())
|
.bizCode(BizCodeEnum.FLIGHT_TASK_PROGRESS.getCode())
|
.build());
|
|
if (receiver.getNeedReply() != null && receiver.getNeedReply() == 1) {
|
messageSender.publish(receivedTopic + TopicConst._REPLY_SUF,
|
CommonTopicResponse.builder()
|
.tid(receiver.getTid())
|
.bid(receiver.getBid())
|
.method(EventsMethodEnum.FLIGHT_TASK_PROGRESS.getMethod())
|
.timestamp(System.currentTimeMillis())
|
.data(RequestsReply.success())
|
.build());
|
}
|
// 进度
|
// if (eventsReceiver.getOutput().getProgress().getPercent() >= 95) {
|
// return;
|
// }
|
String json = "";
|
try {
|
json = mapper.writeValueAsString(eventsReceiver);
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
_dockWebSocketHandler.sendMessageToSubscribers(dockSn, MessageType.STATUS.getType(), json, "");
|
}
|
|
/**
|
* 定时检查并处理航线定时任务
|
* <p>
|
* 该方法每5秒执行一次，从Redis有序集合中获取最早需要执行的任务，检查其执行时间：
|
* 1. 若任务已过期（超过当前时间30秒），则标记为失败并删除
|
* 2. 若任务在允许的时间窗口内（当前时间±30秒），则执行任务并删除
|
* </p>
|
*
|
* <p>任务ID格式：{workspace_id}:{dock_sn}:{job_id}</p>
|
*/
|
@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);
|
// 取到待执行的定时任务ID，获取任务信息websocket发给前端
|
// region 取定时任务信息websocket发给前端
|
// pending status状态待执行
|
EventsReceiver<WaylineTaskProgressReceiver> eventsReceiver = new EventsReceiver<>();
|
WaylineTaskProgressReceiver waylineTaskProgressReceiver = new WaylineTaskProgressReceiver();
|
WaylineTaskProgressExt waylineTaskProgressExt = new WaylineTaskProgressExt();
|
waylineTaskProgressExt.setFlightId(jobArr[1]);
|
waylineTaskProgressReceiver.setStatus("pending");
|
waylineTaskProgressReceiver.setExt(waylineTaskProgressExt);
|
eventsReceiver.setOutput(waylineTaskProgressReceiver);
|
eventsReceiver.setResult(ResponseResult.CODE_SUCCESS);
|
eventsReceiver.setBid(IdUtils.simpleUUID());
|
eventsReceiver.setSn(jobArr[0]);
|
eventsReceiver.setGateway(jobArr[0]);
|
eventsReceiver.setType(MessageType.STATUS.getType() + "-flighttask_progress");
|
eventsReceiver.setMethod(EventsMethodEnum.FLIGHT_TASK_PROGRESS.getMethod());
|
String json = "";
|
try {
|
json = mapper.writeValueAsString(eventsReceiver);
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
_dockWebSocketHandler.sendMessageToSubscribers(jobArr[0], MessageType.STATUS.getType(), json, "");
|
log.info("定时任务自定义状态为待执行，json: {}", json);
|
// endregion
|
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[1])
|
.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[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);
|
}
|
}
|
}
|
}
|
