package com.ard.sample.manage.service.impl; import com.ard.sample.common.error.CommonErrorEnum; import com.ard.sample.common.error.LiveErrorEnum; import com.ard.sample.common.model.ResponseResult; import com.ard.sample.component.mqtt.model.CommonTopicResponse; import com.ard.sample.component.mqtt.model.ServiceReply; 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.*; import com.ard.sample.manage.model.enums.DeviceDomainEnum; import com.ard.sample.manage.model.enums.LiveStreamMethodEnum; import com.ard.sample.manage.model.enums.LiveUrlTypeEnum; import com.ard.sample.manage.model.enums.LiveVideoQualityEnum; import com.ard.sample.manage.model.param.DeviceQueryParam; import com.ard.sample.manage.model.receiver.CapacityDeviceReceiver; import com.ard.sample.manage.model.receiver.LiveCapacityReceiver; import com.ard.sample.manage.service.ICapacityCameraService; import com.ard.sample.manage.service.IDeviceService; import com.ard.sample.manage.service.ILiveStreamService; import com.ard.vtdu.config.MediamtxConfig; import com.ard.vtdu.domain.ArdVtdu; import com.ard.vtdu.service.IArdMediaService; import com.ard.vtdu.service.IArdVtduService; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import org.springframework.util.StringUtils; import javax.annotation.Resource; import java.lang.reflect.Field; import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.stream.Collectors; import static com.ard.sample.component.mqtt.model.TopicConst.*; /** * @author sean.zhou * @version 0.1 * @date 2021/11/22 */ @Slf4j @Service @Transactional public class LiveStreamServiceImpl implements ILiveStreamService { @Autowired private ICapacityCameraService capacityCameraService; @Autowired private IDeviceService deviceService; @Autowired private IMessageSenderService messageSender; @Resource private IArdVtduService ardVtduService; @Resource private MediamtxConfig mediamtxConfig; @Override public List getLiveCapacity() { // Query all devices in this workspace. List devicesList = deviceService.getDevicesByParams( DeviceQueryParam.builder() .domains(List.of(DeviceDomainEnum.SUB_DEVICE.getVal(), DeviceDomainEnum.DOCK.getVal())) .build()); // Query the live capability of each drone. return devicesList.stream() .filter(device -> RedisOpsUtils.checkExist(RedisConst.DEVICE_ONLINE_PREFIX + device.getDeviceSn())) .map(device -> CapacityDeviceDTO.builder() .name(Objects.requireNonNullElse(device.getNickname(), device.getDeviceName())) .sn(device.getDeviceSn()) .camerasList(capacityCameraService.getCapacityCameraByDeviceSn(device.getDeviceSn())) .build()) .collect(Collectors.toList()); } @Override public void saveLiveCapacity(LiveCapacityReceiver liveCapacityReceiver, Long timestamp) { // Solve timing problems if (liveCapacityReceiver == null) { //2025-10-28 zwy,这里有个空指针异常问题,如果liveCapacityReceiver为空,则不应该执行下面的代码。 //机场设备在刚开机时,发送的state状态时,直播能力前三次发送的是空,可能是由于刚开机,等待各设备开机,第四次为正常数据。 return; } for (CapacityDeviceReceiver capacityDeviceReceiver : liveCapacityReceiver.getDeviceList()) { long last = (long) Objects.requireNonNullElse( RedisOpsUtils.get(RedisConst.LIVE_CAPACITY + capacityDeviceReceiver.getSn()), 0L); if (last > timestamp) { return; } capacityCameraService.saveCapacityCameraReceiverList( capacityDeviceReceiver.getCameraList(), capacityDeviceReceiver.getSn(), timestamp); } } @Override public ResponseResult liveStart(LiveTypeDTO liveParam) { String rtmpPublishUrl = "rtmp://" + mediamtxConfig.getPushHost() + ":" + mediamtxConfig.getRtmpPort() + "/" + liveParam.getVideoId(); liveParam.setUrl(rtmpPublishUrl); // Check if this lens is available live. ResponseResult responseResult = this.checkBeforeLive(liveParam.getVideoId()); if (ResponseResult.CODE_SUCCESS != responseResult.getCode()) { return responseResult; } DeviceDTO data = (DeviceDTO) responseResult.getData(); // target topic String respTopic = THING_MODEL_PRE + PRODUCT + data.getDeviceSn() + SERVICES_SUF; ServiceReply receiveReply = this.publishLiveStart(respTopic, liveParam); if (ResponseResult.CODE_SUCCESS != receiveReply.getResult()) { log.info("机场直播开启失败,错误码:{},错误描述:{}", receiveReply.getResult(), LiveErrorEnum.find(receiveReply.getResult()).getErrorMsg()); return ResponseResult.error(receiveReply.getResult(), LiveErrorEnum.find(receiveReply.getResult()).getErrorMsg()); } LiveUrlTypeEnum urlType = LiveUrlTypeEnum.find(liveParam.getUrlType()); LiveDTO live = new LiveDTO(); switch (urlType) { case RTMP: String name = liveParam.getVideoId(); // 拼接地址 String rtspUrl = "rtsp://" + mediamtxConfig.getPlayHost() + ":" + mediamtxConfig.getRtspPort() + "/" + name; // // 拼接 webrtcUrl String rtmpUrl = "rtmp://" + mediamtxConfig.getPlayHost() + ":" + mediamtxConfig.getRtmpPort() + "/" + name; // // 拼接 webrtcUrl String protocol = mediamtxConfig.isHttps() ? "https://" : "http://"; String webrtcUrl = protocol + mediamtxConfig.getPlayHost() + ":" + mediamtxConfig.getWebrtcPort() + "/" + name; // 拼接 // webrtcUrl // 如果没有,插入新的记录 ArdVtdu vtdu = ArdVtdu.builder().name(name).rtspSource("publisher").rtspUrl(rtspUrl).rtmpUrl(rtmpUrl).webrtcUrl(webrtcUrl).build(); // 在停止直播时不删除Vtdu表记录,所以在开启直播时再插入会报错, // 所以要先查询是否存在该记录,如果已经存在记录,直接返回直播地址 // 查询是否已经存在该记录 ArdVtdu ardVtdu = ardVtduService.selectArdVtduByName(vtdu.getName()); int i = 0; log.info("机场视频插入前vtdu:{}",vtdu); if(ardVtdu == null){ i = ardVtduService.insertArdVtdu(vtdu); // 保存到媒体表 log.info("机场视频插入后 i:{}",i); if (i > 0) { live.setRtmpUrl(rtmpUrl); live.setRtspUrl(rtspUrl); live.setWebrtcUrl(webrtcUrl); } } else{ i = ardVtduService.updateArdVtdu(vtdu); log.info("机场视频更新后 i:{}",i); if (i > 0) { live.setRtmpUrl(rtmpUrl); live.setRtspUrl(rtspUrl); live.setWebrtcUrl(webrtcUrl); } } //int i = ardVtduService.insertArdVtdu(vtdu); // 保存到媒体表 // if (i > 0) { // live.setRtmpUrl(rtmpUrl); // live.setRtspUrl(rtspUrl); // live.setWebrtcUrl(webrtcUrl); // } break; case RTSP: Object url = Objects.requireNonNullElse(receiveReply.getOutput(), receiveReply.getInfo()); this.resolveUrlUser(String.valueOf(url), live); break; case UNKNOWN: return ResponseResult.error(LiveErrorEnum.URL_TYPE_NOT_SUPPORTED); } // 开始直播时发送机场相机摄像头默认信息(舱外)到所有在线用户,只取机场sn,发机场的摄像头信息 if(data.getChildDeviceSn()!=null) { String sn = data.getDeviceSn(); deviceService.setRedisCameraPosition(sn, 1); } return ResponseResult.success(live); } @Override public ResponseResult liveStop(String videoId) { ResponseResult responseResult = this.checkBeforeLive(videoId); if (responseResult.getCode() != 0) { return responseResult; } String respTopic = THING_MODEL_PRE + PRODUCT + responseResult.getData().getDeviceSn() + SERVICES_SUF; ServiceReply receiveReply = this.publishLiveStop(respTopic, videoId); if (receiveReply.getResult() != 0) { log.info("videoId:{},直播停止失败,错误码:{},错误描述:{}", videoId, receiveReply.getResult(), LiveErrorEnum.find(receiveReply.getResult()).getErrorMsg()); return ResponseResult.error(LiveErrorEnum.find(receiveReply.getResult())); } //注释掉删除流媒体管理库中的信息 // ardVtduService.deleteArdVtduByName(videoId); return ResponseResult.success(); } @Override public ResponseResult liveSetQuality(LiveTypeDTO liveParam) { if (liveParam.getVideoQuality() == null || LiveVideoQualityEnum.UNKNOWN == LiveVideoQualityEnum.find(liveParam.getVideoQuality())) { return ResponseResult.error(LiveErrorEnum.ERROR_PARAMETERS); } ResponseResult responseResult = this.checkBeforeLive(liveParam.getVideoId()); if (responseResult.getCode() != 0) { return responseResult; } String respTopic = THING_MODEL_PRE + PRODUCT + responseResult.getData().getDeviceSn() + SERVICES_SUF; ServiceReply receiveReply = this.publishLiveSetQuality(respTopic, liveParam); if (ResponseResult.CODE_SUCCESS != receiveReply.getResult()) { return ResponseResult.error(LiveErrorEnum.find(receiveReply.getResult())); } return ResponseResult.success(); } @Override public ResponseResult liveLensChange(LiveTypeDTO liveParam) { log.info("切换镜头参数:{}", liveParam); ResponseResult responseResult = this.checkBeforeLive(liveParam.getVideoId()); if (ResponseResult.CODE_SUCCESS != responseResult.getCode()) { return responseResult; } if (DeviceDomainEnum.GATEWAY.getVal() == responseResult.getData().getDomain()) { return ResponseResult.error(LiveErrorEnum.FUNCTION_NOT_SUPPORT); } String respTopic = THING_MODEL_PRE + PRODUCT + responseResult.getData().getDeviceSn() + SERVICES_SUF; ServiceReply receiveReply = this.publishLiveLensChange(respTopic, liveParam); if (ResponseResult.CODE_SUCCESS != receiveReply.getResult()) { Integer code = receiveReply.getResult(); if(code == 325003) { // 指令响应失败,请重试. 经查询官网,机场3为低配,飞行器不支持红外 return ResponseResult.error(325003000,LiveErrorEnum.COMMAND_FAILED.getErrorMsg()+"请确认无人机是否支持此功能!"); } return ResponseResult.error(LiveErrorEnum.find(receiveReply.getResult())); } return ResponseResult.success(); } @Override public ResponseResult> getLiveLens(String sn) { List videoTypes = new ArrayList<>(); // 检查设备是否在线 if (!deviceService.checkDeviceOnline(sn)) { log.info("SN: {}-设备不在线,请检查网络连接!", sn); // 抛弃出异常,给调用者处理 //throw new Exception("设备不在线,请检查网络连接!"); return ResponseResult.error(CommonErrorEnum.DEVICE_OFFLINE.getErrorCode(),CommonErrorEnum.DEVICE_OFFLINE.getErrorMsg()); } // 读取直播能力拼接参数 List devicesList = getLiveCapacity(); if (devicesList == null || devicesList.isEmpty()) { log.info("SN: {}-未获得设备直播能力!", sn); // 抛弃出异常,给调用者处理 //throw new Exception("未获得设备直播能力!"); return ResponseResult.error(22222200,"未获得设备直播能力!"); } for (CapacityDeviceDTO device : devicesList) { if (!device.getSn().equals(sn)) { continue; } List camerasList = device.getCamerasList(); for (CapacityCameraDTO camera : camerasList) { // 机场3中飞行器,没有name,需要其他条件确定主摄像头参数 for (CapacityVideoDTO video : camera.getVideosList()) { if (video.getSwitchVideoTypes().size() == 1 && camera.getName() == null) { continue; } videoTypes = video.getSwitchVideoTypes(); } } } return ResponseResult.success(videoTypes); } @Override public ResponseResult liveCameraChange(LiveTypeDTO liveParam) { if (liveParam.getCameraPosition() == null) { return ResponseResult.error(LiveErrorEnum.ERROR_PARAMETERS); } ResponseResult responseResult = this.checkBeforeLive(liveParam.getVideoId()); if (ResponseResult.CODE_SUCCESS != responseResult.getCode()) { return responseResult; } if (DeviceDomainEnum.GATEWAY.getVal() == responseResult.getData().getDomain()) { return ResponseResult.error(LiveErrorEnum.FUNCTION_NOT_SUPPORT); } String respTopic = THING_MODEL_PRE + PRODUCT + responseResult.getData().getDeviceSn() + SERVICES_SUF; ServiceReply receiveReply = this.publishLiveCameraChange(respTopic, liveParam); if (ResponseResult.CODE_SUCCESS != receiveReply.getResult()) { return ResponseResult.error(LiveErrorEnum.find(receiveReply.getResult())); } // 将切换相机摄像头信息到redis Integer camera_position = liveParam.getCameraPosition(); String sn = responseResult.getData().getDeviceSn(); deviceService.setRedisCameraPosition(sn, camera_position); return ResponseResult.success(); } private ServiceReply publishLiveLensChange(String respTopic, LiveTypeDTO liveParam) { CommonTopicResponse response = new CommonTopicResponse<>(); response.setTid(UUID.randomUUID().toString()); response.setBid(UUID.randomUUID().toString()); response.setMethod(LiveStreamMethodEnum.LIVE_LENS_CHANGE.getMethod()); response.setData(liveParam); return messageSender.publishWithReply(respTopic, response); } private ServiceReply publishLiveCameraChange(String respTopic, LiveTypeDTO liveParam) { CommonTopicResponse response = new CommonTopicResponse<>(); response.setTid(UUID.randomUUID().toString()); response.setBid(UUID.randomUUID().toString()); response.setMethod(LiveStreamMethodEnum.LIVE_CAMERA_CHANGE.getMethod()); response.setData(liveParam); return messageSender.publishWithReply(respTopic, response); } /** * Check if this lens is available live. * * @param videoId * @return */ private ResponseResult checkBeforeLive(String videoId) { if (!StringUtils.hasText(videoId)) { return ResponseResult.error(LiveErrorEnum.ERROR_PARAMETERS); } String[] videoIdArr = videoId.split("/"); // drone sn / enumeration value of the location where the payload is mounted / payload lens if (videoIdArr.length != 3) { return ResponseResult.error(LiveErrorEnum.ERROR_PARAMETERS); } Optional deviceOpt = deviceService.getDeviceBySn(videoIdArr[0]); // Check if the gateway device connected to this drone exists if (deviceOpt.isEmpty()) { return ResponseResult.error(LiveErrorEnum.NO_AIRCRAFT); } if (DeviceDomainEnum.DOCK.getVal() == deviceOpt.get().getDomain()) { return ResponseResult.success(deviceOpt.get()); } List gatewayList = deviceService.getDevicesByParams( DeviceQueryParam.builder() .childSn(videoIdArr[0]) .build()); if (gatewayList.isEmpty()) { return ResponseResult.error(LiveErrorEnum.NO_FLIGHT_CONTROL); } return ResponseResult.success(gatewayList.get(0)); } /** * When using rtsp live, the account and password are parsed from the information returned by the pilot. * * @param url * @param live */ private void resolveUrlUser(String url, LiveDTO live) { if (!StringUtils.hasText(url)) { return; } int start = url.indexOf("//"); int end = url.lastIndexOf("@"); String user = url.substring(start + 2, end); url = url.replace(user + "@", ""); String[] userArr = user.split(":"); // live.setUsername(userArr[0]); // live.setPassword(userArr[1]); // live.setUrl(url); } /** * When using GB28181 live, url parameters are resolved into objects. * * @param url * @return */ private LiveUrlGB28181DTO urlToGB28181(String url) { String[] arr = url.split("\\=|\\&"); LiveUrlGB28181DTO gb28181 = new LiveUrlGB28181DTO(); try { Class clazz = LiveUrlGB28181DTO.class; for (int i = 0; i < arr.length - 1; i += 2) { Field field = clazz.getDeclaredField(arr[i]); field.setAccessible(true); if (field.getType().equals(Integer.class)) { field.set(gb28181, Integer.valueOf(arr[i + 1])); continue; } field.set(gb28181, arr[i + 1]); } } catch (NoSuchFieldException | IllegalAccessException e) { e.printStackTrace(); } return gb28181; } /** * Send a message to the pilot via mqtt to start the live streaming. * * @param topic * @param liveParam * @return */ private ServiceReply publishLiveStart(String topic, LiveTypeDTO liveParam) { CommonTopicResponse response = new CommonTopicResponse<>(); response.setTid(UUID.randomUUID().toString()); response.setBid(UUID.randomUUID().toString()); response.setData(liveParam); response.setMethod(LiveStreamMethodEnum.LIVE_START_PUSH.getMethod()); return messageSender.publishWithReply(topic, response); } /** * Send a message to the pilot via mqtt to set quality. * * @param respTopic * @param liveParam * @return */ private ServiceReply publishLiveSetQuality(String respTopic, LiveTypeDTO liveParam) { Map data = new ConcurrentHashMap<>(Map.of( "video_id", liveParam.getVideoId(), "video_quality", liveParam.getVideoQuality())); CommonTopicResponse> response = new CommonTopicResponse<>(); response.setTid(UUID.randomUUID().toString()); response.setBid(UUID.randomUUID().toString()); response.setMethod(LiveStreamMethodEnum.LIVE_SET_QUALITY.getMethod()); response.setData(data); return messageSender.publishWithReply(respTopic, response); } /** * Send a message to the pilot via mqtt to stop the live streaming. * * @param topic * @param videoId * @return */ private ServiceReply publishLiveStop(String topic, String videoId) { Map data = new ConcurrentHashMap<>(Map.of("video_id", videoId)); CommonTopicResponse> response = new CommonTopicResponse<>(); response.setTid(UUID.randomUUID().toString()); response.setBid(UUID.randomUUID().toString()); response.setData(data); response.setMethod(LiveStreamMethodEnum.LIVE_STOP_PUSH.getMethod()); return messageSender.publishWithReply(topic, response); } }