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

Merge pull request #845 from 648540858/wvp-28181-2.0-test

Browse Source 
Wvp 28181 2.0 test
This commit is contained in:
648540858 2023-05-09 17:56:29 +08:00 committed by GitHub
commit ccecda7859
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
37 changed files with 1453 additions and 1201 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

126
src/main/java/com/genersoft/iot/vmp/common/InviteInfo.java Normal file
View File

@ -0,0 +1,126 @@
package com.genersoft.iot.vmp.common;
import com.genersoft.iot.vmp.service.bean.SSRCInfo;
/**
* 记录每次发送invite消息的状态
*/
public class InviteInfo {
private String deviceId;
private String channelId;
private String stream;
private SSRCInfo ssrcInfo;
private String receiveIp;
private Integer receivePort;
private String streamMode;
private InviteSessionType type;
private InviteSessionStatus status;
private StreamInfo streamInfo;
public static InviteInfo getinviteInfo(String deviceId, String channelId, String stream, SSRCInfo ssrcInfo,
String receiveIp, Integer receivePort, String streamMode,
InviteSessionType type, InviteSessionStatus status) {
InviteInfo inviteInfo = new InviteInfo();
inviteInfo.setDeviceId(deviceId);
inviteInfo.setChannelId(channelId);
inviteInfo.setStream(stream);
inviteInfo.setSsrcInfo(ssrcInfo);
inviteInfo.setReceiveIp(receiveIp);
inviteInfo.setReceivePort(receivePort);
inviteInfo.setStreamMode(streamMode);
inviteInfo.setType(type);
inviteInfo.setStatus(status);
return inviteInfo;
}
public String getDeviceId() {
return deviceId;
}
public void setDeviceId(String deviceId) {
this.deviceId = deviceId;
}
public String getChannelId() {
return channelId;
}
public void setChannelId(String channelId) {
this.channelId = channelId;
}
public InviteSessionType getType() {
return type;
}
public void setType(InviteSessionType type) {
this.type = type;
}
public InviteSessionStatus getStatus() {
return status;
}
public void setStatus(InviteSessionStatus status) {
this.status = status;
}
public StreamInfo getStreamInfo() {
return streamInfo;
}
public void setStreamInfo(StreamInfo streamInfo) {
this.streamInfo = streamInfo;
}
public String getStream() {
return stream;
}
public void setStream(String stream) {
this.stream = stream;
}
public SSRCInfo getSsrcInfo() {
return ssrcInfo;
}
public void setSsrcInfo(SSRCInfo ssrcInfo) {
this.ssrcInfo = ssrcInfo;
}
public String getReceiveIp() {
return receiveIp;
}
public void setReceiveIp(String receiveIp) {
this.receiveIp = receiveIp;
}
public Integer getReceivePort() {
return receivePort;
}
public void setReceivePort(Integer receivePort) {
this.receivePort = receivePort;
}
public String getStreamMode() {
return streamMode;
}
public void setStreamMode(String streamMode) {
this.streamMode = streamMode;
}
}

11
src/main/java/com/genersoft/iot/vmp/common/InviteSessionStatus.java Normal file
View File

@ -0,0 +1,11 @@
package com.genersoft.iot.vmp.common;
/**
* 标识invite消息发出后的各个状态，
* 收到ok钱停止invite发送cancel，
* 收到200ok后发送BYE停止invite
*/
public enum InviteSessionStatus {
ready,
ok,
}

7
src/main/java/com/genersoft/iot/vmp/common/InviteSessionType.java Normal file
View File

@ -0,0 +1,7 @@
package com.genersoft.iot.vmp.common;
public enum InviteSessionType {
PLAY,
PLAYBACK,
DOWNLOAD
}

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

