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<CapacityDeviceDTO> getLiveCapacity() {
|
|
// Query all devices in this workspace.
|
List<DeviceDTO> 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()) {
|
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<DeviceDTO> 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) {
|
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<DeviceDTO> 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<DeviceDTO> 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<List<String>> getLiveLens(String sn) {
|
List<String> 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<CapacityDeviceDTO> 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<CapacityCameraDTO> 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<DeviceDTO> 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<LiveTypeDTO> 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<LiveTypeDTO> 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<DeviceDTO> 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<DeviceDTO> 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<DeviceDTO> 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<LiveUrlGB28181DTO> 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<LiveTypeDTO> 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<String, Object> data = new ConcurrentHashMap<>(Map.of(
|
"video_id", liveParam.getVideoId(),
|
"video_quality", liveParam.getVideoQuality()));
|
CommonTopicResponse<Map<String, Object>> 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<String, String> data = new ConcurrentHashMap<>(Map.of("video_id", videoId));
|
CommonTopicResponse<Map<String, String>> 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);
|
}
|
|
}
|
