Shikong/wvp-gb28181-project
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: Shikong:f0cba184fea47421993697dcf5138678ae82443f
Branches Tags
Shikong:dev-2.6.9_整理
Shikong:wvp-28181-2.0
...
compare: Shikong:17278523d940d34cd67d084de4a6c71cf035cb9b
Branches Tags
Shikong:wvp-28181-2.0
Shikong:dev-2.6.9_整理

18 Commits
f0cba184fe ... 17278523d9

Author SHA1 Message Date
zxb
17278523d9 修复历史文件拉取异常 2023-08-21 16:55:00 +08:00
zxb
19f7dc21a1 支持配置 assistant 的ip 2023-08-17 13:25:02 +08:00
zxb
869cb772d9 ssrc 码流类型调整 2023-08-16 15:49:56 +08:00
zxb
c8a6ad283c SipRequest fromHeader 为 channelId 时无法正常查询 
DeviceChannelMapper getDeviceByChannelId 查询语句错误修复
 2023-08-15 14:46:41 +08:00
zxb
4bfe6fc17d 心跳过于频繁 添加判断 2023-08-14 17:26:32 +08:00
zxb
5222f89bee 测试 2023-08-11 10:49:31 +08:00
shikong
3e2910d7ea Merge remote-tracking branch 'github/wvp-28181-2.0' into dev-2.6.9_整理 2023-08-11 09:23:21 +08:00
648540858
6291af7d83 Merge branch '2.6.8' into wvp-28181-2.0 2023-08-08 16:06:39 +08:00
648540858
8bb660fa86 解决使用redis集群时获取发送端口失败的问题 2023-08-08 16:06:02 +08:00
zxb
df46dbd653 wvp_device_channel stream_id 长度过短导致国标级联上级点播历史录像出错 
其他奇奇怪怪的bug 调试/修复
 2023-08-08 11:26:51 +08:00
648540858
66fa075a1a 支持对接第三方PS流 2023-08-08 10:01:51 +08:00
648540858
e9b3e659a8 Merge branch '2.6.8' into wvp-28181-2.0 2023-08-08 09:50:31 +08:00
648540858
8830a6046a 支持与第三方对接PS流 2023-08-08 09:49:27 +08:00
648540858
6b169b208c Merge remote-tracking branch 'origin/wvp-28181-2.0' into wvp-28181-2.0 2023-08-07 17:04:11 +08:00
648540858
8b3ea68191 支持全局固定流地址 2023-08-07 17:00:01 +08:00
648540858
ba4de716a5 Merge pull request #989 from dodu2014/wvp-28181-2.0-1 
Update config.md
 2023-08-07 10:05:13 +08:00
dodu2014
1e16b1a51c Update config.md 
修复文档错别字
 2023-08-07 09:55:19 +08:00
648540858
97800b2610 invite消息缓存字符间隔改为使用:代替_,避免scan查询失败 2023-08-04 17:53:12 +08:00
36 changed files with 1138 additions and 207 deletions
Show Stats Expand all files Collapse all files

56
app.sh Normal file
View File

@ -0,0 +1,56 @@
#!/bin/bash
cd $(dirname $0)
exe="java -jar wvp-pro-2.6.9.jar"
arg=""
status(){
count=$(ps aux|grep "$exe"|grep -v grep|wc -l)
if [ $count -eq 0 ] ; then
echo "$exe is stopped"
return
fi
echo "$exe is running"
}
start(){
count=$(ps aux|grep "$exe"|grep -v grep|wc -l)
if [ $count -eq 0 ] ; then
echo "staring $exe"
nohup $exe $arg > nohup.out 2>&1 &
sleep 1
status
exit 0
fi
status
}
stop(){
echo "kill process"
nohup pkill -f "$exe" >/dev/null 2>&1 &
sleep 1
status
}
restart(){
stop
sleep 5
start
}
case $1 in
start)
start
;;
stop)
stop
;;
restart)
restart
;;
status)
status;;
*)
echo "use $0 start|stop|restart|status"
esac

2
doc/_content/introduction/config.md
View File

@ -160,4 +160,4 @@ user-settings:
如果配置信息无误，你可以启动zlm，再启动wvp来测试了，启动成功的话，你可以在wvp的日志下看到zlm已连接的提示。
接下来[部署到服务器](./_content/introduction/deployment.md)， 如何你只是本地运行直接再本地运行即可。
接下来[部署到服务器](./_content/introduction/deployment.md)， 如果你只是本地运行直接在本地运行即可。

13
pom.xml
View File

@ -300,11 +300,11 @@
<scope>test</scope>
</dependency>
<!-- 开发工具 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-devtools</artifactId>
</dependency>
<!--&lt;!&ndash; 开发工具 &ndash;&gt;-->
<!--<dependency>-->
<!-- <groupId>org.springframework.boot</groupId>-->
<!-- <artifactId>spring-boot-devtools</artifactId>-->
<!--</dependency>-->
<!-- lombok -->
<dependency>
@ -328,7 +328,8 @@
<build>
<finalName>${project.artifactId}-${project.version}-${maven.build.timestamp}</finalName>
<!--<finalName>${project.artifactId}-${project.version}-${maven.build.timestamp}</finalName>-->
<finalName>${project.artifactId}-${project.version}</finalName>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>

2
sql/2.6.8升级2.6.9.sql
View File

@ -120,7 +120,7 @@ alter table device_channel
change status status bool default false;
alter table device_channel
change streamId stream_id varchar(50) null;
change streamId stream_id varchar(80) null;
alter table device_channel
change deviceId device_id varchar(50) not null;

2
src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java
View File

