package com.ard.work.component.websocket.handler;
|
|
import com.alibaba.fastjson2.JSON;
|
|
import com.ard.vtdu.config.MediamtxSixHeConfig;
|
import com.ard.vtdu.domain.ArdVtdu;
|
import com.ard.work.component.websocket.utils.WebSocketUtil;
|
import com.ard.work.domain.message.BaseMessage;
|
import com.ard.work.domain.receiver.BoxStreamUrlReceiver;
|
import com.ard.work.service.ITyService;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.stereotype.Service;
|
|
import javax.annotation.Resource;
|
import java.util.Objects;
|
import java.util.Set;
|
import java.util.concurrent.ConcurrentHashMap;
|
import java.util.concurrent.ConcurrentMap;
|
|
@Service
|
@Slf4j
|
public class UavStreamManager {
|
|
private static final String STREAM_LOG_PREFIX = "【stream】";
|
private static final String STREAM_NOT_FOUND_MSG = "Box stream not found: ";
|
private static final String STREAM_CLOSED_MSG = "Stream closed due to no active users: ";
|
private static final String STREAM_IN_USE_MSG = "Stream is still in use, remaining users: ";
|
|
@Resource
|
private ITyService tyService;
|
@Resource
|
private MediamtxSixHeConfig mediamtxSixHeConfig;
|
|
// 盒子拉流地址缓存 key:boxSn value: webrtcUrl
|
public static final ConcurrentMap<String, String> BOX_STREAM_PULL_MAP = new ConcurrentHashMap<>();
|
// 盒子拉流用户集合缓存 key:boxSn value:Set<UserId>
|
public static final ConcurrentMap<String, Set<Long>> STREAM_USER_MAP = new ConcurrentHashMap<>();
|
|
/**
|
* 开始拉流
|
*
|
* @param boxSn 盒子序列号
|
* @param userId 用户ID
|
*/
|
public void subscribe(String boxSn, Long userId) {
|
|
STREAM_USER_MAP.computeIfAbsent(boxSn, k -> ConcurrentHashMap.newKeySet()).add(userId);
|
BoxStreamUrlReceiver streamReceiver = getWebrtcUrl(boxSn);
|
String webrtcUrl = streamReceiver.getRtc();
|
if (Objects.equals(webrtcUrl, "")) {
|
log.warn("{}{}{}", STREAM_LOG_PREFIX, STREAM_NOT_FOUND_MSG, boxSn);
|
BaseMessage<String> baseMessage = new BaseMessage<>("stream", STREAM_NOT_FOUND_MSG + boxSn);
|
sendMessageToUser(userId, JSON.toJSONString(baseMessage));
|
}
|
BOX_STREAM_PULL_MAP.put(boxSn, webrtcUrl);
|
BaseMessage<BoxStreamUrlReceiver> baseMessage = new BaseMessage<>("stream", streamReceiver);
|
sendMessageToUser(userId, JSON.toJSONString(baseMessage));
|
}
|
|
/**
|
* 停止拉流
|
*
|
* @param boxSn 盒子序列号
|
* @param userId 用户ID
|
*/
|
public void unsubscribe(String boxSn, Long userId) {
|
synchronized (this) {
|
Set<Long> users = STREAM_USER_MAP.get(boxSn);
|
if (users == null) {
|
log.warn("{}No active users for box: {}", STREAM_LOG_PREFIX, boxSn);
|
BaseMessage<String> baseMessage = new BaseMessage<>("stream", "No active users for box: " + boxSn);
|
sendMessageToUser(userId, JSON.toJSONString(baseMessage));
|
return;
|
}
|
|
users.remove(userId);
|
String message = "";
|
if (users.isEmpty()) {
|
BOX_STREAM_PULL_MAP.remove(boxSn);
|
STREAM_USER_MAP.remove(boxSn);
|
log.info("{}{}{}", STREAM_LOG_PREFIX, STREAM_CLOSED_MSG, boxSn);
|
message = STREAM_CLOSED_MSG + boxSn;
|
} else {
|
log.info("{}{}{}", STREAM_LOG_PREFIX, STREAM_IN_USE_MSG, users);
|
message = STREAM_IN_USE_MSG + boxSn;
|
}
|
BaseMessage<String> baseMessage = new BaseMessage<>("stream", message);
|
sendMessageToUser(userId, JSON.toJSONString(baseMessage));
|
}
|
}
|
|
/**
|
* 停止拉流
|
*
|
* @param userId 用户ID
|
*/
|
public void stopAllPullStreamByUser(Long userId) {
|
synchronized (this) {
|
STREAM_USER_MAP.forEach((boxSn, users) -> unsubscribe(boxSn, userId));
|
}
|
}
|
|
/**
|
* 发送消息给指定用户
|
*
|
* @param userId 用户ID
|
* @param message 消息内容
|
*/
|
private void sendMessageToUser(Long userId, String message) {
|
try {
|
WebSocketUtil.sendMessage(userId, message);
|
} catch (Exception e) {
|
log.error("{}Failed to send message to user: {}", STREAM_LOG_PREFIX, userId, e);
|
}
|
}
|
|
|
/**
|
* 通过盒子编号获取webrtc地址
|
*
|
* @param boxSn 盒子编号
|
* @return String webrtcUrl
|
*/
|
private BoxStreamUrlReceiver getWebrtcUrl(String boxSn) {
|
BoxStreamUrlReceiver boxLiveUrl = tyService.getBoxLiveUrl(boxSn);
|
if (boxLiveUrl == null) {
|
throw new RuntimeException("未获取到盒子直播地址：" + boxSn);
|
}
|
// 2. 获取原始 URL 字符串
|
String originalUrl = boxLiveUrl.getRtc();
|
if (mediamtxSixHeConfig.getEnable() && originalUrl != null && !originalUrl.isEmpty()) {
|
// --- 开始替换逻辑 ---
|
// 配置你的外网地址
|
String targetHost = mediamtxSixHeConfig.getPlayHost() + ":" + mediamtxSixHeConfig.getWebrtcPort();
|
// 正则解释：
|
// (^https?://) -> 捕获组1：匹配 http:// 或 https://
|
// ([^/:]+) -> 捕获组2：匹配 IP 地址或域名 (非 / 和 : 的字符)
|
// (:\d+)? -> 捕获组3：匹配 :端口号 (可选)
|
// (/.*) -> 捕获组4：匹配后面的路径参数
|
String regex = "(^https?://)([^/:]+)(:\\d+)?(/.*)";
|
// 替换为：协议 + 外网IP:端口 + 路径
|
// $1 代表协议, $4 代表路径
|
String newUrl = originalUrl.replaceAll(regex, "$1" + targetHost + "$4");
|
// 将新 URL 设置回去
|
boxLiveUrl.setRtc(newUrl);
|
}
|
boxLiveUrl.setBoxSn(boxSn);
|
return boxLiveUrl;
|
}
|
}
|