@ -16,8 +16,6 @@ public class VideoManagerConstants {
public static final String MEDIA_SERVERS_ONLINE_PREFIX = "VMP_MEDIA_ONLINE_SERVERS_"; public static final String MEDIA_SERVERS_ONLINE_PREFIX = "VMP_MEDIA_ONLINE_SERVERS_";
public static final String MEDIA_STREAM_PREFIX = "VMP_MEDIA_STREAM";
public static final String DEVICE_PREFIX = "VMP_DEVICE_"; public static final String DEVICE_PREFIX = "VMP_DEVICE_";
// 设备同步完成 // 设备同步完成
@ -28,9 +26,10 @@ public class VideoManagerConstants {
public static final String KEEPLIVEKEY_PREFIX = "VMP_KEEPALIVE_"; public static final String KEEPLIVEKEY_PREFIX = "VMP_KEEPALIVE_";
// TODO 此处多了一个_，暂不修改 // TODO 此处多了一个_，暂不修改
public static final String PLAYER_PREFIX = "VMP_PLAYER_"; public static final String INVITE_PREFIX = "VMP_INVITE";
public static final String PLAY_BLACK_PREFIX = "VMP_PLAYBACK_"; public static final String PLAYER_PREFIX = "VMP_INVITE_PLAY_";
public static final String DOWNLOAD_PREFIX = "VMP_DOWNLOAD_"; public static final String PLAY_BLACK_PREFIX = "VMP_INVITE_PLAYBACK_";
public static final String DOWNLOAD_PREFIX = "VMP_INVITE_DOWNLOAD_";
public static final String PLATFORM_KEEPALIVE_PREFIX = "VMP_PLATFORM_KEEPALIVE_"; public static final String PLATFORM_KEEPALIVE_PREFIX = "VMP_PLATFORM_KEEPALIVE_";

11
src/main/java/com/genersoft/iot/vmp/conf/ProxyServletConfig.java
View File

@ -2,7 +2,6 @@ package com.genersoft.iot.vmp.conf;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.service.IMediaServerService;
import org.apache.catalina.connector.ClientAbortException;
import org.apache.http.HttpHost; import org.apache.http.HttpHost;
import org.apache.http.HttpRequest; import org.apache.http.HttpRequest;
import org.apache.http.HttpResponse; import org.apache.http.HttpResponse;
@ -194,11 +193,11 @@ public class ProxyServletConfig {
} catch (IOException ioException) { } catch (IOException ioException) {
if (ioException instanceof ConnectException) { if (ioException instanceof ConnectException) {
logger.error("录像服务 连接失败"); logger.error("录像服务 连接失败");
}else if (ioException instanceof ClientAbortException) { // }else if (ioException instanceof ClientAbortException) {
/** // /**
* TODO 使用这个代理库实现代理在遇到代理视频文件时，如果是206结果，会遇到报错蛋市目前功能正常， // * TODO 使用这个代理库实现代理在遇到代理视频文件时，如果是206结果，会遇到报错蛋市目前功能正常，
* TODO 暂时去除异常处理。后续使用其他代理框架修改测试 // * TODO 暂时去除异常处理。后续使用其他代理框架修改测试
*/ // */
}else { }else {
logger.error("录像服务 代理失败： ", e); logger.error("录像服务 代理失败： ", e);

2
src/main/java/com/genersoft/iot/vmp/gb28181/bean/InviteStreamType.java
View File

@ -2,7 +2,7 @@ package com.genersoft.iot.vmp.gb28181.bean;
public enum InviteStreamType { public enum InviteStreamType {
PLAY,PLAYBACK,PUSH,PROXY,CLOUD_RECORD_PUSH,CLOUD_RECORD_PROXY PLAY,PLAYBACK,DOWNLOAD,PUSH,PROXY,CLOUD_RECORD_PUSH,CLOUD_RECORD_PROXY
} }

8
src/main/java/com/genersoft/iot/vmp/gb28181/bean/SsrcTransaction.java
View File

@ -1,6 +1,6 @@
package com.genersoft.iot.vmp.gb28181.bean; package com.genersoft.iot.vmp.gb28181.bean;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager; import com.genersoft.iot.vmp.common.InviteSessionType;
public class SsrcTransaction { public class SsrcTransaction {
@ -13,7 +13,7 @@ public class SsrcTransaction {
private SipTransactionInfo sipTransactionInfo; private SipTransactionInfo sipTransactionInfo;
private VideoStreamSessionManager.SessionType type; private InviteSessionType type;
public String getDeviceId() { public String getDeviceId() {
return deviceId; return deviceId;
@ -63,11 +63,11 @@ public class SsrcTransaction {
this.ssrc = ssrc; this.ssrc = ssrc;
} }
public VideoStreamSessionManager.SessionType getType() { public InviteSessionType getType() {
return type; return type;
} }
public void setType(VideoStreamSessionManager.SessionType type) { public void setType(InviteSessionType type) {
this.type = type; this.type = type;
} }

9
src/main/java/com/genersoft/iot/vmp/gb28181/session/VideoStreamSessionManager.java
View File

@ -1,5 +1,6 @@
package com.genersoft.iot.vmp.gb28181.session; package com.genersoft.iot.vmp.gb28181.session;
import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.common.VideoManagerConstants; import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.SipTransactionInfo; import com.genersoft.iot.vmp.gb28181.bean.SipTransactionInfo;
@ -27,12 +28,6 @@ public class VideoStreamSessionManager {
@Autowired @Autowired
private RedisTemplate<Object, Object> redisTemplate; private RedisTemplate<Object, Object> redisTemplate;
public enum SessionType {
play,
playback,
download
}
/** /**
* 添加一个点播/回放的事务信息 * 添加一个点播/回放的事务信息
* 后续可以通过流Id/callID * 后续可以通过流Id/callID
@ -43,7 +38,7 @@ public class VideoStreamSessionManager {
* @param mediaServerId 所使用的流媒体ID * @param mediaServerId 所使用的流媒体ID
* @param response 回复 * @param response 回复
*/ */
public void put(String deviceId, String channelId, String callId, String stream, String ssrc, String mediaServerId, SIPResponse response, SessionType type){ public void put(String deviceId, String channelId, String callId, String stream, String ssrc, String mediaServerId, SIPResponse response, InviteSessionType type){
SsrcTransaction ssrcTransaction = new SsrcTransaction(); SsrcTransaction ssrcTransaction = new SsrcTransaction();
ssrcTransaction.setDeviceId(deviceId); ssrcTransaction.setDeviceId(deviceId);
ssrcTransaction.setChannelId(channelId); ssrcTransaction.setChannelId(channelId);

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

@ -4,7 +4,6 @@ import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException; import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
import com.genersoft.iot.vmp.gb28181.bean.Device; import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceAlarm; import com.genersoft.iot.vmp.gb28181.bean.DeviceAlarm;
import com.genersoft.iot.vmp.gb28181.bean.InviteStreamCallback;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe; import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe; import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
@ -109,7 +108,7 @@ public interface ISIPCommander {
* @param startTime 开始时间,格式要求：yyyy-MM-dd HH:mm:ss * @param startTime 开始时间,格式要求：yyyy-MM-dd HH:mm:ss
* @param endTime 结束时间,格式要求：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 okEvent, SipSubscribe.Event errorEvent) throws InvalidArgumentException, SipException, ParseException; void playbackStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInf, Device device, String channelId, String startTime, String endTime,ZlmHttpHookSubscribe.Event event, SipSubscribe.Event okEvent, SipSubscribe.Event errorEvent) throws InvalidArgumentException, SipException, ParseException;
/** /**
* 请求历史媒体下载 * 请求历史媒体下载
@ -121,7 +120,7 @@ public interface ISIPCommander {
* @param downloadSpeed 下载倍速参数 * @param downloadSpeed 下载倍速参数
*/ */
void downloadStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, void downloadStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId,
String startTime, String endTime, int downloadSpeed, InviteStreamCallback inviteStreamCallback, InviteStreamCallback hookEvent, String startTime, String endTime, int downloadSpeed, ZlmHttpHookSubscribe.Event hookEvent,
SipSubscribe.Event errorEvent,SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException; SipSubscribe.Event errorEvent,SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException;
/** /**

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

@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.gb28181.transmit.cmd.impl; package com.genersoft.iot.vmp.gb28181.transmit.cmd.impl;
import com.alibaba.fastjson2.JSONObject; import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.SipConfig; import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
@ -350,7 +351,7 @@ public class SIPCommander implements ISIPCommander {
// 这里为例避免一个通道的点播只有一个callID这个参数使用一个固定值 // 这里为例避免一个通道的点播只有一个callID这个参数使用一个固定值
ResponseEvent responseEvent = (ResponseEvent) e.event; ResponseEvent responseEvent = (ResponseEvent) e.event;
SIPResponse response = (SIPResponse) responseEvent.getResponse(); SIPResponse response = (SIPResponse) responseEvent.getResponse();
streamSession.put(device.getDeviceId(), channelId, "play", stream, ssrcInfo.getSsrc(), mediaServerItem.getId(), response, VideoStreamSessionManager.SessionType.play); streamSession.put(device.getDeviceId(), channelId, "play", stream, ssrcInfo.getSsrc(), mediaServerItem.getId(), response, InviteSessionType.PLAY);
okEvent.response(e); okEvent.response(e);
}); });
} }
@ -365,11 +366,11 @@ public class SIPCommander implements ISIPCommander {
*/ */
@Override @Override
public void playbackStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, public void playbackStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId,
String startTime, String endTime, InviteStreamCallback inviteStreamCallback, InviteStreamCallback hookEvent, String startTime, String endTime, ZlmHttpHookSubscribe.Event hookEvent,
SipSubscribe.Event okEvent, SipSubscribe.Event errorEvent) throws InvalidArgumentException, SipException, ParseException { SipSubscribe.Event okEvent, SipSubscribe.Event errorEvent) throws InvalidArgumentException, SipException, ParseException {
logger.info("{} 分配的ZLM为: {} [{}:{}]", ssrcInfo.getStream(), mediaServerItem.getSdpIp(), mediaServerItem.getIp(), ssrcInfo.getPort()); logger.info("{} 分配的ZLM为: {} [{}:{}]", ssrcInfo.getStream(), mediaServerItem.getId(), mediaServerItem.getSdpIp(), ssrcInfo.getPort());
String sdpIp; String sdpIp;
if (!ObjectUtils.isEmpty(device.getSdpIp())) { if (!ObjectUtils.isEmpty(device.getSdpIp())) {
sdpIp = device.getSdpIp(); sdpIp = device.getSdpIp();
@ -442,8 +443,7 @@ public class SIPCommander implements ISIPCommander {
// 添加订阅 // 添加订阅
subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject json) -> { subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject json) -> {
if (hookEvent != null) { if (hookEvent != null) {
InviteStreamInfo inviteStreamInfo = new InviteStreamInfo(mediaServerItemInUse, json,sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport()).getCallId(), "rtp", ssrcInfo.getStream()); hookEvent.response(mediaServerItemInUse, json);
hookEvent.call(inviteStreamInfo);
} }
subscribe.removeSubscribe(hookSubscribe); subscribe.removeSubscribe(hookSubscribe);
}); });
@ -452,12 +452,9 @@ public class SIPCommander implements ISIPCommander {
sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), request, errorEvent, event -> { sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), request, errorEvent, event -> {
ResponseEvent responseEvent = (ResponseEvent) event.event; ResponseEvent responseEvent = (ResponseEvent) event.event;
SIPResponse response = (SIPResponse) responseEvent.getResponse(); SIPResponse response = (SIPResponse) responseEvent.getResponse();
streamSession.put(device.getDeviceId(), channelId,sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport()).getCallId(), ssrcInfo.getStream(), ssrcInfo.getSsrc(), mediaServerItem.getId(), response, VideoStreamSessionManager.SessionType.playback); streamSession.put(device.getDeviceId(), channelId,sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport()).getCallId(), ssrcInfo.getStream(), ssrcInfo.getSsrc(), mediaServerItem.getId(), response, InviteSessionType.PLAYBACK);
okEvent.response(event); okEvent.response(event);
}); });
if (inviteStreamCallback != null) {
inviteStreamCallback.call(new InviteStreamInfo(mediaServerItem, null,sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport()).getCallId(), "rtp", ssrcInfo.getStream()));
}
} }
/** /**
@ -472,10 +469,10 @@ public class SIPCommander implements ISIPCommander {
@Override @Override
public void downloadStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, public void downloadStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId,
String startTime, String endTime, int downloadSpeed, String startTime, String endTime, int downloadSpeed,
InviteStreamCallback inviteStreamCallback, InviteStreamCallback hookEvent, ZlmHttpHookSubscribe.Event hookEvent,
SipSubscribe.Event errorEvent,SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException { SipSubscribe.Event errorEvent,SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException {
logger.info("{} 分配的ZLM为: {} [{}:{}]", ssrcInfo.getStream(), mediaServerItem.getSdpIp(), mediaServerItem.getIp(), ssrcInfo.getPort()); logger.info("{} 分配的ZLM为: {} [{}:{}]", ssrcInfo.getStream(), mediaServerItem.getId(), mediaServerItem.getSdpIp(), ssrcInfo.getPort());
String sdpIp; String sdpIp;
if (!ObjectUtils.isEmpty(device.getSdpIp())) { if (!ObjectUtils.isEmpty(device.getSdpIp())) {
sdpIp = device.getSdpIp(); sdpIp = device.getSdpIp();
@ -543,13 +540,13 @@ public class SIPCommander implements ISIPCommander {
content.append("y=" + ssrcInfo.getSsrc() + "\r\n");//ssrc content.append("y=" + ssrcInfo.getSsrc() + "\r\n");//ssrc
logger.debug("此时请求下载信令的ssrc===>{}",ssrcInfo.getSsrc()); logger.debug("此时请求下载信令的ssrc===>{}",ssrcInfo.getSsrc());
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, null, mediaServerItem.getId()); HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId());
// 添加订阅 // 添加订阅
CallIdHeader newCallIdHeader = sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()), device.getTransport()); CallIdHeader newCallIdHeader = sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()), device.getTransport());
String callId= newCallIdHeader.getCallId(); String callId= newCallIdHeader.getCallId();
subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject json) -> { subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject json) -> {
logger.debug("sipc 添加订阅===callId {}",callId); logger.debug("sipc 添加订阅===callId {}",callId);
hookEvent.call(new InviteStreamInfo(mediaServerItem, json,callId, "rtp", ssrcInfo.getStream())); hookEvent.response(mediaServerItemInUse, json);
subscribe.removeSubscribe(hookSubscribe); subscribe.removeSubscribe(hookSubscribe);
hookSubscribe.getContent().put("regist", false); hookSubscribe.getContent().put("regist", false);
hookSubscribe.getContent().put("schema", "rtsp"); hookSubscribe.getContent().put("schema", "rtsp");
@ -567,9 +564,6 @@ public class SIPCommander implements ISIPCommander {
}); });
Request request = headerProvider.createPlaybackInviteRequest(device, channelId, content.toString(), null, SipUtils.getNewFromTag(), null,newCallIdHeader, ssrcInfo.getSsrc()); Request request = headerProvider.createPlaybackInviteRequest(device, channelId, content.toString(), null, SipUtils.getNewFromTag(), null,newCallIdHeader, ssrcInfo.getSsrc());
if (inviteStreamCallback != null) {
inviteStreamCallback.call(new InviteStreamInfo(mediaServerItem, null,callId, "rtp", ssrcInfo.getStream()));
}
sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), request, errorEvent, event -> { sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), request, errorEvent, event -> {
ResponseEvent responseEvent = (ResponseEvent) event.event; ResponseEvent responseEvent = (ResponseEvent) event.event;
@ -580,7 +574,7 @@ public class SIPCommander implements ISIPCommander {
if (ssrcIndex >= 0) { if (ssrcIndex >= 0) {
ssrc = contentString.substring(ssrcIndex + 2, ssrcIndex + 12); ssrc = contentString.substring(ssrcIndex + 2, ssrcIndex + 12);
} }
streamSession.put(device.getDeviceId(), channelId, response.getCallIdHeader().getCallId(), ssrcInfo.getStream(), ssrc, mediaServerItem.getId(), response, VideoStreamSessionManager.SessionType.download); streamSession.put(device.getDeviceId(), channelId, response.getCallIdHeader().getCallId(), ssrcInfo.getStream(), ssrc, mediaServerItem.getId(), response, InviteSessionType.DOWNLOAD);
okEvent.response(event); okEvent.response(event);
}); });
} }

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

@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl; package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;
import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.common.InviteInfo;
import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException; import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
import com.genersoft.iot.vmp.gb28181.bean.Device; import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.InviteStreamType; import com.genersoft.iot.vmp.gb28181.bean.InviteStreamType;
@ -15,6 +16,7 @@ import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorP
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory; import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IDeviceService; import com.genersoft.iot.vmp.service.IDeviceService;
import com.genersoft.iot.vmp.service.IInviteStreamService;
import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.bean.MessageForPushChannel; import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
@ -26,7 +28,9 @@ import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import javax.sip.*; import javax.sip.InvalidArgumentException;
import javax.sip.RequestEvent;
import javax.sip.SipException;
import javax.sip.address.SipURI; import javax.sip.address.SipURI;
import javax.sip.header.CallIdHeader; import javax.sip.header.CallIdHeader;
import javax.sip.header.FromHeader; import javax.sip.header.FromHeader;
@ -52,6 +56,9 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
@Autowired @Autowired
private IRedisCatchStorage redisCatchStorage; private IRedisCatchStorage redisCatchStorage;
@Autowired
private IInviteStreamService inviteStreamService;
@Autowired @Autowired
private IDeviceService deviceService; private IDeviceService deviceService;
@ -136,11 +143,6 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
Device device = storager.queryVideoDeviceByChannelId(platformGbId); Device device = storager.queryVideoDeviceByChannelId(platformGbId);
if (device != null) { if (device != null) {
storager.stopPlay(device.getDeviceId(), channelId); storager.stopPlay(device.getDeviceId(), channelId);
StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(device.getDeviceId(), channelId);
if (streamInfo != null) {
redisCatchStorage.stopPlay(streamInfo);
mediaServerService.closeRTPServer(streamInfo.getMediaServerId(), streamInfo.getStream());
}
SsrcTransaction ssrcTransactionForPlay = streamSession.getSsrcTransaction(device.getDeviceId(), channelId, "play", null); SsrcTransaction ssrcTransactionForPlay = streamSession.getSsrcTransaction(device.getDeviceId(), channelId, "play", null);
if (ssrcTransactionForPlay != null){ if (ssrcTransactionForPlay != null){
if (ssrcTransactionForPlay.getCallId().equals(callIdHeader.getCallId())){ if (ssrcTransactionForPlay.getCallId().equals(callIdHeader.getCallId())){
@ -151,6 +153,14 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
} }
streamSession.remove(device.getDeviceId(), channelId, ssrcTransactionForPlay.getStream()); streamSession.remove(device.getDeviceId(), channelId, ssrcTransactionForPlay.getStream());
} }
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channelId);
if (inviteInfo != null) {
inviteStreamService.removeInviteInfo(inviteInfo);
if (inviteInfo.getStreamInfo() != null) {
mediaServerService.closeRTPServer(inviteInfo.getStreamInfo().getMediaServerId(), inviteInfo.getStream());
}
}
} }
SsrcTransaction ssrcTransactionForPlayBack = streamSession.getSsrcTransaction(device.getDeviceId(), channelId, callIdHeader.getCallId(), null); SsrcTransaction ssrcTransactionForPlayBack = streamSession.getSsrcTransaction(device.getDeviceId(), channelId, callIdHeader.getCallId(), null);
if (ssrcTransactionForPlayBack != null) { if (ssrcTransactionForPlayBack != null) {
@ -160,6 +170,14 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcTransactionForPlayBack.getSsrc()); mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcTransactionForPlayBack.getSsrc());
} }
streamSession.remove(device.getDeviceId(), channelId, ssrcTransactionForPlayBack.getStream()); streamSession.remove(device.getDeviceId(), channelId, ssrcTransactionForPlayBack.getStream());
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAYBACK, device.getDeviceId(), channelId);
if (inviteInfo != null) {
inviteStreamService.removeInviteInfo(inviteInfo);
if (inviteInfo.getStreamInfo() != null) {
mediaServerService.closeRTPServer(inviteInfo.getStreamInfo().getMediaServerId(), inviteInfo.getStream());
}
}
} }
} }

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

@ -1,12 +1,10 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl; package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;
import com.alibaba.fastjson2.JSONObject; import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.DynamicTask; import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.session.SSRCFactory; import com.genersoft.iot.vmp.gb28181.session.SSRCFactory;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver; import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
import com.genersoft.iot.vmp.gb28181.transmit.SIPSender; import com.genersoft.iot.vmp.gb28181.transmit.SIPSender;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform; import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
@ -21,6 +19,8 @@ import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IPlayService; import com.genersoft.iot.vmp.service.IPlayService;
import com.genersoft.iot.vmp.service.IStreamProxyService; import com.genersoft.iot.vmp.service.IStreamProxyService;
import com.genersoft.iot.vmp.service.IStreamPushService; import com.genersoft.iot.vmp.service.IStreamPushService;
import com.genersoft.iot.vmp.service.bean.InviteErrorCallback;
import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
import com.genersoft.iot.vmp.service.bean.MessageForPushChannel; import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
import com.genersoft.iot.vmp.service.bean.SSRCInfo; import com.genersoft.iot.vmp.service.bean.SSRCInfo;
import com.genersoft.iot.vmp.service.redisMsg.RedisGbPlayMsgListener; import com.genersoft.iot.vmp.service.redisMsg.RedisGbPlayMsgListener;
@ -101,9 +101,6 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
@Autowired @Autowired
private SIPProcessorObserver sipProcessorObserver; private SIPProcessorObserver sipProcessorObserver;
@Autowired
private VideoStreamSessionManager sessionManager;
@Autowired @Autowired
private UserSetting userSetting; private UserSetting userSetting;
@ -359,7 +356,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
}else { }else {
streamTypeStr = "UDP"; streamTypeStr = "UDP";
} }
logger.info("[上级点播] 平台：{}， 通道：{}, 收流地址：{}:{}，收流方式：{}, ssrc：{}", username, channelId, addressStr, port, streamTypeStr, ssrc); logger.info("[上级Invite] {}, 平台：{}， 通道：{}, 收流地址：{}:{}，收流方式：{}, ssrc：{}", sessionName, username, channelId, addressStr, port, streamTypeStr, ssrc);
SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId, SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
device.getDeviceId(), channelId, mediaTransmissionTCP, platform.isRtcp()); device.getDeviceId(), channelId, mediaTransmissionTCP, platform.isRtcp());
@ -380,10 +377,10 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
Long finalStartTime = startTime; Long finalStartTime = startTime;
Long finalStopTime = stopTime; Long finalStopTime = stopTime;
ZlmHttpHookSubscribe.Event hookEvent = (mediaServerItemInUSe, responseJSON) -> { InviteErrorCallback<Object> hookEvent = (code, msg, data) -> {
String app = responseJSON.getString("app"); StreamInfo streamInfo = (StreamInfo)data;
String stream = responseJSON.getString("stream"); MediaServerItem mediaServerItemInUSe = mediaServerService.getOne(streamInfo.getMediaServerId());
logger.info("[上级点播]下级已经开始推流。 回复200OK(SDP)， {}/{}", app, stream); logger.info("[上级Invite]下级已经开始推流。 回复200OK(SDP)， {}/{}", streamInfo.getApp(), streamInfo.getStream());
// * 0 等待设备推流上来 // * 0 等待设备推流上来
// * 1 下级已经推流，等待上级平台回复ack // * 1 下级已经推流，等待上级平台回复ack
// * 2 推流中 // * 2 推流中
@ -429,11 +426,13 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
logger.error("[命令发送失败] 国标级联 回复SdpAck", e); logger.error("[命令发送失败] 国标级联 回复SdpAck", e);
} }
}; };
SipSubscribe.Event errorEvent = ((event) -> { InviteErrorCallback<Object> errorEvent = ((statusCode, msg, data) -> {
// 未知错误。直接转发设备点播的错误 // 未知错误。直接转发设备点播的错误
try { try {
Response response = getMessageFactory().createResponse(event.statusCode, evt.getRequest()); if (statusCode > 0) {
sipSender.transmitRequest(request.getLocalAddress().getHostAddress(), response); Response response = getMessageFactory().createResponse(statusCode, evt.getRequest());
sipSender.transmitRequest(request.getLocalAddress().getHostAddress(), response);
}
} catch (ParseException | SipException e) { } catch (ParseException | SipException e) {
logger.error("未处理的异常 ", e); logger.error("未处理的异常 ", e);
} }
@ -446,67 +445,70 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
// 写入redis， 超时时回复 // 写入redis， 超时时回复
redisCatchStorage.updateSendRTPSever(sendRtpItem); redisCatchStorage.updateSendRTPSever(sendRtpItem);
playService.playBack(mediaServerItem, ssrcInfo, device.getDeviceId(), channelId, DateUtil.formatter.format(start), playService.playBack(mediaServerItem, ssrcInfo, device.getDeviceId(), channelId, DateUtil.formatter.format(start),
DateUtil.formatter.format(end), null, result -> { DateUtil.formatter.format(end),
if (result.getCode() != 0) { (code, msg, data) -> {
logger.warn("录像回放失败"); if (code == InviteErrorCode.SUCCESS.getCode()){
if (result.getEvent() != null) { hookEvent.run(code, msg, data);
errorEvent.response(result.getEvent()); }else if (code == InviteErrorCode.ERROR_FOR_SIGNALLING_TIMEOUT.getCode() || code == InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode()){
} logger.info("[录像回放]超时, 用户：{}， 通道：{}", username, channelId);
redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), channelId, callIdHeader.getCallId(), null); redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), channelId, callIdHeader.getCallId(), null);
try { errorEvent.run(code, msg, data);
responseAck(request, Response.REQUEST_TIMEOUT); }else {
} catch (SipException | InvalidArgumentException | ParseException e) { errorEvent.run(code, msg, data);
logger.error("[命令发送失败] 国标级联 录像回放 发送REQUEST_TIMEOUT: {}", e.getMessage());
}
} else {
if (result.getMediaServerItem() != null) {
hookEvent.response(result.getMediaServerItem(), result.getResponse());
}
} }
}); });
} else { }else if ("Download".equalsIgnoreCase(sessionName)) {
sendRtpItem.setPlayType(InviteStreamType.PLAY); // 获取指定的下载速度
SsrcTransaction playTransaction = sessionManager.getSsrcTransaction(device.getDeviceId(), channelId, "play", null); Vector sdpMediaDescriptions = sdp.getMediaDescriptions(true);
if (playTransaction != null) { MediaDescription mediaDescription = null;
Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, "rtp", playTransaction.getStream()); String downloadSpeed = "1";
if (!streamReady) { if (sdpMediaDescriptions.size() > 0) {
boolean hasRtpServer = mediaServerService.checkRtpServer(mediaServerItem, "rtp", playTransaction.getStream()); mediaDescription = (MediaDescription)sdpMediaDescriptions.get(0);
if (hasRtpServer) { }
logger.info("[上级点播]已经开启rtpServer但是尚未收到流，开启监听流的到来"); if (mediaDescription != null) {
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", playTransaction.getStream(), true, "rtsp", mediaServerItem.getId()); downloadSpeed = mediaDescription.getAttribute("downloadspeed");
zlmHttpHookSubscribe.addSubscribe(hookSubscribe, hookEvent);
}else {
playTransaction = null;
}
}
} }
if (playTransaction == null) {
// 被点播的通道目前未被点播，则开始点播
String streamId = null;
if (mediaServerItem.isRtpEnable()) {
streamId = String.format("%s_%s", device.getDeviceId(), channelId);
}
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, ssrc, device.isSsrcCheck(), false, 0, false, device.getStreamModeForParam());
logger.info(JSONObject.toJSONString(ssrcInfo));
sendRtpItem.setStreamId(ssrcInfo.getStream());
// sendRtpItem.setSsrc(ssrcInfo.getSsrc());
// 写入redis， 超时时回复 sendRtpItem.setPlayType(InviteStreamType.DOWNLOAD);
redisCatchStorage.updateSendRTPSever(sendRtpItem); SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, null, device.isSsrcCheck(), true, 0, false, device.getStreamModeForParam());
playService.play(mediaServerItem, ssrcInfo, device, channelId, hookEvent, errorEvent, (code, msg) -> { sendRtpItem.setStreamId(ssrcInfo.getStream());
// 写入redis， 超时时回复
redisCatchStorage.updateSendRTPSever(sendRtpItem);
playService.download(mediaServerItem, ssrcInfo, device.getDeviceId(), channelId, DateUtil.formatter.format(start),
DateUtil.formatter.format(end), Integer.parseInt(downloadSpeed),
(code, msg, data) -> {
if (code == InviteErrorCode.SUCCESS.getCode()){
hookEvent.run(code, msg, data);
}else if (code == InviteErrorCode.ERROR_FOR_SIGNALLING_TIMEOUT.getCode() || code == InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode()){
logger.info("[录像下载]超时, 用户：{}， 通道：{}", username, channelId);
redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), channelId, callIdHeader.getCallId(), null);
errorEvent.run(code, msg, data);
}else {
errorEvent.run(code, msg, data);
}
});
}else {
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(ssrc)).toUpperCase();
}
sendRtpItem.setStreamId(streamId);
redisCatchStorage.updateSendRTPSever(sendRtpItem);
playService.play(mediaServerItem, device.getDeviceId(), channelId, ((code, msg, data) -> {
if (code == InviteErrorCode.SUCCESS.getCode()){
hookEvent.run(code, msg, data);
}else if (code == InviteErrorCode.ERROR_FOR_SIGNALLING_TIMEOUT.getCode() || code == InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode()){
logger.info("[上级点播]超时, 用户：{}， 通道：{}", username, channelId); logger.info("[上级点播]超时, 用户：{}， 通道：{}", username, channelId);
redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), channelId, callIdHeader.getCallId(), null); redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), channelId, callIdHeader.getCallId(), null);
}); errorEvent.run(code, msg, data);
} else { }else {
errorEvent.run(code, msg, data);
}
}));
sendRtpItem.setStreamId(playTransaction.getStream());
// 写入redis， 超时时回复
redisCatchStorage.updateSendRTPSever(sendRtpItem);
JSONObject jsonObject = new JSONObject();
jsonObject.put("app", sendRtpItem.getApp());
jsonObject.put("stream", sendRtpItem.getStreamId());
hookEvent.response(mediaServerItem, jsonObject);
}
} }
} else if (gbStream != null) { } else if (gbStream != null) {
@ -559,7 +561,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
int port, Boolean tcpActive, boolean mediaTransmissionTCP, int port, Boolean tcpActive, boolean mediaTransmissionTCP,
String channelId, String addressStr, String ssrc, String requesterId) { String channelId, String addressStr, String ssrc, String requesterId) {
Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream()); Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream());
if (streamReady) { if (streamReady != null && streamReady) {
// 自平台内容 // 自平台内容
SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId, SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
gbStream.getApp(), gbStream.getStream(), channelId, mediaTransmissionTCP, platform.isRtcp()); gbStream.getApp(), gbStream.getStream(), channelId, mediaTransmissionTCP, platform.isRtcp());
@ -598,7 +600,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
// 推流 // 推流
if (streamPushItem.isSelf()) { if (streamPushItem.isSelf()) {
Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream()); Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream());
if (streamReady) { if (streamReady != null && streamReady) {
// 自平台内容 // 自平台内容
SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId, SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
gbStream.getApp(), gbStream.getStream(), channelId, mediaTransmissionTCP, platform.isRtcp()); gbStream.getApp(), gbStream.getStream(), channelId, mediaTransmissionTCP, platform.isRtcp());

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

@ -80,7 +80,6 @@ public class RegisterRequestProcessor extends SIPRequestProcessorParent implemen
public void process(RequestEvent evt) { public void process(RequestEvent evt) {
try { try {
RequestEventExt evtExt = (RequestEventExt) evt; RequestEventExt evtExt = (RequestEventExt) evt;
String requestAddress = evtExt.getRemoteIpAddress() + ":" + evtExt.getRemotePort();
SIPRequest request = (SIPRequest)evt.getRequest(); SIPRequest request = (SIPRequest)evt.getRequest();
Response response = null; Response response = null;
@ -91,12 +90,13 @@ public class RegisterRequestProcessor extends SIPRequestProcessorParent implemen
AddressImpl address = (AddressImpl) fromHeader.getAddress(); AddressImpl address = (AddressImpl) fromHeader.getAddress();
SipUri uri = (SipUri) address.getURI(); SipUri uri = (SipUri) address.getURI();
String deviceId = uri.getUser(); String deviceId = uri.getUser();
logger.info("[注册请求] 设备：{}, 开始处理: {}", deviceId, requestAddress);
Device device = deviceService.getDevice(deviceId); Device device = deviceService.getDevice(deviceId);
RemoteAddressInfo remoteAddressInfo = SipUtils.getRemoteAddressFromRequest(request, RemoteAddressInfo remoteAddressInfo = SipUtils.getRemoteAddressFromRequest(request,
userSetting.getSipUseSourceIpAsRemoteAddress()); userSetting.getSipUseSourceIpAsRemoteAddress());
logger.info("[注册请求] 设备：{}, 远程地址为: {}:{}", deviceId, remoteAddressInfo.getIp(), remoteAddressInfo.getPort()); String requestAddress = remoteAddressInfo.getIp() + ":" + remoteAddressInfo.getPort();
logger.info("[注册请求] 设备：{}, 开始处理: {}", deviceId, requestAddress);
if (device != null && if (device != null &&
device.getSipTransactionInfo() != null && device.getSipTransactionInfo() != null &&
request.getCallIdHeader().getCallId().equals(device.getSipTransactionInfo().getCallId())) { request.getCallIdHeader().getCallId().equals(device.getSipTransactionInfo().getCallId())) {

50
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/info/InfoRequestProcessor.java
View File

@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.info; package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.info;
import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.common.InviteInfo;
import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe; import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager; import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
@ -9,6 +10,7 @@ import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor; import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent; import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils; import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.service.IInviteStreamService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import gov.nist.javax.sip.message.SIPRequest; import gov.nist.javax.sip.message.SIPRequest;
@ -17,10 +19,12 @@ import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import javax.sip.InvalidArgumentException; import javax.sip.InvalidArgumentException;
import javax.sip.RequestEvent; import javax.sip.RequestEvent;
import javax.sip.SipException; import javax.sip.SipException;
import javax.sip.header.*; import javax.sip.header.CallIdHeader;
import javax.sip.header.ContentTypeHeader;
import javax.sip.message.Response; import javax.sip.message.Response;
import java.text.ParseException; import java.text.ParseException;
@ -43,6 +47,9 @@ public class InfoRequestProcessor extends SIPRequestProcessorParent implements I
@Autowired @Autowired
private IRedisCatchStorage redisCatchStorage; private IRedisCatchStorage redisCatchStorage;
@Autowired
private IInviteStreamService inviteStreamService;
@Autowired @Autowired
private IVideoManagerStorage storager; private IVideoManagerStorage storager;
@ -103,27 +110,30 @@ public class InfoRequestProcessor extends SIPRequestProcessorParent implements I
if ("Application".equalsIgnoreCase(contentType) && "MANSRTSP".equalsIgnoreCase(contentSubType)) { if ("Application".equalsIgnoreCase(contentType) && "MANSRTSP".equalsIgnoreCase(contentSubType)) {
SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(null, null, null, callIdHeader.getCallId()); SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(null, null, null, callIdHeader.getCallId());
String streamId = sendRtpItem.getStreamId(); String streamId = sendRtpItem.getStreamId();
StreamInfo streamInfo = redisCatchStorage.queryPlayback(null, null, streamId, null); InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(InviteSessionType.PLAYBACK, streamId);
if (null == streamInfo) { if (null == inviteInfo) {
responseAck(request, Response.NOT_FOUND, "stream " + streamId + " not found"); responseAck(request, Response.NOT_FOUND, "stream " + streamId + " not found");
return; return;
} }
Device device1 = storager.queryVideoDevice(streamInfo.getDeviceID()); Device device1 = storager.queryVideoDevice(inviteInfo.getDeviceId());
cmder.playbackControlCmd(device1,streamInfo,new String(evt.getRequest().getRawContent()),eventResult -> { if (inviteInfo.getStreamInfo() != null) {
// 失败的回复 cmder.playbackControlCmd(device1,inviteInfo.getStreamInfo(),new String(evt.getRequest().getRawContent()),eventResult -> {
try { // 失败的回复
responseAck(request, eventResult.statusCode, eventResult.msg); try {
} catch (SipException | InvalidArgumentException | ParseException e) { responseAck(request, eventResult.statusCode, eventResult.msg);
logger.error("[命令发送失败] 国标级联 录像控制: {}", e.getMessage()); } catch (SipException | InvalidArgumentException | ParseException e) {
} logger.error("[命令发送失败] 国标级联 录像控制: {}", e.getMessage());
}, eventResult -> { }
// 成功的回复 }, eventResult -> {
try { // 成功的回复
responseAck(request, eventResult.statusCode); try {
} catch (SipException | InvalidArgumentException | ParseException e) { responseAck(request, eventResult.statusCode);
logger.error("[命令发送失败] 国标级联 录像控制: {}", e.getMessage()); } catch (SipException | InvalidArgumentException | ParseException e) {
} logger.error("[命令发送失败] 国标级联 录像控制: {}", e.getMessage());
}); }
});
}
} }
} }
} catch (SipException e) { } catch (SipException e) {

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

@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.notify.cmd; package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.notify.cmd;
import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.common.InviteInfo;
import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException; import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
import com.genersoft.iot.vmp.gb28181.bean.Device; import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform; import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
@ -15,6 +16,7 @@ import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.notify.
import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe; 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.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange; import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.service.IInviteStreamService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import gov.nist.javax.sip.message.SIPRequest; import gov.nist.javax.sip.message.SIPRequest;
@ -64,6 +66,12 @@ public class MediaStatusNotifyMessageHandler extends SIPRequestProcessorParent i
@Autowired @Autowired
private ZlmHttpHookSubscribe subscribe; private ZlmHttpHookSubscribe subscribe;
@Autowired
private IInviteStreamService inviteStreamService;
@Autowired
private VideoStreamSessionManager streamSession;
@Override @Override
public void afterPropertiesSet() throws Exception { public void afterPropertiesSet() throws Exception {
notifyMessageHandler.addHandler(cmdType, this); notifyMessageHandler.addHandler(cmdType, this);
@ -82,17 +90,15 @@ public class MediaStatusNotifyMessageHandler extends SIPRequestProcessorParent i
String NotifyType =getText(rootElement, "NotifyType"); String NotifyType =getText(rootElement, "NotifyType");
if ("121".equals(NotifyType)){ if ("121".equals(NotifyType)){
logger.info("[录像流]推送完毕，收到关流通知"); logger.info("[录像流]推送完毕，收到关流通知");
// 查询是设备
StreamInfo streamInfo = redisCatchStorage.queryDownload(null, null, null, callIdHeader.getCallId());
if (streamInfo != null) {
// 设置进度100%
streamInfo.setProgress(1);
redisCatchStorage.startDownload(streamInfo, callIdHeader.getCallId());
}
// 先从会话内查找 SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(null, null, callIdHeader.getCallId(), null);
SsrcTransaction ssrcTransaction = sessionManager.getSsrcTransaction(null, null, callIdHeader.getCallId(), null); if (ssrcTransaction != null) {
if (ssrcTransaction != null) { // 兼容海康 媒体通知 消息from字段不是设备ID的问题 logger.info("[录像流]推送完毕，关流通知， device: {}, channelId: {}", ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId());
InviteInfo inviteInfo = inviteStreamService.getInviteInfo(InviteSessionType.DOWNLOAD, ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId(), ssrcTransaction.getStream());
if (inviteInfo.getStreamInfo() != null) {
inviteInfo.getStreamInfo().setProgress(1);
inviteStreamService.updateInviteInfo(inviteInfo);
}
try { try {
cmder.streamByeCmd(device, ssrcTransaction.getChannelId(), null, callIdHeader.getCallId()); cmder.streamByeCmd(device, ssrcTransaction.getChannelId(), null, callIdHeader.getCallId());
@ -117,6 +123,8 @@ public class MediaStatusNotifyMessageHandler extends SIPRequestProcessorParent i
logger.error("[命令发送失败] 国标级联 录像播放完毕: {}", e.getMessage()); logger.error("[命令发送失败] 国标级联 录像播放完毕: {}", e.getMessage());
} }
} }
}else {
logger.info("[录像流]推送完毕，关流通知， 但是未找到对应的下载信息");
} }
} }
} }

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

@ -2,6 +2,8 @@ package com.genersoft.iot.vmp.media.zlm;
import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject; import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.InviteInfo;
import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException; import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
@ -22,10 +24,8 @@ import com.genersoft.iot.vmp.media.zlm.dto.hook.*;
import com.genersoft.iot.vmp.service.*; import com.genersoft.iot.vmp.service.*;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.vmanager.bean.DeferredResultEx;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.StreamContent; import com.genersoft.iot.vmp.vmanager.bean.StreamContent;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
@ -70,6 +70,9 @@ public class ZLMHttpHookListener {
@Autowired @Autowired
private IRedisCatchStorage redisCatchStorage; private IRedisCatchStorage redisCatchStorage;
@Autowired
private IInviteStreamService inviteStreamService;
@Autowired @Autowired
private IDeviceService deviceService; private IDeviceService deviceService;
@ -252,7 +255,7 @@ public class ZLMHttpHookListener {
result.setEnable_audio(deviceChannel.isHasAudio()); result.setEnable_audio(deviceChannel.isHasAudio());
} }
// 如果是录像下载就设置视频间隔十秒 // 如果是录像下载就设置视频间隔十秒
if (ssrcTransactionForAll.get(0).getType() == VideoStreamSessionManager.SessionType.download) { if (ssrcTransactionForAll.get(0).getType() == InviteSessionType.DOWNLOAD) {
result.setMp4_max_second(10); result.setMp4_max_second(10);
result.setEnable_audio(true); result.setEnable_audio(true);
result.setEnable_mp4(true); result.setEnable_mp4(true);
@ -342,17 +345,10 @@ public class ZLMHttpHookListener {
} }
if ("rtp".equals(param.getApp()) && !param.isRegist()) { if ("rtp".equals(param.getApp()) && !param.isRegist()) {
StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(param.getStream()); InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, param.getStream());
if (streamInfo != null) { if (inviteInfo != null && (inviteInfo.getType() == InviteSessionType.PLAY || inviteInfo.getType() == InviteSessionType.PLAYBACK)) {
redisCatchStorage.stopPlay(streamInfo); inviteStreamService.removeInviteInfo(inviteInfo);
storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); storager.stopPlay(inviteInfo.getDeviceId(), inviteInfo.getChannelId());
} else {
streamInfo = redisCatchStorage.queryPlayback(null, null,
param.getStream(), null);
if (streamInfo != null) {
redisCatchStorage.stopPlayback(streamInfo.getDeviceID(), streamInfo.getChannelId(),
streamInfo.getStream(), null);
}
} }
} else { } else {
if (!"rtp".equals(param.getApp())) { if (!"rtp".equals(param.getApp())) {
@ -442,7 +438,7 @@ public class ZLMHttpHookListener {
@PostMapping(value = "/on_stream_none_reader", produces = "application/json;charset=UTF-8") @PostMapping(value = "/on_stream_none_reader", produces = "application/json;charset=UTF-8")
public JSONObject onStreamNoneReader(@RequestBody OnStreamNoneReaderHookParam param) { public JSONObject onStreamNoneReader(@RequestBody OnStreamNoneReaderHookParam param) {
logger.info("[ZLM HOOK]流无人观看：{]->{}->{}/{}" + param.getMediaServerId(), param.getSchema(), logger.info("[ZLM HOOK]流无人观看：{}->{}->{}/{}", param.getMediaServerId(), param.getSchema(),
param.getApp(), param.getStream()); param.getApp(), param.getStream());
JSONObject ret = new JSONObject(); JSONObject ret = new JSONObject();
ret.put("code", 0); ret.put("code", 0);
@ -450,13 +446,20 @@ public class ZLMHttpHookListener {
if ("rtp".equals(param.getApp())) { if ("rtp".equals(param.getApp())) {
ret.put("close", userSetting.getStreamOnDemand()); ret.put("close", userSetting.getStreamOnDemand());
// 国标流， 点播/录像回放/录像下载 // 国标流， 点播/录像回放/录像下载
StreamInfo streamInfoForPlayCatch = redisCatchStorage.queryPlayByStreamId(param.getStream()); // StreamInfo streamInfoForPlayCatch = redisCatchStorage.queryPlayByStreamId(param.getStream());
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, param.getStream());
// 点播 // 点播
if (streamInfoForPlayCatch != null) { if (inviteInfo != null) {
// 录像下载
if (inviteInfo.getType() == InviteSessionType.DOWNLOAD) {
ret.put("close", false);
return ret;
}
// 收到无人观看说明流也没有在往上级推送 // 收到无人观看说明流也没有在往上级推送
if (redisCatchStorage.isChannelSendingRTP(streamInfoForPlayCatch.getChannelId())) { if (redisCatchStorage.isChannelSendingRTP(inviteInfo.getChannelId())) {
List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChnnelId( List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChnnelId(
streamInfoForPlayCatch.getChannelId()); inviteInfo.getChannelId());
if (sendRtpItems.size() > 0) { if (sendRtpItems.size() > 0) {
for (SendRtpItem sendRtpItem : sendRtpItems) { for (SendRtpItem sendRtpItem : sendRtpItems) {
ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId()); ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
@ -470,49 +473,22 @@ public class ZLMHttpHookListener {
} }
} }
} }
Device device = deviceService.getDevice(streamInfoForPlayCatch.getDeviceID()); Device device = deviceService.getDevice(inviteInfo.getDeviceId());
if (device != null) { if (device != null) {
try { try {
cmder.streamByeCmd(device, streamInfoForPlayCatch.getChannelId(), if (inviteStreamService.getInviteInfo(inviteInfo.getType(), inviteInfo.getDeviceId(), inviteInfo.getChannelId(), inviteInfo.getStream()) != null) {
streamInfoForPlayCatch.getStream(), null); cmder.streamByeCmd(device, inviteInfo.getChannelId(),
inviteInfo.getStream(), null);
}
} catch (InvalidArgumentException | ParseException | SipException | } catch (InvalidArgumentException | ParseException | SipException |
SsrcTransactionNotFoundException e) { SsrcTransactionNotFoundException e) {
logger.error("[无人观看]点播， 发送BYE失败 {}", e.getMessage()); logger.error("[无人观看]点播， 发送BYE失败 {}", e.getMessage());
} }
} }
redisCatchStorage.stopPlay(streamInfoForPlayCatch); inviteStreamService.removeInviteInfo(inviteInfo.getType(), inviteInfo.getDeviceId(),
storager.stopPlay(streamInfoForPlayCatch.getDeviceID(), streamInfoForPlayCatch.getChannelId()); inviteInfo.getChannelId(), inviteInfo.getStream());
return ret; storager.stopPlay(inviteInfo.getDeviceId(), inviteInfo.getChannelId());
}
// 录像回放
StreamInfo streamInfoForPlayBackCatch = redisCatchStorage.queryPlayback(null, null,
param.getStream(), null);
if (streamInfoForPlayBackCatch != null) {
if (streamInfoForPlayBackCatch.isPause()) {
ret.put("close", false);
} else {
Device device = deviceService.getDevice(streamInfoForPlayBackCatch.getDeviceID());
if (device != null) {
try {
cmder.streamByeCmd(device, streamInfoForPlayBackCatch.getChannelId(),
streamInfoForPlayBackCatch.getStream(), null);
} catch (InvalidArgumentException | ParseException | SipException |
SsrcTransactionNotFoundException e) {
logger.error("[无人观看]回放， 发送BYE失败 {}", e.getMessage());
}
}
redisCatchStorage.stopPlayback(streamInfoForPlayBackCatch.getDeviceID(),
streamInfoForPlayBackCatch.getChannelId(), streamInfoForPlayBackCatch.getStream(), null);
}
return ret;
}
// 录像下载
StreamInfo streamInfoForDownload = redisCatchStorage.queryDownload(null, null,
param.getStream(), null);
// 进行录像下载时无人观看不断流
if (streamInfoForDownload != null) {
ret.put("close", false);
return ret; return ret;
} }
} else { } else {
@ -582,6 +558,7 @@ public class ZLMHttpHookListener {
return defaultResult; return defaultResult;
} }
logger.info("[ZLM HOOK] 流未找到, 发起自动点播：{}->{}->{}/{}", param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream()); logger.info("[ZLM HOOK] 流未找到, 发起自动点播：{}->{}->{}/{}", param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());
RequestMessage msg = new RequestMessage(); RequestMessage msg = new RequestMessage();
String key = DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId; String key = DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId;
boolean exist = resultHolder.exist(key, null); boolean exist = resultHolder.exist(key, null);
@ -589,31 +566,22 @@ public class ZLMHttpHookListener {
String uuid = UUID.randomUUID().toString(); String uuid = UUID.randomUUID().toString();
msg.setId(uuid); msg.setId(uuid);
DeferredResult<HookResult> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue()); DeferredResult<HookResult> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue());
DeferredResultEx<HookResult> deferredResultEx = new DeferredResultEx<>(result);
result.onTimeout(() -> { result.onTimeout(() -> {
logger.info("点播接口等待超时"); logger.info("[ZLM HOOK] 自动点播, 等待超时");
// 释放rtpserver // 释放rtpserver
msg.setData(new HookResult(ErrorCode.ERROR100.getCode(), "点播超时")); msg.setData(new HookResult(ErrorCode.ERROR100.getCode(), "点播超时"));
resultHolder.invokeResult(msg); resultHolder.invokeResult(msg);
}); });
// TODO 在点播未成功的情况下在此调用接口点播会导致返回的流地址ip错误
deferredResultEx.setFilter(result1 -> {
WVPResult<StreamInfo> wvpResult1 = (WVPResult<StreamInfo>) result1;
HookResult resultForEnd = new HookResult();
resultForEnd.setCode(wvpResult1.getCode());
resultForEnd.setMsg(wvpResult1.getMsg());
return resultForEnd;
});
// 录像查询以channelId作为deviceId查询 // 录像查询以channelId作为deviceId查询
resultHolder.put(key, uuid, deferredResultEx); resultHolder.put(key, uuid, result);
if (!exist) { if (!exist) {
playService.play(mediaInfo, deviceId, channelId, null, eventResult -> { playService.play(mediaInfo, deviceId, channelId, (code, message, data) -> {
msg.setData(new HookResult(eventResult.statusCode, eventResult.msg)); msg.setData(new HookResult(code, message));
resultHolder.invokeResult(msg); resultHolder.invokeResult(msg);
}, null); });
} }
return result; return result;
} else { } else {

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

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

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

@ -293,11 +293,14 @@ public class ZLMRTPServerFactory {
if (jsonObject.getInteger("code") == 0) { if (jsonObject.getInteger("code") == 0) {
localPort = jsonObject.getInteger("port"); localPort = jsonObject.getInteger("port");
HookSubscribeForRtpServerTimeout hookSubscribeForRtpServerTimeout = HookSubscribeFactory.on_rtp_server_timeout(ssrc, null, serverItem.getId()); HookSubscribeForRtpServerTimeout hookSubscribeForRtpServerTimeout = HookSubscribeFactory.on_rtp_server_timeout(ssrc, null, serverItem.getId());
// 订阅 zlm启动事件, 新的zlm也会从这里进入系统
hookSubscribe.addSubscribe(hookSubscribeForRtpServerTimeout, hookSubscribe.addSubscribe(hookSubscribeForRtpServerTimeout,
(MediaServerItem mediaServerItem, JSONObject response)->{ (MediaServerItem mediaServerItem, JSONObject response)->{
logger.info("[上级点播] {}->监听端口到期继续保持监听", ssrc); logger.info("[上级点播] {}->监听端口到期继续保持监听", ssrc);
keepPort(serverItem, ssrc); int port = keepPort(serverItem, ssrc);
if (port == 0) {
logger.info("[上级点播] {}->监听端口失败，移除监听", ssrc);
hookSubscribe.removeSubscribe(hookSubscribeForRtpServerTimeout);
}
}); });
logger.info("[上级点播] {}->监听端口: {}", ssrc, localPort); logger.info("[上级点播] {}->监听端口: {}", ssrc, localPort);
}else { }else {
@ -330,6 +333,9 @@ public class ZLMRTPServerFactory {
*/ */
public Boolean isRtpReady(MediaServerItem mediaServerItem, String streamId) { public Boolean isRtpReady(MediaServerItem mediaServerItem, String streamId) {
JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo(mediaServerItem,"rtp", "rtsp", streamId); JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo(mediaServerItem,"rtp", "rtsp", streamId);
if (mediaInfo.getInteger("code") == -2) {
return null;
}
return (mediaInfo.getInteger("code") == 0 && mediaInfo.getBoolean("online")); return (mediaInfo.getInteger("code") == 0 && mediaInfo.getBoolean("online"));
} }
@ -338,8 +344,10 @@ public class ZLMRTPServerFactory {
*/ */
public Boolean isStreamReady(MediaServerItem mediaServerItem, String app, String streamId) { public Boolean isStreamReady(MediaServerItem mediaServerItem, String app, String streamId) {
JSONObject mediaInfo = zlmresTfulUtils.getMediaList(mediaServerItem, app, streamId); JSONObject mediaInfo = zlmresTfulUtils.getMediaList(mediaServerItem, app, streamId);
return mediaInfo != null && (mediaInfo.getInteger("code") == 0 if (mediaInfo == null || (mediaInfo.getInteger("code") == -2)) {
return null;
}
return (mediaInfo.getInteger("code") == 0
&& mediaInfo.getJSONArray("data") != null && mediaInfo.getJSONArray("data") != null
&& mediaInfo.getJSONArray("data").size() > 0); && mediaInfo.getJSONArray("data").size() > 0);
} }

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

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

68
src/main/java/com/genersoft/iot/vmp/service/IInviteStreamService.java Normal file
View File

@ -0,0 +1,68 @@
package com.genersoft.iot.vmp.service;
import com.genersoft.iot.vmp.common.InviteInfo;
import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.service.bean.InviteErrorCallback;
/**
* 记录国标点播的状态，包括实时预览，下载，录像回放
*/
public interface IInviteStreamService {
/**
* 更新点播的状态信息
*/
void updateInviteInfo(InviteInfo inviteInfo);
/**
* 获取点播的状态信息
*/
InviteInfo getInviteInfo(InviteSessionType type,
String deviceId,
String channelId,
String stream);
/**
* 移除点播的状态信息
*/
void removeInviteInfo(InviteSessionType type,
String deviceId,
String channelId,
String stream);
/**
* 移除点播的状态信息
*/
void removeInviteInfo(InviteInfo inviteInfo);
/**
* 移除点播的状态信息
*/
void removeInviteInfoByDeviceAndChannel(InviteSessionType inviteSessionType, String deviceId, String channelId);
/**
* 获取点播的状态信息
*/
InviteInfo getInviteInfoByDeviceAndChannel(InviteSessionType type,
String deviceId,
String channelId);
/**
* 获取点播的状态信息
*/
InviteInfo getInviteInfoByStream(InviteSessionType type, String stream);
/**
* 添加一个invite回调
*/
void once(InviteSessionType type, String deviceId, String channelId, String stream, InviteErrorCallback<Object> callback);
/**
* 调用一个invite回调
*/
void call(InviteSessionType type, String deviceId, String channelId, String stream, int code, String msg, Object data);
/**
* 清空一个设备的所有invite信息
*/
void clearInviteInfo(String deviceId);
}

25
src/main/java/com/genersoft/iot/vmp/service/IPlayService.java
View File

@ -1,16 +1,10 @@
package com.genersoft.iot.vmp.service; package com.genersoft.iot.vmp.service;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.exception.ServiceException; import com.genersoft.iot.vmp.conf.exception.ServiceException;
import com.genersoft.iot.vmp.gb28181.bean.Device; import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.InviteStreamCallback;
import com.genersoft.iot.vmp.gb28181.bean.InviteStreamInfo;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.bean.InviteTimeOutCallback; import com.genersoft.iot.vmp.service.bean.InviteErrorCallback;
import com.genersoft.iot.vmp.service.bean.PlayBackCallback;
import com.genersoft.iot.vmp.service.bean.SSRCInfo; import com.genersoft.iot.vmp.service.bean.SSRCInfo;
import javax.sip.InvalidArgumentException; import javax.sip.InvalidArgumentException;
@ -22,12 +16,9 @@ import java.text.ParseException;
*/ */
public interface IPlayService { public interface IPlayService {
void onPublishHandlerForPlay(MediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId);
void play(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, void play(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId,
ZlmHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent, InviteErrorCallback<Object> callback);
InviteTimeOutCallback timeoutCallback); SSRCInfo play(MediaServerItem mediaServerItem, String deviceId, String channelId, InviteErrorCallback<Object> callback);
void play(MediaServerItem mediaServerItem, String deviceId, String channelId, ZlmHttpHookSubscribe.Event event, SipSubscribe.Event errorEvent, Runnable timeoutCallback);
MediaServerItem getNewMediaServerItem(Device device); MediaServerItem getNewMediaServerItem(Device device);
@ -36,15 +27,13 @@ public interface IPlayService {
*/ */
MediaServerItem getNewMediaServerItemHasAssist(Device device); MediaServerItem getNewMediaServerItemHasAssist(Device device);
void onPublishHandlerForDownload(InviteStreamInfo inviteStreamInfo, String deviceId, String channelId, String toString); void playBack(String deviceId, String channelId, String startTime, String endTime, InviteErrorCallback<Object> callback);
void playBack(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, String deviceId, String channelId, String startTime, String endTime, InviteErrorCallback<Object> callback);
void playBack(String deviceId, String channelId, String startTime, String endTime, InviteStreamCallback infoCallBack, PlayBackCallback playBackCallback);
void playBack(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, String deviceId, String channelId, String startTime, String endTime, InviteStreamCallback infoCallBack, PlayBackCallback hookCallBack);
void zlmServerOffline(String mediaServerId); void zlmServerOffline(String mediaServerId);
void download(String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, InviteStreamCallback infoCallBack, PlayBackCallback playBackCallback); void download(String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, InviteErrorCallback<Object> callback);
void download(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo,String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, InviteStreamCallback infoCallBack, PlayBackCallback hookCallBack); void download(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo,String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, InviteErrorCallback<Object> callback);
StreamInfo getDownLoadInfo(String deviceId, String channelId, String stream); StreamInfo getDownLoadInfo(String deviceId, String channelId, String stream);

6
src/main/java/com/genersoft/iot/vmp/service/bean/InviteErrorCallback.java Normal file
View File

@ -0,0 +1,6 @@
package com.genersoft.iot.vmp.service.bean;
public interface InviteErrorCallback<T> {
void run(int code, String msg, T data);
}

36
src/main/java/com/genersoft/iot/vmp/service/bean/InviteErrorCode.java Normal file
View File

@ -0,0 +1,36 @@
package com.genersoft.iot.vmp.service.bean;
/**
* 全局错误码
*/
public enum InviteErrorCode {
SUCCESS(0, "成功"),
ERROR_FOR_SIGNALLING_TIMEOUT(-1, "信令超时"),
ERROR_FOR_STREAM_TIMEOUT(-2, "收流超时"),
ERROR_FOR_RESOURCE_EXHAUSTION(-3, "资源耗尽"),
ERROR_FOR_CATCH_DATA(-4, "缓存数据异常"),
ERROR_FOR_SIGNALLING_ERROR(-5, "收到信令错误"),
ERROR_FOR_STREAM_PARSING_EXCEPTIONS(-6, "流地址解析错误"),
ERROR_FOR_SDP_PARSING_EXCEPTIONS(-7, "SDP信息解析失败"),
ERROR_FOR_SSRC_UNAVAILABLE(-8, "SSRC不可用"),
ERROR_FOR_RESET_SSRC(-9, "重新设置收流信息失败"),
ERROR_FOR_SIP_SENDING_FAILED(-10, "命令发送失败"),
ERROR_FOR_ASSIST_NOT_READY(-11, "没有可用的assist服务"),
ERROR_FOR_PARAMETER_ERROR(-13, "参数异常");
private final int code;
private final String msg;
InviteErrorCode(int code, String msg) {
this.code = code;
this.msg = msg;
}
public int getCode() {
return code;
}
public String getMsg() {
return msg;
}
}

26
src/main/java/com/genersoft/iot/vmp/service/impl/DeviceChannelServiceImpl.java
View File

@ -1,10 +1,12 @@
package com.genersoft.iot.vmp.service.impl; package com.genersoft.iot.vmp.service.impl;
import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.common.InviteInfo;
import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.gb28181.bean.Device; import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel; import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.utils.Coordtransform; import com.genersoft.iot.vmp.gb28181.utils.Coordtransform;
import com.genersoft.iot.vmp.service.IDeviceChannelService; import com.genersoft.iot.vmp.service.IDeviceChannelService;
import com.genersoft.iot.vmp.service.IInviteStreamService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.dao.DeviceChannelMapper; import com.genersoft.iot.vmp.storager.dao.DeviceChannelMapper;
import com.genersoft.iot.vmp.storager.dao.DeviceMapper; import com.genersoft.iot.vmp.storager.dao.DeviceMapper;
@ -32,6 +34,9 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
@Autowired @Autowired
private IRedisCatchStorage redisCatchStorage; private IRedisCatchStorage redisCatchStorage;
@Autowired
private IInviteStreamService inviteStreamService;
@Autowired @Autowired
private DeviceChannelMapper channelMapper; private DeviceChannelMapper channelMapper;
@ -78,9 +83,10 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
public void updateChannel(String deviceId, DeviceChannel channel) { public void updateChannel(String deviceId, DeviceChannel channel) {
String channelId = channel.getChannelId(); String channelId = channel.getChannelId();
channel.setDeviceId(deviceId); channel.setDeviceId(deviceId);
StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId); // StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
if (streamInfo != null) { InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId);
channel.setStreamId(streamInfo.getStream()); if (inviteInfo != null && inviteInfo.getStreamInfo() != null) {
channel.setStreamId(inviteInfo.getStreamInfo().getStream());
} }
String now = DateUtil.getNow(); String now = DateUtil.getNow();
channel.setUpdateTime(now); channel.setUpdateTime(now);
@ -106,9 +112,9 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
if (channelList.size() == 0) { if (channelList.size() == 0) {
for (DeviceChannel channel : channels) { for (DeviceChannel channel : channels) {
channel.setDeviceId(deviceId); channel.setDeviceId(deviceId);
StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channel.getChannelId()); InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channel.getChannelId());
if (streamInfo != null) { if (inviteInfo != null && inviteInfo.getStreamInfo() != null) {
channel.setStreamId(streamInfo.getStream()); channel.setStreamId(inviteInfo.getStreamInfo().getStream());
} }
String now = DateUtil.getNow(); String now = DateUtil.getNow();
channel.setUpdateTime(now); channel.setUpdateTime(now);
@ -122,9 +128,9 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
} }
for (DeviceChannel channel : channels) { for (DeviceChannel channel : channels) {
channel.setDeviceId(deviceId); channel.setDeviceId(deviceId);
StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channel.getChannelId()); InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channel.getChannelId());
if (streamInfo != null) { if (inviteInfo != null && inviteInfo.getStreamInfo() != null) {
channel.setStreamId(streamInfo.getStream()); channel.setStreamId(inviteInfo.getStreamInfo().getStream());
} }
String now = DateUtil.getNow(); String now = DateUtil.getNow();
channel.setUpdateTime(now); channel.setUpdateTime(now);

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

@ -12,6 +12,7 @@ import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd.CatalogResponseMessageHandler; import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd.CatalogResponseMessageHandler;
import com.genersoft.iot.vmp.service.IDeviceChannelService; import com.genersoft.iot.vmp.service.IDeviceChannelService;
import com.genersoft.iot.vmp.service.IDeviceService; import com.genersoft.iot.vmp.service.IDeviceService;
import com.genersoft.iot.vmp.service.IInviteStreamService;
import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.dao.DeviceChannelMapper; import com.genersoft.iot.vmp.storager.dao.DeviceChannelMapper;
@ -58,6 +59,9 @@ public class DeviceServiceImpl implements IDeviceService {
@Autowired @Autowired
private IRedisCatchStorage redisCatchStorage; private IRedisCatchStorage redisCatchStorage;
@Autowired
private IInviteStreamService inviteStreamService;
@Autowired @Autowired
private DeviceMapper deviceMapper; private DeviceMapper deviceMapper;
@ -97,7 +101,7 @@ public class DeviceServiceImpl implements IDeviceService {
String now = DateUtil.getNow(); String now = DateUtil.getNow();
if (deviceInRedis != null && deviceInDb == null) { if (deviceInRedis != null && deviceInDb == null) {
// redis 存在脏数据 // redis 存在脏数据
redisCatchStorage.clearCatchByDeviceId(device.getDeviceId()); inviteStreamService.clearInviteInfo(device.getDeviceId());
} }
device.setUpdateTime(now); device.setUpdateTime(now);
if (device.getKeepaliveIntervalTime() == 0) { if (device.getKeepaliveIntervalTime() == 0) {
@ -497,8 +501,10 @@ public class DeviceServiceImpl implements IDeviceService {
node.setBasicData(channel); node.setBasicData(channel);
node.setParent(false); node.setParent(false);
if (channel.getChannelId().length() > 8) { if (channel.getChannelId().length() > 8) {
String gbCodeType = channel.getChannelId().substring(10, 13); if (channel.getChannelId().length() > 13) {
node.setParent(gbCodeType.equals(ChannelIdType.BUSINESS_GROUP) || gbCodeType.equals(ChannelIdType.VIRTUAL_ORGANIZATION) ); String gbCodeType = channel.getChannelId().substring(10, 13);
node.setParent(gbCodeType.equals(ChannelIdType.BUSINESS_GROUP) || gbCodeType.equals(ChannelIdType.VIRTUAL_ORGANIZATION) );
}
}else { }else {
node.setParent(true); node.setParent(true);
} }

182
src/main/java/com/genersoft/iot/vmp/service/impl/InviteStreamServiceImpl.java Normal file
View File

@ -0,0 +1,182 @@
package com.genersoft.iot.vmp.service.impl;
import com.alibaba.fastjson2.JSON;
import com.genersoft.iot.vmp.common.InviteInfo;
import com.genersoft.iot.vmp.common.InviteSessionStatus;
import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.service.IInviteStreamService;
import com.genersoft.iot.vmp.service.bean.InviteErrorCallback;
import com.genersoft.iot.vmp.utils.redis.RedisUtil;
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.stereotype.Service;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
@Service
public class InviteStreamServiceImpl implements IInviteStreamService {
private final Logger logger = LoggerFactory.getLogger(InviteStreamServiceImpl.class);
private final Map<String, List<InviteErrorCallback<Object>>> inviteErrorCallbackMap = new ConcurrentHashMap<>();
@Autowired
private RedisTemplate<Object, Object> redisTemplate;
@Override
public void updateInviteInfo(InviteInfo inviteInfo) {
if (inviteInfo == null || (inviteInfo.getDeviceId() == null || inviteInfo.getChannelId() == null)) {
logger.warn("[更新Invite信息]，参数不全： {}", JSON.toJSON(inviteInfo));
return;
}
InviteInfo inviteInfoForUpdate = null;
if (InviteSessionStatus.ready == inviteInfo.getStatus()) {
if (inviteInfo.getDeviceId() == null
|| inviteInfo.getChannelId() == null
|| inviteInfo.getType() == null
|| inviteInfo.getStream() == null
) {
return;
}
inviteInfoForUpdate = inviteInfo;
} else {
InviteInfo inviteInfoInRedis = getInviteInfo(inviteInfo.getType(), inviteInfo.getDeviceId(),
inviteInfo.getChannelId(), inviteInfo.getStream());
if (inviteInfoInRedis == null) {
logger.warn("[更新Invite信息]，未从缓存中读取到Invite信息： deviceId: {}, channel: {}, stream: {}",
inviteInfo.getDeviceId(), inviteInfo.getChannelId(), inviteInfo.getStream());
return;
}
if (inviteInfo.getStreamInfo() != null) {
inviteInfoInRedis.setStreamInfo(inviteInfo.getStreamInfo());
}
if (inviteInfo.getSsrcInfo() != null) {
inviteInfoInRedis.setSsrcInfo(inviteInfo.getSsrcInfo());
}
if (inviteInfo.getStreamMode() != null) {
inviteInfoInRedis.setStreamMode(inviteInfo.getStreamMode());
}
if (inviteInfo.getReceiveIp() != null) {
inviteInfoInRedis.setReceiveIp(inviteInfo.getReceiveIp());
}
if (inviteInfo.getReceivePort() != null) {
inviteInfoInRedis.setReceivePort(inviteInfo.getReceivePort());
}
if (inviteInfo.getStatus() != null) {
inviteInfoInRedis.setStatus(inviteInfo.getStatus());
}
inviteInfoForUpdate = inviteInfoInRedis;
}
String key = VideoManagerConstants.INVITE_PREFIX +
"_" + inviteInfoForUpdate.getType() +
"_" + inviteInfoForUpdate.getDeviceId() +
"_" + inviteInfoForUpdate.getChannelId() +
"_" + inviteInfoForUpdate.getStream();
redisTemplate.opsForValue().set(key, inviteInfoForUpdate);
}
@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 : "*");
List<Object> scanResult = RedisUtil.scan(redisTemplate, key);
if (scanResult.size() != 1) {
return null;
}
return (InviteInfo) redisTemplate.opsForValue().get(scanResult.get(0));
}
@Override
public InviteInfo getInviteInfoByDeviceAndChannel(InviteSessionType type, String deviceId, String channelId) {
return getInviteInfo(type, deviceId, channelId, null);
}
@Override
public InviteInfo getInviteInfoByStream(InviteSessionType type, String stream) {
return getInviteInfo(type, null, null, stream);
}
@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 : "*");
List<Object> scanResult = RedisUtil.scan(redisTemplate, scanKey);
if (scanResult.size() > 0) {
for (Object keyObj : scanResult) {
String key = (String) keyObj;
InviteInfo inviteInfo = (InviteInfo) redisTemplate.opsForValue().get(key);
if (inviteInfo == null) {
continue;
}
redisTemplate.delete(key);
inviteErrorCallbackMap.remove(buildKey(type, deviceId, channelId, inviteInfo.getStream()));
}
}
}
@Override
public void removeInviteInfoByDeviceAndChannel(InviteSessionType inviteSessionType, String deviceId, String channelId) {
removeInviteInfo(inviteSessionType, deviceId, channelId, null);
}
@Override
public void removeInviteInfo(InviteInfo inviteInfo) {
removeInviteInfo(inviteInfo.getType(), inviteInfo.getDeviceId(), inviteInfo.getChannelId(), inviteInfo.getStream());
}
@Override
public void once(InviteSessionType type, String deviceId, String channelId, String stream, InviteErrorCallback<Object> callback) {
String key = buildKey(type, deviceId, channelId, stream);
List<InviteErrorCallback<Object>> callbacks = inviteErrorCallbackMap.get(key);
if (callbacks == null) {
callbacks = new CopyOnWriteArrayList<>();
inviteErrorCallbackMap.put(key, callbacks);
}
callbacks.add(callback);
}
@Override
public void call(InviteSessionType type, String deviceId, String channelId, String stream, int code, String msg, Object data) {
String key = buildKey(type, deviceId, channelId, stream);
List<InviteErrorCallback<Object>> callbacks = inviteErrorCallbackMap.get(key);
if (callbacks == null) {
return;
}
for (InviteErrorCallback<Object> callback : callbacks) {
callback.run(code, msg, data);
}
inviteErrorCallbackMap.remove(key);
}
private String buildKey(InviteSessionType type, String deviceId, String channelId, String stream) {
String key = type + "_" + deviceId + "_" + channelId;
// 如果ssrc未null那么可以实现一个通道只能一次操作，ssrc不为null则可以支持一个通道多次invite
if (stream != null) {
key += ("_" + stream);
}
return key;
}
@Override
public void clearInviteInfo(String deviceId) {
removeInviteInfo(null, deviceId, null, null);
}
}

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

File diff suppressed because it is too large Load Diff

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

@ -264,8 +264,8 @@ public class RedisGbPlayMsgListener implements MessageListener {
return; return;
} }
// 确定流是否在线 // 确定流是否在线
boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, content.getApp(), content.getStream()); Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, content.getApp(), content.getStream());
if (streamReady) { if (streamReady != null && streamReady) {
logger.info("[回复推流信息] {}/{}", content.getApp(), content.getStream()); logger.info("[回复推流信息] {}/{}", content.getApp(), content.getStream());
responseSendItem(mediaServerItem, content, toId, serial); responseSendItem(mediaServerItem, content, toId, serial);
}else { }else {
@ -301,9 +301,6 @@ public class RedisGbPlayMsgListener implements MessageListener {
String key = VideoManagerConstants.VM_MSG_STREAM_PUSH_REQUESTED; String key = VideoManagerConstants.VM_MSG_STREAM_PUSH_REQUESTED;
logger.info("[redis发送通知] 推流被请求 {}: {}/{}", key, messageForPushChannel.getApp(), messageForPushChannel.getStream()); logger.info("[redis发送通知] 推流被请求 {}: {}/{}", key, messageForPushChannel.getApp(), messageForPushChannel.getStream());
redisTemplate.convertAndSend(key, JSON.toJSON(messageForPushChannel)); redisTemplate.convertAndSend(key, JSON.toJSON(messageForPushChannel));
// redisCatchStorage.sendStreamPushRequestedMsg(messageForPushChannel);
} }
} }

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

@ -1,14 +1,16 @@
package com.genersoft.iot.vmp.storager; package com.genersoft.iot.vmp.storager;
import com.alibaba.fastjson2.JSONObject; import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.common.SystemAllInfo; import com.genersoft.iot.vmp.common.SystemAllInfo;
import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.bean.AlarmChannelMessage;
import com.genersoft.iot.vmp.media.zlm.dto.*; import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam; import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo; import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.service.bean.MessageForPushChannel; import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
import com.genersoft.iot.vmp.service.bean.ThirdPartyGB;
import com.genersoft.iot.vmp.storager.dao.dto.PlatformRegisterInfo; import com.genersoft.iot.vmp.storager.dao.dto.PlatformRegisterInfo;
import java.util.List; import java.util.List;
@ -23,42 +25,6 @@ public interface IRedisCatchStorage {
*/ */
Long getCSEQ(); Long getCSEQ();
/**
* 开始播放时将流存入
*
* @param stream 流信息
* @return
*/
boolean startPlay(StreamInfo stream);
/**
* 停止播放时删除
*
* @return
*/
boolean stopPlay(StreamInfo streamInfo);
/**
* 查询播放列表
* @return
*/
StreamInfo queryPlay(StreamInfo streamInfo);
StreamInfo queryPlayByStreamId(String steamId);
StreamInfo queryPlayByDevice(String deviceId, String channelId);
Map<String, StreamInfo> queryPlayByDeviceId(String deviceId);
boolean startPlayback(StreamInfo stream, String callId);
boolean stopPlayback(String deviceId, String channelId, String stream, String callId);
StreamInfo queryPlayback(String deviceId, String channelID, String stream, String callId);
String queryPlaybackForKey(String deviceId, String channelId, String stream, String callId);
void updatePlatformCatchInfo(ParentPlatformCatch parentPlatformCatch); void updatePlatformCatchInfo(ParentPlatformCatch parentPlatformCatch);
ParentPlatformCatch queryPlatformCatchInfo(String platformGbId); ParentPlatformCatch queryPlatformCatchInfo(String platformGbId);
@ -75,8 +41,6 @@ public interface IRedisCatchStorage {
void delPlatformRegisterInfo(String callId); void delPlatformRegisterInfo(String callId);
void cleanPlatformRegisterInfos();
void updateSendRTPSever(SendRtpItem sendRtpItem); void updateSendRTPSever(SendRtpItem sendRtpItem);
/** /**
@ -102,12 +66,6 @@ public interface IRedisCatchStorage {
*/ */
boolean isChannelSendingRTP(String channelId); boolean isChannelSendingRTP(String channelId);
/**
* 清空某个设备的所有缓存
* @param deviceId 设备ID
*/
void clearCatchByDeviceId(String deviceId);
/** /**
* 在redis添加wvp的信息 * 在redis添加wvp的信息
*/ */
@ -148,23 +106,6 @@ public interface IRedisCatchStorage {
*/ */
void removeStream(String mediaServerId, String type); void removeStream(String mediaServerId, String type);
/**
* 开始下载录像时存入
* @param streamInfo
*/
boolean startDownload(StreamInfo streamInfo, String callId);
StreamInfo queryDownload(String deviceId, String channelId, String stream, String callId);
boolean stopDownload(String deviceId, String channelId, String stream, String callId);
/**
* 查找第三方系统留下的国标预设值
* @param queryKey
* @return
*/
ThirdPartyGB queryMemberNoGBId(String queryKey);
List<OnStreamChangedHookParam> getStreams(String mediaServerId, String pull); List<OnStreamChangedHookParam> getStreams(String mediaServerId, String pull);
/** /**

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

@ -2,17 +2,18 @@ package com.genersoft.iot.vmp.storager.impl;
import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject; import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.common.SystemAllInfo; import com.genersoft.iot.vmp.common.SystemAllInfo;
import com.genersoft.iot.vmp.common.VideoManagerConstants; import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.bean.AlarmChannelMessage;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam; import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo; import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo; import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.service.bean.MessageForPushChannel; import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
import com.genersoft.iot.vmp.service.bean.ThirdPartyGB;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.dao.DeviceChannelMapper; import com.genersoft.iot.vmp.storager.dao.DeviceChannelMapper;
import com.genersoft.iot.vmp.storager.dao.dto.PlatformRegisterInfo; import com.genersoft.iot.vmp.storager.dao.dto.PlatformRegisterInfo;
@ -92,241 +93,6 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
} }
} }
/**
* 开始播放时将流存入redis
*/
@Override
public boolean startPlay(StreamInfo stream) {
redisTemplate.opsForValue().set(String.format("%S_%s_%s_%s_%s_%s", VideoManagerConstants.PLAYER_PREFIX, userSetting.getServerId(),
stream.getMediaServerId(), stream.getStream(), stream.getDeviceID(), stream.getChannelId()),
stream);
return true;
}
/**
* 停止播放时从redis删除
*/
@Override
public boolean stopPlay(StreamInfo streamInfo) {
if (streamInfo == null) {
return false;
}
Boolean result = redisTemplate.delete(String.format("%S_%s_%s_%s_%s_%s", VideoManagerConstants.PLAYER_PREFIX,
userSetting.getServerId(),
streamInfo.getMediaServerId(),
streamInfo.getStream(),
streamInfo.getDeviceID(),
streamInfo.getChannelId()));
return result != null && result;
}
/**
* 查询播放列表
*/
@Override
public StreamInfo queryPlay(StreamInfo streamInfo) {
return (StreamInfo)redisTemplate.opsForValue().get(String.format("%S_%s_%s_%s_%s_%s",
VideoManagerConstants.PLAYER_PREFIX,
userSetting.getServerId(),
streamInfo.getMediaServerId(),
streamInfo.getStream(),
streamInfo.getDeviceID(),
streamInfo.getChannelId()));
}
@Override
public StreamInfo queryPlayByStreamId(String streamId) {
List<Object> playLeys = RedisUtil.scan(redisTemplate, String.format("%S_%s_*_%s_*", VideoManagerConstants.PLAYER_PREFIX, userSetting.getServerId(), streamId));
if (playLeys.size() == 0) {
return null;
}
return (StreamInfo)redisTemplate.opsForValue().get(playLeys.get(0).toString());
}
@Override
public StreamInfo queryPlayByDevice(String deviceId, String channelId) {
List<Object> playLeys = RedisUtil.scan(redisTemplate, String.format("%S_%s_*_*_%s_%s", VideoManagerConstants.PLAYER_PREFIX,
userSetting.getServerId(),
deviceId,
channelId));
if (playLeys.size() == 0) {
return null;
}
return (StreamInfo)redisTemplate.opsForValue().get(playLeys.get(0).toString());
}
@Override
public Map<String, StreamInfo> queryPlayByDeviceId(String deviceId) {
Map<String, StreamInfo> streamInfos = new HashMap<>();
List<Object> players = RedisUtil.scan(redisTemplate, String.format("%S_%s_*_*_%s_*", VideoManagerConstants.PLAYER_PREFIX, userSetting.getServerId(),deviceId));
if (players.size() == 0) {
return streamInfos;
}
for (Object player : players) {
String key = (String) player;
StreamInfo streamInfo = JsonUtil.redisJsonToObject(redisTemplate, key, StreamInfo.class);
if (Objects.isNull(streamInfo)) {
continue;
}
streamInfos.put(streamInfo.getDeviceID() + "_" + streamInfo.getChannelId(), streamInfo);
}
return streamInfos;
}
@Override
public boolean startPlayback(StreamInfo stream, String callId) {
redisTemplate.opsForValue().set(String.format("%S_%s_%s_%s_%s_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX,
userSetting.getServerId(), stream.getMediaServerId(), stream.getDeviceID(), stream.getChannelId(), stream.getStream(), callId), stream);
return true;
}
@Override
public boolean startDownload(StreamInfo stream, String callId) {
String key=String.format("%S_%s_%s_%s_%s_%s_%s", VideoManagerConstants.DOWNLOAD_PREFIX,
userSetting.getServerId(), stream.getMediaServerId(), stream.getDeviceID(), stream.getChannelId(), stream.getStream(), callId);
if (stream.getProgress() == 1) {
logger.debug("添加下载缓存==已完成下载=》{}",key);
redisTemplate.opsForValue().set(key, stream);
}else {
logger.debug("添加下载缓存==未完成下载=》{}",key);
Duration duration = Duration.ofSeconds(60*60L);
redisTemplate.opsForValue().set(key, stream, duration);
}
return true;
}
@Override
public boolean stopDownload(String deviceId, String channelId, String stream, String callId) {
DeviceChannel deviceChannel = deviceChannelMapper.queryChannel(deviceId, channelId);
if (deviceChannel != null) {
deviceChannel.setStreamId(null);
deviceChannel.setDeviceId(deviceId);
deviceChannelMapper.update(deviceChannel);
}
if (deviceId == null) {
deviceId = "*";
}
if (channelId == null) {
channelId = "*";
}
if (stream == null) {
stream = "*";
}
if (callId == null) {
callId = "*";
}
String key = String.format("%S_%s_*_%s_%s_%s_%s", VideoManagerConstants.DOWNLOAD_PREFIX,
userSetting.getServerId(),
deviceId,
channelId,
stream,
callId
);
List<Object> scan = RedisUtil.scan(redisTemplate, key);
if (scan.size() > 0) {
for (Object keyObj : scan) {
redisTemplate.delete(keyObj);
}
}
return true;
}
@Override
public boolean stopPlayback(String deviceId, String channelId, String stream, String callId) {
DeviceChannel deviceChannel = deviceChannelMapper.queryChannel(deviceId, channelId);
if (deviceChannel != null) {
deviceChannel.setStreamId(null);
deviceChannel.setDeviceId(deviceId);
deviceChannelMapper.update(deviceChannel);
}
if (deviceId == null) {
deviceId = "*";
}
if (channelId == null) {
channelId = "*";
}
if (stream == null) {
stream = "*";
}
if (callId == null) {
callId = "*";
}
String key = String.format("%S_%s_*_%s_%s_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX,
userSetting.getServerId(),
deviceId,
channelId,
stream,
callId
);
List<Object> scan = RedisUtil.scan(redisTemplate, key);
if (scan.size() > 0) {
for (Object keyObj : scan) {
redisTemplate.delete(keyObj);
}
}
return true;
}
@Override
public StreamInfo queryPlayback(String deviceId, String channelId, String stream, String callId) {
if (stream == null && callId == null) {
return null;
}
if (deviceId == null) {
deviceId = "*";
}
if (channelId == null) {
channelId = "*";
}
if (stream == null) {
stream = "*";
}
if (callId == null) {
callId = "*";
}
String key = String.format("%S_%s_*_%s_%s_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX,
userSetting.getServerId(),
deviceId,
channelId,
stream,
callId
);
List<Object> streamInfoScan = RedisUtil.scan(redisTemplate, key);
if (streamInfoScan.size() > 0) {
return (StreamInfo) redisTemplate.opsForValue().get(streamInfoScan.get(0));
}else {
return null;
}
}
@Override
public String queryPlaybackForKey(String deviceId, String channelId, String stream, String callId) {
if (stream == null && callId == null) {
return null;
}
if (deviceId == null) {
deviceId = "*";
}
if (channelId == null) {
channelId = "*";
}
if (stream == null) {
stream = "*";
}
if (callId == null) {
callId = "*";
}
String key = String.format("%S_%s_*_%s_%s_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX,
userSetting.getServerId(),
deviceId,
channelId,
stream,
callId
);
List<Object> streamInfoScan = RedisUtil.scan(redisTemplate, key);
return (String) streamInfoScan.get(0);
}
@Override @Override
public void updatePlatformCatchInfo(ParentPlatformCatch parentPlatformCatch) { public void updatePlatformCatchInfo(ParentPlatformCatch parentPlatformCatch) {
String key = VideoManagerConstants.PLATFORM_CATCH_PREFIX + userSetting.getServerId() + "_" + parentPlatformCatch.getId(); String key = VideoManagerConstants.PLATFORM_CATCH_PREFIX + userSetting.getServerId() + "_" + parentPlatformCatch.getId();
@ -372,14 +138,6 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
redisTemplate.delete(VideoManagerConstants.PLATFORM_REGISTER_INFO_PREFIX + userSetting.getServerId() + "_" + callId); redisTemplate.delete(VideoManagerConstants.PLATFORM_REGISTER_INFO_PREFIX + userSetting.getServerId() + "_" + callId);
} }
@Override
public void cleanPlatformRegisterInfos() {
List regInfos = RedisUtil.scan(redisTemplate, VideoManagerConstants.PLATFORM_REGISTER_INFO_PREFIX + userSetting.getServerId() + "_" + "*");
for (Object key : regInfos) {
redisTemplate.delete(key.toString());
}
}
@Override @Override
public void updateSendRTPSever(SendRtpItem sendRtpItem) { public void updateSendRTPSever(SendRtpItem sendRtpItem) {
@ -536,36 +294,6 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
return RtpStreams.size() > 0; return RtpStreams.size() > 0;
} }
@Override
public void clearCatchByDeviceId(String deviceId) {
List<Object> playLeys = RedisUtil.scan(redisTemplate, String.format("%S_%s_*_%s_*", VideoManagerConstants.PLAYER_PREFIX,
userSetting.getServerId(),
deviceId));
if (playLeys.size() > 0) {
for (Object key : playLeys) {
redisTemplate.delete(key.toString());
}
}
List<Object> playBackers = RedisUtil.scan(redisTemplate, String.format("%S_%s_*_%s_*_*_*", VideoManagerConstants.PLAY_BLACK_PREFIX,
userSetting.getServerId(),
deviceId));
if (playBackers.size() > 0) {
for (Object key : playBackers) {
redisTemplate.delete(key.toString());
}
}
List<Object> deviceCache = RedisUtil.scan(redisTemplate, String.format("%S%s_%s", VideoManagerConstants.DEVICE_PREFIX,
userSetting.getServerId(),
deviceId));
if (deviceCache.size() > 0) {
for (Object key : deviceCache) {
redisTemplate.delete(key.toString());
}
}
}
@Override @Override
public void updateWVPInfo(JSONObject jsonObject, int time) { public void updateWVPInfo(JSONObject jsonObject, int time) {
String key = VideoManagerConstants.WVP_SERVER_PREFIX + userSetting.getServerId(); String key = VideoManagerConstants.WVP_SERVER_PREFIX + userSetting.getServerId();
@ -597,44 +325,6 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
redisTemplate.delete(key); redisTemplate.delete(key);
} }
@Override
public StreamInfo queryDownload(String deviceId, String channelId, String stream, String callId) {
if (stream == null && callId == null) {
return null;
}
if (deviceId == null) {
deviceId = "*";
}
if (channelId == null) {
channelId = "*";
}
if (stream == null) {
stream = "*";
}
if (callId == null) {
callId = "*";
}
String key = String.format("%S_%s_*_%s_%s_%s_%s", VideoManagerConstants.DOWNLOAD_PREFIX,
userSetting.getServerId(),
deviceId,
channelId,
stream,
callId
);
List<Object> streamInfoScan = RedisUtil.scan(redisTemplate, key);
if (streamInfoScan.size() > 0) {
return (StreamInfo) redisTemplate.opsForValue().get(streamInfoScan.get(0));
}else {
return null;
}
}
@Override
public ThirdPartyGB queryMemberNoGBId(String queryKey) {
String key = VideoManagerConstants.WVP_STREAM_GB_ID_PREFIX + queryKey;
return JsonUtil.redisJsonToObject(redisTemplate, key, ThirdPartyGB.class);
}
@Override @Override
public void removeStream(String mediaServerId, String type) { public void removeStream(String mediaServerId, String type) {
String key = VideoManagerConstants.WVP_SERVER_STREAM_PREFIX + userSetting.getServerId() + "_" + type + "_*_*_" + mediaServerId; String key = VideoManagerConstants.WVP_SERVER_STREAM_PREFIX + userSetting.getServerId() + "_" + type + "_*_*_" + mediaServerId;

6
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/device/DeviceQuery.java
View File

@ -14,6 +14,7 @@ import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander; import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.service.IDeviceChannelService; import com.genersoft.iot.vmp.service.IDeviceChannelService;
import com.genersoft.iot.vmp.service.IDeviceService; import com.genersoft.iot.vmp.service.IDeviceService;
import com.genersoft.iot.vmp.service.IInviteStreamService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.vmanager.bean.BaseTree; import com.genersoft.iot.vmp.vmanager.bean.BaseTree;
@ -62,6 +63,9 @@ public class DeviceQuery {
@Autowired @Autowired
private IRedisCatchStorage redisCatchStorage; private IRedisCatchStorage redisCatchStorage;
@Autowired
private IInviteStreamService inviteStreamService;
@Autowired @Autowired
private SIPCommander cmder; private SIPCommander cmder;
@ -184,7 +188,7 @@ public class DeviceQuery {
// 清除redis记录 // 清除redis记录
boolean isSuccess = deviceService.delete(deviceId); boolean isSuccess = deviceService.delete(deviceId);
if (isSuccess) { if (isSuccess) {
redisCatchStorage.clearCatchByDeviceId(deviceId); inviteStreamService.clearInviteInfo(deviceId);
// 停止此设备的订阅更新 // 停止此设备的订阅更新
Set<String> allKeys = dynamicTask.getAllKeys(); Set<String> allKeys = dynamicTask.getAllKeys();
for (String key : allKeys) { for (String key : allKeys) {

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

@ -2,6 +2,9 @@ package com.genersoft.iot.vmp.vmanager.gb28181.play;
import com.alibaba.fastjson2.JSONArray; import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject; import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.InviteInfo;
import com.genersoft.iot.vmp.common.InviteSessionStatus;
import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException; import com.genersoft.iot.vmp.conf.exception.ControllerException;
@ -14,12 +17,13 @@ import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander; import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils; import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IInviteStreamService;
import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IMediaService; import com.genersoft.iot.vmp.service.IMediaService;
import com.genersoft.iot.vmp.service.IPlayService; import com.genersoft.iot.vmp.service.IPlayService;
import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.vmanager.bean.DeferredResultEx;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.StreamContent; import com.genersoft.iot.vmp.vmanager.bean.StreamContent;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult; import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
@ -59,6 +63,9 @@ public class PlayController {
@Autowired @Autowired
private IRedisCatchStorage redisCatchStorage; private IRedisCatchStorage redisCatchStorage;
@Autowired
private IInviteStreamService inviteStreamService;
@Autowired @Autowired
private ZLMRESTfulUtils zlmresTfulUtils; private ZLMRESTfulUtils zlmresTfulUtils;
@ -88,14 +95,12 @@ public class PlayController {
Device device = storager.queryVideoDevice(deviceId); Device device = storager.queryVideoDevice(deviceId);
MediaServerItem newMediaServerItem = playService.getNewMediaServerItem(device); MediaServerItem newMediaServerItem = playService.getNewMediaServerItem(device);
RequestMessage msg = new RequestMessage(); RequestMessage requestMessage = new RequestMessage();
String key = DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId; String key = DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId;
boolean exist = resultHolder.exist(key, null); requestMessage.setKey(key);
msg.setKey(key);
String uuid = UUID.randomUUID().toString(); String uuid = UUID.randomUUID().toString();
msg.setId(uuid); requestMessage.setId(uuid);
DeferredResult<WVPResult<StreamContent>> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue()); DeferredResult<WVPResult<StreamContent>> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue());
DeferredResultEx<WVPResult<StreamContent>> deferredResultEx = new DeferredResultEx<>(result);
result.onTimeout(()->{ result.onTimeout(()->{
logger.info("点播接口等待超时"); logger.info("点播接口等待超时");
@ -103,32 +108,33 @@ public class PlayController {
WVPResult<StreamInfo> wvpResult = new WVPResult<>(); WVPResult<StreamInfo> wvpResult = new WVPResult<>();
wvpResult.setCode(ErrorCode.ERROR100.getCode()); wvpResult.setCode(ErrorCode.ERROR100.getCode());
wvpResult.setMsg("点播超时"); wvpResult.setMsg("点播超时");
msg.setData(wvpResult); requestMessage.setData(wvpResult);
resultHolder.invokeResult(msg); resultHolder.invokeResult(requestMessage);
}); });
// TODO 在点播未成功的情况下在此调用接口点播会导致返回的流地址ip错误
deferredResultEx.setFilter(result1 -> {
WVPResult<StreamInfo> wvpResult1 = (WVPResult<StreamInfo>)result1;
WVPResult<StreamContent> resultStream = new WVPResult<>();
resultStream.setCode(wvpResult1.getCode());
resultStream.setMsg(wvpResult1.getMsg());
if (wvpResult1.getCode() == ErrorCode.SUCCESS.getCode()) {
StreamInfo data = wvpResult1.getData().clone();
if (userSetting.getUseSourceIpAsStreamIp()) {
data.channgeStreamIp(request.getLocalName());
}
resultStream.setData(new StreamContent(wvpResult1.getData()));
}
return resultStream;
});
// 录像查询以channelId作为deviceId查询 // 录像查询以channelId作为deviceId查询
resultHolder.put(key, uuid, deferredResultEx); resultHolder.put(key, uuid, result);
if (!exist) { playService.play(newMediaServerItem, deviceId, channelId, (code, msg, data) -> {
playService.play(newMediaServerItem, deviceId, channelId, null, null, null); WVPResult<StreamContent> wvpResult = new WVPResult<>();
} if (code == InviteErrorCode.SUCCESS.getCode()) {
wvpResult.setCode(ErrorCode.SUCCESS.getCode());
wvpResult.setMsg(ErrorCode.SUCCESS.getMsg());
if (data != null) {
StreamInfo streamInfo = (StreamInfo)data;
if (userSetting.getUseSourceIpAsStreamIp()) {
streamInfo.channgeStreamIp(request.getLocalName());
}
wvpResult.setData(new StreamContent(streamInfo));
}
}else {
wvpResult.setCode(code);
wvpResult.setMsg(msg);
}
requestMessage.setData(wvpResult);
resultHolder.invokeResult(requestMessage);
});
return result; return result;
} }
@ -149,21 +155,22 @@ public class PlayController {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "设备[" + deviceId + "]不存在"); throw new ControllerException(ErrorCode.ERROR100.getCode(), "设备[" + deviceId + "]不存在");
} }
StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId); InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId);
if (streamInfo == null) { if (inviteInfo == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "点播未找到"); throw new ControllerException(ErrorCode.ERROR100.getCode(), "点播未找到");
} }
if (InviteSessionStatus.ok == inviteInfo.getStatus()) {
try { try {
logger.warn("[停止点播] {}/{}", device.getDeviceId(), channelId); logger.warn("[停止点播] {}/{}", device.getDeviceId(), channelId);
cmder.streamByeCmd(device, channelId, streamInfo.getStream(), null, null); cmder.streamByeCmd(device, channelId, inviteInfo.getStream(), null, null);
} catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) { } catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) {
logger.error("[命令发送失败] 停止点播， 发送BYE: {}", e.getMessage()); logger.error("[命令发送失败] 停止点播， 发送BYE: {}", e.getMessage());
throw new ControllerException(ErrorCode.ERROR100.getCode(), "命令发送失败: " + e.getMessage()); throw new ControllerException(ErrorCode.ERROR100.getCode(), "命令发送失败: " + e.getMessage());
}
} }
redisCatchStorage.stopPlay(streamInfo); inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId);
storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); storager.stopPlay(deviceId, channelId);
JSONObject json = new JSONObject(); JSONObject json = new JSONObject();
json.put("deviceId", deviceId); json.put("deviceId", deviceId);
json.put("channelId", channelId); json.put("channelId", channelId);
@ -178,15 +185,14 @@ public class PlayController {
@Parameter(name = "streamId", description = "视频流ID", required = true) @Parameter(name = "streamId", description = "视频流ID", required = true)
@PostMapping("/convert/{streamId}") @PostMapping("/convert/{streamId}")
public JSONObject playConvert(@PathVariable String streamId) { public JSONObject playConvert(@PathVariable String streamId) {
StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(streamId); // StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(streamId);
if (streamInfo == null) {
streamInfo = redisCatchStorage.queryPlayback(null, null, streamId, null); InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, streamId);
} if (inviteInfo == null || inviteInfo.getStreamInfo() == null) {
if (streamInfo == null) {
logger.warn("视频转码API调用失败！, 视频流已经停止!"); logger.warn("视频转码API调用失败！, 视频流已经停止!");
throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到视频流信息, 视频流可能已经停止"); throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到视频流信息, 视频流可能已经停止");
} }
MediaServerItem mediaInfo = mediaServerService.getOne(streamInfo.getMediaServerId()); MediaServerItem mediaInfo = mediaServerService.getOne(inviteInfo.getStreamInfo().getMediaServerId());
JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaInfo, streamId); JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaInfo, streamId);
if (!rtpInfo.getBoolean("exist")) { if (!rtpInfo.getBoolean("exist")) {
logger.warn("视频转码API调用失败！, 视频流已停止推流!"); logger.warn("视频转码API调用失败！, 视频流已停止推流!");

75
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/playback/PlaybackController.java
View File

@ -1,15 +1,22 @@
package com.genersoft.iot.vmp.vmanager.gb28181.playback; package com.genersoft.iot.vmp.vmanager.gb28181.playback;
import com.genersoft.iot.vmp.common.InviteInfo;
import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException; import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.conf.exception.ServiceException; import com.genersoft.iot.vmp.conf.exception.ServiceException;
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException; import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder; import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage; import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory; import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.service.IInviteStreamService;
import com.genersoft.iot.vmp.service.IPlayService; import com.genersoft.iot.vmp.service.IPlayService;
import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.StreamContent; import com.genersoft.iot.vmp.vmanager.bean.StreamContent;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult; import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
@ -20,17 +27,13 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.ObjectUtils; import org.springframework.util.ObjectUtils;
import org.springframework.web.bind.annotation.CrossOrigin;
import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController; import org.springframework.web.bind.annotation.RestController;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import org.springframework.web.context.request.async.DeferredResult; import org.springframework.web.context.request.async.DeferredResult;
import javax.servlet.http.HttpServletRequest;
import javax.sip.InvalidArgumentException; import javax.sip.InvalidArgumentException;
import javax.sip.SipException; import javax.sip.SipException;
import java.text.ParseException; import java.text.ParseException;
@ -59,6 +62,9 @@ public class PlaybackController {
@Autowired @Autowired
private IRedisCatchStorage redisCatchStorage; private IRedisCatchStorage redisCatchStorage;
@Autowired
private IInviteStreamService inviteStreamService;
@Autowired @Autowired
private IPlayService playService; private IPlayService playService;
@ -74,8 +80,8 @@ public class PlaybackController {
@Parameter(name = "startTime", description = "开始时间", required = true) @Parameter(name = "startTime", description = "开始时间", required = true)
@Parameter(name = "endTime", description = "结束时间", required = true) @Parameter(name = "endTime", description = "结束时间", required = true)
@GetMapping("/start/{deviceId}/{channelId}") @GetMapping("/start/{deviceId}/{channelId}")
public DeferredResult<WVPResult<StreamContent>> start(@PathVariable String deviceId, @PathVariable String channelId, public DeferredResult<WVPResult<StreamContent>> start(HttpServletRequest request, @PathVariable String deviceId, @PathVariable String channelId,
String startTime, String endTime) { String startTime, String endTime) {
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
logger.debug(String.format("设备回放 API调用，deviceId：%s ，channelId：%s", deviceId, channelId)); logger.debug(String.format("设备回放 API调用，deviceId：%s ，channelId：%s", deviceId, channelId));
@ -86,22 +92,31 @@ public class PlaybackController {
DeferredResult<WVPResult<StreamContent>> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue()); DeferredResult<WVPResult<StreamContent>> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue());
resultHolder.put(key, uuid, result); resultHolder.put(key, uuid, result);
WVPResult<StreamContent> wvpResult = new WVPResult<>(); RequestMessage requestMessage = new RequestMessage();
requestMessage.setKey(key);
requestMessage.setId(uuid);
RequestMessage msg = new RequestMessage(); playService.playBack(deviceId, channelId, startTime, endTime,
msg.setKey(key); (code, msg, data)->{
msg.setId(uuid);
playService.playBack(deviceId, channelId, startTime, endTime, null, WVPResult<StreamContent> wvpResult = new WVPResult<>();
playBackResult->{ if (code == InviteErrorCode.SUCCESS.getCode()) {
wvpResult.setCode(playBackResult.getCode()); wvpResult.setCode(ErrorCode.SUCCESS.getCode());
wvpResult.setMsg(playBackResult.getMsg()); wvpResult.setMsg(ErrorCode.SUCCESS.getMsg());
if (playBackResult.getCode() == ErrorCode.SUCCESS.getCode()) {
StreamInfo streamInfo = (StreamInfo)playBackResult.getData(); if (data != null) {
wvpResult.setData(new StreamContent(streamInfo)); StreamInfo streamInfo = (StreamInfo)data;
if (userSetting.getUseSourceIpAsStreamIp()) {
streamInfo.channgeStreamIp(request.getLocalName());
}
wvpResult.setData(new StreamContent(streamInfo));
}
}else {
wvpResult.setCode(code);
wvpResult.setMsg(msg);
} }
msg.setData(wvpResult); requestMessage.setData(wvpResult);
resultHolder.invokeResult(msg); resultHolder.invokeResult(requestMessage);
}); });
return result; return result;
@ -169,14 +184,15 @@ public class PlaybackController {
@GetMapping("/seek/{streamId}/{seekTime}") @GetMapping("/seek/{streamId}/{seekTime}")
public void playSeek(@PathVariable String streamId, @PathVariable long seekTime) { public void playSeek(@PathVariable String streamId, @PathVariable long seekTime) {
logger.info("playSeek: "+streamId+", "+seekTime); logger.info("playSeek: "+streamId+", "+seekTime);
StreamInfo streamInfo = redisCatchStorage.queryPlayback(null, null, streamId, null); InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(InviteSessionType.PLAYBACK, streamId);
if (null == streamInfo) {
if (null == inviteInfo || inviteInfo.getStreamInfo() == null) {
logger.warn("streamId不存在!"); logger.warn("streamId不存在!");
throw new ControllerException(ErrorCode.ERROR400.getCode(), "streamId不存在"); throw new ControllerException(ErrorCode.ERROR400.getCode(), "streamId不存在");
} }
Device device = storager.queryVideoDevice(streamInfo.getDeviceID()); Device device = storager.queryVideoDevice(inviteInfo.getDeviceId());
try { try {
cmder.playSeekCmd(device, streamInfo, seekTime); cmder.playSeekCmd(device, inviteInfo.getStreamInfo(), seekTime);
} catch (InvalidArgumentException | ParseException | SipException e) { } catch (InvalidArgumentException | ParseException | SipException e) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), e.getMessage()); throw new ControllerException(ErrorCode.ERROR100.getCode(), e.getMessage());
} }
@ -188,8 +204,9 @@ public class PlaybackController {
@GetMapping("/speed/{streamId}/{speed}") @GetMapping("/speed/{streamId}/{speed}")
public void playSpeed(@PathVariable String streamId, @PathVariable Double speed) { public void playSpeed(@PathVariable String streamId, @PathVariable Double speed) {
logger.info("playSpeed: "+streamId+", "+speed); logger.info("playSpeed: "+streamId+", "+speed);
StreamInfo streamInfo = redisCatchStorage.queryPlayback(null, null, streamId, null); InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(InviteSessionType.PLAYBACK, streamId);
if (null == streamInfo) {
if (null == inviteInfo || inviteInfo.getStreamInfo() == null) {
logger.warn("streamId不存在!"); logger.warn("streamId不存在!");
throw new ControllerException(ErrorCode.ERROR400.getCode(), "streamId不存在"); throw new ControllerException(ErrorCode.ERROR400.getCode(), "streamId不存在");
} }
@ -197,9 +214,9 @@ public class PlaybackController {
logger.warn("不支持的speed： " + speed); logger.warn("不支持的speed： " + speed);
throw new ControllerException(ErrorCode.ERROR100.getCode(), "不支持的speed（0.25 0.5 1、2、4）"); throw new ControllerException(ErrorCode.ERROR100.getCode(), "不支持的speed（0.25 0.5 1、2、4）");
} }
Device device = storager.queryVideoDevice(streamInfo.getDeviceID()); Device device = storager.queryVideoDevice(inviteInfo.getDeviceId());
try { try {
cmder.playSpeedCmd(device, streamInfo, speed); cmder.playSpeedCmd(device, inviteInfo.getStreamInfo(), speed);
} catch (InvalidArgumentException | ParseException | SipException e) { } catch (InvalidArgumentException | ParseException | SipException e) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), e.getMessage()); throw new ControllerException(ErrorCode.ERROR100.getCode(), e.getMessage());
} }

45
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/record/GBRecordController.java
View File

@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.vmanager.gb28181.record; package com.genersoft.iot.vmp.vmanager.gb28181.record;
import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException; import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException; import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
import com.genersoft.iot.vmp.gb28181.bean.Device; import com.genersoft.iot.vmp.gb28181.bean.Device;
@ -10,6 +11,7 @@ import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander; import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.service.IDeviceService; import com.genersoft.iot.vmp.service.IDeviceService;
import com.genersoft.iot.vmp.service.IPlayService; import com.genersoft.iot.vmp.service.IPlayService;
import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.utils.DateUtil; import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
@ -27,6 +29,7 @@ import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController; import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.context.request.async.DeferredResult; import org.springframework.web.context.request.async.DeferredResult;
import javax.servlet.http.HttpServletRequest;
import javax.sip.InvalidArgumentException; import javax.sip.InvalidArgumentException;
import javax.sip.SipException; import javax.sip.SipException;
import java.text.ParseException; import java.text.ParseException;
@ -55,8 +58,8 @@ public class GBRecordController {
@Autowired @Autowired
private IDeviceService deviceService; private IDeviceService deviceService;
@Autowired
private UserSetting userSetting;
@Operation(summary = "录像查询") @Operation(summary = "录像查询")
@Parameter(name = "deviceId", description = "设备国标编号", required = true) @Parameter(name = "deviceId", description = "设备国标编号", required = true)
@ -119,8 +122,8 @@ public class GBRecordController {
@Parameter(name = "endTime", description = "结束时间", required = true) @Parameter(name = "endTime", description = "结束时间", required = true)
@Parameter(name = "downloadSpeed", description = "下载倍速", required = true) @Parameter(name = "downloadSpeed", description = "下载倍速", required = true)
@GetMapping("/download/start/{deviceId}/{channelId}") @GetMapping("/download/start/{deviceId}/{channelId}")
public DeferredResult<WVPResult<StreamContent>> download(@PathVariable String deviceId, @PathVariable String channelId, public DeferredResult<WVPResult<StreamContent>> download(HttpServletRequest request, @PathVariable String deviceId, @PathVariable String channelId,
String startTime, String endTime, String downloadSpeed) { String startTime, String endTime, String downloadSpeed) {
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
logger.debug(String.format("历史媒体下载 API调用，deviceId：%s，channelId：%s，downloadSpeed：%s", deviceId, channelId, downloadSpeed)); logger.debug(String.format("历史媒体下载 API调用，deviceId：%s，channelId：%s，downloadSpeed：%s", deviceId, channelId, downloadSpeed));
@ -130,22 +133,32 @@ public class GBRecordController {
String key = DeferredResultHolder.CALLBACK_CMD_DOWNLOAD + deviceId + channelId; String key = DeferredResultHolder.CALLBACK_CMD_DOWNLOAD + deviceId + channelId;
DeferredResult<WVPResult<StreamContent>> result = new DeferredResult<>(30000L); DeferredResult<WVPResult<StreamContent>> result = new DeferredResult<>(30000L);
resultHolder.put(key, uuid, result); resultHolder.put(key, uuid, result);
RequestMessage msg = new RequestMessage(); RequestMessage requestMessage = new RequestMessage();
msg.setId(uuid); requestMessage.setId(uuid);
msg.setKey(key); requestMessage.setKey(key);
WVPResult<StreamContent> wvpResult = new WVPResult<>();
playService.download(deviceId, channelId, startTime, endTime, Integer.parseInt(downloadSpeed), null, playBackResult->{ playService.download(deviceId, channelId, startTime, endTime, Integer.parseInt(downloadSpeed),
(code, msg, data)->{
wvpResult.setCode(playBackResult.getCode()); WVPResult<StreamContent> wvpResult = new WVPResult<>();
wvpResult.setMsg(playBackResult.getMsg()); if (code == InviteErrorCode.SUCCESS.getCode()) {
if (playBackResult.getCode() == ErrorCode.SUCCESS.getCode()) { wvpResult.setCode(ErrorCode.SUCCESS.getCode());
StreamInfo streamInfo = (StreamInfo)playBackResult.getData(); wvpResult.setMsg(ErrorCode.SUCCESS.getMsg());
wvpResult.setData(new StreamContent(streamInfo));
if (data != null) {
StreamInfo streamInfo = (StreamInfo)data;
if (userSetting.getUseSourceIpAsStreamIp()) {
streamInfo.channgeStreamIp(request.getLocalName());
}
wvpResult.setData(new StreamContent(streamInfo));
}
}else {
wvpResult.setCode(code);
wvpResult.setMsg(msg);
} }
msg.setData(wvpResult); requestMessage.setData(wvpResult);
resultHolder.invokeResult(msg); resultHolder.invokeResult(requestMessage);
}); });
return result; return result;

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

@ -1,7 +1,8 @@
package com.genersoft.iot.vmp.web.gb28181; package com.genersoft.iot.vmp.web.gb28181;
import com.alibaba.fastjson2.JSONObject; import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.common.InviteInfo;
import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException; import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
import com.genersoft.iot.vmp.gb28181.bean.Device; import com.genersoft.iot.vmp.gb28181.bean.Device;
@ -9,13 +10,18 @@ import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander; import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IDeviceService; import com.genersoft.iot.vmp.service.IDeviceService;
import com.genersoft.iot.vmp.service.IInviteStreamService;
import com.genersoft.iot.vmp.service.IPlayService; import com.genersoft.iot.vmp.service.IPlayService;
import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*; import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.context.request.async.DeferredResult; import org.springframework.web.context.request.async.DeferredResult;
import javax.sip.InvalidArgumentException; import javax.sip.InvalidArgumentException;
@ -45,6 +51,9 @@ public class ApiStreamController {
@Autowired @Autowired
private IRedisCatchStorage redisCatchStorage; private IRedisCatchStorage redisCatchStorage;
@Autowired
private IInviteStreamService inviteStreamService;
@Autowired @Autowired
private IDeviceService deviceService; private IDeviceService deviceService;
@ -111,46 +120,53 @@ public class ApiStreamController {
return resultDeferredResult; return resultDeferredResult;
} }
MediaServerItem newMediaServerItem = playService.getNewMediaServerItem(device); MediaServerItem newMediaServerItem = playService.getNewMediaServerItem(device);
playService.play(newMediaServerItem, serial, code, (mediaServerItem, response)->{
StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(serial, code);
JSONObject result = new JSONObject(); playService.play(newMediaServerItem, serial, code, (errorCode, msg, data) -> {
result.put("StreamID", streamInfo.getStream()); if (errorCode == InviteErrorCode.SUCCESS.getCode()) {
result.put("DeviceID", device.getDeviceId()); InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, serial, code);
result.put("ChannelID", code); if (inviteInfo != null && inviteInfo.getStreamInfo() != null) {
result.put("ChannelName", deviceChannel.getName()); JSONObject result = new JSONObject();
result.put("ChannelCustomName", ""); result.put("StreamID", inviteInfo.getStreamInfo().getStream());
result.put("FLV", streamInfo.getFlv().getUrl()); result.put("DeviceID", device.getDeviceId());
result.put("WS_FLV", streamInfo.getWs_flv().getUrl()); result.put("ChannelID", code);
result.put("RTMP", streamInfo.getRtmp().getUrl()); result.put("ChannelName", deviceChannel.getName());
result.put("HLS", streamInfo.getHls().getUrl()); result.put("ChannelCustomName", "");
result.put("RTSP", streamInfo.getRtsp().getUrl()); result.put("FLV", inviteInfo.getStreamInfo().getFlv().getUrl());
result.put("WEBRTC", streamInfo.getRtc().getUrl()); result.put("WS_FLV", inviteInfo.getStreamInfo().getWs_flv().getUrl());
result.put("CDN", ""); result.put("RTMP", inviteInfo.getStreamInfo().getRtmp().getUrl());
result.put("SnapURL", ""); result.put("HLS", inviteInfo.getStreamInfo().getHls().getUrl());
result.put("Transport", device.getTransport()); result.put("RTSP", inviteInfo.getStreamInfo().getRtsp().getUrl());
result.put("StartAt", ""); result.put("WEBRTC", inviteInfo.getStreamInfo().getRtc().getUrl());
result.put("Duration", ""); result.put("CDN", "");
result.put("SourceVideoCodecName", ""); result.put("SnapURL", "");
result.put("SourceVideoWidth", ""); result.put("Transport", device.getTransport());
result.put("SourceVideoHeight", ""); result.put("StartAt", "");
result.put("SourceVideoFrameRate", ""); result.put("Duration", "");
result.put("SourceAudioCodecName", ""); result.put("SourceVideoCodecName", "");
result.put("SourceAudioSampleRate", ""); result.put("SourceVideoWidth", "");
result.put("AudioEnable", ""); result.put("SourceVideoHeight", "");
result.put("Ondemand", ""); result.put("SourceVideoFrameRate", "");
result.put("InBytes", ""); result.put("SourceAudioCodecName", "");
result.put("InBitRate", ""); result.put("SourceAudioSampleRate", "");
result.put("OutBytes", ""); result.put("AudioEnable", "");
result.put("NumOutputs", ""); result.put("Ondemand", "");
result.put("CascadeSize", ""); result.put("InBytes", "");
result.put("RelaySize", ""); result.put("InBitRate", "");
result.put("ChannelPTZType", "0"); result.put("OutBytes", "");
resultDeferredResult.setResult(result); result.put("NumOutputs", "");
}, (eventResult) -> { result.put("CascadeSize", "");
JSONObject result = new JSONObject(); result.put("RelaySize", "");
result.put("error", "channel[ " + code + " ] " + eventResult.msg); result.put("ChannelPTZType", "0");
resultDeferredResult.setResult(result); resultDeferredResult.setResult(result);
}, null); }
}else {
JSONObject result = new JSONObject();
result.put("error", "channel[ " + code + " ] " + msg);
resultDeferredResult.setResult(result);
}
});
return resultDeferredResult; return resultDeferredResult;
} }
@ -171,8 +187,8 @@ public class ApiStreamController {
){ ){
StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(serial, code); InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, serial, code);
if (streamInfo == null) { if (inviteInfo == null) {
JSONObject result = new JSONObject(); JSONObject result = new JSONObject();
result.put("error","未找到流信息"); result.put("error","未找到流信息");
return result; return result;
@ -184,14 +200,14 @@ public class ApiStreamController {
return result; return result;
} }
try { try {
cmder.streamByeCmd(device, code, streamInfo.getStream(), null); cmder.streamByeCmd(device, code, inviteInfo.getStream(), null);
} catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) { } catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) {
JSONObject result = new JSONObject(); JSONObject result = new JSONObject();
result.put("error","发送BYE失败：" + e.getMessage()); result.put("error","发送BYE失败：" + e.getMessage());
return result; return result;
} }
redisCatchStorage.stopPlay(streamInfo); inviteStreamService.removeInviteInfo(inviteInfo);
storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); storager.stopPlay(inviteInfo.getDeviceId(), inviteInfo.getChannelId());
return null; return null;
} }

30
web_src/src/components/GBRecordDetail.vue
View File

@ -182,9 +182,11 @@
this.playerStyle["height"] = this.winHeight + "px"; this.playerStyle["height"] = this.winHeight + "px";
this.chooseDate = moment().format('YYYY-MM-DD') this.chooseDate = moment().format('YYYY-MM-DD')
this.dateChange(); this.dateChange();
window.addEventListener('beforeunload', this.stopPlayRecord)
}, },
destroyed() { destroyed() {
this.$destroy('recordVideoPlayer'); this.$destroy('recordVideoPlayer');
window.removeEventListener('beforeunload', this.stopPlayRecord)
}, },
methods: { methods: {
dateChange(){ dateChange(){
@ -338,14 +340,18 @@
}); });
}, },
stopPlayRecord: function (callback) { stopPlayRecord: function (callback) {
this.$refs["recordVideoPlayer"].pause(); console.log("停止录像回放")
this.videoUrl = ''; if (this.streamId !== "") {
this.$axios({ this.$refs["recordVideoPlayer"].pause();
method: 'get', this.videoUrl = '';
url: '/api/playback/stop/' + this.deviceId + "/" + this.channelId + "/" + this.streamId this.$axios({
}).then(function (res) { method: 'get',
if (callback) callback() url: '/api/playback/stop/' + this.deviceId + "/" + this.channelId + "/" + this.streamId
}); }).then(function (res) {
if (callback) callback()
});
}
}, },
getDataWidth(item){ getDataWidth(item){
let timeForFile = this.getTimeForFile(item); let timeForFile = this.getTimeForFile(item);
@ -423,8 +429,14 @@
return hStr + ":" + mStr + ":" + sStr return hStr + ":" + mStr + ":" + sStr
}, },
goBack(){ goBack(){
// 如果正在进行录像回放则，发送停止
if (this.streamId !== "") {
this.stopPlayRecord(()=> {
this.streamId = "";
})
}
window.history.go(-1); window.history.go(-1);
} },
} }
}; };
</script> </script>

