forked from uav/ard-uav-server
ard-sample/src/main/java/com/ard/sample/media/service/impl/MediaServiceImpl.java
@@ -1,5 +1,6 @@ package com.ard.sample.media.service.impl; import com.ard.common.config.OssConfiguration; import com.ard.sample.common.error.CommonErrorEnum; import com.ard.sample.common.model.ResponseResult; import com.ard.sample.component.mqtt.model.*; @@ -10,6 +11,7 @@ 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.config.DockConfiguration; import com.ard.sample.manage.model.dto.DeviceDTO; import com.ard.sample.manage.model.enums.UserTypeEnum; import com.ard.sample.manage.service.IDeviceService; @@ -19,9 +21,12 @@ import com.ard.sample.media.model.MediaFileDTO; import com.ard.sample.media.service.IFileService; import com.ard.sample.media.service.IMediaService; import com.ard.sample.wayline.mapper.IWaylineJobHistoryMapper; import com.ard.sample.wayline.model.dto.WaylineJobDTO; import com.ard.sample.wayline.service.IWaylineJobHistoryService; import com.ard.sample.wayline.service.IWaylineJobService; import com.fasterxml.jackson.databind.ObjectMapper; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.mqtt.support.MqttHeaders; @@ -38,6 +43,7 @@ * @version 0.2 * @date 2021/12/9 */ @Slf4j @Service public class MediaServiceImpl implements IMediaService { @@ -90,11 +96,30 @@ } /** * 处理文件上传回调事件 * * 该方法接收文件上传回调消息，进行业务处理并返回响应。主要逻辑包括： * 1. 解析回调数据并构建响应对象 * 2. 检查回调结果，非成功状态直接返回 * 3. 验证Redis中存储的媒体文件计数信息，防止重复处理 * 4. 查询设备信息和作业信息，解析媒体文件 * 5. 发布处理结果到指定主题 * 6. 通知上传计数更新 * * @param receiver 包含文件上传回调数据的通用主题接收器对象 * 包含以下关键字段： * - data: 回调数据(JSON格式) * - gateway: 设备网关标识 * - tid: 事务ID * - bid: 业务ID */ @Override @ServiceActivator(inputChannel = ChannelName.INBOUND_EVENTS_FILE_UPLOAD_CALLBACK, outputChannel = ChannelName.OUTBOUND) public void handleFileUploadCallBack(CommonTopicReceiver receiver) { FileUploadCallback callback = objectMapper.convertValue(receiver.getData(), FileUploadCallback.class); log.info("媒体库回调-原始报文内容：{}",receiver); log.info("媒体库调试-收到文件上传回调：{}",callback); String topic = TopicConst.THING_MODEL_PRE + TopicConst.PRODUCT + receiver.getGateway() + TopicConst.EVENTS_SUF + TopicConst._REPLY_SUF; CommonTopicResponse<RequestsReply> data = CommonTopicResponse.<RequestsReply>builder() @@ -107,6 +132,7 @@ if (callback.getResult() != ResponseResult.CODE_SUCCESS) { messageSenderService.publish(topic, data); log.info("媒体库调试-回调失败：{}",callback); return; } @@ -123,9 +149,12 @@ Optional<WaylineJobDTO> jobOpt = waylineJobService.getJobByJobId(jobId); if (jobOpt.isPresent()) { boolean isSave = parseMediaFile(callback, jobOpt.get()); log.info("媒体库收到数据：{}",callback); if (!isSave) { data.setData(RequestsReply.error(CommonErrorEnum.ILLEGAL_ARGUMENT)); log.info("媒体库保存失败：{}",callback); } } messageSenderService.publish(topic, data); @@ -134,7 +163,7 @@ } /** * update the uploaded count and notify web side * 更新上传的计数并通知网页端 * @param mediaFileCount * @param receiver * @param jobId @@ -172,18 +201,46 @@ .build()); } /** * 解析媒体文件并设置相关属性 * * <p>该函数主要完成两个操作： * 1. 根据任务中的机库序列号设置拍摄媒体的无人机序列号 * 2. 从文件对象键中提取路径并设置到文件对象中 * 最后尝试保存文件并返回是否保存成功 * * @param callback 文件上传回调对象，包含待处理的文件信息 * @param job 航线任务数据传输对象，包含机库序列号等信息 * @return 返回true表示文件保存成功，false表示保存失败 */ private Boolean parseMediaFile(FileUploadCallback callback, WaylineJobDTO job) { // Set the drone sn that shoots the media // 设置拍摄媒体的无人机 Optional<DeviceDTO> dockDTO = deviceService.getDeviceBySn(job.getDockSn()); dockDTO.ifPresent(dock -> callback.getFile().getExt().setSn(dock.getChildDeviceSn())); dockDTO.ifPresent(dock -> callback.getFile().getExt().setPayloadModelKey(dockDTO.get().getDomain()+"-"+dockDTO.get().getType()+"-"+dockDTO.get().getSubType())); // set path String objectKey = callback.getFile().getObjectKey(); callback.getFile().setPath(objectKey.substring(objectKey.indexOf("/") + 1, objectKey.lastIndexOf("/"))); // 设定路径 String objectKey = callback.getFile().getObjectKey()!= null ? callback.getFile().getObjectKey() : ""; String path = String.format("%s/%s/%s", OssConfiguration.endpoint, DockConfiguration.bucket, objectKey); log.info("媒体库上传文件完整路path：{}",path); callback.getFile().setPath(path); // callback.getFile().setPath(objectKey.substring(objectKey.indexOf("/") + 1, objectKey.lastIndexOf("/"))); log.info("保存媒体库数据：{}",callback); return fileService.saveFile(callback.getFile()) > 0; } /** * 处理最高优先级上传飞行任务媒体事件 * * 该方法处理来自MQTT通道的最高优先级上传飞行任务媒体请求，主要功能包括： * 1. 验证请求数据有效性 * 2. 发送成功响应 * 3. 更新Redis中媒体文件计数信息 * 4. 通过WebSocket通知前端 * * @param receiver 包含请求数据的接收器对象，包含方法、bid、tid等信息 * @param headers 消息头信息，包含接收的MQTT主题等元数据 */ @Override @ServiceActivator(inputChannel = ChannelName.INBOUND_EVENTS_HIGHEST_PRIORITY_UPLOAD_FLIGHT_TASK_MEDIA, outputChannel = ChannelName.OUTBOUND) public void handleHighestPriorityUploadFlightTaskMedia(CommonTopicReceiver receiver, MessageHeaders headers) {
