XW-AIOT/wvp-platform
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化拉流代理播放机制

Browse Source
This commit is contained in:
lin
2025-07-01 17:07:10 +08:00
parent 9e8fbea73b
commit f8e5e8f057
14 changed files with 196 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
src/main/java/com/genersoft/iot/vmp/streamProxy/controller/StreamProxyController.java
View File

@@ -7,12 +7,15 @@ import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.conf.security.JwtUtils;
import com.genersoft.iot.vmp.media.bean.MediaServer;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.service.bean.ErrorCallback;
import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
import com.genersoft.iot.vmp.streamProxy.bean.StreamProxy;
import com.genersoft.iot.vmp.streamProxy.bean.StreamProxyParam;
import com.genersoft.iot.vmp.streamProxy.service.IStreamProxyPlayService;
import com.genersoft.iot.vmp.streamProxy.service.IStreamProxyService;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.StreamContent;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
import com.github.pagehelper.PageInfo;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
@@ -20,8 +23,10 @@ import io.swagger.v3.oas.annotations.security.SecurityRequirement;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.context.request.async.DeferredResult;
import javax.servlet.http.HttpServletRequest;
import java.net.MalformedURLException;
@@ -89,7 +94,7 @@ public class StreamProxyController {
})
@PostMapping(value = "/save")
@ResponseBody
public StreamContent save(@RequestBody StreamProxyParam param){
public DeferredResult<WVPResult<StreamContent>> save(HttpServletRequest request, @RequestBody StreamProxyParam param){
log.info("添加代理： " + JSONObject.toJSONString(param));
if (ObjectUtils.isEmpty(param.getMediaServerId())) {
param.setMediaServerId("auto");
@@ -97,18 +102,39 @@ public class StreamProxyController {
if (ObjectUtils.isEmpty(param.getType())) {
param.setType("default");
}
DeferredResult<WVPResult<StreamContent>> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue());
ErrorCallback<StreamInfo> callback = (code, msg, streamInfo) -> {
if (code == InviteErrorCode.SUCCESS.getCode()) {
WVPResult<StreamContent> wvpResult = WVPResult.success();
if (streamInfo != null) {
if (userSetting.getUseSourceIpAsStreamIp()) {
streamInfo=streamInfo.clone();//深拷贝
String host;
try {
URL url=new URL(request.getRequestURL().toString());
host=url.getHost();
} catch (MalformedURLException e) {
host=request.getLocalAddr();
}
streamInfo.changeStreamIp(host);
}
if (!ObjectUtils.isEmpty(streamInfo.getMediaServer().getTranscodeSuffix())
&& !"null".equalsIgnoreCase(streamInfo.getMediaServer().getTranscodeSuffix())) {
streamInfo.setStream(streamInfo.getStream() + "_" + streamInfo.getMediaServer().getTranscodeSuffix());
}
wvpResult.setData(new StreamContent(streamInfo));
}else {
wvpResult.setCode(code);
wvpResult.setMsg(msg);
}
StreamInfo streamInfo = streamProxyService.save(param);
if (param.isEnable()) {
if (streamInfo == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), ErrorCode.ERROR100.getMsg());
result.setResult(wvpResult);
}else {
return new StreamContent(streamInfo);
result.setResult(WVPResult.fail(code, msg));
}
}else {
return null;
}
};
streamProxyService.save(param, callback);
return result;
}
@Operation(summary = "新增代理", security = @SecurityRequirement(name = JwtUtils.HEADER), parameters = {
@@ -193,25 +219,46 @@ public class StreamProxyController {
@ResponseBody
@Operation(summary = "启用代理", security = @SecurityRequirement(name = JwtUtils.HEADER))
@Parameter(name = "id", description = "代理Id", required = true)
public StreamContent start(HttpServletRequest request, int id){
public DeferredResult<WVPResult<StreamContent>> start(HttpServletRequest request, int id){
log.info("播放代理： {}", id);
StreamInfo streamInfo = streamProxyPlayService.start(id, null, null);
if (streamInfo == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), ErrorCode.ERROR100.getMsg());
}else {
if (userSetting.getUseSourceIpAsStreamIp()) {
streamInfo=streamInfo.clone();//深拷贝
String host;
try {
URL url=new URL(request.getRequestURL().toString());
host=url.getHost();
} catch (MalformedURLException e) {
host=request.getLocalAddr();
StreamProxy streamProxy = streamProxyService.getStreamProxy(id);
Assert.notNull(streamProxy, "代理信息不存在");
DeferredResult<WVPResult<StreamContent>> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue());
ErrorCallback<StreamInfo> callback = (code, msg, streamInfo) -> {
if (code == InviteErrorCode.SUCCESS.getCode()) {
WVPResult<StreamContent> wvpResult = WVPResult.success();
if (streamInfo != null) {
if (userSetting.getUseSourceIpAsStreamIp()) {
streamInfo=streamInfo.clone();//深拷贝
String host;
try {
URL url=new URL(request.getRequestURL().toString());
host=url.getHost();
} catch (MalformedURLException e) {
host=request.getLocalAddr();
}
streamInfo.changeStreamIp(host);
}
if (!ObjectUtils.isEmpty(streamInfo.getMediaServer().getTranscodeSuffix())
&& !"null".equalsIgnoreCase(streamInfo.getMediaServer().getTranscodeSuffix())) {
streamInfo.setStream(streamInfo.getStream() + "_" + streamInfo.getMediaServer().getTranscodeSuffix());
}
wvpResult.setData(new StreamContent(streamInfo));
}else {
wvpResult.setCode(code);
wvpResult.setMsg(msg);
}
streamInfo.changeStreamIp(host);
result.setResult(wvpResult);
}else {
result.setResult(WVPResult.fail(code, msg));
}
return new StreamContent(streamInfo);
}
};
streamProxyPlayService.start(id, null, callback);
return result;
}
@GetMapping(value = "/stop")

2
src/main/java/com/genersoft/iot/vmp/streamProxy/dao/StreamProxyMapper.java
View File

@@ -92,5 +92,5 @@ public interface StreamProxyMapper {
" SET pulling=#{pulling}, media_server_id = #{mediaServerId}, " +
" stream_key = #{streamKey} " +
" WHERE id=#{id}")
void addStream(StreamProxy streamProxy);
void updateStream(StreamProxy streamProxy);
}

8
src/main/java/com/genersoft/iot/vmp/streamProxy/service/IStreamProxyPlayService.java
View File

@@ -4,13 +4,13 @@ import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.service.bean.ErrorCallback;
import com.genersoft.iot.vmp.streamProxy.bean.StreamProxy;
import javax.validation.constraints.NotNull;
public interface IStreamProxyPlayService {
StreamInfo start(int id, Boolean record, ErrorCallback<StreamInfo> callback);
void start(int id, Boolean record, ErrorCallback<StreamInfo> callback);
void start(int id, ErrorCallback<StreamInfo> callback);
StreamInfo startProxy(StreamProxy streamProxy);
void startProxy(@NotNull StreamProxy streamProxy, ErrorCallback<StreamInfo> callback);
void stop(int id);

5
src/main/java/com/genersoft/iot/vmp/streamProxy/service/IStreamProxyService.java
View File

@@ -2,6 +2,7 @@ package com.genersoft.iot.vmp.streamProxy.service;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.media.bean.MediaServer;
import com.genersoft.iot.vmp.service.bean.ErrorCallback;
import com.genersoft.iot.vmp.streamProxy.bean.StreamProxy;
import com.genersoft.iot.vmp.streamProxy.bean.StreamProxyParam;
import com.genersoft.iot.vmp.vmanager.bean.ResourceBaseInfo;
@@ -15,7 +16,7 @@ public interface IStreamProxyService {
* 保存视频代理
* @param param
*/
StreamInfo save(StreamProxyParam param);
void save(StreamProxyParam param, ErrorCallback<StreamInfo> callback);
/**
* 分页查询
@@ -38,7 +39,7 @@ public interface IStreamProxyService {
* @param stream
* @return
*/
boolean startByAppAndStream(String app, String stream);
void startByAppAndStream(String app, String stream, ErrorCallback<StreamInfo> callback);
/**
* 停用用视频代理

133
src/main/java/com/genersoft/iot/vmp/streamProxy/service/impl/StreamProxyPlayServiceImpl.java
View File

@@ -4,12 +4,10 @@ import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.media.bean.MediaInfo;
import com.genersoft.iot.vmp.media.bean.MediaServer;
import com.genersoft.iot.vmp.media.event.hook.Hook;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;
import com.genersoft.iot.vmp.media.event.hook.HookType;
import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.service.bean.ErrorCallback;
import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
@@ -20,16 +18,12 @@ import com.genersoft.iot.vmp.streamProxy.service.IStreamProxyPlayService;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
import javax.sip.message.Response;
import javax.validation.constraints.NotNull;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
/**
* 视频代理业务
@@ -56,107 +50,42 @@ public class StreamProxyPlayServiceImpl implements IStreamProxyPlayService {
@Autowired
private IRedisRpcPlayService redisRpcPlayService;
private ConcurrentHashMap<Integer, ErrorCallback<StreamInfo>> callbackMap = new ConcurrentHashMap<>();
private ConcurrentHashMap<Integer, StreamInfo> streamInfoMap = new ConcurrentHashMap<>();
/**
* 流到来的处理
*/
@Async("taskExecutor")
@Transactional
@EventListener
public void onApplicationEvent(MediaArrivalEvent event) {
if ("rtsp".equals(event.getSchema())) {
StreamProxy streamProxy = streamProxyMapper.selectOneByAppAndStream(event.getApp(), event.getStream());
if (streamProxy != null) {
ErrorCallback<StreamInfo> callback = callbackMap.remove(streamProxy.getId());
StreamInfo streamInfo = streamInfoMap.remove(streamProxy.getId());
if (callback != null && streamInfo != null) {
callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo);
}
}
}
}
@Override
public void start(int id, ErrorCallback<StreamInfo> callback) {
StreamProxy streamProxy = streamProxyMapper.select(id);
if (streamProxy == null) {
throw new ControllerException(ErrorCode.ERROR404.getCode(), "代理信息未找到");
}
StreamInfo streamInfo = startProxy(streamProxy);
if (streamInfo == null) {
callback.run(Response.BUSY_HERE, "busy here", null);
return;
}
callbackMap.put(id, callback);
streamInfoMap.put(id, streamInfo);
MediaServer mediaServer = mediaServerService.getOne(streamProxy.getMediaServerId());
if (mediaServer != null) {
MediaInfo mediaInfo = mediaServerService.getMediaInfo(mediaServer, streamProxy.getApp(), streamProxy.getStream());
if (mediaInfo != null) {
callbackMap.remove(id);
streamInfoMap.remove(id);
callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo);
}
}
}
@Override
public StreamInfo start(int id, Boolean record, ErrorCallback<StreamInfo> callback) {
public void start(int id, Boolean record, ErrorCallback<StreamInfo> callback) {
log.info("[拉流代理]， 开始拉流，ID：{}", id);
StreamProxy streamProxy = streamProxyMapper.select(id);
if (streamProxy == null) {
throw new ControllerException(ErrorCode.ERROR404.getCode(), "代理信息未找到");
}
log.info("[拉流代理] 类型： {}， app：{}, stream: {}, 流地址： {}", streamProxy.getType(), streamProxy.getApp(), streamProxy.getStream(), streamProxy.getSrcUrl());
if (record != null) {
streamProxy.setEnableMp4(record);
}
if (streamProxy.getMediaServerId() != null) {
StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStreamWithCheck(streamProxy.getApp(), streamProxy.getStream(), streamProxy.getMediaServerId(), false);
if (streamInfo != null) {
callbackMap.remove(id);
streamInfoMap.remove(id);
callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo);
return streamInfo;
}
}
StreamInfo streamInfo = startProxy(streamProxy);
if (callback != null) {
// 设置流超时的定时任务
String timeOutTaskKey = UUID.randomUUID().toString();
Hook rtpHook = Hook.getInstance(HookType.on_media_arrival, streamProxy.getApp(), streamProxy.getStream(), streamInfo.getMediaServer().getId());
dynamicTask.startDelay(timeOutTaskKey, () -> {
log.info("[拉流代理]， 收流超时，ID：{}", id);
// 收流超时
subscribe.removeSubscribe(rtpHook);
callback.run(InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode(), InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getMsg(), streamInfo);
}, userSetting.getPlayTimeout());
// 开启流到来的监听
subscribe.addSubscribe(rtpHook, (hookData) -> {
dynamicTask.stop(timeOutTaskKey);
// hook响应
callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo);
subscribe.removeSubscribe(rtpHook);
});
}
return streamInfo;
startProxy(streamProxy, callback);
}
@Override
public StreamInfo startProxy(StreamProxy streamProxy){
public void startProxy(@NotNull StreamProxy streamProxy, ErrorCallback<StreamInfo> callback){
if (!streamProxy.isEnable()) {
return null;
callback.run(ErrorCode.ERROR100.getCode(), "代理未启用", null);
return;
}
if (streamProxy.getServerId() == null) {
streamProxy.setServerId(userSetting.getServerId());
}
if (!userSetting.getServerId().equals(streamProxy.getServerId())) {
return redisRpcPlayService.playProxy(streamProxy.getServerId(), streamProxy.getId());
log.info("[拉流代理] 由其他服务{}管理", streamProxy.getServerId());
redisRpcPlayService.playProxy(streamProxy.getServerId(), streamProxy.getId(), callback);
return;
}
if (streamProxy.getMediaServerId() != null) {
StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStreamWithCheck(streamProxy.getApp(), streamProxy.getStream(), streamProxy.getMediaServerId(), null, false);
if (streamInfo != null) {
callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), streamInfo);
return;
}
}
MediaServer mediaServer;
@@ -169,12 +98,32 @@ public class StreamProxyPlayServiceImpl implements IStreamProxyPlayService {
if (mediaServer == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), mediaServerId == null?"未找到可用的媒体节点":"未找到节点" + mediaServerId);
}
StreamInfo streamInfo = mediaServerService.startProxy(mediaServer, streamProxy);
// 设置流超时的定时任务
String timeOutTaskKey = UUID.randomUUID().toString();
Hook rtpHook = Hook.getInstance(HookType.on_media_arrival, streamProxy.getApp(), streamProxy.getStream(), mediaServer.getId());
dynamicTask.startDelay(timeOutTaskKey, () -> {
log.info("[拉流代理] 收流超时，app：{}，stream: {}", streamProxy.getApp(), streamProxy.getStream());
// 收流超时
subscribe.removeSubscribe(rtpHook);
callback.run(InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode(), InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getMsg(), null);
}, userSetting.getPlayTimeout());
// 开启流到来的监听
subscribe.addSubscribe(rtpHook, (hookData) -> {
log.info("[拉流代理] 收流成功，app：{}，stream: {}", hookData.getApp(), hookData.getStream());
dynamicTask.stop(timeOutTaskKey);
StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStream(mediaServer, hookData.getApp(), hookData.getStream(), hookData.getMediaInfo(), null);
// hook响应
callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo);
subscribe.removeSubscribe(rtpHook);
});
mediaServerService.startProxy(mediaServer, streamProxy);
if (mediaServerId == null || !mediaServerId.equals(mediaServer.getId())) {
streamProxy.setMediaServerId(mediaServer.getId());
streamProxyMapper.addStream(streamProxy);
streamProxyMapper.updateStream(streamProxy);
}
return streamInfo;
}
@Override

