Thirdparty
/
wvp-GB28181-pro
mirror of https://github.com/648540858/wvp-GB28181-pro
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'wvp-28181-2.0' 

# Conflicts:
#	src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java
#	src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java
#	src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java
结构优化
648540858 2022-09-08 09:46:35 +08:00
parent aa6bce35c7 60c4f767f0
commit 2f76fa98bb
33 changed files with 330 additions and 114 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -53,7 +53,7 @@ public class ParentPlatform {
/**
* 设备国标编号
*/
@Schema(description = "11111")
@Schema(description = "设备国标编号")
private String deviceGBId;
/**
@ -113,7 +113,6 @@ public class ParentPlatform {
/**
* RTCP流保活
* TODO 预留, 暂不实现
*/
@Schema(description = "RTCP流保活")
private boolean rtcp;

2
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java
View File

@ -108,7 +108,7 @@ public interface ISIPCommander {
* @param startTime 开始时间,格式要求：yyyy-MM-dd HH:mm:ss
* @param endTime 结束时间,格式要求：yyyy-MM-dd HH:mm:ss
*/
void playbackStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInf, Device device, String channelId, String startTime, String endTime,InviteStreamCallback inviteStreamCallback, InviteStreamCallback event, SipSubscribe.Event errorEvent);
void playbackStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInf, Device device, String channelId, String startTime, String endTime,InviteStreamCallback inviteStreamCallback, InviteStreamCallback event, SipSubscribe.Event okEvent, SipSubscribe.Event errorEvent);
/**
* 请求历史媒体下载

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

@ -358,7 +358,7 @@ public class SIPCommander implements ISIPCommander {
// String streamMode = device.getStreamMode().toUpperCase();
logger.info("{} 分配的ZLM为: {} [{}:{}]", stream, mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort());
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", stream, true, "rtmp", mediaServerItem.getId());
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", stream, true, "rtsp", mediaServerItem.getId());
subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject json)->{
if (event != null) {
event.response(mediaServerItemInUse, json);
@ -458,7 +458,7 @@ public class SIPCommander implements ISIPCommander {
@Override
public void playbackStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId,
String startTime, String endTime, InviteStreamCallback inviteStreamCallback, InviteStreamCallback hookEvent,
SipSubscribe.Event errorEvent) {
SipSubscribe.Event okEvent,SipSubscribe.Event errorEvent) {
try {
logger.info("{} 分配的ZLM为: {} [{}:{}]", ssrcInfo.getStream(), mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort());
@ -526,7 +526,7 @@ public class SIPCommander implements ISIPCommander {
CallIdHeader callIdHeader = device.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
: udpSipProvider.getNewCallId();
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtmp", mediaServerItem.getId());
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId());
// 添加订阅
subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject json)->{
if (hookEvent != null) {
@ -537,10 +537,11 @@ public class SIPCommander implements ISIPCommander {
});
Request request = headerProvider.createPlaybackInviteRequest(device, channelId, content.toString(), null, "fromplybck" + tm, null, callIdHeader, ssrcInfo.getSsrc());
transmitRequest(device, request, errorEvent, okEvent -> {
ResponseEvent responseEvent = (ResponseEvent) okEvent.event;
transmitRequest(device, request, errorEvent, event -> {
ResponseEvent responseEvent = (ResponseEvent) event.event;
streamSession.put(device.getDeviceId(), channelId, callIdHeader.getCallId(), ssrcInfo.getStream(), ssrcInfo.getSsrc(), mediaServerItem.getId(), responseEvent.getClientTransaction(), VideoStreamSessionManager.SessionType.playback);
streamSession.put(device.getDeviceId(), channelId, callIdHeader.getCallId(), okEvent.dialog);
streamSession.put(device.getDeviceId(), channelId, callIdHeader.getCallId(), event.dialog);
okEvent.response(event);
});
if (inviteStreamCallback != null) {
inviteStreamCallback.call(new InviteStreamInfo(mediaServerItem, null, callIdHeader.getCallId(), "rtp", ssrcInfo.getStream()));

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

@ -121,6 +121,10 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In
param.put("pt", sendRtpItem.getPt());
param.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0");
param.put("only_audio", sendRtpItem.isOnlyAudio() ? "1" : "0");
if (!sendRtpItem.isTcp() && parentPlatform.isRtcp()) {
// 开启rtcp保活
param.put("udp_rtcp_timeout", "1");
}
JSONObject jsonObject;
if (sendRtpItem.isTcpActive()) {
jsonObject = zlmrtpServerFactory.startSendRtpPassive(mediaInfo, param);

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

@ -104,6 +104,8 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param);
redisCatchStorage.deleteSendRTPServer(platformGbId, sendRtpItem.getChannelId(), callIdHeader.getCallId(), null);
redisCatchStorage.deleteSendRTPServer(platformGbId, channelId, callIdHeader.getCallId(), null);
zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param);
int totalReaderCount = zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId);
if (totalReaderCount <= 0) {
logger.info("收到bye: {} 无其它观看者，通知设备停止推流", streamId);

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

@ -110,6 +110,9 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
@Autowired
private ZLMRESTfulUtils zlmresTfulUtils;
@Autowired
private ZlmHttpHookSubscribe zlmHttpHookSubscribe;
@Autowired
private SIPProcessorObserver sipProcessorObserver;
@ -430,7 +433,14 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
if (playTransaction != null) {
Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, "rtp", playTransaction.getStream());
if (!streamReady) {
playTransaction = null;
boolean hasRtpServer = mediaServerService.checkRtpServer(mediaServerItem, "rtp", playTransaction.getStream());
if (hasRtpServer) {
logger.info("[上级点播]已经开启rtpServer但是尚未收到流，开启监听流的到来");
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", playTransaction.getStream(), true, "rtsp", mediaServerItem.getId());
zlmHttpHookSubscribe.addSubscribe(hookSubscribe, hookEvent);
}else {
playTransaction = null;
}
}
}
if (playTransaction == null) {
@ -593,7 +603,8 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
responseAck(evt, Response.BAD_REQUEST, "channel [" + gbStream.getGbId() + "] offline");
} else if ("push".equals(gbStream.getStreamType())) {
if (!platform.isStartOfflinePush()) {
responseAck(evt, Response.TEMPORARILY_UNAVAILABLE, "channel unavailable");
// 平台设置中关闭了拉起离线的推流则直接回复
responseAck(evt, Response.TEMPORARILY_UNAVAILABLE, "channel stream not pushing");
return;
}
// 发送redis消息以使设备上线
@ -629,7 +640,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
app, stream, channelId, mediaTransmissionTCP);
if (sendRtpItem == null) {
logger.warn("服务器端口资源不足");
logger.warn("上级点时创建sendRTPItem失败，可能是服务器端口资源不足");
try {
responseAck(evt, Response.BUSY_HERE);
} catch (SipException e) {

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

@ -211,7 +211,6 @@ public class SubscribeRequestProcessor extends SIPRequestProcessorParent impleme
}else if (subscribeInfo.getExpires() == 0) {
subscribeHolder.removeCatalogSubscribe(platformId);
}
try {
ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformId);
responseXmlAck(evt, resultXml.toString(), parentPlatform);
@ -219,5 +218,4 @@ public class SubscribeRequestProcessor extends SIPRequestProcessorParent impleme
e.printStackTrace();
}
}
}

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

@ -203,6 +203,12 @@ public class XmlUtil {
return null;
}
deviceChannel.setChannelId(channelId);
int channelTypeCode = Integer.parseInt(channelId.substring(10, 13));
if (channelTypeCode == 136 || channelTypeCode == 137 || channelTypeCode == 138) {
deviceChannel.setHasAudio(true);
}else {
deviceChannel.setHasAudio(false);
}
if (event != null && !event.equals(CatalogEvent.ADD) && !event.equals(CatalogEvent.UPDATE)) {
// 除了ADD和update情况下需要识别全部内容，
return deviceChannel;
@ -396,7 +402,6 @@ public class XmlUtil {
} else {
deviceChannel.setPTZType(Integer.parseInt(XmlUtil.getText(itemDevice, "PTZType")));
}
deviceChannel.setHasAudio(true); // 默认含有音频，播放时再检查是否有音频及是否AAC
return deviceChannel;
}
}

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

@ -50,7 +50,7 @@ public class AssistRESTfulUtils {
if (mediaServerItem == null) {
return null;
}
if (mediaServerItem.getRecordAssistPort() > 0) {
if (mediaServerItem.getRecordAssistPort() <= 0) {
logger.warn("未启用Assist服务");
return null;
}

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

@ -19,8 +19,6 @@ 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.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.ObjectUtils;
import org.springframework.web.bind.annotation.PostMapping;
@ -544,6 +542,8 @@ public class ZLMHttpHookListener {
for (SendRtpItem sendRtpItem : sendRtpItems) {
ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId());
redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(),
sendRtpItem.getCallId(), sendRtpItem.getStreamId());
}
}
}
@ -573,13 +573,19 @@ public class ZLMHttpHookListener {
return ret;
}else {
StreamProxyItem streamProxyItem = streamProxyService.getStreamProxyByAppAndStream(app, streamId);
if (streamProxyItem != null && streamProxyItem.isEnable_remove_none_reader()) {
ret.put("close", true);
streamProxyService.del(app, streamId);
String url = streamProxyItem.getUrl() != null?streamProxyItem.getUrl():streamProxyItem.getSrc_url();
logger.info("[{}/{}]<-[{}] 拉流代理无人观看已经移除", app, streamId, url);
}else {
ret.put("close", false);
if (streamProxyItem != null ) {
if (streamProxyItem.isEnable_remove_none_reader()) {
// 无人观看自动移除
ret.put("close", true);
streamProxyService.del(app, streamId);
String url = streamProxyItem.getUrl() != null?streamProxyItem.getUrl():streamProxyItem.getSrc_url();
logger.info("[{}/{}]<-[{}] 拉流代理无人观看已经移除", app, streamId, url);
}else if (streamProxyItem.isEnable_disable_none_reader()) {
// 无人观看停用
ret.put("close", true);
}else {
ret.put("close", false);
}
}
return ret;
}
@ -626,7 +632,7 @@ public class ZLMHttpHookListener {
@ResponseBody
@PostMapping(value = "/on_server_started", produces = "application/json;charset=UTF-8")
public JSONObject onServerStarted(HttpServletRequest request, @RequestBody JSONObject jsonObject){
if (logger.isDebugEnabled()) {
logger.debug("[ ZLM HOOK ]on_server_started API调用，参数：" + jsonObject.toString());
}
@ -649,6 +655,39 @@ public class ZLMHttpHookListener {
return ret;
}
/**
* 发送rtp(startSendRtp)被动关闭时回调
*/
@ResponseBody
@PostMapping(value = "/on_send_rtp_stopped", produces = "application/json;charset=UTF-8")
public JSONObject onSendRtpStopped(HttpServletRequest request, @RequestBody JSONObject jsonObject){
logger.info("[ ZLM HOOK ]on_send_rtp_stopped API调用，参数：" + jsonObject);
JSONObject ret = new JSONObject();
ret.put("code", 0);
ret.put("msg", "success");
// 查找对应的上级推流，发送停止
String app = jsonObject.getString("app");
if (!"rtp".equals(app)) {
return ret;
}
String stream = jsonObject.getString("stream");
List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByStream(stream);
if (sendRtpItems.size() > 0) {
for (SendRtpItem sendRtpItem : sendRtpItems) {
ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId());
redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(),
sendRtpItem.getCallId(), sendRtpItem.getStreamId());
}
}
return ret;
}
private Map<String, String> urlParamToMap(String params) {
HashMap<String, String> map = new HashMap<>();
if (ObjectUtils.isEmpty(params)) {

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

@ -96,6 +96,10 @@ public class ZLMRTPServerFactory {
if(rtpInfo.getInteger("code") == 0){
if (rtpInfo.getBoolean("exist")) {
result = rtpInfo.getInteger("local_port");
if (result == 0) {
// 此时说明rtpServer已经创建但是流还没有推上来
}
return result;
}
}else if(rtpInfo.getInteger("code") == -2){

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

@ -37,6 +37,9 @@ public class StreamProxyItem extends GbStream {
private boolean enable_mp4;
@Schema(description = "是否 无人观看时删除")
private boolean enable_remove_none_reader;
@Schema(description = "是否 无人观看时不启用")
private boolean enable_disable_none_reader;
@Schema(description = "上级平台国标ID")
private String platformGbId;
@Schema(description = "创建时间")
@ -177,4 +180,11 @@ public class StreamProxyItem extends GbStream {
this.enable_remove_none_reader = enable_remove_none_reader;
}
public boolean isEnable_disable_none_reader() {
return enable_disable_none_reader;
}
public void setEnable_disable_none_reader(boolean enable_disable_none_reader) {
this.enable_disable_none_reader = enable_disable_none_reader;
}
}

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

@ -82,4 +82,6 @@ public interface IMediaServerService {
MediaServerItem getDefaultMediaServer();
void updateMediaServerKeepalive(String mediaServerId, JSONObject data);
boolean checkRtpServer(MediaServerItem mediaServerItem, String rtp, String stream);
}

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

@ -152,9 +152,11 @@ public class MediaServerServiceImpl implements IMediaServerService {
if (streamId == null) {
streamId = String.format("%08x", Integer.parseInt(ssrc)).toUpperCase();
}
int rtpServerPort = mediaServerItem.getRtpProxyPort();
int rtpServerPort;
if (mediaServerItem.isRtpEnable()) {
rtpServerPort = zlmrtpServerFactory.createRTPServer(mediaServerItem, streamId, ssrcCheck?Integer.parseInt(ssrc):0, port);
} else {
rtpServerPort = mediaServerItem.getRtpProxyPort();
}
RedisUtil.set(key, mediaServerItem);
return new SSRCInfo(rtpServerPort, ssrc, streamId);
@ -537,6 +539,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
param.put("hook.on_stream_none_reader",String.format("%s/on_stream_none_reader", hookPrex));
param.put("hook.on_stream_not_found",String.format("%s/on_stream_not_found", hookPrex));
param.put("hook.on_server_keepalive",String.format("%s/on_server_keepalive", hookPrex));
param.put("hook.on_send_rtp_stopped",String.format("%s/on_send_rtp_stopped", 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()));
}else {
@ -686,4 +689,13 @@ public class MediaServerServiceImpl implements IMediaServerService {
}
}
}
@Override
public boolean checkRtpServer(MediaServerItem mediaServerItem, String app, String stream) {
JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaServerItem, stream);
if(rtpInfo.getInteger("code") == 0){
return rtpInfo.getBoolean("exist");
}
return false;
}
}

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

@ -73,7 +73,6 @@ public class MediaServiceImpl implements IMediaService {
}else {
streamInfo = getStreamInfoByAppAndStream(mediaInfo, app, stream, tracks, addr,null, true);
}
}
}
return streamInfo;

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