@ -158,7 +158,9 @@ public class VideoManagerConstants {
public static final String WVP_STREAM_GB_ID_PREFIX = "memberNo_";
public static final String WVP_STREAM_GPS_MSG_PREFIX = "WVP_STREAM_GPS_MSG_";
public static final String WVP_OTHER_SEND_RTP_INFO = "VMP_OTHER_SEND_RTP_INFO_";
public static final String WVP_OTHER_SEND_PS_INFO = "VMP_OTHER_SEND_PS_INFO_";
public static final String WVP_OTHER_RECEIVE_RTP_INFO = "VMP_OTHER_RECEIVE_RTP_INFO_";
public static final String WVP_OTHER_RECEIVE_PS_INFO = "VMP_OTHER_RECEIVE_PS_INFO_";
/**
* Redis Const

6
src/main/java/com/genersoft/iot/vmp/conf/MediaConfig.java
View File

@ -2,6 +2,7 @@ package com.genersoft.iot.vmp.conf;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.utils.DateUtil;
import lombok.Data;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
@ -16,6 +17,7 @@ import java.util.regex.Pattern;
@Configuration("mediaConfig")
@Order(0)
@Data
public class MediaConfig{
private final static Logger logger = LoggerFactory.getLogger(MediaConfig.class);
@ -78,6 +80,9 @@ public class MediaConfig{
@Value("${media.rtp.send-port-range}")
private String rtpSendPortRange;
@Value("${media.record-assist-ip:127.0.0.1}")
private String recordAssistIp = "127.0.0.1";
@Value("${media.record-assist-port:0}")
private Integer recordAssistPort = 0;
@ -210,6 +215,7 @@ public class MediaConfig{
mediaServerItem.setRtpEnable(rtpEnable);
mediaServerItem.setRtpPortRange(rtpPortRange);
mediaServerItem.setSendRtpPortRange(rtpSendPortRange);
mediaServerItem.setRecordAssistIp(recordAssistIp);
mediaServerItem.setRecordAssistPort(recordAssistPort);
mediaServerItem.setHookAliveInterval(30.00f);

28
src/main/java/com/genersoft/iot/vmp/gb28181/conf/ServerLoggerImpl.java
View File

@ -9,12 +9,11 @@ import gov.nist.javax.sip.stack.SIPTransactionStack;
import javax.sip.SipStack;
import java.util.Properties;
@SuppressWarnings("unused")
public class ServerLoggerImpl implements ServerLogger {
private boolean showLog = true;
private SIPTransactionStack sipStack;
protected StackLogger stackLogger;
@Override
@ -27,12 +26,8 @@ public class ServerLoggerImpl implements ServerLogger {
if (!showLog) {
return;
}
StringBuilder stringBuilder = new StringBuilder();
stringBuilder.append(sender? "发送：目标--->" + from:"接收：来自--->" + to)
.append("\r\n")
.append(message);
this.stackLogger.logInfo(stringBuilder.toString());
String log = (sender ? "发送: 目标 => " + to : "接收: 来自 => " + from) + "\r\n" + message;
this.stackLogger.logInfo(log);
}
@Override
@ -40,11 +35,8 @@ public class ServerLoggerImpl implements ServerLogger {
if (!showLog) {
return;
}
StringBuilder stringBuilder = new StringBuilder();
stringBuilder.append(sender? "发送： 目标->" + from :"接收：来自->" + to)
.append("\r\n")
.append(message);
this.stackLogger.logInfo(stringBuilder.toString());
String log = (sender ? "发送: 目标 => " + to : "接收: 来自 => " + from) + "\r\n" + message;
this.stackLogger.logInfo(log);
}
@Override
@ -52,11 +44,8 @@ public class ServerLoggerImpl implements ServerLogger {
if (!showLog) {
return;
}
StringBuilder stringBuilder = new StringBuilder();
stringBuilder.append(sender? "发送： 目标->" + from :"接收：来自->" + to)
.append("\r\n")
.append(message);
this.stackLogger.logInfo(stringBuilder.toString());
String log = (sender ? "发送: 目标 => " + to : "接收: 来自 => " + from) + "\r\n" + message;
this.stackLogger.logInfo(log);
}
@Override
@ -84,8 +73,7 @@ public class ServerLoggerImpl implements ServerLogger {
return;
}
if(sipStack instanceof SIPTransactionStack) {
this.sipStack = (SIPTransactionStack)sipStack;
this.stackLogger = CommonLogger.getLogger(SIPTransactionStack.class);
this.stackLogger = CommonLogger.getLogger(sipStack.getClass());
}
}
}

6
src/main/java/com/genersoft/iot/vmp/gb28181/event/record/RecordEndEventListener.java
View File

@ -35,15 +35,17 @@ public class RecordEndEventListener implements ApplicationListener<RecordEndEven
String channelId = event.getRecordInfo().getChannelId();
int count = event.getRecordInfo().getCount();
int sumNum = event.getRecordInfo().getSumNum();
logger.info("录像查询完成事件触发，deviceId：{}, channelId: {}, 录像数量{}/{}条", event.getRecordInfo().getDeviceId(),
logger.info("录像查询完成事件触发，deviceId：{}, channelId: {}, 录像数量{}/{}条",
event.getRecordInfo().getDeviceId(),
event.getRecordInfo().getChannelId(), count,sumNum);
logger.debug("handlerMap.size => {}", handlerMap.size());
if (handlerMap.size() > 0) {
logger.debug("handlerMap.keys => {}", handlerMap.keySet());
RecordEndEventHandler handler = handlerMap.get(deviceId + channelId);
logger.debug("handler => {}", handler);
if (handler != null){
handler.handler(event.getRecordInfo());
if (count == sumNum){
if (count >= sumNum){
handlerMap.remove(deviceId + channelId);
}
}

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

@ -510,7 +510,8 @@ public class SIPCommander implements ISIPCommander {
content.append("c=IN IP4 " + sdpIp + "\r\n");
content.append("t=" + DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(startTime) + " "
+ DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(endTime) + "\r\n");
logger.debug("起止时间: {} ~ {}",startTime, endTime);
logger.debug("t={} {}",DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(startTime), DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(endTime));
String streamMode = device.getStreamMode().toUpperCase();
if (userSetting.isSeniorSdp()) {

56
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java
View File

@ -19,8 +19,7 @@ import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.utils.GitUtil;
import gov.nist.javax.sip.message.MessageFactoryImpl;
import gov.nist.javax.sip.message.SIPRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.DependsOn;
import org.springframework.lang.Nullable;
@ -37,12 +36,10 @@ import java.text.ParseException;
import java.util.ArrayList;
import java.util.List;
@Slf4j
@Component
@DependsOn("sipLayer")
public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
private final Logger logger = LoggerFactory.getLogger(SIPCommanderFroPlatform.class);
@Autowired
private SIPRequestHeaderPlarformProvider headerProviderPlatformProvider;
@ -113,13 +110,13 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
// 将 callid 写入缓存， 等注册成功可以更新状态
String callIdFromHeader = callIdHeader.getCallId();
PlatformRegisterInfo instance = PlatformRegisterInfo.getInstance(parentPlatform.getServerGBId(), isRegister);
logger.info("callIdFromHeader {}",callIdFromHeader);
logger.info("PlatformRegisterInfo {}",instance);
log.info("callIdFromHeader {}",callIdFromHeader);
log.info("PlatformRegisterInfo {}",instance);
redisCatchStorage.updatePlatformRegisterInfo(callIdFromHeader, instance);
sipSubscribe.addErrorSubscribe(callIdHeader.getCallId(), (event)->{
if (event != null) {
logger.info("向上级平台 [ {} ] 注册发生错误： {} ",
log.info("向上级平台 [ {} ] 注册发生错误： {} ",
parentPlatform.getServerGBId(),
event.msg);
}
@ -218,7 +215,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
}else {
if (channel.getChannelId().length() != 20) {
catalogXml.append("</Item>\r\n");
logger.warn("[编号长度异常] {} 长度错误，请使用20位长度的国标编号,当前长度：{}", channel.getChannelId(), channel.getChannelId().length());
log.warn("[编号长度异常] {} 长度错误，请使用20位长度的国标编号,当前长度：{}", channel.getChannelId(), channel.getChannelId().length());
catalogXml.append("</Item>\r\n");
continue;
}
@ -386,9 +383,9 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
String callId = request.getCallIdHeader().getCallId();
logger.info("[命令发送] 国标级联{} 目录查询回复: 共{}条，已发送{}条", parentPlatform.getServerGBId(),
log.info("[命令发送] 国标级联{} 目录查询回复: 共{}条，已发送{}条", parentPlatform.getServerGBId(),
channels.size(), Math.min(index + parentPlatform.getCatalogGroup(), channels.size()));
logger.debug(catalogXml);
log.debug(catalogXml);
if (sendAfterResponse) {
// 默认按照收到200回复后发送下一条， 如果超时收不到回复，就以30毫秒的间隔直接发送。
dynamicTask.startDelay(timeoutTaskKey, ()->{
@ -397,11 +394,11 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
try {
sendCatalogResponse(channels, parentPlatform, sn, fromTag, indexNext, false);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 目录查询回复: {}", e.getMessage());
log.error("[命令发送失败] 国标级联 目录查询回复: {}", e.getMessage());
}
}, 3000);
sipSender.transmitRequest(parentPlatform.getDeviceIp(), request, eventResult -> {
logger.error("[目录推送失败] 国标级联 platform : {}, code: {}, msg: {}, 停止发送", parentPlatform.getServerGBId(), eventResult.statusCode, eventResult.msg);
log.error("[目录推送失败] 国标级联 platform : {}, code: {}, msg: {}, 停止发送", parentPlatform.getServerGBId(), eventResult.statusCode, eventResult.msg);
dynamicTask.stop(timeoutTaskKey);
}, eventResult -> {
dynamicTask.stop(timeoutTaskKey);
@ -409,12 +406,12 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
try {
sendCatalogResponse(channels, parentPlatform, sn, fromTag, indexNext, true);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 目录查询回复: {}", e.getMessage());
log.error("[命令发送失败] 国标级联 目录查询回复: {}", e.getMessage());
}
});
}else {
sipSender.transmitRequest(parentPlatform.getDeviceIp(), request, eventResult -> {
logger.error("[目录推送失败] 国标级联 platform : {}, code: {}, msg: {}, 停止发送", parentPlatform.getServerGBId(), eventResult.statusCode, eventResult.msg);
log.error("[目录推送失败] 国标级联 platform : {}, code: {}, msg: {}, 停止发送", parentPlatform.getServerGBId(), eventResult.statusCode, eventResult.msg);
dynamicTask.stop(timeoutTaskKey);
}, null);
dynamicTask.startDelay(timeoutTaskKey, ()->{
@ -422,7 +419,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
try {
sendCatalogResponse(channels, parentPlatform, sn, fromTag, indexNext, false);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 目录查询回复: {}", e.getMessage());
log.error("[命令发送失败] 国标级联 目录查询回复: {}", e.getMessage());
}
}, 30);
}
@ -501,8 +498,8 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
if (parentPlatform == null) {
return;
}
if (logger.isDebugEnabled()) {
logger.debug("[发送 移动位置订阅] {}/{}->{},{}", parentPlatform.getServerGBId(), gpsMsgInfo.getId(), gpsMsgInfo.getLng(), gpsMsgInfo.getLat());
if (log.isDebugEnabled()) {
log.debug("[发送 移动位置订阅] {}/{}->{},{}", parentPlatform.getServerGBId(), gpsMsgInfo.getId(), gpsMsgInfo.getLng(), gpsMsgInfo.getLat());
}
String characterSet = parentPlatform.getCharacterSet();
@ -521,7 +518,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
.append("</Notify>\r\n");
sendNotify(parentPlatform, deviceStatusXml.toString(), subscribeInfo, eventResult -> {
logger.error("发送NOTIFY通知消息失败。错误：{} {}", eventResult.statusCode, eventResult.msg);
log.error("发送NOTIFY通知消息失败。错误：{} {}", eventResult.statusCode, eventResult.msg);
}, null);
}
@ -531,7 +528,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
if (parentPlatform == null) {
return;
}
logger.info("[发送报警通知]平台： {}/{}->{},{}: {}", parentPlatform.getServerGBId(), deviceAlarm.getChannelId(),
log.info("[发送报警通知]平台： {}/{}->{},{}: {}", parentPlatform.getServerGBId(), deviceAlarm.getChannelId(),
deviceAlarm.getLongitude(), deviceAlarm.getLatitude(), JSON.toJSONString(deviceAlarm));
String characterSet = parentPlatform.getCharacterSet();
StringBuffer deviceStatusXml = new StringBuffer(600);
@ -579,14 +576,14 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
String catalogXmlContent = getCatalogXmlContentForCatalogAddOrUpdate(parentPlatform, channels,
deviceChannels.size(), type, subscribeInfo);
sendNotify(parentPlatform, catalogXmlContent, subscribeInfo, eventResult -> {
logger.error("发送NOTIFY通知消息失败。错误：{} {}", eventResult.statusCode, eventResult.msg);
log.error("发送NOTIFY通知消息失败。错误：{} {}", eventResult.statusCode, eventResult.msg);
}, (eventResult -> {
try {
sendNotifyForCatalogAddOrUpdate(type, parentPlatform, deviceChannels, subscribeInfo,
finalIndex + parentPlatform.getCatalogGroup());
} catch (InvalidArgumentException | ParseException | NoSuchFieldException | SipException |
IllegalAccessException e) {
logger.error("[命令发送失败] 国标级联 NOTIFY通知: {}", e.getMessage());
log.error("[命令发送失败] 国标级联 NOTIFY通知: {}", e.getMessage());
}
}));
}
@ -666,7 +663,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
|| deviceChannels == null
|| deviceChannels.size() == 0
|| subscribeInfo == null) {
logger.warn("[缺少必要参数]");
log.warn("[缺少必要参数]");
return;
}
@ -685,14 +682,14 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
Integer finalIndex = index;
String catalogXmlContent = getCatalogXmlContentForCatalogOther(parentPlatform, channels, type);
sendNotify(parentPlatform, catalogXmlContent, subscribeInfo, eventResult -> {
logger.error("发送NOTIFY通知消息失败。错误：{} {}", eventResult.statusCode, eventResult.msg);
log.error("发送NOTIFY通知消息失败。错误：{} {}", eventResult.statusCode, eventResult.msg);
}, eventResult -> {
try {
sendNotifyForCatalogOther(type, parentPlatform, deviceChannels, subscribeInfo,
finalIndex + parentPlatform.getCatalogGroup());
} catch (InvalidArgumentException | ParseException | NoSuchFieldException | SipException |
IllegalAccessException e) {
logger.error("[命令发送失败] 国标级联 NOTIFY通知: {}", e.getMessage());
log.error("[命令发送失败] 国标级联 NOTIFY通知: {}", e.getMessage());
}
});
}
@ -725,6 +722,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
}
@Override
public void recordInfo(DeviceChannel deviceChannel, ParentPlatform parentPlatform, String fromTag, RecordInfo recordInfo) throws SipException, InvalidArgumentException, ParseException {
log.debug("deviceChannel => {}, parentPlatform => {}, fromTag => {}, recordInfo => {}", deviceChannel, parentPlatform, fromTag, recordInfo);
if ( parentPlatform ==null) {
return ;
}
@ -811,14 +809,14 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
@Override
public void streamByeCmd(ParentPlatform parentPlatform, SendRtpItem sendRtpItem) throws SipException, InvalidArgumentException, ParseException {
if (sendRtpItem == null ) {
logger.info("[向上级发送BYE]， sendRtpItem 为NULL");
log.info("[向上级发送BYE]， sendRtpItem 为NULL");
return;
}
if (parentPlatform == null) {
logger.info("[向上级发送BYE]， platform 为NULL");
log.info("[向上级发送BYE]， platform 为NULL");
return;
}
logger.info("[向上级发送BYE]， {}/{}", parentPlatform.getServerGBId(), sendRtpItem.getChannelId());
log.info("[向上级发送BYE]， {}/{}", parentPlatform.getServerGBId(), sendRtpItem.getChannelId());
String mediaServerId = sendRtpItem.getMediaServerId();
MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
if (mediaServerItem != null) {
@ -827,7 +825,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
}
SIPRequest byeRequest = headerProviderPlatformProvider.createByeRequest(parentPlatform, sendRtpItem);
if (byeRequest == null) {
logger.warn("[向上级发送bye]：无法创建 byeRequest");
log.warn("[向上级发送bye]：无法创建 byeRequest");
}
sipSender.transmitRequest(parentPlatform.getDeviceIp(),byeRequest);
}

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

@ -459,7 +459,10 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
sendRtpItem.setApp("rtp");
if ("Playback".equalsIgnoreCase(sessionName)) {
sendRtpItem.setPlayType(InviteStreamType.PLAYBACK);
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, null, device.isSsrcCheck(), true, 0, false, device.getStreamModeForParam());
String startTimeStr = DateUtil.urlFormatter.format(start);
String endTimeStr = DateUtil.urlFormatter.format(end);
String stream = device.getDeviceId() + "_" + channelId + "_" + startTimeStr + "_" + endTimeStr;
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, stream, null, device.isSsrcCheck(), true, 0, false, device.getStreamModeForParam());
sendRtpItem.setStreamId(ssrcInfo.getStream());
// 写入redis， 超时时回复
redisCatchStorage.updateSendRTPSever(sendRtpItem);
@ -520,12 +523,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
}
}));
sendRtpItem.setPlayType(InviteStreamType.PLAY);
String streamId = null;
if (mediaServerItem.isRtpEnable()) {
streamId = String.format("%s_%s", device.getDeviceId(), channelId);
}else {
streamId = String.format("%08x", Integer.parseInt(ssrcInfo.getSsrc())).toUpperCase();
}
String streamId = String.format("%s_%s", device.getDeviceId(), channelId);
sendRtpItem.setStreamId(streamId);
sendRtpItem.setSsrc(ssrcInfo.getSsrc());
redisCatchStorage.updateSendRTPSever(sendRtpItem);

14
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/MessageRequestProcessor.java
View File

@ -12,6 +12,7 @@ import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorP
import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.storager.dao.DeviceChannelMapper;
import gov.nist.javax.sip.message.SIPRequest;
import org.dom4j.DocumentException;
import org.dom4j.Element;
@ -27,6 +28,7 @@ import javax.sip.SipException;
import javax.sip.header.CallIdHeader;
import javax.sip.message.Response;
import java.text.ParseException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@ -54,6 +56,9 @@ public class MessageRequestProcessor extends SIPRequestProcessorParent implement
@Autowired
private VideoStreamSessionManager sessionManager;
@Autowired
private DeviceChannelMapper deviceChannelMapper;
@Override
public void afterPropertiesSet() throws Exception {
// 添加消息处理的订阅
@ -79,6 +84,15 @@ public class MessageRequestProcessor extends SIPRequestProcessorParent implement
SIPRequest request = (SIPRequest) evt.getRequest();
// 查询设备是否存在
Device device = redisCatchStorage.getDevice(deviceId);
if(device == null){
String sendBy = sipRequest.getTopmostVia().getSentBy().toString();
List<Device> devicesByChannelId = deviceChannelMapper.getDeviceByChannelId(deviceId);
device = devicesByChannelId.parallelStream().filter((deviceByChannelId)->{
return deviceByChannelId.getHostAddress().equals(sendBy);
}).findFirst().orElse(null);
}
// 查询上级平台是否存在
ParentPlatform parentPlatform = storage.queryParentPlatByServerGBId(deviceId);
try {

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

@ -68,7 +68,8 @@ public class KeepaliveNotifyMessageHandler extends SIPRequestProcessorParent imp
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 心跳回复: {}", e.getMessage());
}
if (DateUtil.getDifferenceForNow(device.getKeepaliveTime()) <= 3000L){
if (device.getKeepaliveTime() != null && DateUtil.getDifferenceForNow(device.getKeepaliveTime()) <= 3000L){
logger.info("[收到心跳] 心跳发送过于频繁，已忽略 device: {}, callId: {}", device.getDeviceId(), request.getCallIdHeader().getCallId());
return;
}

5
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java
View File

@ -8,9 +8,12 @@ import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.IMessageHandler;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.ResponseMessageHandler;
import com.genersoft.iot.vmp.gb28181.utils.XmlUtil;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.utils.UJson;
import gov.nist.javax.sip.message.SIPRequest;
import org.dom4j.Document;
import org.dom4j.DocumentHelper;
import org.dom4j.Element;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -78,7 +81,7 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent
}
taskExecutor.execute(()->{
try {
logger.debug("rootElement => {}",XmlUtil.getElementMap(rootElement));
String sn = getText(rootElement, "SN");
String channelId = getText(rootElement, "DeviceID");
RecordInfo recordInfo = new RecordInfo();

2
src/main/java/com/genersoft/iot/vmp/gb28181/utils/SipUtils.java
View File

@ -263,4 +263,4 @@ public class SipUtils {
}
return localDateTime.format(DateUtil.formatter);
}
}
}

19
src/main/java/com/genersoft/iot/vmp/gb28181/utils/XmlUtil.java
View File

@ -644,4 +644,23 @@ public class XmlUtil {
}
return val;
}
public static Map<String, Object> getElementMap(Element element) {
List<Element> elements = element.elements();
//没有子元素
Map<String, Object> map = new HashMap<>(16);
if (elements.size() == 0) {
String name = element.getName();
String value = element.getText();
map.put(name, value);
} else {
//有子元素
for (Element el : elements) {
String name = el.getName();
map.put(name,getElementMap(el));
}
}
return map;
}
}

94
src/main/java/com/genersoft/iot/vmp/media/zlm/AssistRESTfulUtils.java
View File

@ -21,6 +21,7 @@ public class AssistRESTfulUtils {
private final static Logger logger = LoggerFactory.getLogger(AssistRESTfulUtils.class);
public interface RequestCallback{
void run(JSONObject response);
}
@ -72,51 +73,51 @@ public class AssistRESTfulUtils {
.get()
.url(url)
.build();
if (callback == null) {
try {
Response response = client.newCall(request).execute();
if (callback == null) {
try {
Response response = client.newCall(request).execute();
if (response.isSuccessful()) {
ResponseBody responseBody = response.body();
if (responseBody != null) {
String responseStr = responseBody.string();
responseJSON = JSON.parseObject(responseStr);
}
}else {
response.close();
Objects.requireNonNull(response.body()).close();
}
} catch (ConnectException e) {
logger.error(String.format("连接Assist失败: %s, %s", e.getCause().getMessage(), e.getMessage()));
logger.info("请检查media配置并确认Assist已启动...");
}catch (IOException e) {
logger.error(String.format("[ %s ]请求失败: %s", url, e.getMessage()));
}
}else {
client.newCall(request).enqueue(new Callback(){
@Override
public void onResponse(@NotNull Call call, @NotNull Response response){
if (response.isSuccessful()) {
ResponseBody responseBody = response.body();
if (responseBody != null) {
String responseStr = responseBody.string();
responseJSON = JSON.parseObject(responseStr);
try {
String responseStr = Objects.requireNonNull(response.body()).string();
callback.run(JSON.parseObject(responseStr));
} catch (IOException e) {
logger.error(String.format("[ %s ]请求失败: %s", url, e.getMessage()));
}
}else {
response.close();
Objects.requireNonNull(response.body()).close();
}
} catch (ConnectException e) {
}
@Override
public void onFailure(@NotNull Call call, @NotNull IOException e) {
logger.error(String.format("连接Assist失败: %s, %s", e.getCause().getMessage(), e.getMessage()));
logger.info("请检查media配置并确认Assist已启动...");
}catch (IOException e) {
logger.error(String.format("[ %s ]请求失败: %s", url, e.getMessage()));
}
}else {
client.newCall(request).enqueue(new Callback(){
@Override
public void onResponse(@NotNull Call call, @NotNull Response response){
if (response.isSuccessful()) {
try {
String responseStr = Objects.requireNonNull(response.body()).string();
callback.run(JSON.parseObject(responseStr));
} catch (IOException e) {
logger.error(String.format("[ %s ]请求失败: %s", url, e.getMessage()));
}
}else {
response.close();
Objects.requireNonNull(response.body()).close();
}
}
@Override
public void onFailure(@NotNull Call call, @NotNull IOException e) {
logger.error(String.format("连接Assist失败: %s, %s", e.getCause().getMessage(), e.getMessage()));
logger.info("请检查media配置并确认Assist已启动...");
}
});
}
});
}
@ -145,4 +146,25 @@ public class AssistRESTfulUtils {
return sendGet(mediaServerItem, "api/record/addStreamCallInfo",param, callback);
}
public JSONObject getDateList(MediaServerItem mediaServerItem, String app, String stream, int year, int month) {
Map<String, Object> param = new HashMap<>();
param.put("app", app);
param.put("stream", stream);
param.put("year", year);
param.put("month", month);
return sendGet(mediaServerItem, "api/record/date/list", param, null);
}
public JSONObject getFileList(MediaServerItem mediaServerItem, int page, int count, String app, String stream,
String startTime, String endTime) {
Map<String, Object> param = new HashMap<>();
param.put("app", app);
param.put("stream", stream);
param.put("page", page);
param.put("count", count);
param.put("startTime", startTime);
param.put("endTime", endTime);
return sendGet(mediaServerItem, "api/record/file/listWithDate", param, null);
}
}

42
src/main/java/com/genersoft/iot/vmp/media/zlm/SendRtpPortManager.java
View File

@ -30,7 +30,7 @@ public class SendRtpPortManager {
private final String KEY = "VM_MEDIA_SEND_RTP_PORT_";
public int getNextPort(MediaServerItem mediaServer) {
public synchronized int getNextPort(MediaServerItem mediaServer) {
if (mediaServer == null) {
logger.warn("[发送端口管理] 参数错误，mediaServer为NULL");
return -1;
@ -50,17 +50,15 @@ public class SendRtpPortManager {
String sendRtpPortRange = mediaServer.getSendRtpPortRange();
int startPort;
int endPort;
if (sendRtpPortRange == null) {
logger.warn("{}未设置发送端口默认值，自动使用40000-50000作为端口范围", mediaServer.getId());
if (sendRtpPortRange != null) {
String[] portArray = sendRtpPortRange.split(",");
if (portArray.length != 2 || !NumberUtils.isParsable(portArray[0]) || !NumberUtils.isParsable(portArray[1])) {
logger.warn("{}发送端口配置格式错误，自动使用40000-50000作为端口范围", mediaServer.getId());
logger.warn("{}发送端口配置格式错误，自动使用50000-60000作为端口范围", mediaServer.getId());
startPort = 50000;
endPort = 60000;
}else {
if ( Integer.parseInt(portArray[1]) - Integer.parseInt(portArray[0]) < 1) {
logger.warn("{}发送端口配置错误,结束端口至少比开始端口大一，自动使用40000-50000作为端口范围", mediaServer.getId());
logger.warn("{}发送端口配置错误,结束端口至少比开始端口大一，自动使用50000-60000作为端口范围", mediaServer.getId());
startPort = 50000;
endPort = 60000;
}else {
@ -69,6 +67,7 @@ public class SendRtpPortManager {
}
}
}else {
logger.warn("{}未设置发送端口默认值，自动使用50000-60000作为端口范围", mediaServer.getId());
startPort = 50000;
endPort = 60000;
}
@ -76,10 +75,35 @@ public class SendRtpPortManager {
logger.warn("{}获取redis连接信息失败", mediaServer.getId());
return -1;
}
// RedisAtomicInteger redisAtomicInteger = new RedisAtomicInteger(sendIndexKey , redisTemplate.getConnectionFactory());
// return redisAtomicInteger.getAndUpdate((current)->{
// return getPort(current, startPort, endPort, checkPort-> !sendRtpItemMap.containsKey(checkPort));
// });
return getSendPort(startPort, endPort, sendIndexKey, sendRtpItemMap);
}
private synchronized int getSendPort(int startPort, int endPort, String sendIndexKey, Map<Integer, SendRtpItem> sendRtpItemMap){
RedisAtomicInteger redisAtomicInteger = new RedisAtomicInteger(sendIndexKey , redisTemplate.getConnectionFactory());
return redisAtomicInteger.getAndUpdate((current)->{
return getPort(current, startPort, endPort, checkPort-> !sendRtpItemMap.containsKey(checkPort));
});
if (redisAtomicInteger.get() < startPort) {
redisAtomicInteger.set(startPort);
return startPort;
}else {
int port = redisAtomicInteger.getAndIncrement();
if (port > endPort) {
redisAtomicInteger.set(startPort);
if (sendRtpItemMap.containsKey(startPort)) {
return getSendPort(startPort, endPort, sendIndexKey, sendRtpItemMap);
}else {
return startPort;
}
}
if (sendRtpItemMap.containsKey(port)) {
return getSendPort(startPort, endPort, sendIndexKey, sendRtpItemMap);
}else {
return port;
}
}
}
interface CheckPortCallback{

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

@ -29,6 +29,7 @@ import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.OtherPsSendInfo;
import com.genersoft.iot.vmp.vmanager.bean.OtherRtpSendInfo;
import com.genersoft.iot.vmp.vmanager.bean.StreamContent;
import org.apache.commons.lang3.math.NumberUtils;
@ -194,7 +195,10 @@ public class ZLMHttpHookListener {
String mediaServerId = json.getString("mediaServerId");
MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);
if (mediaInfo == null) {
return new HookResultForOnPublish(200, "success");
}
// 推流鉴权的处理
if (!"rtp".equals(param.getApp())) {
if (userSetting.getPushAuthority()) {
// 推流鉴权
@ -246,11 +250,21 @@ public class ZLMHttpHookListener {
}
});
// 是否录像
if ("rtp".equals(param.getApp())) {
result.setEnable_mp4(userSetting.getRecordSip());
} else {
result.setEnable_mp4(userSetting.isRecordPushLive());
}
// 替换流地址
if ("rtp".equals(param.getApp()) && !mediaInfo.isRtpEnable()) {
String ssrc = String.format("%010d", Long.parseLong(param.getStream(), 16));;
InviteInfo inviteInfo = inviteStreamService.getInviteInfoBySSRC(ssrc);
if (inviteInfo != null) {
result.setStream_replace(inviteInfo.getStream());
logger.info("[ZLM HOOK]推流鉴权 stream: {} 替换为 {}", param.getStream(), inviteInfo.getStream());
}
}
List<SsrcTransaction> ssrcTransactionForAll = sessionManager.getSsrcTransactionForAll(null, null, null, param.getStream());
if (ssrcTransactionForAll != null && ssrcTransactionForAll.size() == 1) {
String deviceId = ssrcTransactionForAll.get(0).getDeviceId();
@ -287,7 +301,10 @@ public class ZLMHttpHookListener {
if (param.getApp().equalsIgnoreCase("rtp")) {
String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "_" + param.getStream();
OtherRtpSendInfo otherRtpSendInfo = (OtherRtpSendInfo)redisTemplate.opsForValue().get(receiveKey);
if (otherRtpSendInfo != null) {
String receiveKeyForPS = VideoManagerConstants.WVP_OTHER_RECEIVE_PS_INFO + userSetting.getServerId() + "_" + param.getStream();
OtherPsSendInfo otherPsSendInfo = (OtherPsSendInfo)redisTemplate.opsForValue().get(receiveKeyForPS);
if (otherRtpSendInfo != null || otherPsSendInfo != null) {
result.setEnable_mp4(true);
}
}
@ -564,7 +581,7 @@ public class ZLMHttpHookListener {
if ("rtp".equals(param.getApp())) {
String[] s = param.getStream().split("_");
if (!mediaInfo.isRtpEnable() || (s.length != 2 && s.length != 4)) {
if ((s.length != 2 && s.length != 4)) {
defaultResult.setResult(HookResult.SUCCESS());
return defaultResult;
}
@ -593,7 +610,6 @@ public class ZLMHttpHookListener {
result.onTimeout(() -> {
logger.info("[ZLM HOOK] 预览流自动点播, 等待超时");
// 释放rtpserver
msg.setData(new HookResult(ErrorCode.ERROR100.getCode(), "点播超时"));
resultHolder.invokeResult(msg);
});

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

@ -42,7 +42,7 @@ public class ZLMServerFactory {
* @param tcpMode 0/null udp 模式，1 tcp 被动模式, 2 tcp 主动模式。
* @return
*/
public int createRTPServer(MediaServerItem mediaServerItem, String streamId, int ssrc, Integer port, Boolean reUsePort, Integer tcpMode) {
public int createRTPServer(MediaServerItem mediaServerItem, String streamId, long ssrc, Integer port, Boolean reUsePort, Integer tcpMode) {
int result = -1;
// 查询此rtp server 是否已经存在
JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaServerItem, streamId);

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

@ -3,8 +3,10 @@ package com.genersoft.iot.vmp.media.zlm.dto;
import com.genersoft.iot.vmp.media.zlm.ZLMServerConfig;
import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Data;
import org.springframework.util.ObjectUtils;
@Data
@Schema(description = "流媒体服务信息")
public class MediaServerItem{
@ -65,6 +67,9 @@ public class MediaServerItem{
@Schema(description = "RTP发流端口范围")
private String sendRtpPortRange;
@Schema(description = "assist服务ip")
private String recordAssistIp;
@Schema(description = "assist服务端口")
private int recordAssistPort;

10
src/main/java/com/genersoft/iot/vmp/media/zlm/dto/hook/HookResultForOnPublish.java
View File

@ -6,6 +6,7 @@ public class HookResultForOnPublish extends HookResult{
private boolean enable_mp4;
private int mp4_max_second;
private String mp4_save_path;
private String stream_replace;
public HookResultForOnPublish() {
}
@ -51,12 +52,21 @@ public class HookResultForOnPublish extends HookResult{
this.mp4_save_path = mp4_save_path;
}
public String getStream_replace() {
return stream_replace;
}
public void setStream_replace(String stream_replace) {
this.stream_replace = stream_replace;
}
@Override
public String toString() {
return "HookResultForOnPublish{" +
"enable_audio=" + enable_audio +
", enable_mp4=" + enable_mp4 +
", mp4_max_second=" + mp4_max_second +
", stream_replace=" + stream_replace +
", mp4_save_path='" + mp4_save_path + '\'' +
'}';
}

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

@ -74,5 +74,13 @@ public interface IInviteStreamService {
int getStreamInfoCount(String mediaServerId);
/**
* 获取MediaServer下的流信息
*/
InviteInfo getInviteInfoBySSRC(String ssrc);
/**
* 更新ssrc
*/
InviteInfo updateInviteInfoForSSRC(InviteInfo inviteInfo, String ssrcInResponse);
}

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

@ -6,6 +6,7 @@ import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.ServerKeepaliveData;
import com.genersoft.iot.vmp.service.bean.MediaServerLoad;
import com.genersoft.iot.vmp.service.bean.SSRCInfo;
import com.genersoft.iot.vmp.vmanager.bean.RecordFile;
import java.util.List;
@ -93,4 +94,14 @@ public interface IMediaServerService {
* @return
*/
MediaServerLoad getLoad(MediaServerItem mediaServerItem);
/**
* 按时间查找录像文件
*/
List<RecordFile> getRecords(String app, String stream, String startTime, String endTime, List<MediaServerItem> mediaServerItems);
/**
* 查找存在录像文件的时间
*/
List<String> getRecordDates(String app, String stream, int year, int month, List<MediaServerItem> mediaServerItems);
}

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

@ -77,10 +77,11 @@ public class InviteStreamServiceImpl implements IInviteStreamService {
}
String key = VideoManagerConstants.INVITE_PREFIX +
"_" + inviteInfoForUpdate.getType() +
"_" + inviteInfoForUpdate.getDeviceId() +
"_" + inviteInfoForUpdate.getChannelId() +
"_" + inviteInfoForUpdate.getStream();
":" + inviteInfoForUpdate.getType() +
":" + inviteInfoForUpdate.getDeviceId() +
":" + inviteInfoForUpdate.getChannelId() +
":" + inviteInfoForUpdate.getStream()+
":" + inviteInfoForUpdate.getSsrcInfo().getSsrc();
redisTemplate.opsForValue().set(key, inviteInfoForUpdate);
}
@ -93,10 +94,11 @@ public class InviteStreamServiceImpl implements IInviteStreamService {
}
removeInviteInfo(inviteInfoInDb);
String key = VideoManagerConstants.INVITE_PREFIX +
"_" + inviteInfo.getType() +
"_" + inviteInfo.getDeviceId() +
"_" + inviteInfo.getChannelId() +
"_" + stream;
":" + inviteInfo.getType() +
":" + inviteInfo.getDeviceId() +
":" + inviteInfo.getChannelId() +
":" + stream +
":" + inviteInfo.getSsrcInfo().getSsrc();
inviteInfoInDb.setStream(stream);
if (inviteInfoInDb.getSsrcInfo() != null) {
inviteInfoInDb.getSsrcInfo().setStream(stream);
@ -108,10 +110,11 @@ public class InviteStreamServiceImpl implements IInviteStreamService {
@Override
public InviteInfo getInviteInfo(InviteSessionType type, String deviceId, String channelId, String stream) {
String key = VideoManagerConstants.INVITE_PREFIX +
"_" + (type != null ? type : "*") +
"_" + (deviceId != null ? deviceId : "*") +
"_" + (channelId != null ? channelId : "*") +
"_" + (stream != null ? stream : "*");
":" + (type != null ? type : "*") +
":" + (deviceId != null ? deviceId : "*") +
":" + (channelId != null ? channelId : "*") +
":" + (stream != null ? stream : "*")
+ ":*";
List<Object> scanResult = RedisUtil.scan(redisTemplate, key);
if (scanResult.size() != 1) {
return null;
@ -133,10 +136,11 @@ public class InviteStreamServiceImpl implements IInviteStreamService {
@Override
public void removeInviteInfo(InviteSessionType type, String deviceId, String channelId, String stream) {
String scanKey = VideoManagerConstants.INVITE_PREFIX +
"_" + (type != null ? type : "*") +
"_" + (deviceId != null ? deviceId : "*") +
"_" + (channelId != null ? channelId : "*") +
"_" + (stream != null ? stream : "*");
":" + (type != null ? type : "*") +
":" + (deviceId != null ? deviceId : "*") +
":" + (channelId != null ? channelId : "*") +
":" + (stream != null ? stream : "*") +
":*";
List<Object> scanResult = RedisUtil.scan(redisTemplate, scanKey);
if (scanResult.size() > 0) {
for (Object keyObj : scanResult) {
@ -174,10 +178,10 @@ public class InviteStreamServiceImpl implements IInviteStreamService {
}
private String buildKey(InviteSessionType type, String deviceId, String channelId, String stream) {
String key = type + "_" + deviceId + "_" + channelId;
String key = type + ":" + deviceId + ":" + channelId;
// 如果ssrc未null那么可以实现一个通道只能一次操作，ssrc不为null则可以支持一个通道多次invite
if (stream != null) {
key += ("_" + stream);
key += (":" + stream);
}
return key;
}
@ -191,7 +195,7 @@ public class InviteStreamServiceImpl implements IInviteStreamService {
@Override
public int getStreamInfoCount(String mediaServerId) {
int count = 0;
String key = VideoManagerConstants.INVITE_PREFIX + "_*_*_*_*";
String key = VideoManagerConstants.INVITE_PREFIX + ":*:*:*:*:*";
List<Object> scanResult = RedisUtil.scan(redisTemplate, key);
if (scanResult.size() == 0) {
return 0;
@ -222,11 +226,42 @@ public class InviteStreamServiceImpl implements IInviteStreamService {
private String buildSubStreamKey(InviteSessionType type, String deviceId, String channelId, String stream) {
String key = type + "_" + "_" + deviceId + "_" + channelId;
String key = type + ":" + ":" + deviceId + ":" + channelId;
// 如果ssrc为null那么可以实现一个通道只能一次操作，ssrc不为null则可以支持一个通道多次invite
if (stream != null) {
key += ("_" + stream);
key += (":" + stream);
}
return key;
}
@Override
public InviteInfo getInviteInfoBySSRC(String ssrc) {
String key = VideoManagerConstants.INVITE_PREFIX + ":*:*:*:*:" + ssrc;
List<Object> scanResult = RedisUtil.scan(redisTemplate, key);
if (scanResult.size() != 1) {
return null;
}
return (InviteInfo) redisTemplate.opsForValue().get(scanResult.get(0));
}
@Override
public InviteInfo updateInviteInfoForSSRC(InviteInfo inviteInfo, String ssrc) {
InviteInfo inviteInfoInDb = getInviteInfo(inviteInfo.getType(), inviteInfo.getDeviceId(), inviteInfo.getChannelId(), inviteInfo.getStream());
if (inviteInfoInDb == null) {
return null;
}
removeInviteInfo(inviteInfoInDb);
String key = VideoManagerConstants.INVITE_PREFIX +
":" + inviteInfo.getType() +
":" + inviteInfo.getDeviceId() +
":" + inviteInfo.getChannelId() +
":" + inviteInfo.getStream() +
":" + inviteInfo.getSsrcInfo().getSsrc();
if (inviteInfoInDb.getSsrcInfo() != null) {
inviteInfoInDb.getSsrcInfo().setSsrc(ssrc);
}
redisTemplate.opsForValue().set(key, inviteInfoInDb);
return inviteInfoInDb;
}
}

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

@ -6,6 +6,7 @@ import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.CommonCallback;
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.MediaConfig;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
@ -24,28 +25,37 @@ import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.utils.JsonUtil;
import com.genersoft.iot.vmp.utils.redis.RedisUtil;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.RecordFile;
import lombok.RequiredArgsConstructor;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
import java.io.File;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
/**
* 媒体服务器节点管理
*/
@Service
@RequiredArgsConstructor
public class MediaServerServiceImpl implements IMediaServerService {
private final static Logger logger = LoggerFactory.getLogger(MediaServerServiceImpl.class);
@ -104,6 +114,11 @@ public class MediaServerServiceImpl implements IMediaServerService {
@Autowired
private RedisTemplate<Object, Object> redisTemplate;
@Qualifier("taskExecutor")
@Autowired
private ThreadPoolTaskExecutor taskExecutor;
private final MediaConfig mediaConfig;
/**
* 初始化
@ -149,7 +164,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
}
if (streamId == null) {
streamId = String.format("%08x", Integer.parseInt(ssrc)).toUpperCase();
streamId = String.format("%08x", Long.parseLong(ssrc)).toUpperCase();
}
int ssrcCheckParam = 0;
if (ssrcCheck && tcpMode > 1) {
@ -158,7 +173,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
}
int rtpServerPort;
if (mediaServerItem.isRtpEnable()) {
rtpServerPort = zlmServerFactory.createRTPServer(mediaServerItem, streamId, (ssrcCheck && tcpMode == 0)?Integer.parseInt(ssrc):0, port, reUsePort, tcpMode);
rtpServerPort = zlmServerFactory.createRTPServer(mediaServerItem, streamId, (ssrcCheck && tcpMode == 0) ? Long.parseLong(ssrc) : 0, port, reUsePort, tcpMode);
} else {
rtpServerPort = mediaServerItem.getRtpProxyPort();
}
@ -576,7 +591,12 @@ public class MediaServerServiceImpl implements IMediaServerService {
param.put("hook.on_send_rtp_stopped",String.format("%s/on_send_rtp_stopped", hookPrex));
param.put("hook.on_rtp_server_timeout",String.format("%s/on_rtp_server_timeout", hookPrex));
if (mediaServerItem.getRecordAssistPort() > 0) {
param.put("hook.on_record_mp4",String.format("http://127.0.0.1:%s/api/record/on_record_mp4", mediaServerItem.getRecordAssistPort()));
if(mediaConfig.getRecordAssistIp().equals("127.0.0.1")){
param.put("hook.on_record_mp4",String.format("http://127.0.0.1:%s/api/record/on_record_mp4", mediaServerItem.getRecordAssistPort()));
} else {
param.put("hook.on_record_mp4",String.format("http://%s:%s/api/record/on_record_mp4", mediaConfig.getRecordAssistIp(),mediaServerItem.getRecordAssistPort()));
}
}else {
param.put("hook.on_record_mp4","");
}
@ -588,7 +608,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
param.put("protocol.continue_push_ms", "3000" );
// 最多等待未初始化的Track时间，单位毫秒，超时之后会忽略未初始化的Track, 设置此选项优化那些音频错误的不规范流，
// 等zlm支持给每个rtpServer设置关闭音频的时候可以不设置此选项
// param.put("general.wait_track_ready_ms", "3000" );
// param.put("general.wait_track_ready_ms", "3000" );
if (mediaServerItem.isRtpEnable() && !ObjectUtils.isEmpty(mediaServerItem.getRtpPortRange())) {
param.put("rtp_proxy.port_range", mediaServerItem.getRtpPortRange().replace(",", "-"));
}
@ -749,4 +769,89 @@ public class MediaServerServiceImpl implements IMediaServerService {
return result;
}
@Override
public List<RecordFile> getRecords(String app, String stream, String startTime, String endTime, List<MediaServerItem> mediaServerItems) {
Assert.notNull(app, "app不存在");
Assert.notNull(stream, "stream不存在");
Assert.notNull(startTime, "startTime不存在");
Assert.notNull(endTime, "endTime不存在");
Assert.notEmpty(mediaServerItems, "流媒体列表为空");
CompletableFuture[] completableFutures = new CompletableFuture[mediaServerItems.size()];
for (int i = 0; i < mediaServerItems.size(); i++) {
completableFutures[i] = getRecordFilesForOne(app, stream, startTime, endTime, mediaServerItems.get(i));
}
List<RecordFile> result = new ArrayList<>();
for (int i = 0; i < completableFutures.length; i++) {
try {
List<RecordFile> list = (List<RecordFile>) completableFutures[i].get();
if (!list.isEmpty()) {
for (int g = 0; g < list.size(); g++) {
list.get(g).setMediaServerId(mediaServerItems.get(i).getId());
}
result.addAll(list);
}
} catch (InterruptedException e) {
throw new RuntimeException(e);
} catch (ExecutionException e) {
throw new RuntimeException(e);
}
}
Comparator<RecordFile> comparator = Comparator.comparing(RecordFile::getFileName);
result.sort(comparator);
return result;
}
@Override
public List<String> getRecordDates(String app, String stream, int year, int month, List<MediaServerItem> mediaServerItems) {
Assert.notNull(app, "app不存在");
Assert.notNull(stream, "stream不存在");
Assert.notEmpty(mediaServerItems, "流媒体列表为空");
CompletableFuture[] completableFutures = new CompletableFuture[mediaServerItems.size()];
for (int i = 0; i < mediaServerItems.size(); i++) {
completableFutures[i] = getRecordDatesForOne(app, stream, year, month, mediaServerItems.get(i));
}
List<String> result = new ArrayList<>();
CompletableFuture.allOf(completableFutures).join();
for (CompletableFuture completableFuture : completableFutures) {
try {
List<String> list = (List<String>) completableFuture.get();
result.addAll(list);
} catch (InterruptedException e) {
throw new RuntimeException(e);
} catch (ExecutionException e) {
throw new RuntimeException(e);
}
}
Collections.sort(result);
return result;
}
@Async
public CompletableFuture<List<String>> getRecordDatesForOne(String app, String stream, int year, int month, MediaServerItem mediaServerItem) {
JSONObject fileListJson = assistRESTfulUtils.getDateList(mediaServerItem, app, stream, year, month);
if (fileListJson != null && !fileListJson.isEmpty()) {
if (fileListJson.getString("code") != null && fileListJson.getInteger("code") == 0) {
JSONArray data = fileListJson.getJSONArray("data");
return CompletableFuture.completedFuture(data.toJavaList(String.class));
}
}
return CompletableFuture.completedFuture(new ArrayList<>());
}
@Async
public CompletableFuture<List<RecordFile>> getRecordFilesForOne(String app, String stream, String startTime, String endTime, MediaServerItem mediaServerItem) {
JSONObject fileListJson = assistRESTfulUtils.getFileList(mediaServerItem, 1, 100000000, app, stream, startTime, endTime);
if (fileListJson != null && !fileListJson.isEmpty()) {
if (fileListJson.getString("code") != null && fileListJson.getInteger("code") == 0) {
JSONObject data = fileListJson.getJSONObject("data");
JSONArray list = data.getJSONArray("list");
if (list != null) {
return CompletableFuture.completedFuture(list.toJavaList(RecordFile.class));
}
}
}
return CompletableFuture.completedFuture(new ArrayList<>());
}
}

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

@ -158,10 +158,7 @@ public class PlayServiceImpl implements IPlayService {
}
}
}
String streamId = null;
if (mediaServerItem.isRtpEnable()) {
streamId = String.format("%s_%s", device.getDeviceId(), channelId);
}
String streamId = String.format("%s_%s", device.getDeviceId(), channelId);;
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, ssrc, device.isSsrcCheck(), false, 0, false, device.getStreamModeForParam());
if (ssrcInfo == null) {
callback.run(InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getCode(), InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getMsg(), null);
@ -457,16 +454,13 @@ public class PlayServiceImpl implements IPlayService {
logger.warn("[录像回放] 单端口收流时不支持TCP主动方式收流 deviceId: {},channelId:{}", deviceId, channelId);
throw new ControllerException(ErrorCode.ERROR100.getCode(), "单端口收流时不支持TCP主动方式收流");
}
String stream = null;
if (newMediaServerItem.isRtpEnable()) {
String startTimeStr = startTime.replace("-", "")
.replace(":", "")
.replace(" ", "");
String endTimeTimeStr = endTime.replace("-", "")
.replace(":", "")
.replace(" ", "");
stream = deviceId + "_" + channelId + "_" + startTimeStr + "_" + endTimeTimeStr;
}
String startTimeStr = startTime.replace("-", "")
.replace(":", "")
.replace(" ", "");
String endTimeTimeStr = endTime.replace("-", "")
.replace(":", "")
.replace(" ", "");
String stream = deviceId + "_" + channelId + "_" + startTimeStr + "_" + endTimeTimeStr;
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, stream, null, device.isSsrcCheck(), true, 0, false, device.getStreamModeForParam());
playBack(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, callback);
}
@ -628,44 +622,13 @@ public class PlayServiceImpl implements IPlayService {
if (ssrcInResponse != null) {
// 单端口
// 重新订阅流上线
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp",
ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId());
subscribe.removeSubscribe(hookSubscribe);
SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(inviteInfo.getDeviceId(),
inviteInfo.getChannelId(), null, inviteInfo.getStream());
streamSession.remove(inviteInfo.getDeviceId(),
inviteInfo.getChannelId(), inviteInfo.getStream());
String stream = String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase();
hookSubscribe.getContent().put("stream", stream);
inviteStreamService.updateInviteInfoForStream(inviteInfo, stream);
inviteStreamService.updateInviteInfoForSSRC(inviteInfo, ssrcInResponse);
streamSession.put(device.getDeviceId(), channelId, ssrcTransaction.getCallId(),
stream, ssrcInResponse, mediaServerItem.getId(), (SIPResponse) responseEvent.getResponse(), inviteSessionType);
subscribe.addSubscribe(hookSubscribe, (mediaServerItemInUse, hookParam) -> {
logger.info("[Invite 200OK] ssrc修正后收到订阅消息： " + hookParam);
dynamicTask.stop(timeOutTaskKey);
subscribe.removeSubscribe(hookSubscribe);
// hook响应
StreamInfo streamInfo = onPublishHandlerForPlay(mediaServerItemInUse, hookParam, device.getDeviceId(), channelId);
if (streamInfo == null){
callback.run(InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(),
InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getMsg(), null);
inviteStreamService.call(inviteSessionType, device.getDeviceId(), channelId, null,
InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(),
InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getMsg(), null);
return;
}
callback.run(InviteErrorCode.SUCCESS.getCode(),
InviteErrorCode.SUCCESS.getMsg(), streamInfo);
inviteStreamService.call(inviteSessionType, device.getDeviceId(), channelId, null,
InviteErrorCode.SUCCESS.getCode(),
InviteErrorCode.SUCCESS.getMsg(),
streamInfo);
if (inviteSessionType == InviteSessionType.PLAY) {
snapOnPlay(mediaServerItemInUse, device.getDeviceId(), channelId, stream);
}
});
inviteInfo.getStream(), ssrcInResponse, mediaServerItem.getId(), (SIPResponse) responseEvent.getResponse(), inviteSessionType);
}
}
}
@ -707,6 +670,7 @@ public class PlayServiceImpl implements IPlayService {
null);
return;
}
logger.debug("startTime => {}, endTime => {}",startTime,endTime);
logger.info("[录像下载] deviceId: {}, channelId: {}, 下载速度：{}, 收流端口：{}, 收流模式：{}, SSRC: {}, SSRC校验：{}", device.getDeviceId(), channelId, downloadSpeed, ssrcInfo.getPort(), device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck());
// 初始化redis中的invite消息状态
InviteInfo inviteInfo = InviteInfo.getInviteInfo(device.getDeviceId(), channelId, ssrcInfo.getStream(), ssrcInfo,
@ -751,6 +715,7 @@ public class PlayServiceImpl implements IPlayService {
}
callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo);
logger.info("[录像下载] 调用成功 deviceId: {}, channelId: {}, 开始时间: {}, 结束时间： {}", device.getDeviceId(), channelId, startTime, endTime);
logger.debug("streamInfo => {}",streamInfo);
};
try {
cmder.downloadStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, downloadSpeed,

2
src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java
View File

@ -419,7 +419,7 @@ public interface DeviceChannelMapper {
"and(latitude_gcj02=0 or latitude_wgs84=0 or longitude_wgs84= 0 or longitude_gcj02 = 0)")
List<DeviceChannel> getChannelsWithoutTransform(String deviceId);
@Select("select de.* from wvp_device de left join wvp_device_channel dc on de.device_id = dc.deviceId where dc.channel_id=#{channelId}")
@Select("select de.* from wvp_device de left join wvp_device_channel dc on de.device_id = dc.device_id where dc.channel_id=#{channelId}")
List<Device> getDeviceByChannelId(String channelId);

3
src/main/java/com/genersoft/iot/vmp/utils/DateUtil.java
View File

@ -109,6 +109,9 @@ public class DateUtil {
}
public static long getDifferenceForNow(String keepaliveTime) {
if(keepaliveTime == null){
return 0;
}
Instant beforeInstant = Instant.from(formatter.parse(keepaliveTime));
return ChronoUnit.MILLIS.between(beforeInstant, Instant.now());
}

137
src/main/java/com/genersoft/iot/vmp/vmanager/bean/OtherPsSendInfo.java Normal file
View File

@ -0,0 +1,137 @@
package com.genersoft.iot.vmp.vmanager.bean;
public class OtherPsSendInfo {
/**
* 发流IP
*/
private String sendLocalIp;
/**
* 发流端口
*/
private int sendLocalPort;
/**
* 收流IP
*/
private String receiveIp;
/**
* 收流端口
*/
private int receivePort;
/**
* 会话ID
*/
private String callId;
/**
* 流ID
*/
private String stream;
/**
* 推流应用名
*/
private String pushApp;
/**
* 推流流ID
*/
private String pushStream;
/**
* 推流SSRC
*/
private String pushSSRC;
public String getSendLocalIp() {
return sendLocalIp;
}
public void setSendLocalIp(String sendLocalIp) {
this.sendLocalIp = sendLocalIp;
}
public int getSendLocalPort() {
return sendLocalPort;
}
public void setSendLocalPort(int sendLocalPort) {
this.sendLocalPort = sendLocalPort;
}
public String getReceiveIp() {
return receiveIp;
}
public void setReceiveIp(String receiveIp) {
this.receiveIp = receiveIp;
}
public int getReceivePort() {
return receivePort;
}
public void setReceivePort(int receivePort) {
this.receivePort = receivePort;
}
public String getCallId() {
return callId;
}
public void setCallId(String callId) {
this.callId = callId;
}
public String getStream() {
return stream;
}
public void setStream(String stream) {
this.stream = stream;
}
public String getPushApp() {
return pushApp;
}
public void setPushApp(String pushApp) {
this.pushApp = pushApp;
}
public String getPushStream() {
return pushStream;
}
public void setPushStream(String pushStream) {
this.pushStream = pushStream;
}
public String getPushSSRC() {
return pushSSRC;
}
public void setPushSSRC(String pushSSRC) {
this.pushSSRC = pushSSRC;
}
@Override
public String toString() {
return "OtherPsSendInfo{" +
"sendLocalIp='" + sendLocalIp + '\'' +
", sendLocalPort=" + sendLocalPort +
", receiveIp='" + receiveIp + '\'' +
", receivePort=" + receivePort +
", callId='" + callId + '\'' +
", stream='" + stream + '\'' +
", pushApp='" + pushApp + '\'' +
", pushStream='" + pushStream + '\'' +
", pushSSRC='" + pushSSRC + '\'' +
'}';
}
}

53
src/main/java/com/genersoft/iot/vmp/vmanager/bean/RecordFile.java Normal file
View File

@ -0,0 +1,53 @@
package com.genersoft.iot.vmp.vmanager.bean;
public class RecordFile {
private String app;
private String stream;
private String fileName;
private String mediaServerId;
private String date;
public String getApp() {
return app;
}
public void setApp(String app) {
this.app = app;
}
public String getStream() {
return stream;
}
public void setStream(String stream) {
this.stream = stream;
}
public String getFileName() {
return fileName;
}
public void setFileName(String fileName) {
this.fileName = fileName;
}
public String getMediaServerId() {
return mediaServerId;
}
public void setMediaServerId(String mediaServerId) {
this.mediaServerId = mediaServerId;
}
public String getDate() {
return date;
}
public void setDate(String date) {
this.date = date;
}
}

324
src/main/java/com/genersoft/iot/vmp/vmanager/ps/PsController.java Normal file
View File

@ -0,0 +1,324 @@
package com.genersoft.iot.vmp.vmanager.ps;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager;
import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForRtpServerTimeout;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnRtpServerTimeoutHookParam;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.utils.redis.RedisUtil;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.OtherPsSendInfo;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.tags.Tag;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.*;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
@SuppressWarnings("rawtypes")
@Tag(name = "第三方PS服务对接")
@RestController
@RequestMapping("/api/ps")
public class PsController {
private final static Logger logger = LoggerFactory.getLogger(PsController.class);
@Autowired
private ZLMServerFactory zlmServerFactory;
@Autowired
private ZlmHttpHookSubscribe hookSubscribe;
@Autowired
private IMediaServerService mediaServerService;
@Autowired
private SendRtpPortManager sendRtpPortManager;
@Autowired
private UserSetting userSetting;
@Autowired
private DynamicTask dynamicTask;
@Autowired
private RedisTemplate<Object, Object> redisTemplate;
@GetMapping(value = "/receive/open")
@ResponseBody
@Operation(summary = "开启收流和获取发流信息")
@Parameter(name = "isSend", description = "是否发送，false时只开启收流， true同时返回推流信息", required = true)
@Parameter(name = "callId", description = "整个过程的唯一标识，为了与后续接口关联", required = true)
@Parameter(name = "ssrc", description = "来源流的SSRC，不传则不校验来源ssrc", required = false)
@Parameter(name = "stream", description = "形成的流的ID", required = true)
@Parameter(name = "tcpMode", description = "收流模式， 0为UDP， 1为TCP被动", required = true)
@Parameter(name = "callBack", description = "回调地址，如果收流超时会通道回调通知，回调为get请求，参数为callId", required = true)
public OtherPsSendInfo openRtpServer(Boolean isSend, @RequestParam(required = false)String ssrc, String callId, String stream, Integer tcpMode, String callBack) {
logger.info("[第三方PS服务对接->开启收流和获取发流信息] isSend->{}, ssrc->{}, callId->{}, stream->{}, tcpMode->{}, callBack->{}",
isSend, ssrc, callId, stream, tcpMode==0?"UDP":"TCP被动", callBack);
MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer();
if (mediaServerItem == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(),"没有可用的MediaServer");
}
if (stream == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(),"stream参数不可为空");
}
if (isSend != null && isSend && callId == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(),"isSend为true时，CallID不能为空");
}
long ssrcInt = 0;
if (ssrc != null) {
try {
ssrcInt = Long.parseLong(ssrc);
}catch (NumberFormatException e) {
throw new ControllerException(ErrorCode.ERROR100.getCode(),"ssrc格式错误");
}
}
String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_PS_INFO + userSetting.getServerId() + "_" + callId + "_" + stream;
int localPort = zlmServerFactory.createRTPServer(mediaServerItem, stream, ssrcInt, null, false, tcpMode);
if (localPort == 0) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "获取端口失败");
}
// 注册回调如果rtp收流超时则通过回调发送通知
if (callBack != null) {
HookSubscribeForRtpServerTimeout hookSubscribeForRtpServerTimeout = HookSubscribeFactory.on_rtp_server_timeout(stream, String.valueOf(ssrcInt), mediaServerItem.getId());
// 订阅 zlm启动事件, 新的zlm也会从这里进入系统
hookSubscribe.addSubscribe(hookSubscribeForRtpServerTimeout,
(mediaServerItemInUse, hookParam)->{
OnRtpServerTimeoutHookParam serverTimeoutHookParam = (OnRtpServerTimeoutHookParam) hookParam;
if (stream.equals(serverTimeoutHookParam.getStream_id())) {
logger.info("[第三方PS服务对接->开启收流和获取发流信息] 等待收流超时 callId->{}, 发送回调", callId);
// 将信息写入redis中，以备后用
redisTemplate.delete(receiveKey);
OkHttpClient.Builder httpClientBuilder = new OkHttpClient.Builder();
OkHttpClient client = httpClientBuilder.build();
String url = callBack + "?callId=" + callId;
Request request = new Request.Builder().get().url(url).build();
try {
client.newCall(request).execute();
} catch (IOException e) {
logger.error("[第三方PS服务对接->开启收流和获取发流信息] 等待收流超时 callId->{}, 发送回调失败", callId, e);
}
hookSubscribe.removeSubscribe(hookSubscribeForRtpServerTimeout);
}
});
}
OtherPsSendInfo otherPsSendInfo = new OtherPsSendInfo();
otherPsSendInfo.setReceiveIp(mediaServerItem.getSdpIp());
otherPsSendInfo.setReceivePort(localPort);
otherPsSendInfo.setCallId(callId);
otherPsSendInfo.setStream(stream);
// 将信息写入redis中，以备后用
redisTemplate.opsForValue().set(receiveKey, otherPsSendInfo);
if (isSend != null && isSend) {
String key = VideoManagerConstants.WVP_OTHER_SEND_PS_INFO + userSetting.getServerId() + "_" + callId;
// 预创建发流信息
int port = sendRtpPortManager.getNextPort(mediaServerItem);
otherPsSendInfo.setSendLocalIp(mediaServerItem.getSdpIp());
otherPsSendInfo.setSendLocalPort(port);
// 将信息写入redis中，以备后用
redisTemplate.opsForValue().set(key, otherPsSendInfo, 300, TimeUnit.SECONDS);
logger.info("[第三方PS服务对接->开启收流和获取发流信息] 结果，callId->{}， {}", callId, otherPsSendInfo);
}
return otherPsSendInfo;
}
@GetMapping(value = "/receive/close")
@ResponseBody
@Operation(summary = "关闭收流")
@Parameter(name = "stream", description = "流的ID", required = true)
public void closeRtpServer(String stream) {
logger.info("[第三方PS服务对接->关闭收流] stream->{}", stream);
MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer();
zlmServerFactory.closeRtpServer(mediaServerItem,stream);
String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_PS_INFO + userSetting.getServerId() + "_*_" + stream;
List<Object> scan = RedisUtil.scan(redisTemplate, receiveKey);
if (!scan.isEmpty()) {
for (Object key : scan) {
// 将信息写入redis中，以备后用
redisTemplate.delete(key);
}
}
}
@GetMapping(value = "/send/start")
@ResponseBody
@Operation(summary = "发送流")
@Parameter(name = "ssrc", description = "发送流的SSRC", required = true)
@Parameter(name = "dstIp", description = "目标收流IP", required = true)
@Parameter(name = "dstPort", description = "目标收流端口", required = true)
@Parameter(name = "app", description = "待发送应用名", required = true)
@Parameter(name = "stream", description = "待发送流Id", required = true)
@Parameter(name = "callId", description = "整个过程的唯一标识，不传则使用随机端口发流", required = true)
@Parameter(name = "isUdp", description = "是否为UDP", required = true)
public void sendRTP(String ssrc,
String dstIp,
Integer dstPort,
String app,
String stream,
String callId,
Boolean isUdp
) {
logger.info("[第三方PS服务对接->发送流] " +
"ssrc->{}, \r\n" +
"dstIp->{}, \n" +
"dstPort->{}, \n" +
"app->{}, \n" +
"stream->{}, \n" +
"callId->{} \n",
ssrc,
dstIp,
dstPort,
app,
stream,
callId);
MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer();
String key = VideoManagerConstants.WVP_OTHER_SEND_PS_INFO + userSetting.getServerId() + "_" + callId;
OtherPsSendInfo sendInfo = (OtherPsSendInfo)redisTemplate.opsForValue().get(key);
if (sendInfo == null) {
sendInfo = new OtherPsSendInfo();
}
sendInfo.setPushApp(app);
sendInfo.setPushStream(stream);
sendInfo.setPushSSRC(ssrc);
Map<String, Object> param;
param = new HashMap<>();
param.put("vhost","__defaultVhost__");
param.put("app",app);
param.put("stream",stream);
param.put("ssrc", ssrc);
param.put("dst_url", dstIp);
param.put("dst_port", dstPort);
String is_Udp = isUdp ? "1" : "0";
param.put("is_udp", is_Udp);
param.put("src_port", sendInfo.getSendLocalPort());
param.put("use_ps", "0");
param.put("only_audio", "1");
Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, app, stream);
if (streamReady) {
JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServerItem, param);
if (jsonObject.getInteger("code") == 0) {
logger.info("[第三方PS服务对接->发送流] 视频流发流成功，callId->{}，param->{}", callId, param);
redisTemplate.opsForValue().set(key, sendInfo);
}else {
redisTemplate.delete(key);
logger.info("[第三方PS服务对接->发送流] 视频流发流失败，callId->{}, {}", callId, jsonObject.getString("msg"));
throw new ControllerException(ErrorCode.ERROR100.getCode(), "[视频流发流失败] " + jsonObject.getString("msg"));
}
}else {
logger.info("[第三方PS服务对接->发送流] 流不存在，等待流上线，callId->{}", callId);
String uuid = UUID.randomUUID().toString();
HookSubscribeForStreamChange hookSubscribeForStreamChange = HookSubscribeFactory.on_stream_changed(app, stream, true, "rtsp", mediaServerItem.getId());
dynamicTask.startDelay(uuid, ()->{
logger.info("[第三方PS服务对接->发送流] 等待流上线超时 callId->{}", callId);
redisTemplate.delete(key);
hookSubscribe.removeSubscribe(hookSubscribeForStreamChange);
}, 10000);
// 订阅 zlm启动事件, 新的zlm也会从这里进入系统
OtherPsSendInfo finalSendInfo = sendInfo;
hookSubscribe.removeSubscribe(hookSubscribeForStreamChange);
hookSubscribe.addSubscribe(hookSubscribeForStreamChange,
(mediaServerItemInUse, response)->{
dynamicTask.stop(uuid);
logger.info("[第三方PS服务对接->发送流] 流上线，开始发流 callId->{}", callId);
try {
Thread.sleep(400);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServerItem, param);
if (jsonObject.getInteger("code") == 0) {
logger.info("[第三方PS服务对接->发送流] 视频流发流成功，callId->{}，param->{}", callId, param);
redisTemplate.opsForValue().set(key, finalSendInfo);
}else {
redisTemplate.delete(key);
logger.info("[第三方PS服务对接->发送流] 视频流发流失败，callId->{}, {}", callId, jsonObject.getString("msg"));
throw new ControllerException(ErrorCode.ERROR100.getCode(), "[视频流发流失败] " + jsonObject.getString("msg"));
}
hookSubscribe.removeSubscribe(hookSubscribeForStreamChange);
});
}
}
@GetMapping(value = "/send/stop")
@ResponseBody
@Operation(summary = "关闭发送流")
@Parameter(name = "callId", description = "整个过程的唯一标识，不传则使用随机端口发流", required = true)
public void closeSendRTP(String callId) {
logger.info("[第三方PS服务对接->关闭发送流] callId->{}", callId);
String key = VideoManagerConstants.WVP_OTHER_SEND_PS_INFO + userSetting.getServerId() + "_" + callId;
OtherPsSendInfo sendInfo = (OtherPsSendInfo)redisTemplate.opsForValue().get(key);
if (sendInfo == null){
throw new ControllerException(ErrorCode.ERROR100.getCode(), "未开启发流");
}
Map<String, Object> param = new HashMap<>();
param.put("vhost","__defaultVhost__");
param.put("app",sendInfo.getPushApp());
param.put("stream",sendInfo.getPushStream());
param.put("ssrc",sendInfo.getPushSSRC());
MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer();
Boolean result = zlmServerFactory.stopSendRtpStream(mediaServerItem, param);
if (!result) {
logger.info("[第三方PS服务对接->关闭发送流] 失败 callId->{}", callId);
throw new ControllerException(ErrorCode.ERROR100.getCode(), "停止发流失败");
}else {
logger.info("[第三方PS服务对接->关闭发送流] 成功 callId->{}", callId);
}
redisTemplate.delete(key);
}
@GetMapping(value = "/getTestPort")
@ResponseBody
public int getTestPort() {
MediaServerItem defaultMediaServer = mediaServerService.getDefaultMediaServer();
// for (int i = 0; i <300; i++) {
// new Thread(() -> {
// int nextPort = sendRtpPortManager.getNextPort(defaultMediaServer);
// try {
// Thread.sleep((int)Math.random()*10);
// } catch (InterruptedException e) {
// throw new RuntimeException(e);
// }
// System.out.println(nextPort);
// }).start();
// }
return sendRtpPortManager.getNextPort(defaultMediaServer);
}
}

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

@ -91,10 +91,10 @@ public class RtpController {
if (isSend != null && isSend && callId == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(),"isSend为true时，CallID不能为空");
}
int ssrcInt = 0;
long ssrcInt = 0;
if (ssrc != null) {
try {
ssrcInt = Integer.parseInt(ssrc);
ssrcInt = Long.parseLong(ssrc);
}catch (NumberFormatException e) {
throw new ControllerException(ErrorCode.ERROR100.getCode(),"ssrc格式错误");
}
@ -198,7 +198,7 @@ public class RtpController {
Boolean isUdp,
@RequestParam(required = false)Integer ptForAudio,
@RequestParam(required = false)Integer ptForVideo
) {
) {
logger.info("[第三方服务对接->发送流] " +
"ssrc->{}, \r\n" +
"dstIpForAudio->{}, \n" +
@ -210,16 +210,16 @@ public class RtpController {
"callId->{}, \n" +
"ptForAudio->{}, \n" +
"ptForVideo->{}",
ssrc,
dstIpForAudio,
dstIpForVideo,
dstPortForAudio,
dstPortForVideo,
app,
stream,
callId,
ptForAudio,
ptForVideo);
ssrc,
dstIpForAudio,
dstIpForVideo,
dstPortForAudio,
dstPortForVideo,
app,
stream,
callId,
ptForAudio,
ptForVideo);
if (!((dstPortForAudio > 0 && !ObjectUtils.isEmpty(dstPortForAudio) || (dstPortForVideo > 0 && !ObjectUtils.isEmpty(dstIpForVideo))))) {
throw new ControllerException(ErrorCode.ERROR400.getCode(), "至少应该存在一组音频或视频发送参数");
}

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

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

122
src/test/resources/application-dev312.yml Normal file
View File

@ -0,0 +1,122 @@
spring:
# 设置接口超时时间
mvc:
async:
request-timeout: 20000
thymeleaf:
cache: false
# [可选]上传文件大小限制
servlet:
multipart:
max-file-size: 10MB
max-request-size: 100MB
# REDIS数据库配置
redis:
# [必须修改] Redis服务器IP, REDIS安装在本机的,使用127.0.0.1
host: 192.168.2.4
# [必须修改] 端口号
port: 6379
# [可选] 数据库 DB
database: 7
# [可选] 访问密码,若你的redis服务器没有设置密码，就不需要用密码去连接
password:
# [可选] 超时时间
timeout: 10000
# mysql数据源
datasource:
type: com.zaxxer.hikari.HikariDataSource
driver-class-name: com.mysql.cj.jdbc.Driver
url: jdbc:mysql://192.168.2.3:3306/wvp?useUnicode=true&characterEncoding=UTF8&rewriteBatchedStatements=true&serverTimezone=PRC&useSSL=false&allowMultiQueries=true
username: root
password: 123456a
hikari:
connection-timeout: 20000 # 是客户端等待连接池连接的最大毫秒数
initialSize: 10 # 连接池初始化连接数
maximum-pool-size: 200 # 连接池最大连接数
minimum-idle: 5 # 连接池最小空闲连接数
idle-timeout: 300000 # 允许连接在连接池中空闲的最长时间（以毫秒为单位）
max-lifetime: 1200000 # 是池中连接关闭后的最长生命周期（以毫秒为单位)
config:
activate:
on-profile: dev312
#[可选] WVP监听的HTTP端口, 网页和接口调用都是这个端口
server:
port: 18978
# [可选] HTTPS配置， 默认不开启
ssl:
# [可选] 是否开启HTTPS访问
enabled: false
# [可选] 证书文件路径，放置在resource/目录下即可，修改xxx为文件名
key-store: classpath:test.monitor.89iot.cn.jks
# [可选] 证书密码
key-store-password: gpf64qmw
# [可选] 证书类型， 默认为jks，根据实际修改
key-store-type: JKS
# 作为28181服务器的配置
sip:
# [必须修改] 本机的IP，对应你的网卡，监听什么ip就是使用什么网卡，
# 如果要监听多张网卡，可以使用逗号分隔多个IP， 例如： 192.168.1.4,10.0.0.4
# 如果不明白，就使用0.0.0.0，大部分情况都是可以的
# 请不要使用127.0.0.1，任何包括localhost在内的域名都是不可以的。
ip: 192.168.3.12
# [可选] 28181服务监听的端口
port: 5060
# 根据国标6.1.2中规定，domain宜采用ID统一编码的前十位编码。国标附录D中定义前8位为中心编码（由省级、市级、区级、基层编号组成，参照GB/T 2260-2007）
# 后两位为行业编码，定义参照附录D.3
# 3701020049标识山东济南历下区 信息行业接入
# [可选]
domain: 4405010000
# [可选]
id: 44050100002000000001
# [可选] 默认设备认证密码，后续扩展使用设备单独密码, 移除密码将不进行校验
password: 123456
# 是否存储alarm信息
alarm: true
#zlm 默认服务器配置
media:
# id: your_server_id
id: amrWMKmbKqoBjRQ9
# [必须修改] zlm服务器的内网IP
ip: 192.168.3.14
# [必须修改] zlm服务器的http.port
http-port: 5081
# [可选] 返回流地址时的ip，置空使用 media.ip
# stream-ip: 192.168.10.241
# [可选] wvp在国标信令中使用的ip，此ip为摄像机可以访问到的ip， 置空使用 media.ip
# sdp-ip: 192.168.10.241
# [可选] zlm服务器的hook所使用的IP, 默认使用sip.ip
hook-ip:
# [可选] zlm服务器的http.sslport, 置空使用zlm配置文件配置
http-ssl-port: 5444
# [可选] zlm服务器的hook.admin_params=secret
# secret: 035c73f7-bb6b-4889-a715-d9eb2d1925cc
secret: 4155cca6-2f9f-11ee-85e6-8de4ce2e7333
# 启用多端口模式, 多端口模式使用端口区分每路流，兼容性更好。 单端口使用流的ssrc区分， 点播超时建议使用多端口测试
rtp:
# [可选] 是否启用多端口模式, 开启后会在portRange范围内选择端口用于媒体流传输
enable: true
# [可选] 在此范围内选择端口用于媒体流传输, 必须提前在zlm上配置该属性，不然自动配置此属性可能不成功
#port-range: 51000,51300 # 端口范围
port-range: 62200,62250
# [可选] 国标级联在此范围内选择端口发送媒体流,
#send-port-range: 51000,51300 # 端口范围
send-port-range: 62200,62250
# 录像辅助服务， 部署此服务可以实现zlm录像的管理与下载， 0 表示不使用
record-assist-port: 18979
# [根据业务需求配置]
user-settings:
# 点播/录像回放 等待超时时间,单位：毫秒
play-timeout: 360000
# [可选] 自动点播， 使用固定流地址进行播放时，如果未点播则自动进行点播, 需要rtp.enable=true
auto-apply-play: true
# 设备/通道状态变化时发送消息
device-status-notify: true
# 跨域配置，配置你访问前端页面的地址即可， 可以配置多个
allowed-origins:
- http://localhost:8080
- http://127.0.0.1:8080
# [可选] 日志配置, 一般不需要改
logging:
config: classpath:logback-spring-local.xml

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

@ -36,6 +36,8 @@ spring:
minimum-idle: 5 # 连接池最小空闲连接数
idle-timeout: 300000 # 允许连接在连接池中空闲的最长时间（以毫秒为单位）
max-lifetime: 1200000 # 是池中连接关闭后的最长生命周期（以毫秒为单位)
profiles:
active: dev312
#[可选] WVP监听的HTTP端口, 网页和接口调用都是这个端口
server:
port: 18978