18
src/main/java/com/genersoft/iot/vmp/streamProxy/service/impl/StreamProxyServiceImpl.java
View File

@@ -15,6 +15,7 @@ import com.genersoft.iot.vmp.media.event.mediaServer.MediaServerOfflineEvent;
import com.genersoft.iot.vmp.media.event.mediaServer.MediaServerOnlineEvent;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OriginType;
import com.genersoft.iot.vmp.service.bean.ErrorCallback;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.streamProxy.bean.StreamProxy;
import com.genersoft.iot.vmp.streamProxy.bean.StreamProxyParam;
@@ -109,7 +110,9 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
// 拉流代理
StreamProxy streamProxyByAppAndStream = getStreamProxyByAppAndStream(event.getApp(), event.getStream());
if (streamProxyByAppAndStream != null && streamProxyByAppAndStream.isEnableDisableNoneReader()) {
startByAppAndStream(event.getApp(), event.getStream());
startByAppAndStream(event.getApp(), event.getStream(), ((code, msg, data) -> {
log.info("[拉流代理] 自动点播成功， app： {}， stream: {}", event.getApp(), event.getStream());
}));
}
}
@@ -136,7 +139,7 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
@Override
@Transactional
public StreamInfo save(StreamProxyParam param) {
public void save(StreamProxyParam param, ErrorCallback<StreamInfo> callback) {
// 兼容旧接口
StreamProxy streamProxyInDb = getStreamProxyByAppAndStream(param.getApp(), param.getStream());
if (streamProxyInDb != null && streamProxyInDb.getPulling() != null && streamProxyInDb.getPulling()) {
@@ -159,9 +162,7 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
}
if (param.isEnable()) {
return playService.startProxy(streamProxy);
} else {
return null;
playService.startProxy(streamProxy, callback);
}
}
@@ -247,13 +248,12 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
@Override
public boolean startByAppAndStream(String app, String stream) {
public void startByAppAndStream(String app, String stream, ErrorCallback<StreamInfo> callback) {
StreamProxy streamProxy = streamProxyMapper.selectOneByAppAndStream(app, stream);
if (streamProxy == null) {
throw new ControllerException(ErrorCode.ERROR404.getCode(), "代理信息未找到");
}
StreamInfo streamInfo = playService.startProxy(streamProxy);
return streamInfo != null;
playService.startProxy(streamProxy, callback);
}
@Override
@@ -406,7 +406,7 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
streamProxy.setPulling(status);
streamProxy.setMediaServerId(mediaServerId);
streamProxy.setUpdateTime(DateUtil.getNow());
streamProxyMapper.addStream(streamProxy);
streamProxyMapper.updateStream(streamProxy);
streamProxy.setGbStatus(status ? "ON" : "OFF");
if (streamProxy.getGbId() > 0) {