@ -193,17 +193,30 @@ public class PlayServiceImpl implements IPlayService {
JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaInfo, streamId);
if(rtpInfo.getInteger("code") == 0){
if (rtpInfo.getBoolean("exist")) {
int localPort = rtpInfo.getInteger("local_port");
if (localPort == 0) {
logger.warn("[点播]，点播时发现rtpServerC存在，但是尚未开始推流");
// 此时说明rtpServer已经创建但是流还没有推上来
WVPResult wvpResult = new WVPResult();
wvpResult.setCode(ErrorCode.ERROR100.getCode());
wvpResult.setMsg("点播已经在进行中，请稍候重试");
msg.setData(wvpResult);
WVPResult wvpResult = new WVPResult();
wvpResult.setCode(ErrorCode.SUCCESS.getCode());
wvpResult.setMsg(ErrorCode.SUCCESS.getMsg());
wvpResult.setData(streamInfo);
msg.setData(wvpResult);
resultHolder.invokeAllResult(msg);
return playResult;
}else {
WVPResult wvpResult = new WVPResult();
wvpResult.setCode(ErrorCode.SUCCESS.getCode());
wvpResult.setMsg(ErrorCode.SUCCESS.getMsg());
wvpResult.setData(streamInfo);
msg.setData(wvpResult);
resultHolder.invokeAllResult(msg);
if (hookEvent != null) {
hookEvent.response(mediaServerItem, JSONObject.parseObject(JSON.toJSONString(streamInfo)));
resultHolder.invokeAllResult(msg);
if (hookEvent != null) {
hookEvent.response(mediaServerItem, JSONObject.parseObject(JSON.toJSONString(streamInfo)));
}
}
}else {
redisCatchStorage.stopPlay(streamInfo);
storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
@ -318,7 +331,7 @@ public class PlayServiceImpl implements IPlayService {
}
logger.info("[点播消息] 收到invite 200, 发现下级自定义了ssrc: {}", ssrcInResponse );
if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) {
logger.info("[SIP 消息] SSRC修正 {}->{}", ssrc, ssrcInResponse);
logger.info("[点播消息] SSRC修正 {}->{}", ssrc, ssrcInResponse);
if (!mediaServerItem.getSsrcConfig().checkSsrc(ssrcInResponse)) {
// ssrc 不可用
@ -468,37 +481,92 @@ public class PlayServiceImpl implements IPlayService {
resultHolder.exist(DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId, uuid);
}, userSetting.getPlayTimeout());
SipSubscribe.Event errorEvent = event -> {
dynamicTask.stop(playBackTimeOutTaskKey);
requestMessage.setData(WVPResult.fail(ErrorCode.ERROR100.getCode(), String.format("回放失败， 错误码： %s, %s", event.statusCode, event.msg)));
playBackResult.setCode(ErrorCode.ERROR100.getCode());
playBackResult.setMsg(String.format("回放失败， 错误码： %s, %s", event.statusCode, event.msg));
playBackResult.setData(requestMessage);
playBackResult.setEvent(event);
playBackCallback.call(playBackResult);
streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
};
InviteStreamCallback hookEvent = (InviteStreamInfo inviteStreamInfo) -> {
logger.info("收到回放订阅消息： " + inviteStreamInfo.getResponse().toJSONString());
dynamicTask.stop(playBackTimeOutTaskKey);
StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId);
if (streamInfo == null) {
logger.warn("设备回放API调用失败！");
playBackResult.setCode(ErrorCode.ERROR100.getCode());
playBackResult.setMsg("设备回放API调用失败！");
playBackCallback.call(playBackResult);
return;
}
redisCatchStorage.startPlayback(streamInfo, inviteStreamInfo.getCallId());
WVPResult<StreamInfo> success = WVPResult.success(streamInfo);
requestMessage.setData(success);
playBackResult.setCode(ErrorCode.SUCCESS.getCode());
playBackResult.setMsg(ErrorCode.SUCCESS.getMsg());
playBackResult.setData(requestMessage);
playBackResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem());
playBackResult.setResponse(inviteStreamInfo.getResponse());
playBackCallback.call(playBackResult);
};
cmder.playbackStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, infoCallBack,
(InviteStreamInfo inviteStreamInfo) -> {
logger.info("收到订阅消息： " + inviteStreamInfo.getResponse().toJSONString());
dynamicTask.stop(playBackTimeOutTaskKey);
StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId);
if (streamInfo == null) {
logger.warn("设备回放API调用失败！");
playBackResult.setCode(ErrorCode.ERROR100.getCode());
playBackResult.setMsg("设备回放API调用失败！");
playBackCallback.call(playBackResult);
return;
hookEvent, eventResult -> {
if (eventResult.type == SipSubscribe.EventResultType.response) {
ResponseEvent responseEvent = (ResponseEvent)eventResult.event;
String contentString = new String(responseEvent.getResponse().getRawContent());
// 获取ssrc
int ssrcIndex = contentString.indexOf("y=");
// 检查是否有y字段
if (ssrcIndex >= 0) {
//ssrc规定长度为10字节，不取余下长度以避免后续还有“f=”字段 TODO 后续对不规范的非10位ssrc兼容
String ssrcInResponse = contentString.substring(ssrcIndex + 2, ssrcIndex + 12);
// 查询到ssrc不一致且开启了ssrc校验则需要针对处理
if (ssrcInfo.getSsrc().equals(ssrcInResponse)) {
return;
}
logger.info("[回放消息] 收到invite 200, 发现下级自定义了ssrc: {}", ssrcInResponse );
if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) {
logger.info("[回放消息] SSRC修正 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse);
if (!mediaServerItem.getSsrcConfig().checkSsrc(ssrcInResponse)) {
// ssrc 不可用
// 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
eventResult.msg = "下级自定义了ssrc,但是此ssrc不可用";
eventResult.statusCode = 400;
errorEvent.response(eventResult);
return;
}
// 单端口模式streamId也有变化，需要重新设置监听
if (!mediaServerItem.isRtpEnable()) {
// 添加订阅
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId());
subscribe.removeSubscribe(hookSubscribe);
hookSubscribe.getContent().put("stream", String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase());
subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject response)->{
logger.info("[ZLM HOOK] ssrc修正后收到订阅消息： " + response.toJSONString());
dynamicTask.stop(playBackTimeOutTaskKey);
// hook响应
onPublishHandlerForPlay(mediaServerItemInUse, response, device.getDeviceId(), channelId, uuid);
hookEvent.call(new InviteStreamInfo(mediaServerItem, null, eventResult.callId, "rtp", ssrcInfo.getStream()));
});
}
// 关闭rtp server
mediaServerService.closeRTPServer(device.getDeviceId(), channelId, ssrcInfo.getStream());
// 重新开启ssrc server
mediaServerService.openRTPServer(mediaServerItem, ssrcInfo.getStream(), ssrcInResponse, device.isSsrcCheck(), true, ssrcInfo.getPort());
}
}
}
redisCatchStorage.startPlayback(streamInfo, inviteStreamInfo.getCallId());
WVPResult<StreamInfo> success = WVPResult.success(streamInfo);
requestMessage.setData(success);
playBackResult.setCode(ErrorCode.SUCCESS.getCode());
playBackResult.setMsg(ErrorCode.SUCCESS.getMsg());
playBackResult.setData(requestMessage);
playBackResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem());
playBackResult.setResponse(inviteStreamInfo.getResponse());
playBackCallback.call(playBackResult);
}, event -> {
dynamicTask.stop(playBackTimeOutTaskKey);
requestMessage.setData(WVPResult.fail(ErrorCode.ERROR100.getCode(), String.format("回放失败， 错误码： %s, %s", event.statusCode, event.msg)));
playBackResult.setCode(ErrorCode.ERROR100.getCode());
playBackResult.setMsg(String.format("回放失败， 错误码： %s, %s", event.statusCode, event.msg));
playBackResult.setData(requestMessage);
playBackResult.setEvent(event);
playBackCallback.call(playBackResult);
streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
});
}, errorEvent);
return result;
}

