package com.ard.tianyu.service.impl;
|
|
import com.ard.tianyu.model.receiver.BoxStreamUrlReceiver;
|
import com.ard.tianyu.service.ITyService;
|
import com.ard.tianyu.websocket.utils.WebSocketUtil;
|
import com.ard.vtdu.domain.ArdVtdu;
|
import com.ard.vtdu.service.IMediaService;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.stereotype.Service;
|
|
import javax.annotation.Resource;
|
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 = STREAM_LOG_PREFIX + "Box stream not found: ";
|
private static final String STREAM_CLOSED_MSG = STREAM_LOG_PREFIX + "Stream closed due to no active users: ";
|
private static final String STREAM_IN_USE_MSG = STREAM_LOG_PREFIX + "Stream is still in use, remaining users: ";
|
|
@Resource
|
private ITyService tyService;
|
@Resource
|
private IMediaService mediaService;
|
|
// 盒子拉流地址缓存 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<String>> STREAM_USER_MAP = new ConcurrentHashMap<>();
|
|
/**
|
* 开始拉流
|
*
|
* @param boxSn 盒子序列号
|
* @param userId 用户ID
|
*/
|
public void startPullStream(String boxSn, String userId) {
|
STREAM_USER_MAP.computeIfAbsent(boxSn, k -> ConcurrentHashMap.newKeySet()).add(userId);
|
|
String webrtcUrl = BOX_STREAM_PULL_MAP.get(boxSn);
|
if (webrtcUrl != null) {
|
log.info("{}Already pulled stream: {}", STREAM_LOG_PREFIX, webrtcUrl);
|
sendMessageToUser(userId, webrtcUrl);
|
return;
|
}
|
|
try {
|
BoxStreamUrlReceiver boxLiveUrl = tyService.getBoxLiveUrl(boxSn);
|
if (boxLiveUrl == null) {
|
log.warn("{}{}{}", STREAM_LOG_PREFIX, STREAM_NOT_FOUND_MSG, boxSn);
|
sendMessageToUser(userId, STREAM_NOT_FOUND_MSG + boxSn);
|
return;
|
}
|
|
ArdVtdu vtdu = ArdVtdu.builder()
|
.name(boxSn)
|
.mode(0)
|
.isCode(0)
|
.rtspSource(boxLiveUrl.getRtsp())
|
.build();
|
ArdVtdu ardVtdu = mediaService.addPath(vtdu);
|
webrtcUrl = ardVtdu.getWebrtcUrl();
|
BOX_STREAM_PULL_MAP.put(boxSn, webrtcUrl);
|
|
log.info("{}First pulled stream: {}", STREAM_LOG_PREFIX, webrtcUrl);
|
sendMessageToUser(userId, webrtcUrl);
|
} catch (Exception e) {
|
log.error("{}Failed to start pull stream for box: {}", STREAM_LOG_PREFIX, boxSn, e);
|
sendMessageToUser(userId, "Failed to start pull stream due to an internal error.");
|
}
|
}
|
|
/**
|
* 停止拉流
|
*
|
* @param boxSn 盒子序列号
|
* @param userId 用户ID
|
*/
|
public void stopPullStream(String boxSn, String userId) {
|
synchronized (this) {
|
Set<String> users = STREAM_USER_MAP.get(boxSn);
|
if (users == null) {
|
log.warn("{}No active users for box: {}", STREAM_LOG_PREFIX, boxSn);
|
sendMessageToUser(userId, "No active users for box: " + boxSn);
|
return;
|
}
|
|
users.remove(userId);
|
if (users.isEmpty()) {
|
mediaService.removePath(boxSn);
|
BOX_STREAM_PULL_MAP.remove(boxSn);
|
STREAM_USER_MAP.remove(boxSn);
|
log.info("{}{}{}", STREAM_LOG_PREFIX, STREAM_CLOSED_MSG, boxSn);
|
sendMessageToUser(userId, STREAM_CLOSED_MSG + boxSn);
|
} else {
|
log.info("{}{}{}", STREAM_LOG_PREFIX, STREAM_IN_USE_MSG, users);
|
sendMessageToUser(userId, STREAM_IN_USE_MSG + users);
|
}
|
}
|
}
|
/**
|
* 停止拉流
|
*
|
* @param userId 用户ID
|
*/
|
public void stopAllPullStreamByUser(String userId) {
|
synchronized (this) {
|
STREAM_USER_MAP.forEach((boxSn, users) -> stopPullStream(boxSn, userId));
|
}
|
}
|
|
/**
|
* 发送消息给指定用户
|
*
|
* @param userId 用户ID
|
* @param message 消息内容
|
*/
|
private void sendMessageToUser(String userId, String message) {
|
try {
|
WebSocketUtil.sendMessage(userId, message);
|
} catch (Exception e) {
|
log.error("{}Failed to send message to user: {}", STREAM_LOG_PREFIX, userId, e);
|
}
|
}
|
}
|