19
web_src/src/components/dialog/recordDownload.vue
View File

@ -7,6 +7,7 @@
</el-col> </el-col>
<el-col :span="6" > <el-col :span="6" >
<el-button icon="el-icon-download" v-if="percentage < 100" size="mini" title="点击下载可将以缓存部分下载到本地" @click="download()">停止缓存并下载</el-button> <el-button icon="el-icon-download" v-if="percentage < 100" size="mini" title="点击下载可将以缓存部分下载到本地" @click="download()">停止缓存并下载</el-button>
<el-button icon="el-icon-download" v-if="downloadFile" size="mini" title="点击下载" @click="downloadFileClientEvent()">点击下载</el-button>
</el-col> </el-col>
</el-row> </el-row>
</el-dialog> </el-dialog>
@ -21,7 +22,7 @@ import moment from "moment";
export default { export default {
name: 'recordDownload', name: 'recordDownload',
created() { created() {
window.addEventListener('beforeunload', this.stopDownloadRecord)
}, },
data() { data() {
@ -39,7 +40,8 @@ export default {
taskId: null, taskId: null,
getProgressRun: false, getProgressRun: false,
getProgressForFileRun: false, getProgressForFileRun: false,
timer: null timer: null,
downloadFile: null,
}; };
}, },
@ -187,8 +189,9 @@ export default {
this.percentage = parseFloat(res.data.data[0].percentage)*100 this.percentage = parseFloat(res.data.data[0].percentage)*100
if (res.data.data[0].percentage === '1') { if (res.data.data[0].percentage === '1') {
this.getProgressForFileRun = false; this.getProgressForFileRun = false;
window.open(res.data.data[0].downloadFile) this.downloadFile = res.data.data[0].downloadFile
this.close(); this.title = "文件处理完成，点击按扭下载"
// window.open(res.data.data[0].downloadFile)
}else { }else {
if (callback)callback() if (callback)callback()
} }
@ -196,7 +199,13 @@ export default {
}).catch(function (error) { }).catch(function (error) {
console.log(error); console.log(error);
}); });
} },
downloadFileClientEvent: function (){
window.open(this.downloadFile )
}
},
destroyed() {
window.removeEventListener('beforeunload', this.stopDownloadRecord)
} }
}; };
</script> </script>