1
src/main/java/com/genersoft/iot/vmp/service/impl/RedisPushStreamListMsgListener.java
View File

@ -53,7 +53,6 @@ public class RedisPushStreamListMsgListener implements MessageListener {
boolean contains = allAppAndStream.contains(app + stream);
//不存在就添加
if (!contains) {
streamPushItem.setStatus(false);
streamPushItem.setStreamType("push");
streamPushItem.setCreateTime(DateUtil.getNow());
streamPushItem.setMediaServerId(mediaServerService.getDefaultMediaServer().getId());

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

@ -116,7 +116,7 @@ public class StreamPushUploadFileHandler extends AnalysisEventListener<StreamPus
streamPushItem.setApp(streamPushExcelDto.getApp());
streamPushItem.setStream(streamPushExcelDto.getStream());
streamPushItem.setGbId(streamPushExcelDto.getGbId());
streamPushItem.setStatus(false);
streamPushItem.setStatus(streamPushExcelDto.getStatus());
streamPushItem.setStreamType("push");
streamPushItem.setCreateTime(DateUtil.getNow());
streamPushItem.setMediaServerId(defaultMediaServerId);

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

@ -236,4 +236,6 @@ public interface IRedisCatchStorage {
void sendStreamPushRequestedMsgForStatus();
List<SendRtpItem> querySendRTPServerByChnnelId(String channelId);
List<SendRtpItem> querySendRTPServerByStream(String stream);
}

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

@ -143,15 +143,12 @@ public interface DeviceChannelMapper {
@Update(value = {"UPDATE device_channel SET status=0 WHERE deviceId=#{deviceId}"})
void offlineByDeviceId(String deviceId);
@Update(value = {"UPDATE device_channel SET status=1 WHERE deviceId=#{deviceId} AND channelId=#{channelId}"})
void online(String deviceId, String channelId);
@Insert("<script> " +
"insert into device_channel " +
"(channelId, deviceId, name, manufacture, model, owner, civilCode, block, subCount, " +
" address, parental, parentId, safetyWay, registerWay, certNum, certifiable, errCode, secrecy, " +
" ipAddress, port, password, PTZType, status, streamId, longitude, latitude, longitudeGcj02, latitudeGcj02, " +
" longitudeWgs84, latitudeWgs84, createTime, updateTime, businessGroupId, gpsTime) " +
" longitudeWgs84, latitudeWgs84, hasAudio, createTime, updateTime, businessGroupId, gpsTime) " +
"values " +
"<foreach collection='addChannels' index='index' item='item' separator=','> " +
"('${item.channelId}', '${item.deviceId}', '${item.name}', '${item.manufacture}', '${item.model}', " +
@ -160,7 +157,7 @@ public interface DeviceChannelMapper {
"'${item.certNum}', ${item.certifiable}, ${item.errCode}, '${item.secrecy}', " +
"'${item.ipAddress}', ${item.port}, '${item.password}', ${item.PTZType}, ${item.status}, " +
"'${item.streamId}', ${item.longitude}, ${item.latitude},${item.longitudeGcj02}, " +
"${item.latitudeGcj02},${item.longitudeWgs84}, ${item.latitudeWgs84},'${item.createTime}', '${item.updateTime}', " +
"${item.latitudeGcj02},${item.longitudeWgs84}, ${item.latitudeWgs84}, ${item.hasAudio},'${item.createTime}', '${item.updateTime}', " +
"'${item.businessGroupId}', '${item.gpsTime}') " +
"</foreach> " +
"ON DUPLICATE KEY UPDATE " +
@ -193,11 +190,15 @@ public interface DeviceChannelMapper {
"latitudeGcj02=VALUES(latitudeGcj02), " +
"longitudeWgs84=VALUES(longitudeWgs84), " +
"latitudeWgs84=VALUES(latitudeWgs84), " +
"hasAudio=VALUES(hasAudio), " +
"businessGroupId=VALUES(businessGroupId), " +
"gpsTime=VALUES(gpsTime)" +
"</script>")
int batchAdd(List<DeviceChannel> addChannels);
@Update(value = {"UPDATE device_channel SET status=1 WHERE deviceId=#{deviceId} AND channelId=#{channelId}"})
void online(String deviceId, String channelId);
@Update({"<script>" +
"<foreach collection='updateChannels' item='item' separator=';'>" +
" UPDATE" +
@ -341,4 +342,7 @@ public interface DeviceChannelMapper {
" left join platform_catalog pc on pgc.catalogId = pc.id and pgc.platformId = pc.platformId" +
" where pgc.platformId=#{serverGBId}")
List<DeviceChannel> queryChannelWithCatalog(String serverGBId);
@Select("select * from device_channel where deviceId = #{deviceId}")
List<DeviceChannel> queryAllChannels(String deviceId);
}

4
src/main/java/com/genersoft/iot/vmp/storager/dao/PlatformGbStreamMapper.java
View File

@ -23,10 +23,10 @@ public interface PlatformGbStreamMapper {
@Insert("<script> " +
"INSERT into platform_gb_stream " +
"(gbStreamId, platformId, catalogId) " +
"(gbStreamId, platformId, catalogId,status) " +
"values " +
"<foreach collection='streamPushItems' index='index' item='item' separator=','> " +
"(${item.gbStreamId}, '${item.platformId}', '${item.catalogId}')" +
"(${item.gbStreamId}, '${item.platformId}', '${item.catalogId}'), '${item.status}')" +
"</foreach> " +
"</script>")
int batchAdd(List<StreamPushItem> streamPushItems);

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

@ -77,7 +77,7 @@ public interface StreamPushMapper {
"1=1 " +
" <if test='query != null'> AND (st.app LIKE '%${query}%' OR st.stream LIKE '%${query}%' OR gs.gbId LIKE '%${query}%' OR gs.name LIKE '%${query}%')</if> " +
" <if test='pushing == true' > AND (gs.gbId is null OR st.pushIng=1)</if>" +
" <if test='pushing == false' > AND st.pushIng=0</if>" +
" <if test='pushing == false' > AND (gs.pushIng is null OR st.pushIng=0) </if>" +
" <if test='mediaServerId != null' > AND st.mediaServerId=#{mediaServerId} </if>" +
"order by st.createTime desc" +
" </script>"})

18
src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java
View File

@ -387,6 +387,24 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
return result;
}
@Override
public List<SendRtpItem> querySendRTPServerByStream(String stream) {
if (stream == null) {
return null;
}
String platformGbId = "*";
String callId = "*";
String channelId = "*";
String key = VideoManagerConstants.PLATFORM_SEND_RTP_INFO_PREFIX + userSetting.getServerId() + "_" + platformGbId
+ "_" + channelId + "_" + stream + "_" + callId;
List<Object> scan = RedisUtil.scan(key);
List<SendRtpItem> result = new ArrayList<>();
for (Object o : scan) {
result.add((SendRtpItem) RedisUtil.get((String) o));
}
return result;
}
@Override
public List<SendRtpItem> querySendRTPServer(String platformGbId) {
if (platformGbId == null) {

20
src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java
View File

@ -111,11 +111,11 @@ public class VideoManagerStorageImpl implements IVideoManagerStorage {
if (CollectionUtils.isEmpty(deviceChannelList)) {
return false;
}
List<DeviceChannel> allChannelInPlay = deviceChannelMapper.getAllChannelInPlay();
Map<String,DeviceChannel> allChannelMapInPlay = new ConcurrentHashMap<>();
if (allChannelInPlay.size() > 0) {
for (DeviceChannel deviceChannel : allChannelInPlay) {
allChannelMapInPlay.put(deviceChannel.getChannelId(), deviceChannel);
List<DeviceChannel> allChannels = deviceChannelMapper.queryAllChannels(deviceId);
Map<String,DeviceChannel> allChannelMap = new ConcurrentHashMap<>();
if (allChannels.size() > 0) {
for (DeviceChannel deviceChannel : allChannels) {
allChannelMap.put(deviceChannel.getChannelId(), deviceChannel);
}
}
TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
@ -123,15 +123,17 @@ public class VideoManagerStorageImpl implements IVideoManagerStorage {
List<DeviceChannel> channels = new ArrayList<>();
StringBuilder stringBuilder = new StringBuilder();
Map<String, Integer> subContMap = new HashMap<>();
if (deviceChannelList.size() > 1) {
if (deviceChannelList.size() > 0) {
// 数据去重
Set<String> gbIdSet = new HashSet<>();
for (DeviceChannel deviceChannel : deviceChannelList) {
if (!gbIdSet.contains(deviceChannel.getChannelId())) {
gbIdSet.add(deviceChannel.getChannelId());
if (allChannelMapInPlay.containsKey(deviceChannel.getChannelId())) {
deviceChannel.setStreamId(allChannelMapInPlay.get(deviceChannel.getChannelId()).getStreamId());
if (allChannelMap.containsKey(deviceChannel.getChannelId())) {
deviceChannel.setStreamId(allChannelMap.get(deviceChannel.getChannelId()).getStreamId());
deviceChannel.setHasAudio(allChannelMap.get(deviceChannel.getChannelId()).isHasAudio());
}
channels.add(deviceChannel);
if (!ObjectUtils.isEmpty(deviceChannel.getParentId())) {
if (subContMap.get(deviceChannel.getParentId()) == null) {
@ -153,8 +155,6 @@ public class VideoManagerStorageImpl implements IVideoManagerStorage {
}
}
}else {
channels = deviceChannelList;
}
if (stringBuilder.length() > 0) {
logger.info("[目录查询]收到的数据存在重复： {}" , stringBuilder);

15
src/main/java/com/genersoft/iot/vmp/vmanager/bean/StreamPushExcelDto.java
View File

@ -22,6 +22,9 @@ public class StreamPushExcelDto {
@ExcelProperty("目录ID")
private String catalogId;
@ExcelProperty("在线状态")
private boolean status;
public String getName() {
return name;
}
@ -70,4 +73,16 @@ public class StreamPushExcelDto {
public void setCatalogId(String catalogId) {
this.catalogId = catalogId;
}
public boolean isStatus() {
return status;
}
public boolean getStatus() {
return status;
}
public void setStatus(boolean status) {
this.status = status;
}
}

1
src/main/java/com/genersoft/iot/vmp/vmanager/user/UserController.java
View File

@ -43,6 +43,7 @@ public class UserController {
private IRoleService roleService;
@GetMapping("/login")
@PostMapping("/login")
@Operation(summary = "登录")
@Parameter(name = "username", description = "用户名", required = true)
@Parameter(name = "password", description = "密码（32位md5加密）", required = true)

5
src/main/resources/logback-spring-local.xml
View File

@ -98,6 +98,11 @@
<appender-ref ref="STDOUT" />
</root>
<logger name="wvp" level="debug" additivity="true">
<appender-ref ref="RollingFileError"/>
<appender-ref ref="RollingFile"/>
</logger>
<logger name="GB28181_SIP" level="debug" additivity="true">
<appender-ref ref="RollingFileError"/>
<appender-ref ref="sipRollingFile"/>

3
web_src/src/components/ParentPlatformList.vue
View File

@ -143,7 +143,8 @@ export default {
});
},
chooseChannel: function(platform) {
this.$refs.chooseChannelDialog.openDialog(platform.serverGBId, platform.name, platform.catalogId, platform.treeType, this.initData)
console.log("platform.name: " + platform.name)
this.$refs.chooseChannelDialog.openDialog(platform.serverGBId,platform.deviceGBId, platform.name, platform.catalogId, platform.treeType, this.initData)
},
initData: function() {
this.getPlatformList();

7
web_src/src/components/dialog/chooseChannel.vue
View File

@ -8,7 +8,7 @@
<el-tab-pane label="目录结构" name="catalog">
<el-container>
<el-main v-bind:style="{backgroundColor: '#FFF', maxHeight: winHeight + 'px'}">
<chooseChannelForCatalog ref="chooseChannelForCatalog" :platformId=platformId :platformName=platformName :defaultCatalogId=defaultCatalogId :catalogIdChange="catalogIdChange" :treeType=treeType ></chooseChannelForCatalog>
<chooseChannelForCatalog ref="chooseChannelForCatalog" :platformId=platformId :platformDeviceId=platformDeviceId :platformName=platformName :defaultCatalogId=defaultCatalogId :catalogIdChange="catalogIdChange" :treeType=treeType ></chooseChannelForCatalog>
</el-main>
</el-container>
</el-tab-pane>
@ -60,6 +60,7 @@ export default {
tabActiveName: "gbChannel",
catalogTabActiveName: "catalog",
platformId: "",
platformDeviceId: "",
catalogId: "",
catalogName: "",
currentCatalogId: "",
@ -73,8 +74,10 @@ export default {
};
},
methods: {
openDialog(platformId, platformName, defaultCatalogId, treeType, closeCallback) {
openDialog(platformId, platformDeviceId, platformName, defaultCatalogId, treeType, closeCallback) {
console.log("defaultCatalogId: " + defaultCatalogId)
this.platformId = platformId
this.platformDeviceId = platformDeviceId
this.platformName = platformName
this.defaultCatalogId = defaultCatalogId
this.showDialog = true

7
web_src/src/components/dialog/chooseChannelForCatalog.vue
View File

@ -38,7 +38,7 @@
import catalogEdit from './catalogEdit.vue'
export default {
name: 'chooseChannelForCatalog',
props: ['platformId', 'platformName', 'defaultCatalogId', 'catalogIdChange', 'treeType'],
props: ['platformId', 'platformDeviceId', 'platformName', 'defaultCatalogId', 'catalogIdChange', 'treeType'],
created() {
this.chooseId = this.defaultCatalogId;
this.defaultCatalogIdSign = this.defaultCatalogId;
@ -171,6 +171,7 @@ export default {
});
},
loadNode: function(node, resolve){
console.log("this.platformDeviceId： " + this.platformDeviceId)
if (node.level === 0) {
resolve([
{
@ -179,7 +180,7 @@ export default {
type: -1
},{
name: this.platformName,
id: this.platformId,
id: this.platformDeviceId,
type: 0
}
]);
@ -298,6 +299,8 @@ export default {
return false;
},
nodeClickHandler: function (data, node, tree){
console.log(data)
console.log(node)
this.chooseId = data.id;
this.chooseName = data.name;
if (this.catalogIdChange)this.catalogIdChange(this.chooseId, this.chooseName);

16
web_src/src/components/dialog/getCatalog.vue
View File

@ -77,6 +77,7 @@ export default {
},
methods: {
openDialog(catalogIdResult) {
console.log(this.chooseId)
this.showDialog = true
this.catalogIdResult = catalogIdResult
},
@ -107,9 +108,6 @@ export default {
},
loadNode: function(node, resolve){
if (node.level === 0) {
this.$axios({
method:"get",
@ -124,7 +122,7 @@ export default {
resolve([
{
name: this.platformName,
id: this.platformId,
id: res.data.data.deviceGBId,
type: 0
}
]);
@ -142,9 +140,19 @@ export default {
this.chooseId = data.id;
},
close: function() {
this.chooseId = null;
this.showDialog = false;
},
submit: function() {
console.log(this.chooseId)
if (this.chooseId === null) {
this.$message({
showClose: true,
message: '未选择任何节点,',
type: 'warning'
});
return;
}
if (this.catalogIdResult)this.catalogIdResult(this.chooseId)
this.showDialog = false;
},

41
web_src/src/components/dialog/platformEdit.vue
View File

@ -37,13 +37,13 @@
<el-form-item label="本地端口" prop="devicePort">
<el-input v-model="platform.devicePort" :disabled="true" type="number"></el-input>
</el-form-item>
<el-form-item label="SIP认证用户名" prop="username">
<el-input v-model="platform.username"></el-input>
</el-form-item>
</el-form>
</el-col>
<el-col :span="12">
<el-form ref="platform2" :rules="rules" :model="platform" label-width="160px">
<el-form-item label="SIP认证用户名" prop="username">
<el-input v-model="platform.username"></el-input>
</el-form-item>
<el-form-item label="行政区划" prop="administrativeDivision">
<el-input v-model="platform.administrativeDivision" clearable></el-input>
</el-form-item>
@ -79,7 +79,7 @@
</el-select>
</el-form-item>
<el-form-item label="目录结构" prop="treeType" >
<el-select v-model="platform.treeType" style="width: 100%" >
<el-select v-model="platform.treeType" style="width: 100%" @change="treeTypeChange">
<el-option key="WGS84" label="行政区划" value="CivilCode"></el-option>
<el-option key="GCJ02" label="业务分组" value="BusinessGroup"></el-option>
</el-select>
@ -98,6 +98,7 @@
<el-checkbox label="启用" v-model="platform.enable" @change="checkExpires"></el-checkbox>
<el-checkbox label="云台控制" v-model="platform.ptz"></el-checkbox>
<el-checkbox label="拉起离线推流" v-model="platform.startOfflinePush"></el-checkbox>
<el-checkbox label="RTCP保活" v-model="platform.rtcp" @change="rtcpCheckBoxChange"></el-checkbox>
</el-form-item>
<el-form-item>
<el-button type="primary" @click="onSubmit">{{
@ -251,21 +252,7 @@ export default {
},
onSubmit: function () {
if (this.onSubmit_text === "保存") {
this.$confirm("修改目录结构会导致关联目录与通道数据被清空", '提示', {
dangerouslyUseHTMLString: true,
confirmButtonText: '确定',
cancelButtonText: '取消',
center: true,
type: 'warning'
}).then(() => {
this.saveForm()
}).catch(() => {
});
}else {
this.saveForm()
}
this.saveForm()
},
saveForm: function (){
this.$axios({
@ -343,6 +330,22 @@ export default {
if (this.platform.enable && this.platform.expires == "0") {
this.platform.expires = "300";
}
},
rtcpCheckBoxChange: function (result){
if (result) {
this.$message({
showClose: true,
message: "开启RTCP保活需要上级平台支持，可以避免无效推流",
type: "warning",
});
}
},
treeTypeChange: function (){
this.$message({
showClose: true,
message: "修改目录结构会导致关联目录与通道数据被清空，保存后生效",
type: "warning",
});
}
},
};

BIN
web_src/static/file/推流通道导入.zip
View File

Binary file not shown.