aunix/wvp-GB28181-pro
1
0
Fork 0
mirror of https://gitee.com/pan648540858/wvp-GB28181-pro.git synced 2026-05-25 06:27:50 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

优化RTP服务器相关代码，统一使用MediaApp枚举，简化参数传递，调整配置以提高兼容性

Browse Source
This commit is contained in:
648540858 2026-02-20 22:33:26 +08:00
parent b1b4d065a5
commit 803c47a430
30 changed files with 391 additions and 300 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/main/java/com/genersoft/iot/vmp/common/enums/MediaApp.java Normal file
View File

@ -0,0 +1,8 @@
package com.genersoft.iot.vmp.common.enums;
public class MediaApp {
public final static String GB28181 = "rtp";
public final static String JT1078 = "1078";
}

3
src/main/java/com/genersoft/iot/vmp/gb28181/bean/SendRtpInfo.java
View File

@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.gb28181.bean; package com.genersoft.iot.vmp.gb28181.bean;
import com.genersoft.iot.vmp.common.enums.ChannelDataType; import com.genersoft.iot.vmp.common.enums.ChannelDataType;
import com.genersoft.iot.vmp.common.enums.MediaApp;
import com.genersoft.iot.vmp.media.bean.MediaServer; import com.genersoft.iot.vmp.media.bean.MediaServer;
import com.genersoft.iot.vmp.service.bean.RequestPushStreamMsg; import com.genersoft.iot.vmp.service.bean.RequestPushStreamMsg;
import lombok.Data; import lombok.Data;
@ -199,7 +200,7 @@ public class SendRtpInfo {
sendRtpItem.setChannelId(channelId); sendRtpItem.setChannelId(channelId);
sendRtpItem.setTcp(isTcp); sendRtpItem.setTcp(isTcp);
sendRtpItem.setRtcp(rtcp); sendRtpItem.setRtcp(rtcp);
sendRtpItem.setApp("rtp"); sendRtpItem.setApp(MediaApp.GB28181);
sendRtpItem.setLocalPort(localPort); sendRtpItem.setLocalPort(localPort);
sendRtpItem.setServerId(serverId); sendRtpItem.setServerId(serverId);
sendRtpItem.setMediaServerId(mediaServer.getId()); sendRtpItem.setMediaServerId(mediaServer.getId());

2
src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/DeviceServiceImpl.java
View File

@ -248,7 +248,7 @@ public class DeviceServiceImpl implements IDeviceService, CommandLineRunner {
if (ssrcTransactions != null && !ssrcTransactions.isEmpty()) { if (ssrcTransactions != null && !ssrcTransactions.isEmpty()) {
for (SsrcTransaction ssrcTransaction : ssrcTransactions) { for (SsrcTransaction ssrcTransaction : ssrcTransactions) {
mediaServerService.releaseSsrc(ssrcTransaction.getMediaServerId(), ssrcTransaction.getSsrc()); mediaServerService.releaseSsrc(ssrcTransaction.getMediaServerId(), ssrcTransaction.getSsrc());
mediaServerService.closeRTPServer(ssrcTransaction.getMediaServerId(), ssrcTransaction.getStream()); mediaServerService.closeRTPServer(ssrcTransaction.getMediaServerId(), ssrcTransaction.getApp(), ssrcTransaction.getStream());
sessionManager.removeByCallId(ssrcTransaction.getCallId()); sessionManager.removeByCallId(ssrcTransaction.getCallId());
} }
} }

3
src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/InviteStreamServiceImpl.java
View File

@ -2,6 +2,7 @@ package com.genersoft.iot.vmp.gb28181.service.impl;
import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSON;
import com.genersoft.iot.vmp.common.*; import com.genersoft.iot.vmp.common.*;
import com.genersoft.iot.vmp.common.enums.MediaApp;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.Device; import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.dao.DeviceChannelMapper; import com.genersoft.iot.vmp.gb28181.dao.DeviceChannelMapper;
@ -49,7 +50,7 @@ public class InviteStreamServiceImpl implements IInviteStreamService {
@Async("taskExecutor") @Async("taskExecutor")
@EventListener @EventListener
public void onApplicationEvent(MediaDepartureEvent event) { public void onApplicationEvent(MediaDepartureEvent event) {
if ("rtsp".equals(event.getSchema()) && "rtp".equals(event.getApp())) { if ("rtsp".equals(event.getSchema()) && MediaApp.GB28181.equals(event.getApp())) {
InviteInfo inviteInfo = getInviteInfoByStream(null, event.getStream()); InviteInfo inviteInfo = getInviteInfoByStream(null, event.getStream());
if (inviteInfo != null && (inviteInfo.getType() == InviteSessionType.PLAY || inviteInfo.getType() == InviteSessionType.PLAYBACK)) { if (inviteInfo != null && (inviteInfo.getType() == InviteSessionType.PLAY || inviteInfo.getType() == InviteSessionType.PLAYBACK)) {
removeInviteInfo(inviteInfo); removeInviteInfo(inviteInfo);

13
src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/PlatformServiceImpl.java
View File

@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.gb28181.service.impl; package com.genersoft.iot.vmp.gb28181.service.impl;
import com.genersoft.iot.vmp.common.*; import com.genersoft.iot.vmp.common.*;
import com.genersoft.iot.vmp.common.enums.MediaApp;
import com.genersoft.iot.vmp.conf.DynamicTask; import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.SipConfig; import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
@ -618,7 +619,7 @@ public class PlatformServiceImpl implements IPlatformService, CommandLineRunner
} else { } else {
tcpMode = 0; tcpMode = 0;
} }
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, null, ssrcCheck, false, null, true, false, false, tcpMode); SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, MediaApp.GB28181, streamId, null, ssrcCheck, false, null, true, false, false, tcpMode);
if (ssrcInfo == null || ssrcInfo.getPort() < 0) { if (ssrcInfo == null || ssrcInfo.getPort() < 0) {
log.info("[国标级联] 发起语音喊话 开启端口监听失败， platform: {}, channel： {}", platform.getServerGBId(), channel.getGbDeviceId()); log.info("[国标级联] 发起语音喊话 开启端口监听失败， platform: {}, channel： {}", platform.getServerGBId(), channel.getGbDeviceId());
SipSubscribe.EventResult<Object> eventResult = new SipSubscribe.EventResult<>(); SipSubscribe.EventResult<Object> eventResult = new SipSubscribe.EventResult<>();
@ -650,7 +651,7 @@ public class PlatformServiceImpl implements IPlatformService, CommandLineRunner
} finally { } finally {
timeoutCallback.run(1, "收流超时"); timeoutCallback.run(1, "收流超时");
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream()); mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getApp(), ssrcInfo.getStream());
sessionManager.removeByStream(ssrcInfo.getApp(), ssrcInfo.getStream()); sessionManager.removeByStream(ssrcInfo.getApp(), ssrcInfo.getStream());
} }
} }
@ -724,7 +725,7 @@ public class PlatformServiceImpl implements IPlatformService, CommandLineRunner
log.info("[Invite 200OK] SSRC修正 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse); log.info("[Invite 200OK] SSRC修正 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse);
// 释放ssrc // 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
Boolean result = mediaServerService.updateRtpServerSSRC(mediaServerItem, ssrcInfo.getStream(), ssrcInResponse); Boolean result = mediaServerService.updateRtpServerSSRC(mediaServerItem, ssrcInfo.getApp(), ssrcInfo.getStream(), ssrcInResponse);
if (!result) { if (!result) {
try { try {
log.warn("[Invite 200OK] 更新ssrc失败，停止喊话 {}/{}", platform.getServerGBId(), channel.getGbDeviceId()); log.warn("[Invite 200OK] 更新ssrc失败，停止喊话 {}/{}", platform.getServerGBId(), channel.getGbDeviceId());
@ -826,12 +827,12 @@ public class PlatformServiceImpl implements IPlatformService, CommandLineRunner
} }
log.info("[TCP主动连接对方] serverGbId: {}, channelId: {}, 连接对方的地址：{}:{}, SSRC: {}, SSRC校验：{}", log.info("[TCP主动连接对方] serverGbId: {}, channelId: {}, 连接对方的地址：{}:{}, SSRC: {}, SSRC校验：{}",
platform.getServerGBId(), channel.getGbDeviceId(), sdp.getConnection().getAddress(), port, ssrcInfo.getSsrc(), ssrcCheck); platform.getServerGBId(), channel.getGbDeviceId(), sdp.getConnection().getAddress(), port, ssrcInfo.getSsrc(), ssrcCheck);
Boolean result = mediaServerService.connectRtpServer(mediaServerItem, sdp.getConnection().getAddress(), port, ssrcInfo.getStream()); Boolean result = mediaServerService.connectRtpServer(mediaServerItem, sdp.getConnection().getAddress(), port, ssrcInfo.getApp(), ssrcInfo.getStream());
log.info("[TCP主动连接对方] 结果： {}", result); log.info("[TCP主动连接对方] 结果： {}", result);
} catch (SdpException e) { } catch (SdpException e) {
log.error("[TCP主动连接对方] serverGbId: {}, channelId: {}, 解析200OK的SDP信息失败", platform.getServerGBId(), channel.getGbDeviceId(), e); log.error("[TCP主动连接对方] serverGbId: {}, channelId: {}, 解析200OK的SDP信息失败", platform.getServerGBId(), channel.getGbDeviceId(), e);
dynamicTask.stop(timeOutTaskKey); dynamicTask.stop(timeOutTaskKey);
mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream()); mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getApp(), ssrcInfo.getStream());
// 释放ssrc // 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
@ -855,7 +856,7 @@ public class PlatformServiceImpl implements IPlatformService, CommandLineRunner
} catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) { } catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) {
log.warn("[消息发送失败] 停止语音对讲， 平台：{}，通道：{}", platform.getId(), channel.getGbDeviceId() ); log.warn("[消息发送失败] 停止语音对讲， 平台：{}，通道：{}", platform.getId(), channel.getGbDeviceId() );
} finally { } finally {
mediaServerService.closeRTPServer(mediaServerItem, stream); mediaServerService.closeRTPServer(mediaServerItem, app, stream);
InviteInfo inviteInfo = inviteStreamService.getInviteInfo(null, channel.getGbId(), stream); InviteInfo inviteInfo = inviteStreamService.getInviteInfo(null, channel.getGbId(), stream);
if (inviteInfo != null) { if (inviteInfo != null) {
// 释放ssrc // 释放ssrc

94
src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/PlayServiceImpl.java
View File

@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.gb28181.service.impl; package com.genersoft.iot.vmp.gb28181.service.impl;
import com.genersoft.iot.vmp.common.*; import com.genersoft.iot.vmp.common.*;
import com.genersoft.iot.vmp.common.enums.MediaApp;
import com.genersoft.iot.vmp.conf.DynamicTask; import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException; import com.genersoft.iot.vmp.conf.exception.ControllerException;
@ -231,7 +232,7 @@ public class PlayServiceImpl implements IPlayService {
} }
} }
} }
}else if ("rtp".equals(event.getApp())) { }else if (MediaApp.GB28181.equals(event.getApp())) {
// 释放ssrc // 释放ssrc
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, event.getStream()); InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, event.getStream());
if (inviteInfo != null && inviteInfo.getStatus() == InviteSessionStatus.ok if (inviteInfo != null && inviteInfo.getStatus() == InviteSessionStatus.ok
@ -249,7 +250,7 @@ public class PlayServiceImpl implements IPlayService {
@Async("taskExecutor") @Async("taskExecutor")
@EventListener @EventListener
public void onApplicationEvent(MediaNotFoundEvent event) { public void onApplicationEvent(MediaNotFoundEvent event) {
if (!"rtp".equals(event.getApp())) { if (!MediaApp.GB28181.equals(event.getApp())) {
return; return;
} }
String[] s = event.getStream().split("_"); String[] s = event.getStream().split("_");
@ -324,9 +325,9 @@ public class PlayServiceImpl implements IPlayService {
return play(mediaServerItem, device, channel, ssrc, userSetting.getRecordSip(), callback); return play(mediaServerItem, device, channel, ssrc, userSetting.getRecordSip(), callback);
} }
private SSRCInfo play(MediaServer mediaServerItem, Device device, DeviceChannel channel, String ssrc, Boolean record, private SSRCInfo play(MediaServer mediaServer, Device device, DeviceChannel channel, String ssrc, Boolean record,
ErrorCallback<StreamInfo> callback) { ErrorCallback<StreamInfo> callback) {
if (mediaServerItem == null ) { if (mediaServer == null ) {
if (callback != null) { if (callback != null) {
callback.run(InviteErrorCode.ERROR_FOR_PARAMETER_ERROR.getCode(), callback.run(InviteErrorCode.ERROR_FOR_PARAMETER_ERROR.getCode(),
InviteErrorCode.ERROR_FOR_PARAMETER_ERROR.getMsg(), InviteErrorCode.ERROR_FOR_PARAMETER_ERROR.getMsg(),
@ -340,7 +341,7 @@ public class PlayServiceImpl implements IPlayService {
if (inviteInfoInCatch.getStreamInfo() == null) { if (inviteInfoInCatch.getStreamInfo() == null) {
// 释放生成的ssrc，使用上一次申请的 // 释放生成的ssrc，使用上一次申请的
ssrcFactory.releaseSsrc(mediaServerItem.getId(), ssrc); ssrcFactory.releaseSsrc(mediaServer.getId(), ssrc);
// 点播发起了但是尚未成功, 仅注册回调等待结果即可 // 点播发起了但是尚未成功, 仅注册回调等待结果即可
inviteStreamService.once(InviteSessionType.PLAY, channel.getId(), null, callback); inviteStreamService.once(InviteSessionType.PLAY, channel.getId(), null, callback);
log.info("[点播开始] 已经请求中，等待结果， deviceId: {}, channelId({}): {}", device.getDeviceId(), channel.getDeviceId(), channel.getId()); log.info("[点播开始] 已经请求中，等待结果， deviceId: {}, channelId({}): {}", device.getDeviceId(), channel.getDeviceId(), channel.getId());
@ -357,7 +358,7 @@ public class PlayServiceImpl implements IPlayService {
return inviteInfoInCatch.getSsrcInfo(); return inviteInfoInCatch.getSsrcInfo();
} }
MediaServer mediaInfo = streamInfo.getMediaServer(); MediaServer mediaInfo = streamInfo.getMediaServer();
Boolean ready = mediaServerService.isStreamReady(mediaInfo, "rtp", streamId); Boolean ready = mediaServerService.isStreamReady(mediaInfo, MediaApp.GB28181, streamId);
if (ready != null && ready) { if (ready != null && ready) {
if(callback != null) { if(callback != null) {
callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo); callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo);
@ -377,20 +378,30 @@ public class PlayServiceImpl implements IPlayService {
} }
} }
// 获取mediaServer可用的ssrc
final String finalSsrc;
if (ssrc != null) {
finalSsrc = ssrc;
}else {
finalSsrc = ssrcFactory.getPlaySsrc(mediaServer.getId());
}
String streamId = String.format("%s_%s", device.getDeviceId(), channel.getDeviceId()); String streamId = String.format("%s_%s", device.getDeviceId(), channel.getDeviceId());
int tcpMode = device.getStreamMode().equals("TCP-ACTIVE")? 2: (device.getStreamMode().equals("TCP-PASSIVE")? 1:0); int tcpMode = device.getStreamMode().equals("TCP-ACTIVE")? 2: (device.getStreamMode().equals("TCP-PASSIVE")? 1:0);
RTPServerParam rtpServerParam = new RTPServerParam(); RTPServerParam rtpServerParam = new RTPServerParam();
rtpServerParam.setMediaServerItem(mediaServerItem); rtpServerParam.setMediaServer(mediaServer);
rtpServerParam.setApp(MediaApp.GB28181);
rtpServerParam.setStreamId(streamId); rtpServerParam.setStreamId(streamId);
rtpServerParam.setPresetSsrc(ssrc); if (device.isSsrcCheck()) {
rtpServerParam.setSsrcCheck(device.isSsrcCheck()); rtpServerParam.setSsrc(ssrc);
rtpServerParam.setPlayback(false); }
rtpServerParam.setPort(0); rtpServerParam.setPort(0);
rtpServerParam.setTcpMode(tcpMode); rtpServerParam.setTcpMode(tcpMode);
rtpServerParam.setOnlyAuto(false); rtpServerParam.setOnlyAuto(false);
rtpServerParam.setDisableAudio(!channel.isHasAudio()); rtpServerParam.setDisableAudio(!channel.isHasAudio());
SSRCInfo ssrcInfo = receiveRtpServerService.openRTPServer(rtpServerParam, (code, msg, result) -> {
int port = receiveRtpServerService.openRTPServer(rtpServerParam, (code, msg, result) -> {
if (code == InviteErrorCode.SUCCESS.getCode() && result != null && result.getHookData() != null) { if (code == InviteErrorCode.SUCCESS.getCode() && result != null && result.getHookData() != null) {
// hook响应 // hook响应
@ -422,14 +433,14 @@ public class PlayServiceImpl implements IPlayService {
} }
inviteStreamService.call(InviteSessionType.PLAY, channel.getId(), null, code, msg, null); inviteStreamService.call(InviteSessionType.PLAY, channel.getId(), null, code, msg, null);
inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, channel.getId()); inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, channel.getId());
SsrcTransaction ssrcTransaction = sessionManager.getSsrcTransactionByStream("rtp", streamId); SsrcTransaction ssrcTransaction = sessionManager.getSsrcTransactionByStream(MediaApp.GB28181, streamId);
if (ssrcTransaction != null) { if (ssrcTransaction != null) {
try { try {
cmder.streamByeCmd(device, channel.getDeviceId(),"rtp", streamId, null, null); cmder.streamByeCmd(device, channel.getDeviceId(),MediaApp.GB28181, streamId, null, null);
} catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) { } catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) {
log.error("[点播超时]， 发送BYE失败 {}", e.getMessage()); log.error("[点播超时]， 发送BYE失败 {}", e.getMessage());
} finally { } finally {
sessionManager.removeByStream("rtp", streamId); sessionManager.removeByStream(MediaApp.GB28181, streamId);
} }
} }
} }
@ -448,8 +459,8 @@ public class PlayServiceImpl implements IPlayService {
ssrcInfo.getSsrc(), device.isSsrcCheck()); ssrcInfo.getSsrc(), device.isSsrcCheck());
// 初始化redis中的invite消息状态 // 初始化redis中的invite消息状态
InviteInfo inviteInfo = InviteInfo.getInviteInfo(device.getDeviceId(), channel.getId(), ssrcInfo.getStream(), ssrcInfo, mediaServerItem.getId(), InviteInfo inviteInfo = InviteInfo.getInviteInfo(device.getDeviceId(), channel.getId(), ssrcInfo.getStream(), ssrcInfo, mediaServer.getId(),
mediaServerItem.getSdpIp(), ssrcInfo.getPort(), device.getStreamMode(), InviteSessionType.PLAY, mediaServer.getSdpIp(), ssrcInfo.getPort(), device.getStreamMode(), InviteSessionType.PLAY,
InviteSessionStatus.ready, userSetting.getRecordSip()); InviteSessionStatus.ready, userSetting.getRecordSip());
if (record != null) { if (record != null) {
inviteInfo.setRecord(record); inviteInfo.setRecord(record);
@ -460,12 +471,12 @@ public class PlayServiceImpl implements IPlayService {
inviteStreamService.updateInviteInfo(inviteInfo); inviteStreamService.updateInviteInfo(inviteInfo);
try { try {
cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channel, (eventResult) -> { cmder.playStreamCmd(mediaServer, ssrcInfo, device, channel, (eventResult) -> {
// 处理收到200ok后的TCP主动连接以及SSRC不一致的问题 // 处理收到200ok后的TCP主动连接以及SSRC不一致的问题
InviteOKHandler(eventResult, ssrcInfo, mediaServerItem, device, channel, callback, inviteInfo, InviteSessionType.PLAY); InviteOKHandler(eventResult, ssrcInfo, mediaServer, device, channel, callback, inviteInfo, InviteSessionType.PLAY);
}, (event) -> { }, (event) -> {
log.info("[点播失败]{}:{} deviceId: {}, channelId:{}",event.statusCode, event.msg, device.getDeviceId(), channel.getDeviceId()); log.info("[点播失败]{}:{} deviceId: {}, channelId:{}",event.statusCode, event.msg, device.getDeviceId(), channel.getDeviceId());
receiveRtpServerService.closeRTPServer(mediaServerItem, ssrcInfo); receiveRtpServerService.closeRTPServer(mediaServer, ssrcInfo);
sessionManager.removeByStream(ssrcInfo.getApp(), ssrcInfo.getStream()); sessionManager.removeByStream(ssrcInfo.getApp(), ssrcInfo.getStream());
if (callback != null) { if (callback != null) {
@ -478,7 +489,7 @@ public class PlayServiceImpl implements IPlayService {
}, userSetting.getPlayTimeout().longValue()); }, userSetting.getPlayTimeout().longValue());
} catch (InvalidArgumentException | SipException | ParseException e) { } catch (InvalidArgumentException | SipException | ParseException e) {
log.error("[命令发送失败] 点播消息: {}", e.getMessage()); log.error("[命令发送失败] 点播消息: {}", e.getMessage());
receiveRtpServerService.closeRTPServer(mediaServerItem, ssrcInfo); receiveRtpServerService.closeRTPServer(mediaServer, ssrcInfo);
sessionManager.removeByStream(ssrcInfo.getApp(), ssrcInfo.getStream()); sessionManager.removeByStream(ssrcInfo.getApp(), ssrcInfo.getStream());
if (callback != null) { if (callback != null) {
callback.run(InviteErrorCode.ERROR_FOR_SIP_SENDING_FAILED.getCode(), callback.run(InviteErrorCode.ERROR_FOR_SIP_SENDING_FAILED.getCode(),
@ -595,7 +606,7 @@ public class PlayServiceImpl implements IPlayService {
}, (event) -> { }, (event) -> {
dynamicTask.stop(timeOutTaskKey); dynamicTask.stop(timeOutTaskKey);
mediaServerService.closeRTPServer(mediaServerItem, sendRtpInfo.getStream()); mediaServerService.closeRTPServer(mediaServerItem, sendRtpInfo.getApp(), sendRtpInfo.getStream());
// 释放ssrc // 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpInfo.getSsrc()); mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpInfo.getSsrc());
sessionManager.removeByStream(sendRtpInfo.getApp(), sendRtpInfo.getStream()); sessionManager.removeByStream(sendRtpInfo.getApp(), sendRtpInfo.getStream());
@ -605,7 +616,7 @@ public class PlayServiceImpl implements IPlayService {
log.error("[命令发送失败] 对讲消息: {}", e.getMessage()); log.error("[命令发送失败] 对讲消息: {}", e.getMessage());
dynamicTask.stop(timeOutTaskKey); dynamicTask.stop(timeOutTaskKey);
mediaServerService.closeRTPServer(mediaServerItem, sendRtpInfo.getStream()); mediaServerService.closeRTPServer(mediaServerItem, sendRtpInfo.getApp(), sendRtpInfo.getStream());
// 释放ssrc // 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpInfo.getSsrc()); mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpInfo.getSsrc());
@ -647,7 +658,7 @@ public class PlayServiceImpl implements IPlayService {
} }
} }
log.info("[TCP主动连接对方] deviceId: {}, channelId: {}, 连接对方的地址：{}:{}, 收流模式：{}, SSRC: {}, SSRC校验：{}", device.getDeviceId(), channel.getDeviceId(), sdp.getConnection().getAddress(), port, device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck()); log.info("[TCP主动连接对方] deviceId: {}, channelId: {}, 连接对方的地址：{}:{}, 收流模式：{}, SSRC: {}, SSRC校验：{}", device.getDeviceId(), channel.getDeviceId(), sdp.getConnection().getAddress(), port, device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck());
Boolean result = mediaServerService.connectRtpServer(mediaServerItem, sdp.getConnection().getAddress(), port, ssrcInfo.getStream()); Boolean result = mediaServerService.connectRtpServer(mediaServerItem, sdp.getConnection().getAddress(), port, ssrcInfo.getApp(), ssrcInfo.getStream());
log.info("[TCP主动连接对方] 结果： {}" , result); log.info("[TCP主动连接对方] 结果： {}" , result);
if (!result) { if (!result) {
// 主动连接失败，结束流程， 清理数据 // 主动连接失败，结束流程， 清理数据
@ -686,7 +697,7 @@ public class PlayServiceImpl implements IPlayService {
String fileName = deviceId + "_" + channelId + ".jpg"; String fileName = deviceId + "_" + channelId + ".jpg";
// 请求截图 // 请求截图
log.info("[请求截图]: " + fileName); log.info("[请求截图]: " + fileName);
mediaServerService.getSnap(mediaServerItemInuse, "rtp", stream, 15, 1, path, fileName); mediaServerService.getSnap(mediaServerItemInuse, MediaApp.GB28181, stream, 15, 1, path, fileName);
} }
public StreamInfo onPublishHandlerForPlay(MediaServer mediaServerItem, MediaInfo mediaInfo, Device device, DeviceChannel channel) { public StreamInfo onPublishHandlerForPlay(MediaServer mediaServerItem, MediaInfo mediaInfo, Device device, DeviceChannel channel) {
@ -779,7 +790,8 @@ public class PlayServiceImpl implements IPlayService {
int tcpMode = device.getStreamMode().equals("TCP-ACTIVE")? 2: (device.getStreamMode().equals("TCP-PASSIVE")? 1:0); int tcpMode = device.getStreamMode().equals("TCP-ACTIVE")? 2: (device.getStreamMode().equals("TCP-PASSIVE")? 1:0);
RTPServerParam rtpServerParam = new RTPServerParam(); RTPServerParam rtpServerParam = new RTPServerParam();
rtpServerParam.setMediaServerItem(mediaServerItem); rtpServerParam.setMediaServer(mediaServerItem);
rtpServerParam.setApp(MediaApp.GB28181);
rtpServerParam.setStreamId(stream); rtpServerParam.setStreamId(stream);
rtpServerParam.setSsrcCheck(device.isSsrcCheck()); rtpServerParam.setSsrcCheck(device.isSsrcCheck());
rtpServerParam.setPlayback(true); rtpServerParam.setPlayback(true);
@ -805,14 +817,14 @@ public class PlayServiceImpl implements IPlayService {
} }
inviteStreamService.call(InviteSessionType.PLAYBACK, channel.getId(), null, code, msg, null); inviteStreamService.call(InviteSessionType.PLAYBACK, channel.getId(), null, code, msg, null);
inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAYBACK, channel.getId()); inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAYBACK, channel.getId());
SsrcTransaction ssrcTransaction = sessionManager.getSsrcTransactionByStream("rtp", stream); SsrcTransaction ssrcTransaction = sessionManager.getSsrcTransactionByStream(MediaApp.GB28181, stream);
if (ssrcTransaction != null) { if (ssrcTransaction != null) {
try { try {
cmder.streamByeCmd(device, channel.getDeviceId(),"rtp", stream, null, null); cmder.streamByeCmd(device, channel.getDeviceId(),MediaApp.GB28181, stream, null, null);
} catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) { } catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) {
log.error("[录像回放] 发送BYE失败 {}", e.getMessage()); log.error("[录像回放] 发送BYE失败 {}", e.getMessage());
} finally { } finally {
sessionManager.removeByStream("rtp", stream); sessionManager.removeByStream(MediaApp.GB28181, stream);
} }
} }
} }
@ -902,7 +914,7 @@ public class PlayServiceImpl implements IPlayService {
log.info("[Invite 200OK] SSRC修正 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse); log.info("[Invite 200OK] SSRC修正 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse);
// 释放ssrc // 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
Boolean result = mediaServerService.updateRtpServerSSRC(mediaServerItem, ssrcInfo.getStream(), ssrcInResponse); Boolean result = mediaServerService.updateRtpServerSSRC(mediaServerItem, ssrcInfo.getApp(), ssrcInfo.getStream(), ssrcInResponse);
if (!result) { if (!result) {
try { try {
log.warn("[Invite 200OK] 更新ssrc失败，停止点播 {}/{}", device.getDeviceId(), channel.getDeviceId()); log.warn("[Invite 200OK] 更新ssrc失败，停止点播 {}/{}", device.getDeviceId(), channel.getDeviceId());
@ -940,14 +952,14 @@ public class PlayServiceImpl implements IPlayService {
if (ssrcInResponse != null) { if (ssrcInResponse != null) {
// 单端口 // 单端口
// 重新订阅流上线 // 重新订阅流上线
SsrcTransaction ssrcTransaction = sessionManager.getSsrcTransactionByStream("rtp", inviteInfo.getStream()); SsrcTransaction ssrcTransaction = sessionManager.getSsrcTransactionByStream(MediaApp.GB28181, inviteInfo.getStream());
sessionManager.removeByStream("rtp", inviteInfo.getStream()); sessionManager.removeByStream(MediaApp.GB28181, inviteInfo.getStream());
inviteStreamService.updateInviteInfoForSSRC(inviteInfo, ssrcInResponse); inviteStreamService.updateInviteInfoForSSRC(inviteInfo, ssrcInResponse);
ssrcTransaction.setDeviceId(device.getDeviceId()); ssrcTransaction.setDeviceId(device.getDeviceId());
ssrcTransaction.setChannelId(ssrcTransaction.getChannelId()); ssrcTransaction.setChannelId(ssrcTransaction.getChannelId());
ssrcTransaction.setCallId(ssrcTransaction.getCallId()); ssrcTransaction.setCallId(ssrcTransaction.getCallId());
ssrcTransaction.setSsrc(ssrcInResponse); ssrcTransaction.setSsrc(ssrcInResponse);
ssrcTransaction.setApp("rtp"); ssrcTransaction.setApp(MediaApp.GB28181);
ssrcTransaction.setStream(inviteInfo.getStream()); ssrcTransaction.setStream(inviteInfo.getStream());
ssrcTransaction.setMediaServerId(mediaServerItem.getId()); ssrcTransaction.setMediaServerId(mediaServerItem.getId());
ssrcTransaction.setSipTransactionInfo(new SipTransactionInfo((SIPResponse) responseEvent.getResponse())); ssrcTransaction.setSipTransactionInfo(new SipTransactionInfo((SIPResponse) responseEvent.getResponse()));
@ -990,7 +1002,7 @@ public class PlayServiceImpl implements IPlayService {
int tcpMode = device.getStreamMode().equals("TCP-ACTIVE")? 2: (device.getStreamMode().equals("TCP-PASSIVE")? 1:0); int tcpMode = device.getStreamMode().equals("TCP-ACTIVE")? 2: (device.getStreamMode().equals("TCP-PASSIVE")? 1:0);
// 录像下载不使用固定流地址，固定流地址会导致如果开始时间与结束时间一致时文件错误的叠加在一起 // 录像下载不使用固定流地址，固定流地址会导致如果开始时间与结束时间一致时文件错误的叠加在一起
RTPServerParam rtpServerParam = new RTPServerParam(); RTPServerParam rtpServerParam = new RTPServerParam();
rtpServerParam.setMediaServerItem(mediaServer); rtpServerParam.setMediaServer(mediaServer);
rtpServerParam.setSsrcCheck(device.isSsrcCheck()); rtpServerParam.setSsrcCheck(device.isSsrcCheck());
rtpServerParam.setPlayback(true); rtpServerParam.setPlayback(true);
rtpServerParam.setPort(0); rtpServerParam.setPort(0);
@ -1081,7 +1093,7 @@ public class PlayServiceImpl implements IPlayService {
inviteStreamService.updateInviteInfo(inviteInfoForNew, 60*15L); inviteStreamService.updateInviteInfo(inviteInfoForNew, 60*15L);
} }
}; };
Hook hook = Hook.getInstance(HookType.on_record_mp4, "rtp", ssrcInfo.getStream(), mediaServer.getId()); Hook hook = Hook.getInstance(HookType.on_record_mp4, MediaApp.GB28181, ssrcInfo.getStream(), mediaServer.getId());
// 设置过期时间，下载失败时自动处理订阅数据 // 设置过期时间，下载失败时自动处理订阅数据
hook.setExpireTime(System.currentTimeMillis() + 24 * 60 * 60 * 1000); hook.setExpireTime(System.currentTimeMillis() + 24 * 60 * 60 * 1000);
subscribe.addSubscribe(hook, hookEventForRecord); subscribe.addSubscribe(hook, hookEventForRecord);
@ -1101,7 +1113,7 @@ public class PlayServiceImpl implements IPlayService {
InviteInfo inviteInfo = inviteStreamService.getInviteInfo(InviteSessionType.DOWNLOAD, channel.getId(), stream); InviteInfo inviteInfo = inviteStreamService.getInviteInfo(InviteSessionType.DOWNLOAD, channel.getId(), stream);
if (inviteInfo == null) { if (inviteInfo == null) {
String app = "rtp"; String app = MediaApp.GB28181;
StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(app, stream); StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(app, stream);
if (streamAuthorityInfo != null) { if (streamAuthorityInfo != null) {
List<CloudRecordItem> allList = cloudRecordService.getAllList(null, app, stream, null, null, null, streamAuthorityInfo.getCallId(), null); List<CloudRecordItem> allList = cloudRecordService.getAllList(null, app, stream, null, null, null, streamAuthorityInfo.getCallId(), null);
@ -1143,7 +1155,7 @@ public class PlayServiceImpl implements IPlayService {
log.warn("[获取下载进度] 查询录像信息时发现节点不存在"); log.warn("[获取下载进度] 查询录像信息时发现节点不存在");
return null; return null;
} }
String app = "rtp"; String app = MediaApp.GB28181;
Long duration = mediaServerService.updateDownloadProcess(mediaServerItem, app, stream); Long duration = mediaServerService.updateDownloadProcess(mediaServerItem, app, stream);
if (duration == null || duration == 0) { if (duration == null || duration == 0) {
inviteInfo.getStreamInfo().setProgress(0); inviteInfo.getStreamInfo().setProgress(0);
@ -1186,7 +1198,7 @@ public class PlayServiceImpl implements IPlayService {
public StreamInfo onPublishHandler(MediaServer mediaServerItem, MediaInfo mediaInfo, Device device, DeviceChannel channel) { public StreamInfo onPublishHandler(MediaServer mediaServerItem, MediaInfo mediaInfo, Device device, DeviceChannel channel) {
StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStream(mediaServerItem, "rtp", mediaInfo.getStream(), mediaInfo, null); StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStream(mediaServerItem, MediaApp.GB28181, mediaInfo.getStream(), mediaInfo, null);
streamInfo.setDeviceId(device.getDeviceId()); streamInfo.setDeviceId(device.getDeviceId());
streamInfo.setChannelId(channel.getId()); streamInfo.setChannelId(channel.getId());
return streamInfo; return streamInfo;
@ -1557,7 +1569,7 @@ public class PlayServiceImpl implements IPlayService {
SendRtpInfo sendRtpInfo = sendRtpServerService.queryByChannelId(channel.getId(), device.getDeviceId()); SendRtpInfo sendRtpInfo = sendRtpServerService.queryByChannelId(channel.getId(), device.getDeviceId());
if (sendRtpInfo != null) { if (sendRtpInfo != null) {
MediaServer mediaServer = mediaServerService.getOne(sendRtpInfo.getMediaServerId()); MediaServer mediaServer = mediaServerService.getOne(sendRtpInfo.getMediaServerId());
Boolean streamReady = mediaServerService.isStreamReady(mediaServer, "rtp", sendRtpInfo.getReceiveStream()); Boolean streamReady = mediaServerService.isStreamReady(mediaServer, MediaApp.GB28181, sendRtpInfo.getReceiveStream());
if (streamReady) { if (streamReady) {
log.warn("[语音对讲] 进行中： {}", channel.getDeviceId()); log.warn("[语音对讲] 进行中： {}", channel.getDeviceId());
event.call("语音对讲进行中"); event.call("语音对讲进行中");
@ -1634,7 +1646,7 @@ public class PlayServiceImpl implements IPlayService {
String path = "snap"; String path = "snap";
// 请求截图 // 请求截图
log.info("[请求截图]: " + fileName); log.info("[请求截图]: " + fileName);
mediaServerService.getSnap(mediaServer, "rtp", inviteInfo.getStreamInfo().getStream(), 15, 1, path, fileName); mediaServerService.getSnap(mediaServer, MediaApp.GB28181, inviteInfo.getStreamInfo().getStream(), 15, 1, path, fileName);
File snapFile = new File(path + File.separator + fileName); File snapFile = new File(path + File.separator + fileName);
if (snapFile.exists()) { if (snapFile.exists()) {
errorCallback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), snapFile.getAbsoluteFile()); errorCallback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), snapFile.getAbsoluteFile());
@ -1676,7 +1688,7 @@ public class PlayServiceImpl implements IPlayService {
if (InviteSessionStatus.ok == inviteInfo.getStatus()) { if (InviteSessionStatus.ok == inviteInfo.getStatus()) {
try { try {
log.info("[停止点播/回放/下载] {}/{}", device.getDeviceId(), channel.getDeviceId()); log.info("[停止点播/回放/下载] {}/{}", device.getDeviceId(), channel.getDeviceId());
cmder.streamByeCmd(device, channel.getDeviceId(), "rtp", inviteInfo.getStream(), null, null); cmder.streamByeCmd(device, channel.getDeviceId(), MediaApp.GB28181, inviteInfo.getStream(), null, null);
} catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) { } catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) {
log.error("[命令发送失败] 停止点播/回放/下载， 发送BYE: {}", e.getMessage()); log.error("[命令发送失败] 停止点播/回放/下载， 发送BYE: {}", e.getMessage());
throw new ControllerException(ErrorCode.ERROR100.getCode(), "命令发送失败: " + e.getMessage()); throw new ControllerException(ErrorCode.ERROR100.getCode(), "命令发送失败: " + e.getMessage());
@ -1709,7 +1721,7 @@ public class PlayServiceImpl implements IPlayService {
if (InviteSessionStatus.ok == inviteInfo.getStatus()) { if (InviteSessionStatus.ok == inviteInfo.getStatus()) {
try { try {
log.info("[停止点播/回放/下载] {}/{}", device.getDeviceId(), channel.getDeviceId()); log.info("[停止点播/回放/下载] {}/{}", device.getDeviceId(), channel.getDeviceId());
cmder.streamByeCmd(device, channel.getDeviceId(), "rtp", inviteInfo.getStream(), null, null); cmder.streamByeCmd(device, channel.getDeviceId(), MediaApp.GB28181, inviteInfo.getStream(), null, null);
} catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) { } catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) {
log.warn("[命令发送失败] 停止点播/回放/下载， 发送BYE: {}", e.getMessage()); log.warn("[命令发送失败] 停止点播/回放/下载， 发送BYE: {}", e.getMessage());
} }

7
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
View File

@ -2,6 +2,7 @@ package com.genersoft.iot.vmp.gb28181.transmit.cmd.impl;
import com.genersoft.iot.vmp.common.InviteSessionType; import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.common.enums.MediaApp;
import com.genersoft.iot.vmp.conf.SipConfig; import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException; import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
@ -505,7 +506,7 @@ public class SIPCommander implements ISIPCommander {
} }
log.info("[语音喊话] {} 分配的ZLM为: {} [{}:{}]", stream, mediaServerItem.getId(), mediaServerItem.getIp(), sendRtpItem.getPort()); log.info("[语音喊话] {} 分配的ZLM为: {} [{}:{}]", stream, mediaServerItem.getId(), mediaServerItem.getIp(), sendRtpItem.getPort());
Hook hook = Hook.getInstance(HookType.on_media_arrival, "rtp", stream, mediaServerItem.getId()); Hook hook = Hook.getInstance(HookType.on_media_arrival, MediaApp.GB28181, stream, mediaServerItem.getId());
subscribe.addSubscribe(hook, (hookData) -> { subscribe.addSubscribe(hook, (hookData) -> {
if (event != null) { if (event != null) {
event.response(hookData); event.response(hookData);
@ -515,7 +516,7 @@ public class SIPCommander implements ISIPCommander {
CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()), device.getTransport()); CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()), device.getTransport());
callIdHeader.setCallId(callId); callIdHeader.setCallId(callId);
Hook publishHook = Hook.getInstance(HookType.on_publish, "rtp", stream, mediaServerItem.getId()); Hook publishHook = Hook.getInstance(HookType.on_publish, MediaApp.GB28181, stream, mediaServerItem.getId());
subscribe.addSubscribe(publishHook, (hookData) -> { subscribe.addSubscribe(publishHook, (hookData) -> {
if (eventForPush != null) { if (eventForPush != null) {
eventForPush.response(hookData); eventForPush.response(hookData);
@ -1388,7 +1389,7 @@ public class SIPCommander implements ISIPCommander {
@Override @Override
public void playbackControlCmd(Device device, DeviceChannel channel, String stream, String content, SipSubscribe.Event errorEvent, SipSubscribe.Event okEvent) throws SipException, InvalidArgumentException, ParseException { public void playbackControlCmd(Device device, DeviceChannel channel, String stream, String content, SipSubscribe.Event errorEvent, SipSubscribe.Event okEvent) throws SipException, InvalidArgumentException, ParseException {
SsrcTransaction ssrcTransaction = sessionManager.getSsrcTransactionByStream("rtp", stream); SsrcTransaction ssrcTransaction = sessionManager.getSsrcTransactionByStream(MediaApp.GB28181, stream);
if (ssrcTransaction == null) { if (ssrcTransaction == null) {
log.info("[回放控制]未找到视频流信息，设备：{}, 流ID: {}", device.getDeviceId(), stream); log.info("[回放控制]未找到视频流信息，设备：{}, 流ID: {}", device.getDeviceId(), stream);
return; return;

7
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderForPlatform.java
View File

@ -2,6 +2,7 @@ package com.genersoft.iot.vmp.gb28181.transmit.cmd.impl;
import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSON;
import com.genersoft.iot.vmp.common.InviteSessionType; import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.common.enums.MediaApp;
import com.genersoft.iot.vmp.conf.DynamicTask; import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException; import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
@ -641,7 +642,7 @@ public class SIPCommanderForPlatform implements ISIPCommanderForPlatform {
MediaServer mediaServerItem = mediaServerService.getOne(mediaServerId); MediaServer mediaServerItem = mediaServerService.getOne(mediaServerId);
if (mediaServerItem != null) { if (mediaServerItem != null) {
mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpItem.getSsrc()); mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpItem.getSsrc());
mediaServerService.closeRTPServer(mediaServerItem, sendRtpItem.getStream()); mediaServerService.closeRTPServer(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStream());
} }
SIPRequest byeRequest = headerProviderPlatformProvider.createByeRequest(platform, sendRtpItem, channel); SIPRequest byeRequest = headerProviderPlatformProvider.createByeRequest(platform, sendRtpItem, channel);
if (byeRequest == null) { if (byeRequest == null) {
@ -664,7 +665,7 @@ public class SIPCommanderForPlatform implements ISIPCommanderForPlatform {
} }
mediaServerService.releaseSsrc(ssrcTransaction.getMediaServerId(), ssrcTransaction.getSsrc()); mediaServerService.releaseSsrc(ssrcTransaction.getMediaServerId(), ssrcTransaction.getSsrc());
mediaServerService.closeRTPServer(ssrcTransaction.getMediaServerId(), ssrcTransaction.getStream()); mediaServerService.closeRTPServer(ssrcTransaction.getMediaServerId(), ssrcTransaction.getApp(), ssrcTransaction.getStream());
sessionManager.removeByStream(ssrcTransaction.getApp(), ssrcTransaction.getStream()); sessionManager.removeByStream(ssrcTransaction.getApp(), ssrcTransaction.getStream());
Request byteRequest = headerProviderPlatformProvider.createByteRequest(platform, channel.getGbDeviceId(), ssrcTransaction.getSipTransactionInfo()); Request byteRequest = headerProviderPlatformProvider.createByteRequest(platform, channel.getGbDeviceId(), ssrcTransaction.getSipTransactionInfo());
@ -705,7 +706,7 @@ public class SIPCommanderForPlatform implements ISIPCommanderForPlatform {
} }
log.info("{} 分配的ZLM为: {} [{}:{}]", stream, mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort()); log.info("{} 分配的ZLM为: {} [{}:{}]", stream, mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort());
Hook hook = Hook.getInstance(HookType.on_media_arrival, "rtp", stream, mediaServerItem.getId()); Hook hook = Hook.getInstance(HookType.on_media_arrival, MediaApp.GB28181, stream, mediaServerItem.getId());
subscribe.addSubscribe(hook, (hookData) -> { subscribe.addSubscribe(hook, (hookData) -> {
if (event != null) { if (event != null) {
event.response(hookData); event.response(hookData);

2
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java
View File

@ -228,7 +228,7 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
deviceChannelService.stopPlay(channel.getId()); deviceChannelService.stopPlay(channel.getId());
inviteStreamService.removeInviteInfo(inviteInfo); inviteStreamService.removeInviteInfo(inviteInfo);
if (inviteInfo.getStreamInfo() != null) { if (inviteInfo.getStreamInfo() != null) {
mediaServerService.closeRTPServer(inviteInfo.getStreamInfo().getMediaServer(), inviteInfo.getStreamInfo().getStream()); mediaServerService.closeRTPServer(inviteInfo.getStreamInfo().getMediaServer(), inviteInfo.getStreamInfo().getApp(), inviteInfo.getStreamInfo().getStream());
} }
} }
break; break;

3
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/MediaStatusNotifyMessageHandler.java
View File

@ -2,6 +2,7 @@ package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.notify
import com.genersoft.iot.vmp.common.InviteInfo; import com.genersoft.iot.vmp.common.InviteInfo;
import com.genersoft.iot.vmp.common.InviteSessionType; import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.common.enums.MediaApp;
import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.service.*; import com.genersoft.iot.vmp.gb28181.service.*;
import com.genersoft.iot.vmp.gb28181.session.SipInviteSessionManager; import com.genersoft.iot.vmp.gb28181.session.SipInviteSessionManager;
@ -98,7 +99,7 @@ public class MediaStatusNotifyMessageHandler extends SIPRequestProcessorParent i
playService.stop(inviteInfo); playService.stop(inviteInfo);
} }
// 去除监听流注销自动停止下载的监听 // 去除监听流注销自动停止下载的监听
Hook hook = Hook.getInstance(HookType.on_media_arrival, "rtp", ssrcTransaction.getStream(), ssrcTransaction.getMediaServerId()); Hook hook = Hook.getInstance(HookType.on_media_arrival, MediaApp.GB28181, ssrcTransaction.getStream(), ssrcTransaction.getMediaServerId());
subscribe.removeSubscribe(hook); subscribe.removeSubscribe(hook);
if (ssrcTransaction.getPlatformId() != null) { if (ssrcTransaction.getPlatformId() != null) {
// 如果级联播放，需要给上级发送此通知 TODO 多个上级同时观看一个下级 可能存在停错的问题，需要将点播CallId进行上下级绑定 // 如果级联播放，需要给上级发送此通知 TODO 多个上级同时观看一个下级 可能存在停错的问题，需要将点播CallId进行上下级绑定

102
src/main/java/com/genersoft/iot/vmp/jt1078/service/impl/jt1078PlayServiceImpl.java
View File

@ -3,14 +3,13 @@ package com.genersoft.iot.vmp.jt1078.service.impl;
import com.genersoft.iot.vmp.common.CommonCallback; import com.genersoft.iot.vmp.common.CommonCallback;
import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.common.VideoManagerConstants; import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.common.enums.MediaApp;
import com.genersoft.iot.vmp.conf.DynamicTask; import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException; import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.conf.ftpServer.FtpSetting;
import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.jt1078.bean.*; import com.genersoft.iot.vmp.jt1078.bean.*;
import com.genersoft.iot.vmp.jt1078.cmd.JT1078Template; import com.genersoft.iot.vmp.jt1078.cmd.JT1078Template;
import com.genersoft.iot.vmp.jt1078.event.FtpUploadEvent;
import com.genersoft.iot.vmp.jt1078.proc.request.J1205; import com.genersoft.iot.vmp.jt1078.proc.request.J1205;
import com.genersoft.iot.vmp.jt1078.proc.response.*; import com.genersoft.iot.vmp.jt1078.proc.response.*;
import com.genersoft.iot.vmp.jt1078.service.Ijt1078PlayService; import com.genersoft.iot.vmp.jt1078.service.Ijt1078PlayService;
@ -26,11 +25,12 @@ import com.genersoft.iot.vmp.media.event.media.MediaDepartureEvent;
import com.genersoft.iot.vmp.media.event.media.MediaNotFoundEvent; import com.genersoft.iot.vmp.media.event.media.MediaNotFoundEvent;
import com.genersoft.iot.vmp.media.event.mediaServer.MediaSendRtpStoppedEvent; import com.genersoft.iot.vmp.media.event.mediaServer.MediaSendRtpStoppedEvent;
import com.genersoft.iot.vmp.media.service.IMediaServerService; import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IReceiveRtpServerService;
import com.genersoft.iot.vmp.service.ISendRtpServerService; import com.genersoft.iot.vmp.service.ISendRtpServerService;
import com.genersoft.iot.vmp.service.bean.ErrorCallback; import com.genersoft.iot.vmp.service.bean.ErrorCallback;
import com.genersoft.iot.vmp.service.bean.InviteErrorCode; import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
import com.genersoft.iot.vmp.service.bean.RTPServerParam;
import com.genersoft.iot.vmp.service.bean.SSRCInfo; import com.genersoft.iot.vmp.service.bean.SSRCInfo;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.utils.DateUtil; import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.utils.MediaServerUtils; import com.genersoft.iot.vmp.utils.MediaServerUtils;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
@ -74,6 +74,9 @@ public class jt1078PlayServiceImpl implements Ijt1078PlayService {
@Autowired @Autowired
private IMediaServerService mediaServerService; private IMediaServerService mediaServerService;
@Autowired
private IReceiveRtpServerService receiveRtpServerService;
@Autowired @Autowired
private DynamicTask dynamicTask; private DynamicTask dynamicTask;
@ -197,7 +200,6 @@ public class jt1078PlayServiceImpl implements Ijt1078PlayService {
private void play(JTDevice device, JTChannel channel, int type, CommonCallback<WVPResult<StreamInfo>> callback) { private void play(JTDevice device, JTChannel channel, int type, CommonCallback<WVPResult<StreamInfo>> callback) {
String phoneNumber = device.getPhoneNumber(); String phoneNumber = device.getPhoneNumber();
int channelId = channel.getChannelId(); int channelId = channel.getChannelId();
String app = "1078";
String stream = phoneNumber + "_" + channelId; String stream = phoneNumber + "_" + channelId;
// 检查流是否已经存在，存在则返回 // 检查流是否已经存在，存在则返回
String playKey = VideoManagerConstants.INVITE_INFO_1078_PLAY + phoneNumber + ":" + channelId; String playKey = VideoManagerConstants.INVITE_INFO_1078_PLAY + phoneNumber + ":" + channelId;
@ -208,7 +210,7 @@ public class jt1078PlayServiceImpl implements Ijt1078PlayService {
MediaServer mediaServer = streamInfo.getMediaServer(); MediaServer mediaServer = streamInfo.getMediaServer();
if (mediaServer != null) { if (mediaServer != null) {
// 查询流是否存在，不存在则删除缓存数据 // 查询流是否存在，不存在则删除缓存数据
MediaInfo mediaInfo = mediaServerService.getMediaInfo(mediaServer, app, streamInfo.getStream()); MediaInfo mediaInfo = mediaServerService.getMediaInfo(mediaServer, MediaApp.JT1078, streamInfo.getStream());
if (mediaInfo != null) { if (mediaInfo != null) {
log.info("[JT-点播] 点播已经存在，直接返回， phoneNumber： {}， channelId： {}", phoneNumber, channelId); log.info("[JT-点播] 点播已经存在，直接返回， phoneNumber： {}， channelId： {}", phoneNumber, channelId);
for (CommonCallback<WVPResult<StreamInfo>> errorCallback : errorCallbacks) { for (CommonCallback<WVPResult<StreamInfo>> errorCallback : errorCallbacks) {
@ -233,48 +235,56 @@ public class jt1078PlayServiceImpl implements Ijt1078PlayService {
} }
return; return;
} }
// 设置hook监听
Hook hook = Hook.getInstance(HookType.on_media_arrival, app, stream, mediaServer.getId());
subscribe.addSubscribe(hook, (hookData) -> {
dynamicTask.stop(playKey);
log.info("[JT-点播] 点播成功， 手机号： {}， 通道： {}", phoneNumber, channelId);
// TODO 发送9105 实时音视频传输状态通知， 通知丢包率
StreamInfo info = onPublishHandler(mediaServer, hookData, phoneNumber, channelId);
for (CommonCallback<WVPResult<StreamInfo>> errorCallback : errorCallbacks) {
if (errorCallback == null) {
continue;
}
errorCallback.run(new WVPResult<>(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), info));
}
subscribe.removeSubscribe(hook);
redisTemplate.opsForValue().set(playKey, info);
// 截图
String path = "snap";
String fileName = phoneNumber + "_" + channelId + ".jpg";
// 请求截图
log.info("[请求截图]: " + fileName);
mediaServerService.getSnap(mediaServer, app, stream, 15, 1, path, fileName);
});
// 开启收流端口 // 开启收流端口
SSRCInfo ssrcInfo = mediaServerService.openJTTServer(mediaServer, stream, null, false, !channel.isHasAudio(), 1); RTPServerParam rtpServerParam = new RTPServerParam();
rtpServerParam.setMediaServer(mediaServer);
rtpServerParam.setApp(MediaApp.JT1078);
rtpServerParam.setStreamId(stream);
rtpServerParam.setSsrcCheck(false);
rtpServerParam.setPlayback(false);
rtpServerParam.setPort(0);
rtpServerParam.setTcpMode(1); // 1 表示tcp被动
rtpServerParam.setOnlyAuto(false);
rtpServerParam.setDisableAudio(!channel.isHasAudio());
SSRCInfo ssrcInfo = receiveRtpServerService.openRTPServer(rtpServerParam, (code, msg, result) -> {
if (code == InviteErrorCode.SUCCESS.getCode() && result != null && result.getHookData() != null) {
// hook响应
log.info("[JT-点播] 点播成功， 手机号： {}， 通道： {}", phoneNumber, channelId);
// TODO 发送9105 实时音视频传输状态通知， 通知丢包率
StreamInfo info = onPublishHandler(mediaServer, result.getHookData(), phoneNumber, channelId);
for (CommonCallback<WVPResult<StreamInfo>> errorCallback : errorCallbacks) {
if (errorCallback == null) {
continue;
}
errorCallback.run(new WVPResult<>(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), info));
}
redisTemplate.opsForValue().set(playKey, info);
// 截图
String path = "snap";
String fileName = phoneNumber + "_" + channelId + ".jpg";
// 请求截图
log.info("[请求截图]: " + fileName);
mediaServerService.getSnap(mediaServer, MediaApp.JT1078, stream, 15, 1, path, fileName);
}else {
if (callback != null) {
callback.run(WVPResult.fail(code, msg));
}
log.info("[JT-点播] 超时， phoneNumber： {}， channelId： {}", phoneNumber, channelId);
for (CommonCallback<WVPResult<StreamInfo>> errorCallback : errorCallbacks) {
errorCallback.run(new WVPResult<>(InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode(),
InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getMsg(), null));
}
stopPlay(phoneNumber, channelId);
}
});
if (ssrcInfo == null) { if (ssrcInfo == null) {
stopPlay(phoneNumber, channelId); stopPlay(phoneNumber, channelId);
return; return;
} }
// 设置超时监听
dynamicTask.startDelay(playKey, () -> {
log.info("[JT-点播] 超时， phoneNumber： {}， channelId： {}", phoneNumber, channelId);
for (CommonCallback<WVPResult<StreamInfo>> errorCallback : errorCallbacks) {
errorCallback.run(new WVPResult<>(InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode(),
InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getMsg(), null));
}
mediaServerService.closeJTTServer(mediaServer, stream, null);
subscribe.removeSubscribe(hook);
stopPlay(phoneNumber, channelId);
}, userSetting.getPlayTimeout());
log.info("[JT-点播] phoneNumber： {}， channelId： {}，IP: {}, 端口： {}", phoneNumber, channelId, mediaServer.getSdpIp(), ssrcInfo.getPort()); log.info("[JT-点播] phoneNumber： {}， channelId： {}，IP: {}, 端口： {}", phoneNumber, channelId, mediaServer.getSdpIp(), ssrcInfo.getPort());
J9101 j9101 = new J9101(); J9101 j9101 = new J9101();
j9101.setChannel(channelId); j9101.setChannel(channelId);
@ -287,7 +297,7 @@ public class jt1078PlayServiceImpl implements Ijt1078PlayService {
} }
public StreamInfo onPublishHandler(MediaServer mediaServerItem, HookData hookData, String phoneNumber, Integer channelId) { public StreamInfo onPublishHandler(MediaServer mediaServerItem, HookData hookData, String phoneNumber, Integer channelId) {
StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStream(mediaServerItem, "1078", hookData.getStream(), hookData.getMediaInfo(), null); StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStream(mediaServerItem, MediaApp.JT1078, hookData.getStream(), hookData.getMediaInfo(), null);
streamInfo.setDeviceId(phoneNumber); streamInfo.setDeviceId(phoneNumber);
streamInfo.setChannelId(channelId); streamInfo.setChannelId(channelId);
return streamInfo; return streamInfo;
@ -317,7 +327,7 @@ public class jt1078PlayServiceImpl implements Ijt1078PlayService {
// 删除缓存数据 // 删除缓存数据
if (streamInfo != null) { if (streamInfo != null) {
// 关闭rtpServer // 关闭rtpServer
mediaServerService.closeJTTServer(streamInfo.getMediaServer(), streamInfo.getStream(), null); receiveRtpServerService.closeRTPServer(streamInfo.getMediaServer(), new SSRCInfo(streamInfo.getApp(), streamInfo.getStream()));
redisTemplate.delete(playKey); redisTemplate.delete(playKey);
} }
@ -425,12 +435,12 @@ public class jt1078PlayServiceImpl implements Ijt1078PlayService {
StreamInfo streamInfo = (StreamInfo) redisTemplate.opsForValue().get(playbackKey); StreamInfo streamInfo = (StreamInfo) redisTemplate.opsForValue().get(playbackKey);
if (streamInfo != null) { if (streamInfo != null) {
mediaServerService.closeJTTServer(streamInfo.getMediaServer(), streamInfo.getStream(), null); mediaServerService.closeJTTServer(streamInfo.getMediaServer(), streamInfo.getStream());
// 清理数据 // 清理数据
redisTemplate.delete(playbackKey); redisTemplate.delete(playbackKey);
} }
String app = "1078"; String app = MediaApp.JT1078;
String stream = String.format("%s_%s_%s_%s", phoneNumber, channelId, String stream = String.format("%s_%s_%s_%s", phoneNumber, channelId,
DateUtil.yyyy_MM_dd_HH_mm_ssToUrl(startTime), DateUtil.yyyy_MM_dd_HH_mm_ssToUrl(endTime)); DateUtil.yyyy_MM_dd_HH_mm_ssToUrl(startTime), DateUtil.yyyy_MM_dd_HH_mm_ssToUrl(endTime));
MediaServer mediaServer; MediaServer mediaServer;
@ -615,7 +625,7 @@ public class jt1078PlayServiceImpl implements Ijt1078PlayService {
sendRtpInfo.setReceiveStream(stream + "_talk"); sendRtpInfo.setReceiveStream(stream + "_talk");
// 设置hook监听 // 设置hook监听
Hook hook = Hook.getInstance(HookType.on_media_arrival, "1078", sendRtpInfo.getReceiveStream(), mediaServer.getId()); Hook hook = Hook.getInstance(HookType.on_media_arrival, MediaApp.JT1078, sendRtpInfo.getReceiveStream(), mediaServer.getId());
subscribe.addSubscribe(hook, (hookData) -> { subscribe.addSubscribe(hook, (hookData) -> {
log.info("[JT-对讲] 对讲连接建立， phoneNumber： {}， channelId： {}", phoneNumber, channelId); log.info("[JT-对讲] 对讲连接建立， phoneNumber： {}， channelId： {}", phoneNumber, channelId);
subscribe.removeSubscribe(hook); subscribe.removeSubscribe(hook);

5
src/main/java/com/genersoft/iot/vmp/media/abl/ABLHttpHookListener.java
View File

@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.media.abl; package com.genersoft.iot.vmp.media.abl;
import com.alibaba.fastjson2.JSONObject; import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.enums.MediaApp;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException; import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.gb28181.service.IDeviceService; import com.genersoft.iot.vmp.gb28181.service.IDeviceService;
@ -194,14 +195,14 @@ public class ABLHttpHookListener {
logger.info("[ABL HOOK] 码流不到达通知：{}->{}/{}", param.getMediaServerId(), param.getApp(), param.getStream()); logger.info("[ABL HOOK] 码流不到达通知：{}->{}/{}", param.getMediaServerId(), param.getApp(), param.getStream());
try { try {
if ("rtp".equals(param.getApp())) { if (MediaApp.GB28181.equals(param.getApp())) {
return HookResult.SUCCESS(); return HookResult.SUCCESS();
} }
MediaRtpServerTimeoutEvent event = new MediaRtpServerTimeoutEvent(this); MediaRtpServerTimeoutEvent event = new MediaRtpServerTimeoutEvent(this);
MediaServer mediaServerItem = mediaServerService.getOne(param.getMediaServerId()); MediaServer mediaServerItem = mediaServerService.getOne(param.getMediaServerId());
if (mediaServerItem != null) { if (mediaServerItem != null) {
event.setMediaServer(mediaServerItem); event.setMediaServer(mediaServerItem);
event.setApp("rtp"); event.setApp(MediaApp.GB28181);
applicationEventPublisher.publishEvent(event); applicationEventPublisher.publishEvent(event);
} }
}catch (Exception e) { }catch (Exception e) {

49
src/main/java/com/genersoft/iot/vmp/media/abl/ABLMediaNodeServerService.java
View File

@ -5,6 +5,7 @@ import com.genersoft.iot.vmp.common.CommonCallback;
import com.genersoft.iot.vmp.common.InviteInfo; import com.genersoft.iot.vmp.common.InviteInfo;
import com.genersoft.iot.vmp.common.InviteSessionType; import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.common.enums.MediaApp;
import com.genersoft.iot.vmp.conf.SipConfig; import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException; import com.genersoft.iot.vmp.conf.exception.ControllerException;
@ -68,40 +69,40 @@ public class ABLMediaNodeServerService implements IMediaNodeServerService {
} }
@Override @Override
public int createRTPServer(MediaServer mediaServer, String stream, long ssrc, Integer port, Boolean onlyAuto, Boolean disableAudio, Boolean reUsePort, Integer tcpMode) { public int createRTPServer(MediaServer mediaServer, String app, String stream, long ssrc, Integer port, Boolean onlyAuto, Boolean disableAudio, Boolean reUsePort, Integer tcpMode) {
Boolean recordSip = userSetting.getRecordSip(); Boolean recordSip = userSetting.getRecordSip();
return ablresTfulUtils.openRtpServer(mediaServer, "rtp", stream, 96, port, tcpMode, disableAudio?1:0, recordSip, false); return ablresTfulUtils.openRtpServer(mediaServer, app, stream, 96, port, tcpMode, disableAudio?1:0, recordSip, false);
} }
@Override @Override
public void closeRtpServer(MediaServer mediaServer, String streamId, CommonCallback<Boolean> callback) { public void closeRtpServer(MediaServer mediaServer, String app, String stream, CommonCallback<Boolean> callback) {
if (mediaServer == null) { if (mediaServer == null) {
return; return;
} }
ABLResult result = ablresTfulUtils.closeStreams(mediaServer, "rtp", streamId); ABLResult result = ablresTfulUtils.closeStreams(mediaServer, app, stream);
logger.info("关闭RTP Server " + result); logger.info("关闭RTP Server " + result);
if (result.getCode() != 0) { if (result.getCode() != 0) {
logger.error("[closeRtpServer] 失败: {}", result.getMemo()); logger.error("[closeRtpServer] 失败: {}", result.getMemo());
} }
} }
@Override // @Override
public int createJTTServer(MediaServer mediaServer, String stream, Integer port, Boolean disableVideo, Boolean disableAudio, Integer tcpMode) { // public int createJTTServer(MediaServer mediaServer, String stream, Integer port, Boolean disableVideo, Boolean disableAudio, Integer tcpMode) {
Boolean recordSip = userSetting.getRecordSip(); // Boolean recordSip = userSetting.getRecordSip();
return ablresTfulUtils.openRtpServer(mediaServer, "1078", stream, 96, port, tcpMode, disableAudio?1:0, recordSip, true); // return ablresTfulUtils.openRtpServer(mediaServer, MediaApp.JT1078, stream, 96, port, tcpMode, disableAudio?1:0, recordSip, true);
} // }
//
@Override // @Override
public void closeJTTServer(MediaServer mediaServer, String streamId, CommonCallback<Boolean> callback) { // public void closeJTTServer(MediaServer mediaServer, String streamId, CommonCallback<Boolean> callback) {
if (mediaServer == null) { // if (mediaServer == null) {
return; // return;
} // }
ABLResult result = ablresTfulUtils.closeStreams(mediaServer, "1078", streamId); // ABLResult result = ablresTfulUtils.closeStreams(mediaServer, MediaApp.JT1078, streamId);
logger.info("关闭JT-RTP Server " + result); // logger.info("关闭JT-RTP Server " + result);
if (result.getCode() != 0) { // if (result.getCode() != 0) {
logger.error("[JT-closeRtpServer] 失败: {}", result.getMemo()); // logger.error("[JT-closeRtpServer] 失败: {}", result.getMemo());
} // }
} // }
@Override @Override
public void closeStreams(MediaServer mediaServer, String app, String streamId) { public void closeStreams(MediaServer mediaServer, String app, String streamId) {
@ -112,7 +113,7 @@ public class ABLMediaNodeServerService implements IMediaNodeServerService {
} }
@Override @Override
public Boolean updateRtpServerSSRC(MediaServer mediaServerItem, String streamId, String ssrc) { public Boolean updateRtpServerSSRC(MediaServer mediaServerItem, String app, String streamId, String ssrc) {
return null; return null;
} }
@ -256,7 +257,7 @@ public class ABLMediaNodeServerService implements IMediaNodeServerService {
} }
@Override @Override
public Boolean connectRtpServer(MediaServer mediaServerItem, String address, int port, String stream) { public Boolean connectRtpServer(MediaServer mediaServerItem, String address, int port, String app, String stream) {
logger.warn("[abl-connectRtpServer] 未实现"); logger.warn("[abl-connectRtpServer] 未实现");
return null; return null;
} }
@ -443,7 +444,7 @@ public class ABLMediaNodeServerService implements IMediaNodeServerService {
@Override @Override
public List<String> listRtpServer(MediaServer mediaServer) { public List<String> listRtpServer(MediaServer mediaServer) {
ABLResult ablResult = ablresTfulUtils.getMediaList(mediaServer, "rtp", null); ABLResult ablResult = ablresTfulUtils.getMediaList(mediaServer, MediaApp.GB28181, null);
if (ablResult.getCode() != 0) { if (ablResult.getCode() != 0) {
return null; return null;
} }

14
src/main/java/com/genersoft/iot/vmp/media/service/IMediaNodeServerService.java
View File

@ -15,19 +15,19 @@ import java.util.List;
import java.util.Map; import java.util.Map;
public interface IMediaNodeServerService { public interface IMediaNodeServerService {
int createRTPServer(MediaServer mediaServer, String streamId, long ssrc, Integer port, Boolean onlyAuto, Boolean disableAudio, Boolean reUsePort, Integer tcpMode); int createRTPServer(MediaServer mediaServer, String app, String stream, long ssrc, Integer port, Boolean onlyAuto, Boolean disableAudio, Boolean reUsePort, Integer tcpMode);
void closeRtpServer(MediaServer mediaServer, String streamId, CommonCallback<Boolean> callback); void closeRtpServer(MediaServer mediaServer, String app, String stream, CommonCallback<Boolean> callback);
int createJTTServer(MediaServer mediaServer, String streamId, Integer port, Boolean disableVideo, Boolean disableAudio, Integer tcpMode); // int createJTTServer(MediaServer mediaServer, String streamId, Integer port, Boolean disableVideo, Boolean disableAudio, Integer tcpMode);
//
void closeJTTServer(MediaServer mediaServer, String streamId, CommonCallback<Boolean> callback); // void closeJTTServer(MediaServer mediaServer, String streamId, CommonCallback<Boolean> callback);
void closeStreams(MediaServer mediaServer, String app, String stream); void closeStreams(MediaServer mediaServer, String app, String stream);
Boolean updateRtpServerSSRC(MediaServer mediaServer, String stream, String ssrc); Boolean updateRtpServerSSRC(MediaServer mediaServer, String app, String stream, String ssrc);
boolean checkNodeId(MediaServer mediaServer); boolean checkNodeId(MediaServer mediaServer);
@ -43,7 +43,7 @@ public interface IMediaNodeServerService {
List<StreamInfo> getMediaList(MediaServer mediaServer, String app, String stream, String callId); List<StreamInfo> getMediaList(MediaServer mediaServer, String app, String stream, String callId);
Boolean connectRtpServer(MediaServer mediaServer, String address, int port, String stream); Boolean connectRtpServer(MediaServer mediaServer, String address, int port, String app, String stream);
void getSnap(MediaServer mediaServer, String app, String stream, int timeoutSec, int expireSec, String path, String fileName); void getSnap(MediaServer mediaServer, String app, String stream, int timeoutSec, int expireSec, String path, String fileName);

20
src/main/java/com/genersoft/iot/vmp/media/service/IMediaServerService.java
View File

@ -34,20 +34,20 @@ public interface IMediaServerService {
void updateVmServer(List<MediaServer> mediaServerItemList); void updateVmServer(List<MediaServer> mediaServerItemList);
SSRCInfo openRTPServer(MediaServer mediaServerItem, String streamId, String presetSsrc, boolean ssrcCheck, SSRCInfo openRTPServer(MediaServer mediaServerItem, String app, String streamId, String presetSsrc, boolean ssrcCheck,
boolean isPlayback, Integer port, Boolean onlyAuto, Boolean disableAudio, Boolean reUsePort, Integer tcpMode); boolean isPlayback, Integer port, Boolean onlyAuto, Boolean disableAudio, Boolean reUsePort, Integer tcpMode);
void closeRTPServer(MediaServer mediaServerItem, String streamId); void closeRTPServer(MediaServer mediaServerItem, String app, String streamId);
void closeRTPServer(MediaServer mediaServerItem, String streamId, CommonCallback<Boolean> callback); void closeRTPServer(MediaServer mediaServerItem, String app, String streamId, CommonCallback<Boolean> callback);
SSRCInfo openJTTServer(MediaServer mediaServerItem, String streamId, Integer port, Boolean disableVideo, Boolean disableAudio, Integer tcpMode); // SSRCInfo openJTTServer(MediaServer mediaServerItem, String streamId, Integer port, Boolean disableVideo, Boolean disableAudio, Integer tcpMode);
//
// void closeJTTServer(MediaServer mediaServerItem, String streamId, CommonCallback<Boolean> callback);
void closeJTTServer(MediaServer mediaServerItem, String streamId, CommonCallback<Boolean> callback); Boolean updateRtpServerSSRC(MediaServer mediaServerItem, String app, String streamId, String ssrc);
Boolean updateRtpServerSSRC(MediaServer mediaServerItem, String streamId, String ssrc); void closeRTPServer(String mediaServerId, String app, String streamId);
void closeRTPServer(String mediaServerId, String streamId);
void clearRTPServer(MediaServer mediaServerItem); void clearRTPServer(MediaServer mediaServerItem);
@ -89,7 +89,7 @@ public interface IMediaServerService {
List<StreamInfo> getMediaList(MediaServer mediaInfo, String app, String stream, String callId); List<StreamInfo> getMediaList(MediaServer mediaInfo, String app, String stream, String callId);
Boolean connectRtpServer(MediaServer mediaServerItem, String address, int port, String stream); Boolean connectRtpServer(MediaServer mediaServerItem, String address, int port, String app, String stream);
void getSnap(MediaServer mediaServer, String app, String stream, int timeoutSec, int expireSec, String path, String fileName); void getSnap(MediaServer mediaServer, String app, String stream, int timeoutSec, int expireSec, String path, String fileName);
@ -162,7 +162,7 @@ public interface IMediaServerService {
StreamInfo getMediaByAppAndStream(String app, String stream); StreamInfo getMediaByAppAndStream(String app, String stream);
int createRTPServer(MediaServer mediaServerItem, String streamId, long ssrc, Integer port, boolean onlyAuto, boolean disableAudio, boolean reUsePort, Integer tcpMode); int createRTPServer(MediaServer mediaServerItem, String app, String streamId, long ssrc, Integer port, boolean onlyAuto, boolean disableAudio, boolean reUsePort, Integer tcpMode);
List<String> listRtpServer(MediaServer mediaServer); List<String> listRtpServer(MediaServer mediaServer);

98
src/main/java/com/genersoft/iot/vmp/media/service/impl/MediaServerServiceImpl.java
View File

@ -164,7 +164,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
@Override @Override
public SSRCInfo openRTPServer(MediaServer mediaServer, String streamId, String presetSsrc, boolean ssrcCheck, public SSRCInfo openRTPServer(MediaServer mediaServer, String app, String streamId, String presetSsrc, boolean ssrcCheck,
boolean isPlayback, Integer port, Boolean onlyAuto, Boolean disableAudio, Boolean reUsePort, Integer tcpMode) { boolean isPlayback, Integer port, Boolean onlyAuto, Boolean disableAudio, Boolean reUsePort, Integer tcpMode) {
if (mediaServer == null || mediaServer.getId() == null) { if (mediaServer == null || mediaServer.getId() == null) {
log.info("[openRTPServer] 失败, mediaServer == null || mediaServer.getId() == null"); log.info("[openRTPServer] 失败, mediaServer == null || mediaServer.getId() == null");
@ -196,15 +196,15 @@ public class MediaServerServiceImpl implements IMediaServerService {
log.info("[openRTPServer] 失败, mediaServer的类型： {}，未找到对应的实现类", mediaServer.getType()); log.info("[openRTPServer] 失败, mediaServer的类型： {}，未找到对应的实现类", mediaServer.getType());
return null; return null;
} }
rtpServerPort = mediaNodeServerService.createRTPServer(mediaServer, streamId, ssrcCheck ? Long.parseLong(ssrc) : 0, port, onlyAuto, disableAudio, reUsePort, tcpMode); rtpServerPort = mediaNodeServerService.createRTPServer(mediaServer, app, streamId, ssrcCheck ? Long.parseLong(ssrc) : 0, port, onlyAuto, disableAudio, reUsePort, tcpMode);
} else { } else {
rtpServerPort = mediaServer.getRtpProxyPort(); rtpServerPort = mediaServer.getRtpProxyPort();
} }
return new SSRCInfo(rtpServerPort, ssrc, "rtp", streamId, null); return new SSRCInfo(rtpServerPort, ssrc, app, streamId);
} }
@Override @Override
public int createRTPServer(MediaServer mediaServer, String streamId, long ssrc, Integer port, boolean onlyAuto, boolean disableAudio, boolean reUsePort, Integer tcpMode) { public int createRTPServer(MediaServer mediaServer, String app, String streamId, long ssrc, Integer port, boolean onlyAuto, boolean disableAudio, boolean reUsePort, Integer tcpMode) {
int rtpServerPort; int rtpServerPort;
if (mediaServer.isRtpEnable()) { if (mediaServer.isRtpEnable()) {
IMediaNodeServerService mediaNodeServerService = nodeServerServiceMap.get(mediaServer.getType()); IMediaNodeServerService mediaNodeServerService = nodeServerServiceMap.get(mediaServer.getType());
@ -212,33 +212,33 @@ public class MediaServerServiceImpl implements IMediaServerService {
log.info("[openRTPServer] 失败, mediaServer的类型： {}，未找到对应的实现类", mediaServer.getType()); log.info("[openRTPServer] 失败, mediaServer的类型： {}，未找到对应的实现类", mediaServer.getType());
return 0; return 0;
} }
rtpServerPort = mediaNodeServerService.createRTPServer(mediaServer, streamId, ssrc, port, onlyAuto, disableAudio, reUsePort, tcpMode); rtpServerPort = mediaNodeServerService.createRTPServer(mediaServer, app, streamId, ssrc, port, onlyAuto, disableAudio, reUsePort, tcpMode);
} else { } else {
rtpServerPort = mediaServer.getRtpProxyPort(); rtpServerPort = mediaServer.getRtpProxyPort();
} }
return rtpServerPort; return rtpServerPort;
} }
@Override // @Override
public SSRCInfo openJTTServer(MediaServer mediaServer, @NotNull String streamId, Integer port, Boolean disableVideo, Boolean disableAudio, Integer tcpMode) { // public SSRCInfo openJTTServer(MediaServer mediaServer, @NotNull String streamId, Integer port, Boolean disableVideo, Boolean disableAudio, Integer tcpMode) {
if (mediaServer == null || mediaServer.getId() == null) { // if (mediaServer == null || mediaServer.getId() == null) {
log.info("[openJTTServer] 失败, mediaServer == null || mediaServer.getId() == null"); // log.info("[openJTTServer] 失败, mediaServer == null || mediaServer.getId() == null");
return null; // return null;
} // }
//
int rtpServerPort; // int rtpServerPort;
if (mediaServer.isRtpEnable()) { // if (mediaServer.isRtpEnable()) {
IMediaNodeServerService mediaNodeServerService = nodeServerServiceMap.get(mediaServer.getType()); // IMediaNodeServerService mediaNodeServerService = nodeServerServiceMap.get(mediaServer.getType());
if (mediaNodeServerService == null) { // if (mediaNodeServerService == null) {
log.info("[openJTTServer] 失败, mediaServer的类型： {}，未找到对应的实现类", mediaServer.getType()); // log.info("[openJTTServer] 失败, mediaServer的类型： {}，未找到对应的实现类", mediaServer.getType());
return null; // return null;
} // }
rtpServerPort = mediaNodeServerService.createJTTServer(mediaServer, streamId, port, disableVideo, disableAudio, tcpMode); // rtpServerPort = mediaNodeServerService.createJTTServer(mediaServer, streamId, port, disableVideo, disableAudio, tcpMode);
} else { // } else {
rtpServerPort = mediaServer.getJttProxyPort(); // rtpServerPort = mediaServer.getJttProxyPort();
} // }
return new SSRCInfo(rtpServerPort, null, "1078", streamId, null); // return new SSRCInfo(rtpServerPort, null, "1078", streamId, null);
} // }
@Override @Override
public List<String> listRtpServer(MediaServer mediaServer) { public List<String> listRtpServer(MediaServer mediaServer) {
@ -251,7 +251,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
} }
@Override @Override
public void closeRTPServer(MediaServer mediaServer, String streamId) { public void closeRTPServer(MediaServer mediaServer, String app, String streamId) {
if (mediaServer == null) { if (mediaServer == null) {
return; return;
} }
@ -260,11 +260,11 @@ public class MediaServerServiceImpl implements IMediaServerService {
log.info("[closeRTPServer] 失败, mediaServer的类型： {}，未找到对应的实现类", mediaServer.getType()); log.info("[closeRTPServer] 失败, mediaServer的类型： {}，未找到对应的实现类", mediaServer.getType());
return; return;
} }
mediaNodeServerService.closeRtpServer(mediaServer, streamId, null); mediaNodeServerService.closeRtpServer(mediaServer, app, streamId, null);
} }
@Override @Override
public void closeRTPServer(MediaServer mediaServer, String streamId, CommonCallback<Boolean> callback) { public void closeRTPServer(MediaServer mediaServer, String app, String streamId, CommonCallback<Boolean> callback) {
if (mediaServer == null) { if (mediaServer == null) {
callback.run(false); callback.run(false);
return; return;
@ -274,42 +274,42 @@ public class MediaServerServiceImpl implements IMediaServerService {
log.info("[closeRTPServer] 失败, mediaServer的类型： {}，未找到对应的实现类", mediaServer.getType()); log.info("[closeRTPServer] 失败, mediaServer的类型： {}，未找到对应的实现类", mediaServer.getType());
return; return;
} }
mediaNodeServerService.closeRtpServer(mediaServer, streamId, callback); mediaNodeServerService.closeRtpServer(mediaServer, app, streamId, callback);
} }
@Override @Override
public void closeRTPServer(String mediaServerId, String streamId) { public void closeRTPServer(String mediaServerId, String app, String streamId) {
MediaServer mediaServer = this.getOne(mediaServerId); MediaServer mediaServer = this.getOne(mediaServerId);
if (mediaServer == null) { if (mediaServer == null) {
return; return;
} }
if (mediaServer.isRtpEnable()) { if (mediaServer.isRtpEnable()) {
closeRTPServer(mediaServer, streamId); closeRTPServer(mediaServer, app, streamId);
} }
IMediaNodeServerService mediaNodeServerService = nodeServerServiceMap.get(mediaServer.getType()); IMediaNodeServerService mediaNodeServerService = nodeServerServiceMap.get(mediaServer.getType());
if (mediaNodeServerService == null) { if (mediaNodeServerService == null) {
log.info("[closeRTPServer] 失败, mediaServer的类型： {}，未找到对应的实现类", mediaServer.getType()); log.info("[closeRTPServer] 失败, mediaServer的类型： {}，未找到对应的实现类", mediaServer.getType());
return; return;
} }
mediaNodeServerService.closeStreams(mediaServer, "rtp", streamId); mediaNodeServerService.closeStreams(mediaServer, app, streamId);
} }
@Override // @Override
public void closeJTTServer(MediaServer mediaServer, String streamId, CommonCallback<Boolean> callback) { // public void closeJTTServer(MediaServer mediaServer, String streamId, CommonCallback<Boolean> callback) {
if (mediaServer == null) { // if (mediaServer == null) {
callback.run(false); // callback.run(false);
return; // return;
} // }
IMediaNodeServerService mediaNodeServerService = nodeServerServiceMap.get(mediaServer.getType()); // IMediaNodeServerService mediaNodeServerService = nodeServerServiceMap.get(mediaServer.getType());
if (mediaNodeServerService == null) { // if (mediaNodeServerService == null) {
log.info("[closeJTTServer] 失败, mediaServer的类型： {}，未找到对应的实现类", mediaServer.getType()); // log.info("[closeJTTServer] 失败, mediaServer的类型： {}，未找到对应的实现类", mediaServer.getType());
return; // return;
} // }
mediaNodeServerService.closeJTTServer(mediaServer, streamId, callback); // mediaNodeServerService.closeJTTServer(mediaServer, streamId, callback);
} // }
@Override @Override
public Boolean updateRtpServerSSRC(MediaServer mediaServer, String streamId, String ssrc) { public Boolean updateRtpServerSSRC(MediaServer mediaServer, String app, String streamId, String ssrc) {
if (mediaServer == null) { if (mediaServer == null) {
return false; return false;
} }
@ -318,7 +318,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
log.info("[updateRtpServerSSRC] 失败, mediaServer的类型： {}，未找到对应的实现类", mediaServer.getType()); log.info("[updateRtpServerSSRC] 失败, mediaServer的类型： {}，未找到对应的实现类", mediaServer.getType());
return false; return false;
} }
return mediaNodeServerService.updateRtpServerSSRC(mediaServer, streamId, ssrc); return mediaNodeServerService.updateRtpServerSSRC(mediaServer, app, streamId, ssrc);
} }
@Override @Override
@ -716,13 +716,13 @@ public class MediaServerServiceImpl implements IMediaServerService {
} }
@Override @Override
public Boolean connectRtpServer(MediaServer mediaServer, String address, int port, String stream) { public Boolean connectRtpServer(MediaServer mediaServer, String address, int port, String app, String stream) {
IMediaNodeServerService mediaNodeServerService = nodeServerServiceMap.get(mediaServer.getType()); IMediaNodeServerService mediaNodeServerService = nodeServerServiceMap.get(mediaServer.getType());
if (mediaNodeServerService == null) { if (mediaNodeServerService == null) {
log.info("[connectRtpServer] 失败, mediaServer的类型： {}，未找到对应的实现类", mediaServer.getType()); log.info("[connectRtpServer] 失败, mediaServer的类型： {}，未找到对应的实现类", mediaServer.getType());
return false; return false;
} }
return mediaNodeServerService.connectRtpServer(mediaServer, address, port, stream); return mediaNodeServerService.connectRtpServer(mediaServer, address, port, app, stream);
} }
@Override @Override

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

@ -2,6 +2,7 @@ package com.genersoft.iot.vmp.media.zlm;
import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject; import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.enums.MediaApp;
import com.genersoft.iot.vmp.conf.MediaConfig; import com.genersoft.iot.vmp.conf.MediaConfig;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.media.bean.MediaServer; import com.genersoft.iot.vmp.media.bean.MediaServer;
@ -262,7 +263,7 @@ public class ZLMHttpHookListener {
log.info("[ZLM HOOK] rtp发送关闭：{}->{}/{}", param.getMediaServerId(), param.getApp(), param.getStream()); log.info("[ZLM HOOK] rtp发送关闭：{}->{}/{}", param.getMediaServerId(), param.getApp(), param.getStream());
// 查找对应的上级推流，发送停止 // 查找对应的上级推流，发送停止
if (!"rtp".equals(param.getApp())) { if (!MediaApp.GB28181.equals(param.getApp())) {
return HookResult.SUCCESS(); return HookResult.SUCCESS();
} }
try { try {
@ -293,7 +294,7 @@ public class ZLMHttpHookListener {
MediaServer mediaServerItem = mediaServerService.getOne(param.getMediaServerId()); MediaServer mediaServerItem = mediaServerService.getOne(param.getMediaServerId());
if (mediaServerItem != null) { if (mediaServerItem != null) {
event.setMediaServer(mediaServerItem); event.setMediaServer(mediaServerItem);
event.setApp("rtp"); event.setApp(MediaApp.GB28181);
applicationEventPublisher.publishEvent(event); applicationEventPublisher.publishEvent(event);
} }
}catch (Exception e) { }catch (Exception e) {

32
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMMediaNodeServerService.java
View File

@ -47,24 +47,24 @@ public class ZLMMediaNodeServerService implements IMediaNodeServerService {
private HookSubscribe subscribe; private HookSubscribe subscribe;
@Override @Override
public int createRTPServer(MediaServer mediaServer, String streamId, long ssrc, Integer port, Boolean onlyAuto, Boolean disableAudio, Boolean reUsePort, Integer tcpMode) { public int createRTPServer(MediaServer mediaServer, String app, String stream, long ssrc, Integer port, Boolean onlyAuto, Boolean disableAudio, Boolean reUsePort, Integer tcpMode) {
return zlmServerFactory.createRTPServer(mediaServer, "rtp", streamId, ssrc, port, onlyAuto, disableAudio, reUsePort, tcpMode); return zlmServerFactory.createRTPServer(mediaServer, app, stream, ssrc, port, onlyAuto, disableAudio, reUsePort, tcpMode);
} }
@Override @Override
public void closeRtpServer(MediaServer mediaServer, String streamId, CommonCallback<Boolean> callback) { public void closeRtpServer(MediaServer mediaServer, String app, String stream, CommonCallback<Boolean> callback) {
zlmServerFactory.closeRtpServer(mediaServer, streamId, callback); zlmServerFactory.closeRtpServer(mediaServer, app, stream, callback);
} }
@Override // @Override
public int createJTTServer(MediaServer mediaServer, String streamId, Integer port, Boolean disableVideo, Boolean disableAudio, Integer tcpMode) { // public int createJTTServer(MediaServer mediaServer, String streamId, Integer port, Boolean disableVideo, Boolean disableAudio, Integer tcpMode) {
return zlmServerFactory.createRTPServer(mediaServer, "1078", streamId, 0, port, disableVideo, disableAudio, false, tcpMode); // return zlmServerFactory.createRTPServer(mediaServer, "1078", streamId, 0, port, disableVideo, disableAudio, false, tcpMode);
} // }
@Override // @Override
public void closeJTTServer(MediaServer mediaServer, String streamId, CommonCallback<Boolean> callback) { // public void closeJTTServer(MediaServer mediaServer, String streamId, CommonCallback<Boolean> callback) {
zlmServerFactory.closeRtpServer(mediaServer, streamId, callback); // zlmServerFactory.closeRtpServer(mediaServer, streamId, callback);
} // }
@Override @Override
public void closeStreams(MediaServer mediaServer, String app, String stream) { public void closeStreams(MediaServer mediaServer, String app, String stream) {
@ -72,8 +72,8 @@ public class ZLMMediaNodeServerService implements IMediaNodeServerService {
} }
@Override @Override
public Boolean updateRtpServerSSRC(MediaServer mediaServer, String streamId, String ssrc) { public Boolean updateRtpServerSSRC(MediaServer mediaServer, String app, String streamId, String ssrc) {
return zlmServerFactory.updateRtpServerSSRC(mediaServer, streamId, ssrc); return zlmServerFactory.updateRtpServerSSRC(mediaServer, app, streamId, ssrc);
} }
@Override @Override
@ -208,8 +208,8 @@ public class ZLMMediaNodeServerService implements IMediaNodeServerService {
} }
@Override @Override
public Boolean connectRtpServer(MediaServer mediaServer, String address, int port, String stream) { public Boolean connectRtpServer(MediaServer mediaServer, String address, int port, String app, String stream) {
ZLMResult<?> zlmResult = zlmresTfulUtils.connectRtpServer(mediaServer, address, port, stream); ZLMResult<?> zlmResult = zlmresTfulUtils.connectRtpServer(mediaServer, address, port, app, stream);
log.info("[TCP主动连接对方] 结果： {}", zlmResult); log.info("[TCP主动连接对方] 结果： {}", zlmResult);
return zlmResult.getCode() == 0; return zlmResult.getCode() == 0;
} }

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

@ -694,10 +694,11 @@ public class ZLMRESTfulUtils {
} }
} }
public ZLMResult<?> connectRtpServer(MediaServer mediaServer, String dst_url, int dst_port, String stream_id) { public ZLMResult<?> connectRtpServer(MediaServer mediaServer, String dst_url, int dst_port, String app, String stream_id) {
Map<String, Object> param = new HashMap<>(1); Map<String, Object> param = new HashMap<>(1);
param.put("dst_url", dst_url); param.put("dst_url", dst_url);
param.put("dst_port", dst_port); param.put("dst_port", dst_port);
param.put("app", app);
param.put("stream_id", stream_id); param.put("stream_id", stream_id);
String response = sendPost(mediaServer, "connectRtpServer", param, null); String response = sendPost(mediaServer, "connectRtpServer", param, null);
if (response == null) { if (response == null) {
@ -712,9 +713,10 @@ public class ZLMRESTfulUtils {
} }
} }
public ZLMResult<?> updateRtpServerSSRC(MediaServer mediaServer, String streamId, String ssrc) { public ZLMResult<?> updateRtpServerSSRC(MediaServer mediaServer, String app, String streamId, String ssrc) {
Map<String, Object> param = new HashMap<>(1); Map<String, Object> param = new HashMap<>(1);
param.put("ssrc", ssrc); param.put("ssrc", ssrc);
param.put("app", app);
param.put("stream_id", streamId); param.put("stream_id", streamId);
String response = sendPost(mediaServer, "updateRtpServerSSRC", param, null); String response = sendPost(mediaServer, "updateRtpServerSSRC", param, null);

10
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMServerFactory.java
View File

@ -98,10 +98,11 @@ public class ZLMServerFactory {
return result; return result;
} }
public boolean closeRtpServer(MediaServer serverItem, String streamId) { public boolean closeRtpServer(MediaServer serverItem, String app, String streamId) {
boolean result = false; boolean result = false;
if (serverItem !=null){ if (serverItem !=null){
Map<String, Object> param = new HashMap<>(); Map<String, Object> param = new HashMap<>();
param.put("app", app);
param.put("stream_id", streamId); param.put("stream_id", streamId);
ZLMResult<?> zlmResult = zlmresTfulUtils.closeRtpServer(serverItem, param); ZLMResult<?> zlmResult = zlmresTfulUtils.closeRtpServer(serverItem, param);
if (zlmResult != null ) { if (zlmResult != null ) {
@ -118,7 +119,7 @@ public class ZLMServerFactory {
return result; return result;
} }
public void closeRtpServer(MediaServer serverItem, String streamId, CommonCallback<Boolean> callback) { public void closeRtpServer(MediaServer serverItem, String app, String streamId, CommonCallback<Boolean> callback) {
if (serverItem == null) { if (serverItem == null) {
if (callback != null) { if (callback != null) {
callback.run(false); callback.run(false);
@ -126,6 +127,7 @@ public class ZLMServerFactory {
return; return;
} }
Map<String, Object> param = new HashMap<>(); Map<String, Object> param = new HashMap<>();
param.put("app", app);
param.put("stream_id", streamId); param.put("stream_id", streamId);
zlmresTfulUtils.closeRtpServer(serverItem, param, zlmResult -> { zlmresTfulUtils.closeRtpServer(serverItem, param, zlmResult -> {
if (zlmResult.getCode() == 0) { if (zlmResult.getCode() == 0) {
@ -223,9 +225,9 @@ public class ZLMServerFactory {
return zlmResult; return zlmResult;
} }
public Boolean updateRtpServerSSRC(MediaServer mediaServerItem, String streamId, String ssrc) { public Boolean updateRtpServerSSRC(MediaServer mediaServerItem, String app, String streamId, String ssrc) {
boolean result = false; boolean result = false;
ZLMResult<?> zlmResult = zlmresTfulUtils.updateRtpServerSSRC(mediaServerItem, streamId, ssrc); ZLMResult<?> zlmResult = zlmresTfulUtils.updateRtpServerSSRC(mediaServerItem, app, streamId, ssrc);
if (zlmResult.getCode() == 0) { if (zlmResult.getCode() == 0) {
result= true; result= true;
log.info("[更新RTPServer] 成功"); log.info("[更新RTPServer] 成功");

7
src/main/java/com/genersoft/iot/vmp/service/IReceiveRtpServerService.java
View File

@ -1,13 +1,16 @@
package com.genersoft.iot.vmp.service; package com.genersoft.iot.vmp.service;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.gb28181.bean.OpenRTPServerResult; import com.genersoft.iot.vmp.gb28181.bean.OpenRTPServerResult;
import com.genersoft.iot.vmp.media.bean.MediaServer; import com.genersoft.iot.vmp.media.bean.MediaServer;
import com.genersoft.iot.vmp.media.event.hook.HookData;
import com.genersoft.iot.vmp.service.bean.ErrorCallback; import com.genersoft.iot.vmp.service.bean.ErrorCallback;
import com.genersoft.iot.vmp.service.bean.RTPServerParam; import com.genersoft.iot.vmp.service.bean.RTPServerParam;
import com.genersoft.iot.vmp.service.bean.SSRCInfo; import com.genersoft.iot.vmp.service.bean.SSRCInfo;
public interface IReceiveRtpServerService { public interface IReceiveRtpServerService {
SSRCInfo openRTPServer(RTPServerParam rtpServerParam, ErrorCallback<OpenRTPServerResult> callback);
void closeRTPServer(MediaServer mediaServer, SSRCInfo ssrcInfo); int openRTPServer(RTPServerParam rtpServerParam, ErrorCallback<HookData> callback);
void closeRTPServer(MediaServer mediaServer, String app, String stream);
} }

33
src/main/java/com/genersoft/iot/vmp/service/bean/RTPServerParam.java
View File

@ -2,15 +2,25 @@ package com.genersoft.iot.vmp.service.bean;
import com.genersoft.iot.vmp.media.bean.MediaServer; import com.genersoft.iot.vmp.media.bean.MediaServer;
import lombok.Data; import lombok.Data;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
@Data @Getter
@Setter
@NoArgsConstructor
public class RTPServerParam { public class RTPServerParam {
private MediaServer mediaServerItem; /**
* 使用的流媒体
*/
private MediaServer mediaServer;
private String app;
private String streamId; private String streamId;
private String presetSsrc; /**
private boolean ssrcCheck; * 开启rtpServer时使用的ssrc，开启rtpServer时会根据这个ssrc进行校验，如果不填则不校验
private boolean playback; */
private Long ssrc;
private Integer port; private Integer port;
private boolean onlyAuto; private boolean onlyAuto;
private boolean disableAudio; private boolean disableAudio;
@ -21,5 +31,16 @@ public class RTPServerParam {
*/ */
private Integer tcpMode; private Integer tcpMode;
public RTPServerParam(MediaServer mediaServer, String app, String streamId, Long ssrc, Integer port,
boolean onlyAuto, boolean disableAudio, boolean reUsePort, Integer tcpMode) {
this.mediaServer = mediaServer;
this.app = app;
this.streamId = streamId;
this.ssrc = ssrc;
this.port = port;
this.onlyAuto = onlyAuto;
this.disableAudio = disableAudio;
this.reUsePort = reUsePort;
this.tcpMode = tcpMode;
}
} }

5
src/main/java/com/genersoft/iot/vmp/service/bean/SSRCInfo.java
View File

@ -9,13 +9,12 @@ public class SSRCInfo {
private String ssrc; private String ssrc;
private String app; private String app;
private String Stream; private String Stream;
private String timeOutTaskKey;
public SSRCInfo(int port, String ssrc, String app, String stream, String timeOutTaskKey) { public SSRCInfo(int port, String ssrc, String app, String stream) {
this.port = port; this.port = port;
this.ssrc = ssrc; this.ssrc = ssrc;
this.app = app; this.app = app;
this.Stream = stream; this.Stream = stream;
this.timeOutTaskKey = timeOutTaskKey;
} }
} }

13
src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java
View File

@ -4,6 +4,7 @@ import com.genersoft.iot.vmp.common.InviteInfo;
import com.genersoft.iot.vmp.common.InviteSessionStatus; import com.genersoft.iot.vmp.common.InviteSessionStatus;
import com.genersoft.iot.vmp.common.InviteSessionType; import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.common.VideoManagerConstants; import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.common.enums.MediaApp;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException; import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel; import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
@ -82,7 +83,7 @@ public class MediaServiceImpl implements IMediaService {
if (app == null || stream == null) { if (app == null || stream == null) {
return false; return false;
} }
if ("rtp".equals(app)) { if (MediaApp.GB28181.equals(app)) {
return true; return true;
} }
StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(app, stream); StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(app, stream);
@ -95,7 +96,7 @@ public class MediaServiceImpl implements IMediaService {
@Override @Override
public ResultForOnPublish authenticatePublish(MediaServer mediaServer, String app, String stream, String params) { public ResultForOnPublish authenticatePublish(MediaServer mediaServer, String app, String stream, String params) {
// 推流鉴权的处理 // 推流鉴权的处理
if (!"rtp".equals(app) && !"1078".equals(app) ) { if (!MediaApp.GB28181.equals(app) && !MediaApp.JT1078.equals(app) ) {
if ("talk".equals(app) && stream.endsWith("_talk")) { if ("talk".equals(app) && stream.endsWith("_talk")) {
ResultForOnPublish result = new ResultForOnPublish(); ResultForOnPublish result = new ResultForOnPublish();
result.setEnable_mp4(false); result.setEnable_mp4(false);
@ -156,7 +157,7 @@ public class MediaServiceImpl implements IMediaService {
result.setEnable_audio(true); result.setEnable_audio(true);
// 国标流 // 国标流
if ("rtp".equals(app)) { if (MediaApp.GB28181.equals(app)) {
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, stream); InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, stream);
@ -223,7 +224,7 @@ public class MediaServiceImpl implements IMediaService {
}else { }else {
result.setEnable_mp4(userSetting.getRecordPushLive()); result.setEnable_mp4(userSetting.getRecordPushLive());
} }
if (app.equalsIgnoreCase("rtp")) { if (app.equalsIgnoreCase(MediaApp.GB28181)) {
String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "_" + stream; String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "_" + stream;
OtherRtpSendInfo otherRtpSendInfo = (OtherRtpSendInfo) redisTemplate.opsForValue().get(receiveKey); OtherRtpSendInfo otherRtpSendInfo = (OtherRtpSendInfo) redisTemplate.opsForValue().get(receiveKey);
@ -243,7 +244,7 @@ public class MediaServiceImpl implements IMediaService {
return false; return false;
} }
// 国标类型的流 // 国标类型的流
if ("rtp".equals(app)) { if (MediaApp.GB28181.equals(app)) {
result = userSetting.getStreamOnDemand(); result = userSetting.getStreamOnDemand();
// 国标流， 点播/录像回放/录像下载 // 国标流， 点播/录像回放/录像下载
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, stream); InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, stream);
@ -260,7 +261,7 @@ public class MediaServiceImpl implements IMediaService {
} }
return result; return result;
} }
}else if ("1078".equals(app)) { }else if (MediaApp.JT1078.equals(app)) {
// 判断是否是1078播放类型 // 判断是否是1078播放类型
JTMediaStreamType jtMediaStreamType = ijt1078Service.checkStreamFromJt(stream); JTMediaStreamType jtMediaStreamType = ijt1078Service.checkStreamFromJt(stream);
if (jtMediaStreamType != null) { if (jtMediaStreamType != null) {

124
src/main/java/com/genersoft/iot/vmp/service/impl/RtpServerServiceImpl.java
View File

@ -1,12 +1,16 @@
package com.genersoft.iot.vmp.service.impl; package com.genersoft.iot.vmp.service.impl;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.common.enums.MediaApp;
import com.genersoft.iot.vmp.conf.DynamicTask; import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.OpenRTPServerResult; import com.genersoft.iot.vmp.gb28181.bean.OpenRTPServerResult;
import com.genersoft.iot.vmp.gb28181.session.SSRCFactory; import com.genersoft.iot.vmp.gb28181.session.SSRCFactory;
import com.genersoft.iot.vmp.gb28181.session.SipInviteSessionManager; import com.genersoft.iot.vmp.gb28181.session.SipInviteSessionManager;
import com.genersoft.iot.vmp.media.bean.MediaInfo;
import com.genersoft.iot.vmp.media.bean.MediaServer; 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.Hook;
import com.genersoft.iot.vmp.media.event.hook.HookData;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribe; 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.hook.HookType;
import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent; import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent;
@ -23,12 +27,15 @@ import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Async; import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import java.util.Objects;
import java.util.UUID; import java.util.UUID;
@Slf4j @Slf4j
@Service @Service
public class RtpServerServiceImpl implements IReceiveRtpServerService { public class RtpServerServiceImpl implements IReceiveRtpServerService {
private final static String TIMEOUT_TASK_KEY_PREFIX = "RTP_SERVER_TIMEOUT_TASK";
@Autowired @Autowired
private IMediaServerService mediaServerService; private IMediaServerService mediaServerService;
@ -65,98 +72,111 @@ public class RtpServerServiceImpl implements IReceiveRtpServerService {
} }
@Override public SSRCInfo openGbRTPServer(MediaServer mediaServer, String streamId, String presetSSRC, int tcpMode,
public SSRCInfo openRTPServer(RTPServerParam rtpServerParam, ErrorCallback<OpenRTPServerResult> callback) { boolean playback, boolean ssrcCheck, boolean onlyAuto, boolean disableAuto, boolean reUsePort,
ErrorCallback<OpenRTPServerResult> callback ) {
if (callback == null) { if (callback == null) {
log.warn("[开启RTP收流] 失败，回调为NULL"); log.warn("[开启国标RTP收流] 失败，回调为NULL");
return null; return null;
} }
if (rtpServerParam.getMediaServerItem() == null) { if (mediaServer == null) {
log.warn("[开启RTP收流] 失败，媒体节点为NULL"); log.warn("[开启国标RTP收流] 失败，媒体节点为NULL");
return null; return null;
} }
// 获取mediaServer可用的ssrc // 获取mediaServer可用的ssrc
final String ssrc; final String ssrc;
if (rtpServerParam.getPresetSsrc() != null) { if (presetSSRC != null) {
ssrc = rtpServerParam.getPresetSsrc(); ssrc = presetSSRC;
}else { }else {
if (rtpServerParam.isPlayback()) { if (playback) {
ssrc = ssrcFactory.getPlayBackSsrc(rtpServerParam.getMediaServerItem().getId()); ssrc = ssrcFactory.getPlayBackSsrc(mediaServer.getId());
}else { }else {
ssrc = ssrcFactory.getPlaySsrc(rtpServerParam.getMediaServerItem().getId()); ssrc = ssrcFactory.getPlaySsrc(mediaServer.getId());
} }
} }
final String streamId; if (streamId == null) {
if (rtpServerParam.getStreamId() == null) {
streamId = String.format("%08x", Long.parseLong(ssrc)).toUpperCase(); streamId = String.format("%08x", Long.parseLong(ssrc)).toUpperCase();
}else {
streamId = rtpServerParam.getStreamId();
} }
if (rtpServerParam.isSsrcCheck() && rtpServerParam.getTcpMode() > 0) { if (ssrcCheck && tcpMode > 0) {
// 目前zlm不支持 tcp模式更新ssrc，暂时关闭ssrc校验 // 目前zlm不支持 tcp模式更新ssrc，暂时关闭ssrc校验
log.warn("[openRTPServer] 平台对接时下级可能自定义ssrc，但是tcp模式zlm收流目前无法更新ssrc，可能收流超时，此时请使用udp收流或者关闭ssrc校验"); log.warn("[openRTPServer] 平台对接时下级可能自定义ssrc，但是tcp模式zlm收流目前无法更新ssrc，可能收流超时，此时请使用udp收流或者关闭ssrc校验");
} }
int rtpServerPort;
if (rtpServerParam.getMediaServerItem().isRtpEnable()) { SSRCInfo ssrcInfo = new SSRCInfo(0, ssrc, MediaApp.GB28181, streamId);
rtpServerPort = mediaServerService.createRTPServer(rtpServerParam.getMediaServerItem(), streamId, RTPServerParam rtpServerParam = new RTPServerParam(mediaServer, MediaApp.GB28181, streamId, ssrcCheck ? Long.parseLong(ssrc): 0L, null, onlyAuto, disableAuto, reUsePort, tcpMode);
rtpServerParam.isSsrcCheck() ? Long.parseLong(ssrc) : 0, rtpServerParam.getPort(), rtpServerParam.isOnlyAuto(), int rtpServerPort = openRTPServer(rtpServerParam, ((code, msg, data) -> {
rtpServerParam.isDisableAudio(), rtpServerParam.isReUsePort(), rtpServerParam.getTcpMode()); if (code == InviteErrorCode.SUCCESS.getCode()) {
} else { OpenRTPServerResult openRTPServerResult = new OpenRTPServerResult();
rtpServerPort = rtpServerParam.getMediaServerItem().getRtpProxyPort(); openRTPServerResult.setHookData(data);
} openRTPServerResult.setSsrcInfo(ssrcInfo);
if (rtpServerPort == 0) { callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), openRTPServerResult);
callback.run(InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getCode(), "开启RTPServer失败", null); } else {
// 释放ssrc // 释放ssrc
if (rtpServerParam.getPresetSsrc() == null) { if (presetSSRC == null) {
ssrcFactory.releaseSsrc(rtpServerParam.getMediaServerItem().getId(), ssrc); ssrcFactory.releaseSsrc(mediaServer.getId(), ssrc);
}
callback.run(code, msg, null);
} }
return null; }));
ssrcInfo.setPort(rtpServerPort);
return new SSRCInfo(rtpServerPort, ssrc, MediaApp.GB28181, streamId);
}
@Override
public int openRTPServer(RTPServerParam rtpServerParam, ErrorCallback<HookData> callback) {
if (callback == null) {
log.warn("[开启RTP收流] 失败，回调为NULL");
return -1;
}
if (rtpServerParam.getMediaServer() == null) {
log.warn("[开启RTP收流] 失败，媒体节点为NULL");
return -1;
} }
// 设置流超时的定时任务 // 设置流超时的定时任务
String timeOutTaskKey = UUID.randomUUID().toString(); String timeOutTaskKey = String.format("%s_%s_%s_%s", TIMEOUT_TASK_KEY_PREFIX, rtpServerParam.getMediaServer().getId(), rtpServerParam.getApp(), rtpServerParam.getStreamId());
SSRCInfo ssrcInfo = new SSRCInfo(rtpServerPort, ssrc, "rtp", streamId, timeOutTaskKey); Hook rtpHook = Hook.getInstance(HookType.on_media_arrival, rtpServerParam.getApp(), rtpServerParam.getStreamId(), rtpServerParam.getMediaServer().getId());
OpenRTPServerResult openRTPServerResult = new OpenRTPServerResult();
openRTPServerResult.setSsrcInfo(ssrcInfo);
Hook rtpHook = Hook.getInstance(HookType.on_media_arrival, ssrcInfo.getApp(), streamId, rtpServerParam.getMediaServerItem().getId());
dynamicTask.startDelay(timeOutTaskKey, () -> { dynamicTask.startDelay(timeOutTaskKey, () -> {
// 收流超时 // 收流超时
// 释放ssrc
if (rtpServerParam.getPresetSsrc() == null) {
ssrcFactory.releaseSsrc(rtpServerParam.getMediaServerItem().getId(), ssrc);
}
// 关闭收流端口 // 关闭收流端口
mediaServerService.closeRTPServer(rtpServerParam.getMediaServerItem(), streamId); mediaServerService.closeRTPServer(rtpServerParam.getMediaServer(), rtpServerParam.getApp(), rtpServerParam.getStreamId());
subscribe.removeSubscribe(rtpHook); subscribe.removeSubscribe(rtpHook);
callback.run(InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode(), InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getMsg(), openRTPServerResult); callback.run(InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode(), InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getMsg(), null);
}, userSetting.getPlayTimeout()); }, userSetting.getPlayTimeout());
// 开启流到来的监听 // 开启流到来的监听
subscribe.addSubscribe(rtpHook, (hookData) -> { subscribe.addSubscribe(rtpHook, (hookData) -> {
dynamicTask.stop(timeOutTaskKey); dynamicTask.stop(timeOutTaskKey);
// hook响应 // hook响应
openRTPServerResult.setHookData(hookData); callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), hookData);
callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), openRTPServerResult);
subscribe.removeSubscribe(rtpHook); subscribe.removeSubscribe(rtpHook);
}); });
return ssrcInfo; int rtpServerPort;
if (rtpServerParam.getMediaServer().isRtpEnable()) {
rtpServerPort = mediaServerService.createRTPServer(rtpServerParam.getMediaServer(), rtpServerParam.getApp(), rtpServerParam.getStreamId(),
Objects.requireNonNullElse(rtpServerParam.getSsrc(), 0L), rtpServerParam.getPort(), rtpServerParam.isOnlyAuto(),
rtpServerParam.isDisableAudio(), rtpServerParam.isReUsePort(), rtpServerParam.getTcpMode());
} else {
rtpServerPort = rtpServerParam.getMediaServer().getRtpProxyPort();
}
if (rtpServerPort == 0) {
callback.run(InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getCode(), "开启RTPServer失败", null);
return -1;
}
return rtpServerPort;
} }
@Override @Override
public void closeRTPServer(MediaServer mediaServer, SSRCInfo ssrcInfo) { public void closeRTPServer(MediaServer mediaServer, String app, String stream) {
if (mediaServer == null) { if (mediaServer == null) {
return; return;
} }
if (ssrcInfo.getTimeOutTaskKey() != null) { String timeOutTaskKey = String.format("%s_%s_%s_%s", TIMEOUT_TASK_KEY_PREFIX, mediaServer.getId(), app, stream);
dynamicTask.stop(ssrcInfo.getTimeOutTaskKey()); if (dynamicTask.contains(timeOutTaskKey)) {
dynamicTask.stop(timeOutTaskKey);
} }
if (ssrcInfo.getSsrc() != null) { mediaServerService.closeRTPServer(mediaServer, app, stream);
// 释放ssrc
ssrcFactory.releaseSsrc(mediaServer.getId(), ssrcInfo.getSsrc());
}
mediaServerService.closeRTPServer(mediaServer, ssrcInfo.getStream());
} }
} }

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

@ -3,6 +3,7 @@ package com.genersoft.iot.vmp.streamProxy.service.impl;
import com.alibaba.fastjson2.JSONObject; import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.common.enums.ChannelDataType; import com.genersoft.iot.vmp.common.enums.ChannelDataType;
import com.genersoft.iot.vmp.common.enums.MediaApp;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException; import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.gb28181.bean.CommonGBChannel; import com.genersoft.iot.vmp.gb28181.bean.CommonGBChannel;
@ -103,7 +104,7 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
@Async("taskExecutor") @Async("taskExecutor")
@EventListener @EventListener
public void onApplicationEvent(MediaNotFoundEvent event) { public void onApplicationEvent(MediaNotFoundEvent event) {
if ("rtp".equals(event.getApp())) { if (MediaApp.GB28181.equals(event.getApp())) {
return; return;
} }
// 拉流代理 // 拉流代理

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

@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.vmanager.ps; package com.genersoft.iot.vmp.vmanager.ps;
import com.genersoft.iot.vmp.common.VideoManagerConstants; import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.common.enums.MediaApp;
import com.genersoft.iot.vmp.conf.DynamicTask; import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException; import com.genersoft.iot.vmp.conf.exception.ControllerException;
@ -92,14 +93,14 @@ public class PsController {
} }
} }
String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_PS_INFO + userSetting.getServerId() + "_" + callId + "_" + stream; String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_PS_INFO + userSetting.getServerId() + "_" + callId + "_" + stream;
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServer, stream, ssrcInt + "", false, false, null, false, false, false, tcpMode); SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServer, MediaApp.GB28181, stream, ssrcInt + "", false, false, null, false, false, false, tcpMode);
if (ssrcInfo.getPort() == 0) { if (ssrcInfo.getPort() == 0) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "获取端口失败"); throw new ControllerException(ErrorCode.ERROR100.getCode(), "获取端口失败");
} }
// 注册回调如果rtp收流超时则通过回调发送通知 // 注册回调如果rtp收流超时则通过回调发送通知
if (callBack != null) { if (callBack != null) {
Hook hook = Hook.getInstance(HookType.on_rtp_server_timeout, "rtp", stream, mediaServer.getId()); Hook hook = Hook.getInstance(HookType.on_rtp_server_timeout, MediaApp.GB28181, stream, mediaServer.getId());
// 订阅 zlm启动事件, 新的zlm也会从这里进入系统 // 订阅 zlm启动事件, 新的zlm也会从这里进入系统
hookSubscribe.addSubscribe(hook, hookSubscribe.addSubscribe(hook,
(hookData)->{ (hookData)->{
@ -149,7 +150,7 @@ public class PsController {
public void closeRtpServer(String stream) { public void closeRtpServer(String stream) {
log.info("[第三方PS服务对接->关闭收流] stream->{}", stream); log.info("[第三方PS服务对接->关闭收流] stream->{}", stream);
MediaServer mediaServerItem = mediaServerService.getDefaultMediaServer(); MediaServer mediaServerItem = mediaServerService.getDefaultMediaServer();
mediaServerService.closeRTPServer(mediaServerItem, stream); mediaServerService.closeRTPServer(mediaServerItem, MediaApp.GB28181, stream);
String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_PS_INFO + userSetting.getServerId() + "_*_" + stream; String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_PS_INFO + userSetting.getServerId() + "_*_" + stream;
List<Object> scan = RedisUtil.scan(redisTemplate, receiveKey); List<Object> scan = RedisUtil.scan(redisTemplate, receiveKey);
if (!scan.isEmpty()) { if (!scan.isEmpty()) {

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

@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.vmanager.rtp; package com.genersoft.iot.vmp.vmanager.rtp;
import com.genersoft.iot.vmp.common.VideoManagerConstants; import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.common.enums.MediaApp;
import com.genersoft.iot.vmp.conf.DynamicTask; import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException; import com.genersoft.iot.vmp.conf.exception.ControllerException;
@ -92,14 +93,14 @@ public class RtpController {
} }
} }
String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "_" + callId + "_" + stream; String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "_" + callId + "_" + stream;
SSRCInfo ssrcInfoForVideo = mediaServerService.openRTPServer(mediaServer, stream, ssrcInt + "",false,false, null, false, false, false, tcpMode); SSRCInfo ssrcInfoForVideo = mediaServerService.openRTPServer(mediaServer, MediaApp.GB28181, stream, ssrcInt + "",false,false, null, false, false, false, tcpMode);
SSRCInfo ssrcInfoForAudio = mediaServerService.openRTPServer(mediaServer, stream + "_a", ssrcInt + "", false, false, null, false,false,false, tcpMode); SSRCInfo ssrcInfoForAudio = mediaServerService.openRTPServer(mediaServer, MediaApp.GB28181,stream + "_a", ssrcInt + "", false, false, null, false,false,false, tcpMode);
if (ssrcInfoForVideo.getPort() == 0 || ssrcInfoForAudio.getPort() == 0) { if (ssrcInfoForVideo.getPort() == 0 || ssrcInfoForAudio.getPort() == 0) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "获取端口失败"); throw new ControllerException(ErrorCode.ERROR100.getCode(), "获取端口失败");
} }
// 注册回调如果rtp收流超时则通过回调发送通知 // 注册回调如果rtp收流超时则通过回调发送通知
if (callBack != null) { if (callBack != null) {
Hook hook = Hook.getInstance(HookType.on_rtp_server_timeout, "rtp", stream, mediaServer.getId()); Hook hook = Hook.getInstance(HookType.on_rtp_server_timeout, MediaApp.GB28181, stream, mediaServer.getId());
// 订阅 zlm启动事件, 新的zlm也会从这里进入系统 // 订阅 zlm启动事件, 新的zlm也会从这里进入系统
hookSubscribe.addSubscribe(hook, hookSubscribe.addSubscribe(hook,
(hookData)->{ (hookData)->{
@ -152,8 +153,8 @@ public class RtpController {
public void closeRtpServer(String stream) { public void closeRtpServer(String stream) {
log.info("[第三方服务对接->关闭收流] stream->{}", stream); log.info("[第三方服务对接->关闭收流] stream->{}", stream);
MediaServer mediaServerItem = mediaServerService.getDefaultMediaServer(); MediaServer mediaServerItem = mediaServerService.getDefaultMediaServer();
mediaServerService.closeRTPServer(mediaServerItem, stream); mediaServerService.closeRTPServer(mediaServerItem, MediaApp.GB28181, stream);
mediaServerService.closeRTPServer(mediaServerItem, stream+ "_a"); mediaServerService.closeRTPServer(mediaServerItem, MediaApp.GB28181, stream+ "_a");
String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "_*_" + stream; String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "_*_" + stream;
List<Object> scan = RedisUtil.scan(redisTemplate, receiveKey); List<Object> scan = RedisUtil.scan(redisTemplate, receiveKey);
if (scan.size() > 0) { if (scan.size() > 0) {

3
src/main/java/com/genersoft/iot/vmp/web/gb28181/ApiStreamController.java
View File

@ -4,6 +4,7 @@ import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.InviteInfo; import com.genersoft.iot.vmp.common.InviteInfo;
import com.genersoft.iot.vmp.common.InviteSessionType; import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.common.enums.MediaApp;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException; import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
import com.genersoft.iot.vmp.gb28181.bean.Device; import com.genersoft.iot.vmp.gb28181.bean.Device;
@ -246,7 +247,7 @@ public class ApiStreamController {
} }
try { try {
cmder.streamByeCmd(device, code, "rtp", inviteInfo.getStream(), null, null); cmder.streamByeCmd(device, code, MediaApp.GB28181, inviteInfo.getStream(), null, null);
} catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) { } catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) {
JSONObject result = new JSONObject(); JSONObject result = new JSONObject();
result.put("error","发送BYE失败：" + e.getMessage()); result.put("error","发送BYE失败：" + e.getMessage());

2
src/main/resources/application.yml
View File

@ -2,4 +2,4 @@ spring:
application: application:
name: wvp name: wvp
profiles: profiles:
active: 274-dev active: dev