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

Merge branch 'refs/heads/master' into develop-add-api-key

Browse Source
This commit is contained in:
leesam
2024-04-10 20:49:44 +08:00
parent b6d70c6807 0f55872dbb
commit 16b7e4a7ef
124 changed files with 4940 additions and 3762 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
src/main/java/com/genersoft/iot/vmp/vmanager/bean/StreamContent.java
View File

@@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.vmanager.bean;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.media.bean.MediaInfo;
import com.genersoft.iot.vmp.service.bean.DownloadFileInfo;
import io.swagger.v3.oas.annotations.media.Schema;
@@ -86,7 +87,7 @@ public class StreamContent {
private String mediaServerId;
@Schema(description = "流编码信息")
private Object tracks;
private MediaInfo mediaInfo;
@Schema(description = "开始时间")
private String startTime;
@@ -170,7 +171,7 @@ public class StreamContent {
}
this.mediaServerId = streamInfo.getMediaServerId();
this.tracks = streamInfo.getTracks();
this.mediaInfo = streamInfo.getMediaInfo();
this.startTime = streamInfo.getStartTime();
this.endTime = streamInfo.getEndTime();
this.progress = streamInfo.getProgress();
@@ -388,12 +389,12 @@ public class StreamContent {
this.mediaServerId = mediaServerId;
}
public Object getTracks() {
return tracks;
public MediaInfo getMediaInfo() {
return mediaInfo;
}
public void setTracks(Object tracks) {
this.tracks = tracks;
public void setMediaInfo(MediaInfo mediaInfo) {
this.mediaInfo = mediaInfo;
}
public String getStartTime() {

18
src/main/java/com/genersoft/iot/vmp/vmanager/cloudRecord/CloudRecordController.java
View File

@@ -3,9 +3,9 @@ package com.genersoft.iot.vmp.vmanager.cloudRecord;
import com.alibaba.fastjson2.JSONArray;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.conf.security.JwtUtils;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.bean.MediaServer;
import com.genersoft.iot.vmp.service.ICloudRecordService;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.service.bean.CloudRecordItem;
import com.genersoft.iot.vmp.service.bean.DownloadFileInfo;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
@@ -67,16 +67,16 @@ public class CloudRecordController {
if (ObjectUtils.isEmpty(month)) {
month = calendar.get(Calendar.MONTH) + 1;
}
List<MediaServerItem> mediaServerItems;
List<MediaServer> mediaServerItems;
if (!ObjectUtils.isEmpty(mediaServerId)) {
mediaServerItems = new ArrayList<>();
MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
MediaServer mediaServerItem = mediaServerService.getOne(mediaServerId);
if (mediaServerItem == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到流媒体: " + mediaServerId);
}
mediaServerItems.add(mediaServerItem);
} else {
mediaServerItems = mediaServerService.getAll();
mediaServerItems = mediaServerService.getAllOnlineList();
}
if (mediaServerItems.isEmpty()) {
return new ArrayList<>();
@@ -110,16 +110,16 @@ public class CloudRecordController {
logger.info("[云端录像] 查询 app->{}, stream->{}, mediaServerId->{}, page->{}, count->{}, startTime->{}, endTime->{}",
app, stream, mediaServerId, page, count, startTime, endTime);
List<MediaServerItem> mediaServerItems;
List<MediaServer> mediaServerItems;
if (!ObjectUtils.isEmpty(mediaServerId)) {
mediaServerItems = new ArrayList<>();
MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
MediaServer mediaServerItem = mediaServerService.getOne(mediaServerId);
if (mediaServerItem == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到流媒体: " + mediaServerId);
}
mediaServerItems.add(mediaServerItem);
} else {
mediaServerItems = mediaServerService.getAll();
mediaServerItems = mediaServerService.getAllOnlineList();
}
if (mediaServerItems.isEmpty()) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "当前无流媒体");
@@ -162,7 +162,7 @@ public class CloudRecordController {
@RequestParam(required = false) String callId,
@RequestParam(required = false) String remoteHost
){
MediaServerItem mediaServerItem;
MediaServer mediaServerItem;
if (mediaServerId == null) {
mediaServerItem = mediaServerService.getDefaultMediaServer();
}else {

14
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/media/MediaController.java
View File

@@ -5,6 +5,7 @@ import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.conf.security.JwtUtils;
import com.genersoft.iot.vmp.conf.security.SecurityUtils;
import com.genersoft.iot.vmp.conf.security.dto.LoginUser;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo;
import com.genersoft.iot.vmp.service.IMediaService;
import com.genersoft.iot.vmp.service.IStreamProxyService;
@@ -35,11 +36,12 @@ public class MediaController {
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Autowired
private IMediaService mediaService;
@Autowired
private IStreamProxyService streamProxyService;
@Autowired
private IMediaServerService mediaServerService;
/**
* 根据应用名和流id获取播放地址
@@ -85,9 +87,9 @@ public class MediaController {
String host = request.getHeader("Host");
String localAddr = host.split(":")[0];
logger.info("使用{}作为返回流的ip", localAddr);
streamInfo = mediaService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, localAddr, authority);
streamInfo = mediaServerService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, localAddr, authority);
}else {
streamInfo = mediaService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, authority);
streamInfo = mediaServerService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, authority);
}
if (streamInfo != null){
@@ -105,9 +107,9 @@ public class MediaController {
String host = request.getHeader("Host");
String localAddr = host.split(":")[0];
logger.info("使用{}作为返回流的ip", localAddr);
streamInfo = mediaService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, localAddr, authority);
streamInfo = mediaServerService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, localAddr, authority);
}else {
streamInfo = mediaService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, authority);
streamInfo = mediaServerService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, authority);
}
if (streamInfo != null){
return new StreamContent(streamInfo);

75
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java
View File

@@ -2,7 +2,6 @@ package com.genersoft.iot.vmp.vmanager.gb28181.play;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.InviteInfo;
import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.UserSetting;
@@ -14,14 +13,11 @@ import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.bean.MediaServer;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IInviteStreamService;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IMediaService;
import com.genersoft.iot.vmp.service.IPlayService;
import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.vmanager.bean.AudioBroadcastResult;
@@ -65,24 +61,15 @@ public class PlayController {
@Autowired
private IVideoManagerStorage storager;
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Autowired
private IInviteStreamService inviteStreamService;
@Autowired
private ZLMRESTfulUtils zlmresTfulUtils;
@Autowired
private DeferredResultHolder resultHolder;
@Autowired
private IPlayService playService;
@Autowired
private IMediaService mediaService;
@Autowired
private IMediaServerService mediaServerService;
@@ -99,7 +86,7 @@ public class PlayController {
logger.info("[开始点播] deviceId：{}, channelId：{}, ", deviceId, channelId);
// 获取可用的zlm
Device device = storager.queryVideoDevice(deviceId);
MediaServerItem newMediaServerItem = playService.getNewMediaServerItem(device);
MediaServer newMediaServerItem = playService.getNewMediaServerItem(device);
RequestMessage requestMessage = new RequestMessage();
String key = DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId;
@@ -181,50 +168,6 @@ public class PlayController {
json.put("channelId", channelId);
return json;
}
/**
* 将不是h264的视频通过ffmpeg 转码为h264 + aac
* @param streamId 流ID
*/
@Operation(summary = "将不是h264的视频通过ffmpeg 转码为h264 + aac", security = @SecurityRequirement(name = JwtUtils.HEADER))
@Parameter(name = "streamId", description = "视频流ID", required = true)
@PostMapping("/convert/{streamId}")
public JSONObject playConvert(@PathVariable String streamId) {
// StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(streamId);
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, streamId);
if (inviteInfo == null || inviteInfo.getStreamInfo() == null) {
logger.warn("视频转码API调用失败！, 视频流已经停止!");
throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到视频流信息, 视频流可能已经停止");
}
MediaServerItem mediaInfo = mediaServerService.getOne(inviteInfo.getStreamInfo().getMediaServerId());
JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaInfo, streamId);
if (!rtpInfo.getBoolean("exist")) {
logger.warn("视频转码API调用失败！, 视频流已停止推流!");
throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到视频流信息, 视频流可能已停止推流");
} else {
String dstUrl = String.format("rtmp://%s:%s/convert/%s", "127.0.0.1", mediaInfo.getRtmpPort(),
streamId );
String srcUrl = String.format("rtsp://%s:%s/rtp/%s", "127.0.0.1", mediaInfo.getRtspPort(), streamId);
JSONObject jsonObject = zlmresTfulUtils.addFFmpegSource(mediaInfo, srcUrl, dstUrl, "1000000", true, false, null);
logger.info(jsonObject.toJSONString());
if (jsonObject != null && jsonObject.getInteger("code") == 0) {
JSONObject data = jsonObject.getJSONObject("data");
if (data != null) {
JSONObject result = new JSONObject();
result.put("key", data.getString("key"));
StreamInfo streamInfoResult = mediaService.getStreamInfoByAppAndStreamWithCheck("convert", streamId, mediaInfo.getId(), false);
result.put("StreamInfo", streamInfoResult);
return result;
}else {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "转码失败");
}
}else {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "转码失败");
}
}
}
/**
* 结束转码
*/
@@ -236,18 +179,12 @@ public class PlayController {
if (mediaServerId == null) {
throw new ControllerException(ErrorCode.ERROR400.getCode(), "流媒体：" + mediaServerId + "不存在" );
}
MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);
MediaServer mediaInfo = mediaServerService.getOne(mediaServerId);
if (mediaInfo == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "使用的流媒体已经停止运行" );
}else {
JSONObject jsonObject = zlmresTfulUtils.delFFmpegSource(mediaInfo, key);
logger.info(jsonObject.toJSONString());
if (jsonObject != null && jsonObject.getInteger("code") == 0) {
JSONObject data = jsonObject.getJSONObject("data");
if (data == null || data.getBoolean("flag") == null || !data.getBoolean("flag")) {
throw new ControllerException(ErrorCode.ERROR100 );
}
}else {
Boolean deleted = mediaServerService.delFFmpegSource(mediaInfo, key);
if (!deleted) {
throw new ControllerException(ErrorCode.ERROR100 );
}
}

49
src/main/java/com/genersoft/iot/vmp/vmanager/ps/PsController.java
View File

@@ -6,15 +6,13 @@ 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.conf.security.JwtUtils;
import com.genersoft.iot.vmp.media.event.hook.Hook;
import com.genersoft.iot.vmp.media.event.hook.HookType;
import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager;
import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForRtpServerTimeout;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnRtpServerTimeoutHookParam;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;
import com.genersoft.iot.vmp.media.bean.MediaServer;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.utils.redis.RedisUtil;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.OtherPsSendInfo;
@@ -50,7 +48,7 @@ public class PsController {
private ZLMServerFactory zlmServerFactory;
@Autowired
private ZlmHttpHookSubscribe hookSubscribe;
private HookSubscribe hookSubscribe;
@Autowired
private IMediaServerService mediaServerService;
@@ -83,7 +81,7 @@ public class PsController {
logger.info("[第三方PS服务对接->开启收流和获取发流信息] isSend->{}, ssrc->{}, callId->{}, stream->{}, tcpMode->{}, callBack->{}",
isSend, ssrc, callId, stream, tcpMode==0?"UDP":"TCP被动", callBack);
MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer();
MediaServer mediaServerItem = mediaServerService.getDefaultMediaServer();
if (mediaServerItem == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(),"没有可用的MediaServer");
}
@@ -108,12 +106,11 @@ public class PsController {
}
// 注册回调如果rtp收流超时则通过回调发送通知
if (callBack != null) {
HookSubscribeForRtpServerTimeout hookSubscribeForRtpServerTimeout = HookSubscribeFactory.on_rtp_server_timeout(stream, String.valueOf(ssrcInt), mediaServerItem.getId());
Hook hook = Hook.getInstance(HookType.on_rtp_server_timeout, "rtp", stream, mediaServerItem.getId());
// 订阅 zlm启动事件, 新的zlm也会从这里进入系统
hookSubscribe.addSubscribe(hookSubscribeForRtpServerTimeout,
(mediaServerItemInUse, hookParam)->{
OnRtpServerTimeoutHookParam serverTimeoutHookParam = (OnRtpServerTimeoutHookParam) hookParam;
if (stream.equals(serverTimeoutHookParam.getStream_id())) {
hookSubscribe.addSubscribe(hook,
(hookData)->{
if (stream.equals(hookData.getStream())) {
logger.info("[第三方PS服务对接->开启收流和获取发流信息] 等待收流超时 callId->{}, 发送回调", callId);
// 将信息写入redis中，以备后用
redisTemplate.delete(receiveKey);
@@ -126,7 +123,7 @@ public class PsController {
} catch (IOException e) {
logger.error("[第三方PS服务对接->开启收流和获取发流信息] 等待收流超时 callId->{}, 发送回调失败", callId, e);
}
hookSubscribe.removeSubscribe(hookSubscribeForRtpServerTimeout);
hookSubscribe.removeSubscribe(hook);
}
});
}
@@ -158,7 +155,7 @@ public class PsController {
@Parameter(name = "stream", description = "流的ID", required = true)
public void closeRtpServer(String stream) {
logger.info("[第三方PS服务对接->关闭收流] stream->{}", stream);
MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer();
MediaServer mediaServerItem = mediaServerService.getDefaultMediaServer();
zlmServerFactory.closeRtpServer(mediaServerItem,stream);
String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_PS_INFO + userSetting.getServerId() + "_*_" + stream;
List<Object> scan = RedisUtil.scan(redisTemplate, receiveKey);
@@ -201,7 +198,7 @@ public class PsController {
app,
stream,
callId);
MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer();
MediaServer mediaServerItem = mediaServerService.getDefaultMediaServer();
String key = VideoManagerConstants.WVP_OTHER_SEND_PS_INFO + userSetting.getServerId() + "_" + callId;
OtherPsSendInfo sendInfo = (OtherPsSendInfo)redisTemplate.opsForValue().get(key);
if (sendInfo == null) {
@@ -227,7 +224,7 @@ public class PsController {
param.put("src_port", sendInfo.getSendLocalPort());
Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, app, stream);
Boolean streamReady = mediaServerService.isStreamReady(mediaServerItem, app, stream);
if (streamReady) {
JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServerItem, param);
if (jsonObject.getInteger("code") == 0) {
@@ -241,18 +238,18 @@ public class PsController {
}else {
logger.info("[第三方PS服务对接->发送流] 流不存在，等待流上线，callId->{}", callId);
String uuid = UUID.randomUUID().toString();
HookSubscribeForStreamChange hookSubscribeForStreamChange = HookSubscribeFactory.on_stream_changed(app, stream, true, "rtsp", mediaServerItem.getId());
Hook hook = Hook.getInstance(HookType.on_media_arrival, app, stream, mediaServerItem.getId());
dynamicTask.startDelay(uuid, ()->{
logger.info("[第三方PS服务对接->发送流] 等待流上线超时 callId->{}", callId);
redisTemplate.delete(key);
hookSubscribe.removeSubscribe(hookSubscribeForStreamChange);
hookSubscribe.removeSubscribe(hook);
}, 10000);
// 订阅 zlm启动事件, 新的zlm也会从这里进入系统
OtherPsSendInfo finalSendInfo = sendInfo;
hookSubscribe.removeSubscribe(hookSubscribeForStreamChange);
hookSubscribe.addSubscribe(hookSubscribeForStreamChange,
(mediaServerItemInUse, response)->{
hookSubscribe.removeSubscribe(hook);
hookSubscribe.addSubscribe(hook,
(hookData)->{
dynamicTask.stop(uuid);
logger.info("[第三方PS服务对接->发送流] 流上线，开始发流 callId->{}", callId);
try {
@@ -269,7 +266,7 @@ public class PsController {
logger.info("[第三方PS服务对接->发送流] 视频流发流失败，callId->{}, {}", callId, jsonObject.getString("msg"));
throw new ControllerException(ErrorCode.ERROR100.getCode(), "[视频流发流失败] " + jsonObject.getString("msg"));
}
hookSubscribe.removeSubscribe(hookSubscribeForStreamChange);
hookSubscribe.removeSubscribe(hook);
});
}
}
@@ -290,7 +287,7 @@ public class PsController {
param.put("app",sendInfo.getPushApp());
param.put("stream",sendInfo.getPushStream());
param.put("ssrc",sendInfo.getPushSSRC());
MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer();
MediaServer mediaServerItem = mediaServerService.getDefaultMediaServer();
Boolean result = zlmServerFactory.stopSendRtpStream(mediaServerItem, param);
if (!result) {
logger.info("[第三方PS服务对接->关闭发送流] 失败 callId->{}", callId);
@@ -305,7 +302,7 @@ public class PsController {
@GetMapping(value = "/getTestPort")
@ResponseBody
public int getTestPort() {
MediaServerItem defaultMediaServer = mediaServerService.getDefaultMediaServer();
MediaServer defaultMediaServer = mediaServerService.getDefaultMediaServer();
// for (int i = 0; i <300; i++) {
// new Thread(() -> {

55
src/main/java/com/genersoft/iot/vmp/vmanager/rtp/RtpController.java
View File

@@ -6,15 +6,13 @@ 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.conf.security.JwtUtils;
import com.genersoft.iot.vmp.media.event.hook.Hook;
import com.genersoft.iot.vmp.media.event.hook.HookType;
import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager;
import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForRtpServerTimeout;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnRtpServerTimeoutHookParam;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;
import com.genersoft.iot.vmp.media.bean.MediaServer;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.utils.redis.RedisUtil;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.OtherRtpSendInfo;
@@ -54,7 +52,7 @@ public class RtpController {
private final static Logger logger = LoggerFactory.getLogger(RtpController.class);
@Autowired
private ZlmHttpHookSubscribe hookSubscribe;
private HookSubscribe hookSubscribe;
@Autowired
private IMediaServerService mediaServerService;
@@ -83,7 +81,7 @@ public class RtpController {
logger.info("[第三方服务对接->开启收流和获取发流信息] isSend->{}, ssrc->{}, callId->{}, stream->{}, tcpMode->{}, callBack->{}",
isSend, ssrc, callId, stream, tcpMode==0?"UDP":"TCP被动", callBack);
MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer();
MediaServer mediaServerItem = mediaServerService.getDefaultMediaServer();
if (mediaServerItem == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(),"没有可用的MediaServer");
}
@@ -109,12 +107,11 @@ public class RtpController {
}
// 注册回调如果rtp收流超时则通过回调发送通知
if (callBack != null) {
HookSubscribeForRtpServerTimeout hookSubscribeForRtpServerTimeout = HookSubscribeFactory.on_rtp_server_timeout(stream, String.valueOf(ssrcInt), mediaServerItem.getId());
Hook hook = Hook.getInstance(HookType.on_rtp_server_timeout, "rtp", stream, mediaServerItem.getId());
// 订阅 zlm启动事件, 新的zlm也会从这里进入系统
hookSubscribe.addSubscribe(hookSubscribeForRtpServerTimeout,
(mediaServerItemInUse, hookParam)->{
OnRtpServerTimeoutHookParam serverTimeoutHookParam = (OnRtpServerTimeoutHookParam) hookParam;
if (stream.equals(serverTimeoutHookParam.getStream_id())) {
hookSubscribe.addSubscribe(hook,
(hookData)->{
if (stream.equals(hookData.getStream())) {
logger.info("[开启收流和获取发流信息] 等待收流超时 callId->{}, 发送回调", callId);
OkHttpClient.Builder httpClientBuilder = new OkHttpClient.Builder();
OkHttpClient client = httpClientBuilder.build();
@@ -125,7 +122,7 @@ public class RtpController {
} catch (IOException e) {
logger.error("[第三方服务对接->开启收流和获取发流信息] 等待收流超时 callId->{}, 发送回调失败", callId, e);
}
hookSubscribe.removeSubscribe(hookSubscribeForRtpServerTimeout);
hookSubscribe.removeSubscribe(hook);
}
});
}
@@ -162,7 +159,7 @@ public class RtpController {
@Parameter(name = "stream", description = "流的ID", required = true)
public void closeRtpServer(String stream) {
logger.info("[第三方服务对接->关闭收流] stream->{}", stream);
MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer();
MediaServer mediaServerItem = mediaServerService.getDefaultMediaServer();
zlmServerFactory.closeRtpServer(mediaServerItem,stream);
zlmServerFactory.closeRtpServer(mediaServerItem,stream + "_a");
String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "_*_" + stream;
@@ -225,7 +222,7 @@ public class RtpController {
if (!((dstPortForAudio > 0 && !ObjectUtils.isEmpty(dstPortForAudio) || (dstPortForVideo > 0 && !ObjectUtils.isEmpty(dstIpForVideo))))) {
throw new ControllerException(ErrorCode.ERROR400.getCode(), "至少应该存在一组音频或视频发送参数");
}
MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer();
MediaServer mediaServer = mediaServerService.getDefaultMediaServer();
String key = VideoManagerConstants.WVP_OTHER_SEND_RTP_INFO + userSetting.getServerId() + "_" + callId;
OtherRtpSendInfo sendInfo = (OtherRtpSendInfo)redisTemplate.opsForValue().get(key);
if (sendInfo == null) {
@@ -278,10 +275,10 @@ public class RtpController {
paramForVideo = null;
}
Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, app, stream);
Boolean streamReady = mediaServerService.isStreamReady(mediaServer, app, stream);
if (streamReady) {
if (paramForVideo != null) {
JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServerItem, paramForVideo);
JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServer, paramForVideo);
if (jsonObject.getInteger("code") == 0) {
logger.info("[第三方服务对接->发送流] 视频流发流成功，callId->{}，param->{}", callId, paramForVideo);
redisTemplate.opsForValue().set(key, sendInfo);
@@ -292,7 +289,7 @@ public class RtpController {
}
}
if(paramForAudio != null) {
JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServerItem, paramForAudio);
JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServer, paramForAudio);
if (jsonObject.getInteger("code") == 0) {
logger.info("[第三方服务对接->发送流] 音频流发流成功，callId->{}，param->{}", callId, paramForAudio);
redisTemplate.opsForValue().set(key, sendInfo);
@@ -305,18 +302,18 @@ public class RtpController {
}else {
logger.info("[第三方服务对接->发送流] 流不存在，等待流上线，callId->{}", callId);
String uuid = UUID.randomUUID().toString();
HookSubscribeForStreamChange hookSubscribeForStreamChange = HookSubscribeFactory.on_stream_changed(app, stream, true, "rtsp", mediaServerItem.getId());
Hook hook = Hook.getInstance(HookType.on_media_arrival, app, stream, mediaServer.getId());
dynamicTask.startDelay(uuid, ()->{
logger.info("[第三方服务对接->发送流] 等待流上线超时 callId->{}", callId);
redisTemplate.delete(key);
hookSubscribe.removeSubscribe(hookSubscribeForStreamChange);
hookSubscribe.removeSubscribe(hook);
}, 10000);
// 订阅 zlm启动事件, 新的zlm也会从这里进入系统
OtherRtpSendInfo finalSendInfo = sendInfo;
hookSubscribe.removeSubscribe(hookSubscribeForStreamChange);
hookSubscribe.addSubscribe(hookSubscribeForStreamChange,
(mediaServerItemInUse, response)->{
hookSubscribe.removeSubscribe(hook);
hookSubscribe.addSubscribe(hook,
(hookData)->{
dynamicTask.stop(uuid);
logger.info("[第三方服务对接->发送流] 流上线，开始发流 callId->{}", callId);
try {
@@ -325,7 +322,7 @@ public class RtpController {
throw new RuntimeException(e);
}
if (paramForVideo != null) {
JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServerItem, paramForVideo);
JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServer, paramForVideo);
if (jsonObject.getInteger("code") == 0) {
logger.info("[第三方服务对接->发送流] 视频流发流成功，callId->{}，param->{}", callId, paramForVideo);
redisTemplate.opsForValue().set(key, finalSendInfo);
@@ -336,7 +333,7 @@ public class RtpController {
}
}
if(paramForAudio != null) {
JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServerItem, paramForAudio);
JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServer, paramForAudio);
if (jsonObject.getInteger("code") == 0) {
logger.info("[第三方服务对接->发送流] 音频流发流成功，callId->{}，param->{}", callId, paramForAudio);
redisTemplate.opsForValue().set(key, finalSendInfo);
@@ -346,7 +343,7 @@ public class RtpController {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "[音频流发流失败] " + jsonObject.getString("msg"));
}
}
hookSubscribe.removeSubscribe(hookSubscribeForStreamChange);
hookSubscribe.removeSubscribe(hook);
});
}
}
@@ -367,7 +364,7 @@ public class RtpController {
param.put("app",sendInfo.getPushApp());
param.put("stream",sendInfo.getPushStream());
param.put("ssrc",sendInfo.getPushSSRC());
MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer();
MediaServer mediaServerItem = mediaServerService.getDefaultMediaServer();
Boolean result = zlmServerFactory.stopSendRtpStream(mediaServerItem, param);
if (!result) {
logger.info("[第三方服务对接->关闭发送流] 失败 callId->{}", callId);

39
src/main/java/com/genersoft/iot/vmp/vmanager/server/ServerController.java
View File

@@ -9,10 +9,8 @@ import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.VersionInfo;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.conf.security.JwtUtils;
import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager;
import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.dto.IHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.media.bean.MediaServer;
import com.genersoft.iot.vmp.service.*;
import com.genersoft.iot.vmp.service.bean.MediaServerLoad;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
@@ -39,8 +37,6 @@ import java.util.List;
@RequestMapping("/api/server")
public class ServerController {
@Autowired
private ZlmHttpHookSubscribe zlmHttpHookSubscribe;
@Autowired
private IMediaServerService mediaServerService;
@@ -75,21 +71,19 @@ public class ServerController {
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Autowired
private SendRtpPortManager sendRtpPortManager;
@GetMapping(value = "/media_server/list")
@ResponseBody
@Operation(summary = "流媒体服务列表", security = @SecurityRequirement(name = JwtUtils.HEADER))
public List<MediaServerItem> getMediaServerList() {
public List<MediaServer> getMediaServerList() {
return mediaServerService.getAll();
}
@GetMapping(value = "/media_server/online/list")
@ResponseBody
@Operation(summary = "在线流媒体服务列表", security = @SecurityRequirement(name = JwtUtils.HEADER))
public List<MediaServerItem> getOnlineMediaServerList() {
public List<MediaServer> getOnlineMediaServerList() {
return mediaServerService.getAllOnline();
}
@@ -97,7 +91,7 @@ public class ServerController {
@ResponseBody
@Operation(summary = "停止视频回放", security = @SecurityRequirement(name = JwtUtils.HEADER))
@Parameter(name = "id", description = "流媒体服务ID", required = true)
public MediaServerItem getMediaServer(@PathVariable String id) {
public MediaServer getMediaServer(@PathVariable String id) {
return mediaServerService.getOne(id);
}
@@ -107,8 +101,8 @@ public class ServerController {
@Parameter(name = "secret", description = "流媒体服务secret", required = true)
@GetMapping(value = "/media_server/check")
@ResponseBody
public MediaServerItem checkMediaServer(@RequestParam String ip, @RequestParam int port, @RequestParam String secret) {
return mediaServerService.checkMediaServer(ip, port, secret);
public MediaServer checkMediaServer(@RequestParam String ip, @RequestParam int port, @RequestParam String secret, @RequestParam String type) {
return mediaServerService.checkMediaServer(ip, port, secret, type);
}
@Operation(summary = "测试流媒体录像管理服务", security = @SecurityRequirement(name = JwtUtils.HEADER))
@@ -127,8 +121,8 @@ public class ServerController {
@Parameter(name = "mediaServerItem", description = "流媒体信息", required = true)
@PostMapping(value = "/media_server/save")
@ResponseBody
public void saveMediaServer(@RequestBody MediaServerItem mediaServerItem) {
MediaServerItem mediaServerItemInDatabase = mediaServerService.getOne(mediaServerItem.getId());
public void saveMediaServer(@RequestBody MediaServer mediaServerItem) {
MediaServer mediaServerItemInDatabase = mediaServerService.getOneFromDatabase(mediaServerItem.getId());
if (mediaServerItemInDatabase != null) {
mediaServerService.update(mediaServerItem);
@@ -142,11 +136,7 @@ public class ServerController {
@DeleteMapping(value = "/media_server/delete")
@ResponseBody
public void deleteMediaServer(@RequestParam String id) {
if (mediaServerService.getOne(id) == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到此节点");
}
mediaServerService.delete(id);
mediaServerService.deleteDb(id);
}
@@ -220,13 +210,6 @@ public class ServerController {
return jsonObject;
}
@GetMapping(value = "/hooks")
@ResponseBody
@Operation(summary = "获取当前所有hook")
public List<IHookSubscribe> getHooks() {
return zlmHttpHookSubscribe.getAll();
}
@GetMapping(value = "/system/info")
@ResponseBody
@Operation(summary = "获取系统信息")
@@ -241,11 +224,11 @@ public class ServerController {
@Operation(summary = "获取负载信息")
public List<MediaServerLoad> getMediaLoad() {
List<MediaServerLoad> result = new ArrayList<>();
List<MediaServerItem> allOnline = mediaServerService.getAllOnline();
List<MediaServer> allOnline = mediaServerService.getAllOnline();
if (allOnline.size() == 0) {
return result;
}else {
for (MediaServerItem mediaServerItem : allOnline) {
for (MediaServer mediaServerItem : allOnline) {
result.add(mediaServerService.getLoad(mediaServerItem));
}
}

9
src/main/java/com/genersoft/iot/vmp/vmanager/streamProxy/StreamProxyController.java
View File

@@ -7,9 +7,9 @@ import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.conf.security.JwtUtils;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.bean.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IStreamProxyService;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.StreamContent;
@@ -27,6 +27,7 @@ import org.springframework.util.ObjectUtils;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.context.request.async.DeferredResult;
import java.util.Map;
import java.util.UUID;
@SuppressWarnings("rawtypes")
@@ -135,10 +136,10 @@ public class StreamProxyController {
@ResponseBody
@Operation(summary = "获取ffmpeg.cmd模板", security = @SecurityRequirement(name = JwtUtils.HEADER))
@Parameter(name = "mediaServerId", description = "流媒体ID", required = true)
public JSONObject getFFmpegCMDs(@RequestParam String mediaServerId){
public Map<String, String> getFFmpegCMDs(@RequestParam String mediaServerId){
logger.debug("获取节点[ {} ]ffmpeg.cmd模板", mediaServerId );
MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
MediaServer mediaServerItem = mediaServerService.getOne(mediaServerId);
if (mediaServerItem == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "流媒体： " + mediaServerId + "未找到");
}

4
src/main/java/com/genersoft/iot/vmp/vmanager/streamPush/StreamPushController.java
View File

@@ -13,7 +13,7 @@ import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IMediaService;
import com.genersoft.iot.vmp.service.IStreamPushService;
import com.genersoft.iot.vmp.service.impl.StreamPushUploadFileHandler;
@@ -249,7 +249,7 @@ public class StreamPushController {
if (push != null && !push.isSelf()) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "来自其他平台的推流信息");
}
StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, authority);
StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, authority);
if (streamInfo == null){
throw new ControllerException(ErrorCode.ERROR100.getCode(), "获取播放地址失败");
}