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.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 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 sameList(List vtduses, List names) { return vtduses.stream().filter(vtdu -> names.contains(vtdu.getName())).collect(Collectors.toList()); } //需要删除的(流媒体中多的) @Override public List diffListToDel(List vtduses, List names) { // 先获取本地库非 publisher 的名称集合 Set localNames = vtduses.stream() .map(ArdVtdu::getName) .collect(Collectors.toSet()); // 过滤 MediaMTX 流中非 publisher 流,并判断是否在本地库中 List toDelete = names.stream() .filter(n -> !localNames.contains(n)) // MediaMTX 有,但本地库没有 .collect(Collectors.toList()); return toDelete; } //需要新增的(流媒体中少的) @Override public List diffListToAdd(List vtduses, List 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 vtdus, List 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 nameList = ardMediaService.getNameList(); List 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 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 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 username = loginUser.getUsername(); ArdVtdu ardVtdu = ArdVtdu.builder() .name("user_" + userId) // 用 userId 作为流名称 .rtspSource("publisher") // 或 "publisher",按业务区分 .build(); // 本地库插入 insertArdVtdu(ardVtdu); log.info("用户【{}】登录,立即加入流媒体", username); } else { log.warn("登录事件 principal 不是 LoginUser,无法获取 userId"); } } }