package com.ard.vtdu.service.impl;
|
|
import com.ard.common.constant.CacheConstants;
|
import com.ard.common.core.redis.RedisCache;
|
import com.ard.common.utils.DateUtils;
|
import com.ard.vtdu.domain.ArdVtdu;
|
import com.ard.vtdu.domain.Conf;
|
import com.ard.vtdu.mapper.ArdVtduMapper;
|
import com.ard.vtdu.service.IArdVtduService;
|
import com.ard.vtdu.service.IArdMediaService;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.context.event.EventListener;
|
import org.springframework.scheduling.annotation.Scheduled;
|
import org.springframework.security.authentication.event.AuthenticationSuccessEvent;
|
import com.ard.common.core.domain.model.LoginUser;
|
import org.springframework.security.authentication.event.LogoutSuccessEvent;
|
import org.springframework.security.core.Authentication;
|
import org.springframework.stereotype.Service;
|
|
import javax.annotation.PostConstruct;
|
import javax.annotation.Resource;
|
import java.util.*;
|
import java.util.stream.Collectors;
|
|
/**
|
* 流媒体管理Service业务层处理
|
*
|
* @author ard
|
* @date 2023-08-29
|
*/
|
@Service
|
@Slf4j
|
public class ArdVtduServiceImpl implements IArdVtduService {
|
@Resource
|
private ArdVtduMapper ardVtduMapper;
|
@Resource
|
private IArdMediaService ardMediaService;
|
@Resource
|
private RedisCache redisCache;
|
|
@PostConstruct
|
public void initMediaMTX() {
|
//清空mediamtx流媒体数据
|
ardMediaService.clearAll();
|
}
|
|
/**
|
* 查询流媒体管理
|
*
|
* @param name 流媒体管理主键
|
* @return 流媒体管理
|
*/
|
@Override
|
public ArdVtdu selectArdVtduByName(String name) {
|
return ardVtduMapper.selectArdVtduByName(name);
|
}
|
|
/**
|
* 查询流媒体管理列表
|
*
|
* @param ardVtdu 流媒体管理
|
* @return 流媒体管理
|
*/
|
@Override
|
public List<ArdVtdu> selectArdVtduList(ArdVtdu ardVtdu) {
|
return ardVtduMapper.selectArdVtduList(ardVtdu);
|
}
|
|
/**
|
* 新增流媒体管理
|
*
|
* @param ardVtdu 流媒体管理
|
* @return 结果
|
*/
|
@Override
|
public int insertArdVtdu(ArdVtdu ardVtdu) {
|
ardVtdu.setCreateTime(DateUtils.getNowDate());
|
ardVtdu = ardMediaService.addPath(ardVtdu);
|
return ardVtduMapper.insertArdVtdu(ardVtdu);
|
}
|
|
/**
|
* 修改流媒体管理
|
*
|
* @param ardVtdu 流媒体管理
|
* @return 结果
|
*/
|
@Override
|
public int updateArdVtdu(ArdVtdu ardVtdu) {
|
ardVtdu.setUpdateTime(DateUtils.getNowDate());
|
ardVtdu = ardMediaService.editPath(ardVtdu);
|
ardVtdu.setUpdateTime(DateUtils.getNowDate());
|
return ardVtduMapper.updateArdVtdu(ardVtdu);
|
}
|
|
/**
|
* 批量删除流媒体管理
|
*
|
* @param names 需要删除的数据主键集合
|
* @return 结果
|
*/
|
@Override
|
public int deleteArdVtduByNames(String[] names) {
|
for (String name : names) {
|
log.debug("流媒体【" + name + "】通道删除");
|
}
|
ardMediaService.removePath(names);
|
return ardVtduMapper.deleteArdVtduByNames(names);
|
}
|
|
/**
|
* 删除流媒体管理
|
*
|
* @param name 流媒体通道名称
|
* @return 结果
|
*/
|
@Override
|
public int deleteArdVtduByName(String name) {
|
ardMediaService.removePath(name);
|
log.debug("流媒体【" + name + "】通道删除");
|
return ardVtduMapper.deleteArdVtduByName(name);
|
}
|
|
/**
|
* 删除流媒体管理信息
|
*
|
* @param cameraId 相机ID
|
* @return 结果
|
*/
|
@Override
|
public int deleteArdVtduByCameraId(String cameraId) {
|
return ardVtduMapper.deleteArdVtduByCameraId(cameraId);
|
}
|
|
/**
|
* 清除所有流媒体数据
|
*
|
* @author 刘苏义
|
* @date 2024/7/10 13:31
|
*/
|
@Override
|
public void clearAllArdVtdu() {
|
ardVtduMapper.clearAllArdVtdu();
|
}
|
|
|
|
|
//需要更新的（流媒体和vtdu相同）
|
@Override
|
public List<ArdVtdu> sameList(List<ArdVtdu> vtduses, List<String> names) {
|
return vtduses.stream().filter(vtdu -> names.contains(vtdu.getName())).collect(Collectors.toList());
|
}
|
|
//需要删除的（流媒体中多的）
|
@Override
|
public List<String> diffListToDel(List<ArdVtdu> vtduses, List<String> names) {
|
// 先获取本地库非 publisher 的名称集合
|
Set<String> localNames = vtduses.stream()
|
.map(ArdVtdu::getName)
|
.collect(Collectors.toSet());
|
|
// 过滤 MediaMTX 流中非 publisher 流，并判断是否在本地库中
|
List<String> toDelete = names.stream()
|
.filter(n -> !localNames.contains(n)) // MediaMTX 有，但本地库没有
|
.collect(Collectors.toList());
|
|
return toDelete;
|
}
|
|
|
//需要新增的（流媒体中少的）
|
@Override
|
public List<ArdVtdu> diffListToAdd(List<ArdVtdu> vtduses, List<String> names) {
|
return vtduses.stream().filter(vtdu -> !names.contains(vtdu.getName())).collect(Collectors.toList());
|
}
|
|
/**
|
* @return
|
* @Author 刘苏义
|
* @Description 同步本地vtdu库和流媒体中的数据
|
* @Date 2024/7/10 15:26
|
* @Param vtdus vtdu库的集合
|
* @Param names 流媒体的name集合
|
*/
|
@Override
|
public void asyncMediaMTX(List<ArdVtdu> vtdus, List<String> names) {
|
//需要更新的数据,参数顺序注意
|
sameList(vtdus, names).forEach(vtdu -> {
|
Conf conf = ardMediaService.getPathConfig(vtdu.getName());
|
if (conf == null || conf.getSource() == null) {
|
return;
|
}
|
String currentRtspSource = vtdu.getRtspSource(); // 提前获取需要比对的RTSP源
|
String confRtspSource = conf.getSource();
|
|
// 比较 RTSP 源是否一致
|
boolean isRtspSourceChanged = !Objects.equals(confRtspSource, currentRtspSource);
|
// 如果 RTSP 源发生变化，则更新路径
|
if (isRtspSourceChanged) {
|
ardMediaService.editPath(vtdu);
|
}
|
});
|
//需要删除的数据
|
diffListToDel(vtdus, names).forEach(name -> {
|
ardMediaService.removePath(name);
|
});
|
//需要新增的数据
|
diffListToAdd(vtdus, names).forEach(vtdu -> {
|
ardMediaService.addPath(vtdu);
|
});
|
}
|
|
/**
|
* 同步mediamtx
|
*/
|
@Scheduled(cron = "0/10 * * * * ?")
|
public void syncMediamtx() {
|
log.debug("定时同步MediaMTX任务");
|
try {
|
List<String> nameList = ardMediaService.getNameList();
|
List<ArdVtdu> ardVtduList = this.selectArdVtduList(new ArdVtdu());
|
asyncMediaMTX(ardVtduList, nameList);
|
} catch (Exception e) {
|
log.error("同步MediaMTX任务失败", e);
|
}
|
}
|
|
/**
|
* 定时同步流媒体用户
|
*/
|
@Scheduled(fixedRate = 5000)
|
public void syncStreamMedia() {
|
|
// Redis：在线 userId（1001）
|
Set<String> onlineUserIds = redisCache.getSetKey(CacheConstants.LOGIN_USERID_KEY)
|
.stream()
|
.map(k -> k.substring(CacheConstants.LOGIN_USERID_KEY.length()))
|
.collect(Collectors.toSet());
|
|
// DB：当前流媒体 userId（从 user_1001 提取 1001）
|
Set<String> currentStreamUserIds = ardVtduMapper
|
.selectArdVtduList(new ArdVtdu())
|
.stream()
|
.filter(v -> "publisher".equals(v.getRtspSource()))
|
.map(ArdVtdu::getName)
|
.filter(name -> name != null && name.startsWith("user_"))
|
.map(name -> name.substring("user_".length()))
|
.collect(Collectors.toSet());
|
|
// 1️⃣ 新增：Redis 有，DB 没有
|
onlineUserIds.stream()
|
.filter(id -> !currentStreamUserIds.contains(id))
|
.forEach(id -> {
|
ArdVtdu vtdu = ArdVtdu.builder()
|
.name("user_" + id)
|
.rtspSource("publisher")
|
.build();
|
insertArdVtdu(vtdu);
|
log.info("用户【{}】添加到流媒体", id);
|
});
|
|
// 2️⃣ 删除：DB 有，Redis 没有
|
currentStreamUserIds.stream()
|
.filter(id -> !onlineUserIds.contains(id))
|
.forEach(id -> {
|
deleteArdVtduByName("user_" + id);
|
log.info("用户【{}】移除流媒体", id);
|
});
|
}
|
|
@EventListener
|
public void handleLoginSuccess(AuthenticationSuccessEvent event) {
|
Object principal = event.getAuthentication().getPrincipal();
|
if (principal instanceof LoginUser) {
|
LoginUser loginUser = (LoginUser) principal; // 传统强制类型转换
|
Long userId = loginUser.getUser().getUserId();
|
String name="user_" + userId;
|
// 1️⃣ 判断数据库是否已存在
|
ArdVtdu existing = ardVtduMapper.selectArdVtduByName(name);
|
if (existing != null) {
|
log.info("流媒体【{}】已存在，跳过插入", name);
|
return;
|
}
|
ArdVtdu ardVtdu = ArdVtdu.builder()
|
.name(name) // 用 userId 作为流名称
|
.rtspSource("publisher") // 或 "publisher"，按业务区分
|
.build();
|
// 本地库插入
|
insertArdVtdu(ardVtdu);
|
log.info("用户【{}】登录，立即加入流媒体", name);
|
}
|
}
|
@EventListener
|
public void handleLogout(LogoutSuccessEvent event) {
|
Object principal = event.getAuthentication().getPrincipal();
|
if (principal instanceof LoginUser) {
|
LoginUser loginUser = (LoginUser) principal; // 传统强制类型转换
|
Long userId = loginUser.getUser().getUserId();
|
String name="user_" + userId;
|
deleteArdVtduByName(name);
|
log.info("用户【{}】注销，立即移除流媒体", name);
|
}
|
}
|
}
|
