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:
648540858
2023-05-25 17:28:57 +08:00
parent d295f3c512 512b0cc8f5
commit 30ae9e929f
129 changed files with 5909 additions and 2399 deletions
Download Patch File Download Diff File Expand all files Collapse all files

430
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
View File

@@ -2,12 +2,16 @@ package com.genersoft.iot.vmp.media.zlm;
import com.alibaba.fastjson2.JSON;
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;
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.gb28181.session.SSRCFactory;
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;
@@ -21,10 +25,8 @@ import com.genersoft.iot.vmp.media.zlm.dto.hook.*;
import com.genersoft.iot.vmp.service.*;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.vmanager.bean.DeferredResultEx;
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 org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -57,14 +59,14 @@ public class ZLMHttpHookListener {
@Autowired
private SIPCommander cmder;
@Autowired
private ISIPCommanderForPlatform commanderFroPlatform;
@Autowired
private ISIPCommanderForPlatform commanderFroPlatform;
@Autowired
private AudioBroadcastManager audioBroadcastManager;
@Autowired
private AudioBroadcastManager audioBroadcastManager;
@Autowired
private ZLMRTPServerFactory zlmrtpServerFactory;
@Autowired
private ZLMRTPServerFactory zlmrtpServerFactory;
@Autowired
private IPlayService playService;
@@ -75,6 +77,9 @@ public class ZLMHttpHookListener {
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Autowired
private IInviteStreamService inviteStreamService;
@Autowired
private IDeviceService deviceService;
@@ -111,6 +116,9 @@ public class ZLMHttpHookListener {
@Autowired
private AssistRESTfulUtils assistRESTfulUtils;
@Autowired
private SSRCFactory ssrcFactory;
@Qualifier("taskExecutor")
@Autowired
private ThreadPoolTaskExecutor taskExecutor;
@@ -255,13 +263,13 @@ public class ZLMHttpHookListener {
result.setEnable_audio(deviceChannel.isHasAudio());
}
// 如果是录像下载就设置视频间隔十秒
if (ssrcTransactionForAll.get(0).getType() == VideoStreamSessionManager.SessionType.download) {
if (ssrcTransactionForAll.get(0).getType() == InviteSessionType.DOWNLOAD) {
result.setMp4_max_second(10);
result.setEnable_audio(true);
result.setEnable_mp4(true);
}
// 如果是talk对讲，则默认获取声音
if (ssrcTransactionForAll.get(0).getType() == VideoStreamSessionManager.SessionType.talk) {
if (ssrcTransactionForAll.get(0).getType() == InviteSessionType.TALK) {
result.setEnable_audio(true);
}
@@ -269,7 +277,7 @@ public class ZLMHttpHookListener {
if (mediaInfo.getRecordAssistPort() > 0 && userSetting.getRecordPath() == null) {
logger.info("推流时发现尚未设置录像路径，从assist服务中读取");
JSONObject info = assistRESTfulUtils.getInfo(mediaInfo, null);
if (info != null && info.getInteger("code") != null && info.getInteger("code") == 0 ) {
if (info != null && info.getInteger("code") != null && info.getInteger("code") == 0) {
JSONObject dataJson = info.getJSONObject("data");
if (dataJson != null) {
String recordPath = dataJson.getString("record");
@@ -301,29 +309,30 @@ public class ZLMHttpHookListener {
logger.info("[ZLM HOOK] 流注销, {}->{}->{}/{}", param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());
}
JSONObject ret = new JSONObject();
ret.put("code", 0);
ret.put("msg", "success");
JSONObject ret = new JSONObject();
ret.put("code", 0);
ret.put("msg", "success");
MediaServerItem mediaInfo = mediaServerService.getOne(param.getMediaServerId());
JSONObject json = (JSONObject) JSON.toJSON(param);
taskExecutor.execute(() -> {
ZlmHttpHookSubscribe.Event subscribe = this.subscribe.sendNotify(HookType.on_stream_changed, json);
if (mediaInfo == null) {
logger.info("[ZLM HOOK] 流变化未找到ZLM, {}", param.getMediaServerId());
return;
}
if (subscribe != null) {
if (mediaInfo != null) {
subscribe.response(mediaInfo, json);
}
subscribe.response(mediaInfo, json);
}
List<OnStreamChangedHookParam.MediaTrack> tracks = param.getTracks();
// TODO 重构此处逻辑
boolean isPush = false;
if (param.isRegist()) {
// 处理流注册的鉴权信息
if (param.getOriginType() == OriginType.RTMP_PUSH.ordinal()
|| param.getOriginType() == OriginType.RTSP_PUSH.ordinal()
|| param.getOriginType() == OriginType.RTC_PUSH.ordinal()) {
isPush = true;
StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream());
if (streamAuthorityInfo == null) {
streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(param);
@@ -337,123 +346,122 @@ public class ZLMHttpHookListener {
redisCatchStorage.removeStreamAuthorityInfo(param.getApp(), param.getStream());
}
if ("rtsp".equals(param.getSchema())){
logger.info("流变化：注册->{}, app->{}, stream->{}", param.isRegist(), param.getApp(), param.getStream());
if (param.isRegist()) {
mediaServerService.addCount(param.getMediaServerId());
}else {
mediaServerService.removeCount(param.getMediaServerId());
}
if (param.getOriginType() == OriginType.PULL.ordinal()
|| param.getOriginType() == OriginType.FFMPEG_PULL.ordinal()) {
// 设置拉流代理上线/离线
streamProxyService.updateStatus(param.isRegist(), param.getApp(), param.getStream());
}
if ("rtp".equals(param.getApp()) && !param.isRegist() ) {
StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(param.getStream());
if (streamInfo!=null){
redisCatchStorage.stopPlay(streamInfo);
storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
}else{
streamInfo = redisCatchStorage.queryPlayback(null, null, param.getStream(), null);
if (streamInfo != null) {
redisCatchStorage.stopPlayback(streamInfo.getDeviceID(), streamInfo.getChannelId(),
streamInfo.getStream(), null);
}
}
}else if ("broadcast".equals(param.getApp())){
// 语音对讲推流 stream需要满足格式deviceId_channelId
if (param.getStream().indexOf("_") > 0) {
String[] streamArray = param.getStream().split("_");
if (streamArray.length == 2) {
String deviceId = streamArray[0];
String channelId = streamArray[1];
Device device = deviceService.getDevice(deviceId);
if (device != null) {
if (param.isRegist()) {
if (audioBroadcastManager.exit(deviceId, channelId)) {
if ("rtsp".equals(param.getSchema())) {
logger.info("流变化：注册->{}, app->{}, stream->{}", param.isRegist(), param.getApp(), param.getStream());
if (param.isRegist()) {
mediaServerService.addCount(param.getMediaServerId());
} else {
mediaServerService.removeCount(param.getMediaServerId());
}
int updateStatusResult = streamProxyService.updateStatus(param.isRegist(), param.getApp(), param.getStream());
if (updateStatusResult > 0) {
}
if ("rtp".equals(param.getApp()) && !param.isRegist()) {
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, param.getStream());
if (inviteInfo != null && (inviteInfo.getType() == InviteSessionType.PLAY || inviteInfo.getType() == InviteSessionType.PLAYBACK)) {
inviteStreamService.removeInviteInfo(inviteInfo);
storager.stopPlay(inviteInfo.getDeviceId(), inviteInfo.getChannelId());
}
} else if ("broadcast".equals(param.getApp())) {
// 语音对讲推流 stream需要满足格式deviceId_channelId
if (param.getStream().indexOf("_") > 0) {
String[] streamArray = param.getStream().split("_");
if (streamArray.length == 2) {
String deviceId = streamArray[0];
String channelId = streamArray[1];
Device device = deviceService.getDevice(deviceId);
if (device != null) {
if (param.isRegist()) {
if (audioBroadcastManager.exit(deviceId, channelId)) {
playService.stopAudioBroadcast(deviceId, channelId);
}
// 开启语音对讲通道
try {
playService.audioBroadcastCmd(device, channelId, mediaInfo, param.getApp(), param.getStream(), 60, false, (msg) -> {
logger.info("[语音对讲] 通道建立成功, device: {}, channel: {}", deviceId, channelId);
});
} catch (InvalidArgumentException | ParseException | SipException e) {
logger.error("[命令发送失败] 语音对讲: {}", e.getMessage());
}
} else {
// 流注销
playService.stopAudioBroadcast(deviceId, channelId);
}
// 开启语音对讲通道
try {
playService.audioBroadcastCmd(device, channelId, mediaInfo, param.getApp(), param.getStream(), 60, false, (msg)->{
} else {
logger.info("[语音对讲] 未找到设备：{}", deviceId);
}
}
}
} else if ("talk".equals(param.getApp())) {
// 语音对讲推流 stream需要满足格式deviceId_channelId
if (param.getStream().indexOf("_") > 0) {
String[] streamArray = param.getStream().split("_");
if (streamArray.length == 2) {
String deviceId = streamArray[0];
String channelId = streamArray[1];
Device device = deviceService.getDevice(deviceId);
if (device != null) {
if (param.isRegist()) {
if (audioBroadcastManager.exit(deviceId, channelId)) {
playService.stopAudioBroadcast(deviceId, channelId);
}
// 开启语音对讲通道
playService.talkCmd(device, channelId, mediaInfo, param.getStream(), (msg) -> {
logger.info("[语音对讲] 通道建立成功, device: {}, channel: {}", deviceId, channelId);
});
} catch (InvalidArgumentException | ParseException | SipException e) {
logger.error("[命令发送失败] 语音对讲: {}", e.getMessage());
} else {
// 流注销
playService.stopTalk(device, channelId, param.isRegist());
}
}else {
// 流注销
playService.stopAudioBroadcast(deviceId, channelId);
} else {
logger.info("[语音对讲] 未找到设备：{}", deviceId);
}
} else{
logger.info("[语音对讲] 未找到设备：{}", deviceId);
}
}
}
}else if ("talk".equals(param.getApp())){
// 语音对讲推流 stream需要满足格式deviceId_channelId
if (param.getStream().indexOf("_") > 0) {
String[] streamArray = param.getStream().split("_");
if (streamArray.length == 2) {
String deviceId = streamArray[0];
String channelId = streamArray[1];
Device device = deviceService.getDevice(deviceId);
if (device != null) {
} else {
if (!"rtp".equals(param.getApp())) {
String type = OriginType.values()[param.getOriginType()].getType();
MediaServerItem mediaServerItem = mediaServerService.getOne(param.getMediaServerId());
if (mediaServerItem != null) {
if (param.isRegist()) {
if (audioBroadcastManager.exit(deviceId, channelId)) {
playService.stopAudioBroadcast(deviceId, channelId);
StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream());
String callId = null;
if (streamAuthorityInfo != null) {
callId = streamAuthorityInfo.getCallId();
}
// 开启语音对讲通道
playService.talkCmd(device, channelId, mediaInfo, param.getStream(), (msg)->{
logger.info("[语音对讲] 通道建立成功, device: {}, channel: {}", deviceId, channelId);
});
}else {
// 流注销
playService.stopTalk(device, channelId, param.isRegist());
}
} else{
logger.info("[语音对讲] 未找到设备：{}", deviceId);
}
}
}
}else{
if (!"rtp".equals(param.getApp())){
String type = OriginType.values()[param.getOriginType()].getType();
MediaServerItem mediaServerItem = mediaServerService.getOne(param.getMediaServerId());
if (mediaServerItem != null){
if (param.isRegist()) {
StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream());
String callId = null;
if (streamAuthorityInfo != null) {
callId = streamAuthorityInfo.getCallId();
}
StreamInfo streamInfoByAppAndStream = mediaService.getStreamInfoByAppAndStream(mediaServerItem,
param.getApp(), param.getStream(), param.getTracks(), callId);
param.setStreamInfo(new StreamContent(streamInfoByAppAndStream));
redisCatchStorage.addStream(mediaServerItem, type, param.getApp(), param.getStream(), param);
if (param.getOriginType() == OriginType.RTSP_PUSH.ordinal()
|| param.getOriginType() == OriginType.RTMP_PUSH.ordinal()
|| param.getOriginType() == OriginType.RTC_PUSH.ordinal() ) {
param.setSeverId(userSetting.getServerId());
zlmMediaListManager.addPush(param);
}
}else {
// 兼容流注销时类型从redis记录获取
OnStreamChangedHookParam onStreamChangedHookParam = redisCatchStorage.getStreamInfo(
param.getApp(), param.getStream(), param.getMediaServerId());
if (onStreamChangedHookParam != null) {
type = OriginType.values()[onStreamChangedHookParam.getOriginType()].getType();
redisCatchStorage.removeStream(mediaServerItem.getId(), type, param.getApp(), param.getStream());
}
GbStream gbStream = storager.getGbStream(param.getApp(), param.getStream());
if (gbStream != null) {
StreamInfo streamInfoByAppAndStream = mediaService.getStreamInfoByAppAndStream(mediaServerItem,
param.getApp(), param.getStream(), param.getTracks(), callId);
param.setStreamInfo(new StreamContent(streamInfoByAppAndStream));
redisCatchStorage.addStream(mediaServerItem, type, param.getApp(), param.getStream(), param);
if (param.getOriginType() == OriginType.RTSP_PUSH.ordinal()
|| param.getOriginType() == OriginType.RTMP_PUSH.ordinal()
|| param.getOriginType() == OriginType.RTC_PUSH.ordinal()) {
param.setSeverId(userSetting.getServerId());
zlmMediaListManager.addPush(param);
}
} else {
// 兼容流注销时类型从redis记录获取
OnStreamChangedHookParam onStreamChangedHookParam = redisCatchStorage.getStreamInfo(
param.getApp(), param.getStream(), param.getMediaServerId());
if (onStreamChangedHookParam != null) {
type = OriginType.values()[onStreamChangedHookParam.getOriginType()].getType();
redisCatchStorage.removeStream(mediaServerItem.getId(), type, param.getApp(), param.getStream());
}
GbStream gbStream = storager.getGbStream(param.getApp(), param.getStream());
if (gbStream != null) {
// eventPublisher.catalogEventPublishForStream(null, gbStream, CatalogEvent.OFF);
}
zlmMediaListManager.removeMedia(param.getApp(), param.getStream());
}
GbStream gbStream = storager.getGbStream(param.getApp(), param.getStream());
if (gbStream != null) {
eventPublisher.catalogEventPublishForStream(null, gbStream, param.isRegist() ? CatalogEvent.ON : CatalogEvent.OFF);
}
if (type != null) {
// 发送流变化redis消息
JSONObject jsonObject = new JSONObject();
@@ -466,38 +474,35 @@ public class ZLMHttpHookListener {
}
}
}
}
if (!param.isRegist()) {
List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByStream(param.getStream());
if (sendRtpItems.size() > 0) {
for (SendRtpItem sendRtpItem : sendRtpItems) {
if (sendRtpItem.getApp().equals(param.getApp())) {
String platformId = sendRtpItem.getPlatformId();
ParentPlatform platform = storager.queryParentPlatByServerGBId(platformId);
Device device = deviceService.getDevice(platformId);
if (!param.isRegist()) {
List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByStream(param.getStream());
if (sendRtpItems.size() > 0) {
for (SendRtpItem sendRtpItem : sendRtpItems) {
if (sendRtpItem != null && sendRtpItem.getApp().equals(param.getApp())) {
String platformId = sendRtpItem.getPlatformId();
ParentPlatform platform = storager.queryParentPlatByServerGBId(platformId);
Device device = deviceService.getDevice(platformId);
if (platform != null) {
try {
try {
if (platform != null) {
commanderFroPlatform.streamByeCmd(platform, sendRtpItem);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage());
}
} else {
try {
redisCatchStorage.deleteSendRTPServer(platformId, sendRtpItem.getChannelId(),
sendRtpItem.getCallId(), sendRtpItem.getStream());
} else {
cmder.streamByeCmd(device, sendRtpItem.getChannelId(), param.getStream(), sendRtpItem.getCallId());
} catch (SipException | InvalidArgumentException | ParseException |
SsrcTransactionNotFoundException e) {
logger.error("[命令发送失败] 发送BYE: {}", e.getMessage());
}
} catch (SipException | InvalidArgumentException | ParseException |
SsrcTransactionNotFoundException e) {
logger.error("[命令发送失败] 发送BYE: {}", e.getMessage());
}
}
}
}
}
}
}
});
return HookResult.SUCCESS();
}
@@ -508,82 +513,65 @@ public class ZLMHttpHookListener {
@PostMapping(value = "/on_stream_none_reader", produces = "application/json;charset=UTF-8")
public JSONObject onStreamNoneReader(@RequestBody OnStreamNoneReaderHookParam param) {
logger.info("[ZLM HOOK]流无人观看：{]->{}->{}/{}" + param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());
JSONObject ret = new JSONObject();
ret.put("code", 0);
// 国标类型的流
if ("rtp".equals(param.getApp())){
ret.put("close", userSetting.getStreamOnDemand());
// 国标流， 点播/录像回放/录像下载
StreamInfo streamInfoForPlayCatch = redisCatchStorage.queryPlayByStreamId(param.getStream());
// 点播
if (streamInfoForPlayCatch != null) {
// 收到无人观看说明流也没有在往上级推送
if (redisCatchStorage.isChannelSendingRTP(streamInfoForPlayCatch.getChannelId())) {
List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChnnelId(streamInfoForPlayCatch.getChannelId());
if (sendRtpItems.size() > 0) {
for (SendRtpItem sendRtpItem : sendRtpItems) {
ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
if (parentPlatform == null) {
continue;
}
try {
commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId());
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage());
}
redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(),
sendRtpItem.getCallId(), sendRtpItem.getStream());
}
}
}
Device device = deviceService.getDevice(streamInfoForPlayCatch.getDeviceID());
if (device != null) {
try {
cmder.streamByeCmd(device, streamInfoForPlayCatch.getChannelId(),
streamInfoForPlayCatch.getStream(), null);
} catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) {
logger.error("[无人观看]点播， 发送BYE失败 {}", e.getMessage());
}
}
logger.info("[ZLM HOOK]流无人观看：{}->{}->{}/{}", param.getMediaServerId(), param.getSchema(),
param.getApp(), param.getStream());
JSONObject ret = new JSONObject();
ret.put("code", 0);
// 国标类型的流
if ("rtp".equals(param.getApp())) {
ret.put("close", userSetting.getStreamOnDemand());
// 国标流， 点播/录像回放/录像下载
// StreamInfo streamInfoForPlayCatch = redisCatchStorage.queryPlayByStreamId(param.getStream());
redisCatchStorage.stopPlay(streamInfoForPlayCatch);
storager.stopPlay(streamInfoForPlayCatch.getDeviceID(), streamInfoForPlayCatch.getChannelId());
return ret;
}
// 录像回放
StreamInfo streamInfoForPlayBackCatch = redisCatchStorage.queryPlayback(null, null, param.getStream(), null);
if (streamInfoForPlayBackCatch != null) {
if (streamInfoForPlayBackCatch.isPause()) {
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, param.getStream());
// 点播
if (inviteInfo != null) {
// 录像下载
if (inviteInfo.getType() == InviteSessionType.DOWNLOAD) {
ret.put("close", false);
} else {
Device device = deviceService.getDevice(streamInfoForPlayBackCatch.getDeviceID());
if (device != null) {
try {
cmder.streamByeCmd(device, streamInfoForPlayBackCatch.getChannelId(),
streamInfoForPlayBackCatch.getStream(), null);
} catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) {
logger.error("[无人观看]回放， 发送BYE失败 {}", e.getMessage());
return ret;
}
// 收到无人观看说明流也没有在往上级推送
if (redisCatchStorage.isChannelSendingRTP(inviteInfo.getChannelId())) {
List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChnnelId(
inviteInfo.getChannelId());
if (sendRtpItems.size() > 0) {
for (SendRtpItem sendRtpItem : sendRtpItems) {
ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
try {
commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId());
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage());
}
redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(),
sendRtpItem.getCallId(), sendRtpItem.getStream());
}
}
redisCatchStorage.stopPlayback(streamInfoForPlayBackCatch.getDeviceID(),
streamInfoForPlayBackCatch.getChannelId(), streamInfoForPlayBackCatch.getStream(), null);
}
return ret;
}
// 录像下载
StreamInfo streamInfoForDownload = redisCatchStorage.queryDownload(null, null, param.getStream(), null);
// 进行录像下载时无人观看不断流
if (streamInfoForDownload != null) {
ret.put("close", false);
Device device = deviceService.getDevice(inviteInfo.getDeviceId());
if (device != null) {
try {
if (inviteStreamService.getInviteInfo(inviteInfo.getType(), inviteInfo.getDeviceId(), inviteInfo.getChannelId(), inviteInfo.getStream()) != null) {
cmder.streamByeCmd(device, inviteInfo.getChannelId(),
inviteInfo.getStream(), null);
}
} catch (InvalidArgumentException | ParseException | SipException |
SsrcTransactionNotFoundException e) {
logger.error("[无人观看]点播， 发送BYE失败 {}", e.getMessage());
}
}
inviteStreamService.removeInviteInfo(inviteInfo.getType(), inviteInfo.getDeviceId(),
inviteInfo.getChannelId(), inviteInfo.getStream());
storager.stopPlay(inviteInfo.getDeviceId(), inviteInfo.getChannelId());
return ret;
}
SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(null, null, param.getStream(), null);
if ("talk".equals(sendRtpItem.getApp())){
if ("talk".equals(sendRtpItem.getApp())) {
ret.put("close", false);
return ret;
}
}else if ("talk".equals(param.getApp()) || "broadcast".equals(param.getApp())){
} else if ("talk".equals(param.getApp()) || "broadcast".equals(param.getApp())) {
ret.put("close", false);
} else {
// 非国标流 推流/拉流代理
@@ -652,6 +640,7 @@ public class ZLMHttpHookListener {
return defaultResult;
}
logger.info("[ZLM HOOK] 流未找到, 发起自动点播：{}->{}->{}/{}", param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());
RequestMessage msg = new RequestMessage();
String key = DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId;
boolean exist = resultHolder.exist(key, null);
@@ -659,31 +648,22 @@ public class ZLMHttpHookListener {
String uuid = UUID.randomUUID().toString();
msg.setId(uuid);
DeferredResult<HookResult> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue());
DeferredResultEx<HookResult> deferredResultEx = new DeferredResultEx<>(result);
result.onTimeout(() -> {
logger.info("点播接口等待超时");
logger.info("[ZLM HOOK] 自动点播, 等待超时");
// 释放rtpserver
msg.setData(new HookResult(ErrorCode.ERROR100.getCode(), "点播超时"));
resultHolder.invokeResult(msg);
});
// TODO 在点播未成功的情况下在此调用接口点播会导致返回的流地址ip错误
deferredResultEx.setFilter(result1 -> {
WVPResult<StreamInfo> wvpResult1 = (WVPResult<StreamInfo>) result1;
HookResult resultForEnd = new HookResult();
resultForEnd.setCode(wvpResult1.getCode());
resultForEnd.setMsg(wvpResult1.getMsg());
return resultForEnd;
});
// 录像查询以channelId作为deviceId查询
resultHolder.put(key, uuid, deferredResultEx);
resultHolder.put(key, uuid, result);
if (!exist) {
playService.play(mediaInfo, deviceId, channelId, null, eventResult -> {
msg.setData(new HookResult(eventResult.statusCode, eventResult.msg));
playService.play(mediaInfo, deviceId, channelId, (code, message, data) -> {
msg.setData(new HookResult(code, message));
resultHolder.invokeResult(msg);
}, null);
});
}
return result;
} else {
@@ -740,6 +720,7 @@ public class ZLMHttpHookListener {
if (sendRtpItems.size() > 0) {
for (SendRtpItem sendRtpItem : sendRtpItems) {
ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
ssrcFactory.releaseSsrc(sendRtpItem.getMediaServerId(), sendRtpItem.getSsrc());
try {
commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId());
} catch (SipException | InvalidArgumentException | ParseException e) {
@@ -759,7 +740,8 @@ public class ZLMHttpHookListener {
*/
@ResponseBody
@PostMapping(value = "/on_rtp_server_timeout", produces = "application/json;charset=UTF-8")
public HookResult onRtpServerTimeout(HttpServletRequest request, @RequestBody OnRtpServerTimeoutHookParam param) {
public HookResult onRtpServerTimeout(HttpServletRequest request, @RequestBody OnRtpServerTimeoutHookParam
param) {
logger.info("[ZLM HOOK] rtpServer收流超时：{}->{}({})", param.getMediaServerId(), param.getStream_id(), param.getSsrc());
taskExecutor.execute(() -> {

3
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMMediaListManager.java
View File

@@ -97,7 +97,8 @@ public class ZLMMediaListManager {
public void sendStreamEvent(String app, String stream, String mediaServerId) {
MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
// 查看推流状态
if (zlmrtpServerFactory.isStreamReady(mediaServerItem, app, stream)) {
Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, app, stream);
if (streamReady != null && streamReady) {
ChannelOnlineEvent channelOnlineEventLister = getChannelOnlineEventLister(app, stream);
if (channelOnlineEventLister != null) {
try {

21
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java
View File

@@ -25,6 +25,8 @@ public class ZLMRESTfulUtils {
private OkHttpClient client;
public interface RequestCallback{
void run(JSONObject response);
}
@@ -279,6 +281,10 @@ public class ZLMRESTfulUtils {
return sendPost(mediaServerItem, "closeRtpServer",param, null);
}
public void closeRtpServer(MediaServerItem mediaServerItem, Map<String, Object> param, RequestCallback callback) {
sendPost(mediaServerItem, "closeRtpServer",param, callback);
}
public JSONObject listRtpServer(MediaServerItem mediaServerItem) {
return sendPost(mediaServerItem, "listRtpServer",null, null);
}
@@ -353,4 +359,19 @@ public class ZLMRESTfulUtils {
param.put("stream_id", streamId);
return sendPost(mediaServerItem, "resumeRtpCheck",param, null);
}
public JSONObject connectRtpServer(MediaServerItem mediaServerItem, String dst_url, int dst_port, String stream_id) {
Map<String, Object> param = new HashMap<>(1);
param.put("dst_url", dst_url);
param.put("dst_port", dst_port);
param.put("stream_id", stream_id);
return sendPost(mediaServerItem, "connectRtpServer",param, null);
}
public JSONObject updateRtpServerSSRC(MediaServerItem mediaServerItem, String streamId, String ssrc) {
Map<String, Object> param = new HashMap<>(1);
param.put("ssrc", ssrc);
param.put("stream_id", streamId);
return sendPost(mediaServerItem, "updateRtpServerSSRC",param, null);
}
}

81
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java
View File

@@ -3,6 +3,7 @@ package com.genersoft.iot.vmp.media.zlm;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.CommonCallback;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
@@ -92,7 +93,17 @@ public class ZLMRTPServerFactory {
return result;
}
public int createRTPServer(MediaServerItem mediaServerItem, String streamId, int ssrc, Integer port, Boolean onlyAuto) {
/**
* 开启rtpServer
* @param mediaServerItem zlm服务实例
* @param streamId 流Id
* @param ssrc ssrc
* @param port 端口， 0/null为使用随机
* @param reUsePort 是否重用端口
* @param tcpMode 0/null udp 模式，1 tcp 被动模式, 2 tcp 主动模式。
* @return
*/
public int createRTPServer(MediaServerItem mediaServerItem, String streamId, int ssrc, Integer port, Boolean onlyAuto, Boolean reUsePort, Integer tcpMode) {
int result = -1;
// 查询此rtp server 是否已经存在
JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaServerItem, streamId);
@@ -108,7 +119,7 @@ public class ZLMRTPServerFactory {
JSONObject jsonObject = zlmresTfulUtils.closeRtpServer(mediaServerItem, param);
if (jsonObject != null ) {
if (jsonObject.getInteger("code") == 0) {
return createRTPServer(mediaServerItem, streamId, ssrc, port, onlyAuto);
return createRTPServer(mediaServerItem, streamId, ssrc, port,onlyAuto, reUsePort, tcpMode);
}else {
logger.warn("[开启rtpServer], 重启RtpServer错误");
}
@@ -122,8 +133,14 @@ public class ZLMRTPServerFactory {
Map<String, Object> param = new HashMap<>();
param.put("enable_tcp", 1);
if (tcpMode == null) {
tcpMode = 0;
}
param.put("tcp_mode", tcpMode);
param.put("stream_id", streamId);
if (reUsePort != null) {
param.put("re_use_port", reUsePort?"1":"0");
}
// 推流端口设置0则使用随机端口
if (port == null) {
param.put("port", 0);
@@ -169,6 +186,31 @@ public class ZLMRTPServerFactory {
return result;
}
public void closeRtpServer(MediaServerItem serverItem, String streamId, CommonCallback<Boolean> callback) {
if (serverItem == null) {
callback.run(false);
return;
}
Map<String, Object> param = new HashMap<>();
param.put("stream_id", streamId);
zlmresTfulUtils.closeRtpServer(serverItem, param, jsonObject -> {
if (jsonObject != null ) {
if (jsonObject.getInteger("code") == 0) {
callback.run(jsonObject.getInteger("hit") == 1);
return;
}else {
logger.error("关闭RTP Server 失败: " + jsonObject.getString("msg"));
}
}else {
// 检查ZLM状态
logger.error("关闭RTP Server 失败: 请检查ZLM服务");
}
callback.run(false);
});
}
/**
* 创建一个国标推流
@@ -256,11 +298,14 @@ public class ZLMRTPServerFactory {
if (jsonObject.getInteger("code") == 0) {
localPort = jsonObject.getInteger("port");
HookSubscribeForRtpServerTimeout hookSubscribeForRtpServerTimeout = HookSubscribeFactory.on_rtp_server_timeout(ssrc, null, serverItem.getId());
// 订阅 zlm启动事件, 新的zlm也会从这里进入系统
hookSubscribe.addSubscribe(hookSubscribeForRtpServerTimeout,
(MediaServerItem mediaServerItem, JSONObject response)->{
logger.info("[保持端口] {}->监听端口到期继续保持监听", ssrc);
keepPort(serverItem, ssrc);
logger.info("[上级点播] {}->监听端口到期继续保持监听", ssrc);
int port = keepPort(serverItem, ssrc);
if (port == 0) {
logger.info("[上级点播] {}->监听端口失败，移除监听", ssrc);
hookSubscribe.removeSubscribe(hookSubscribeForRtpServerTimeout);
}
});
logger.info("[保持端口] {}->监听端口: {}", ssrc, localPort);
logger.info("[保持端口] {}->监听端口: {}", ssrc, localPort);
@@ -305,6 +350,9 @@ public class ZLMRTPServerFactory {
*/
public Boolean isRtpReady(MediaServerItem mediaServerItem, String streamId) {
JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo(mediaServerItem,"rtp", "rtsp", streamId);
if (mediaInfo.getInteger("code") == -2) {
return null;
}
return (mediaInfo.getInteger("code") == 0 && mediaInfo.getBoolean("online"));
}
@@ -313,8 +361,10 @@ public class ZLMRTPServerFactory {
*/
public Boolean isStreamReady(MediaServerItem mediaServerItem, String app, String streamId) {
JSONObject mediaInfo = zlmresTfulUtils.getMediaList(mediaServerItem, app, streamId);
return mediaInfo != null && (mediaInfo.getInteger("code") == 0
if (mediaInfo == null || (mediaInfo.getInteger("code") == -2)) {
return null;
}
return (mediaInfo.getInteger("code") == 0
&& mediaInfo.getJSONArray("data") != null
&& mediaInfo.getJSONArray("data").size() > 0);
}
@@ -406,4 +456,19 @@ public class ZLMRTPServerFactory {
}
return startSendRtpStreamResult;
}
public Boolean updateRtpServerSSRC(MediaServerItem mediaServerItem, String streamId, String ssrc) {
boolean result = false;
JSONObject jsonObject = zlmresTfulUtils.updateRtpServerSSRC(mediaServerItem, streamId, ssrc);
if (jsonObject == null) {
logger.error("[更新RTPServer] 失败: 请检查ZLM服务");
} else if (jsonObject.getInteger("code") == 0) {
result= true;
logger.info("[更新RTPServer] 成功");
} else {
logger.error("[更新RTPServer] 失败: {}, streamId：{}，ssrc：{}->\r\n{}",jsonObject.getString("msg"),
streamId, ssrc, jsonObject);
}
return result;
}
}

9
src/main/java/com/genersoft/iot/vmp/media/zlm/ZlmHttpHookSubscribe.java
View File

@@ -100,7 +100,10 @@ public class ZlmHttpHookSubscribe {
if (!CollectionUtils.isEmpty(entriesToRemove)) {
for (Map.Entry<IHookSubscribe, ZlmHttpHookSubscribe.Event> entry : entriesToRemove) {
entries.remove(entry);
eventMap.remove(entry.getKey());
}
if (eventMap.size() == 0) {
allSubscribes.remove(hookSubscribe.getHookType());
}
}
@@ -136,9 +139,9 @@ public class ZlmHttpHookSubscribe {
/**
* 对订阅数据进行过期清理
*/
@Scheduled(cron="0 0/5 * * * ?") //每5分钟执行一次
// @Scheduled(cron="0 0/5 * * * ?") //每5分钟执行一次
@Scheduled(fixedRate = 2 * 1000)
public void execute(){
Instant instant = Instant.now().minusMillis(TimeUnit.MINUTES.toMillis(5));
int total = 0;
for (HookType hookType : allSubscribes.keySet()) {

29
src/main/java/com/genersoft/iot/vmp/media/zlm/dto/MediaServerItem.java
View File

@@ -1,13 +1,10 @@
package com.genersoft.iot.vmp.media.zlm.dto;
import com.genersoft.iot.vmp.gb28181.session.SsrcConfig;
import com.genersoft.iot.vmp.media.zlm.ZLMServerConfig;
import io.swagger.v3.oas.annotations.media.Schema;
import org.springframework.util.ObjectUtils;
import java.util.HashMap;
@Schema(description = "流媒体服务信息")
public class MediaServerItem{
@@ -80,20 +77,10 @@ public class MediaServerItem{
@Schema(description = "是否是默认ZLM")
private boolean defaultServer;
@Schema(description = "SSRC信息")
private SsrcConfig ssrcConfig;
@Schema(description = "当前使用到的端口")
private int currentPort;
/**
* 每一台ZLM都有一套独立的SSRC列表
* 在ApplicationCheckRunner里对mediaServerSsrcMap进行初始化
*/
@Schema(description = "ID")
private HashMap<String, SsrcConfig> mediaServerSsrcMap;
public MediaServerItem() {
}
@@ -279,22 +266,6 @@ public class MediaServerItem{
this.updateTime = updateTime;
}
public HashMap<String, SsrcConfig> getMediaServerSsrcMap() {
return mediaServerSsrcMap;
}
public void setMediaServerSsrcMap(HashMap<String, SsrcConfig> mediaServerSsrcMap) {
this.mediaServerSsrcMap = mediaServerSsrcMap;
}
public SsrcConfig getSsrcConfig() {
return ssrcConfig;
}
public void setSsrcConfig(SsrcConfig ssrcConfig) {
this.ssrcConfig = ssrcConfig;
}
public int getCurrentPort() {
return currentPort;
}

4
src/main/java/com/genersoft/iot/vmp/media/zlm/dto/hook/HookResult.java
View File

@@ -18,6 +18,10 @@ public class HookResult {
return new HookResult(0, "success");
}
public static HookResult Fail(){
return new HookResult(-1, "fail");
}
public int getCode() {
return code;
}