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

Merge branch 'wvp-28181-2.0' into wvp-pro-record 

# Conflicts:
#	src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java
#	src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java
#	src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookSubscribe.java
#	src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
pull/375/head
648540858 2022-03-03 16:28:32 +08:00
parent 1dcdbc3742 2eb1ca2d94
commit 14174ff400
38 changed files with 312 additions and 234 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
sql/mysql.sql
View File

@ -415,6 +415,7 @@ DROP TABLE IF EXISTS `stream_proxy`;
CREATE TABLE `stream_proxy` ( CREATE TABLE `stream_proxy` (
`id` int NOT NULL AUTO_INCREMENT, `id` int NOT NULL AUTO_INCREMENT,
`type` varchar(50) COLLATE utf8mb4_general_ci NOT NULL, `type` varchar(50) COLLATE utf8mb4_general_ci NOT NULL,
`name` varchar(255) COLLATE utf8mb4_general_ci NOT NULL,
`app` varchar(255) COLLATE utf8mb4_general_ci NOT NULL, `app` varchar(255) COLLATE utf8mb4_general_ci NOT NULL,
`stream` varchar(255) COLLATE utf8mb4_general_ci NOT NULL, `stream` varchar(255) COLLATE utf8mb4_general_ci NOT NULL,
`url` varchar(255) COLLATE utf8mb4_general_ci DEFAULT NULL, `url` varchar(255) COLLATE utf8mb4_general_ci DEFAULT NULL,

10
src/main/java/com/genersoft/iot/vmp/common/StreamInfo.java
View File

@ -5,7 +5,7 @@ import com.alibaba.fastjson.JSONArray;
public class StreamInfo { public class StreamInfo {
private String app; private String app;
private String streamId; private String stream;
private String deviceID; private String deviceID;
private String channelId; private String channelId;
private String flv; private String flv;
@ -153,12 +153,12 @@ public class StreamInfo {
this.ws_ts = ws_ts; this.ws_ts = ws_ts;
} }
public String getStreamId() { public String getStream() {
return streamId; return stream;
} }
public void setStreamId(String streamId) { public void setStream(String stream) {
this.streamId = streamId; this.stream = stream;
} }
public String getRtc() { public String getRtc() {

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

@ -29,6 +29,7 @@ public class VideoManagerConstants {
// 此处多了一个_，暂不修改 // 此处多了一个_，暂不修改
public static final String PLAYER_PREFIX = "VMP_PLAYER_"; public static final String PLAYER_PREFIX = "VMP_PLAYER_";
public static final String PLAY_BLACK_PREFIX = "VMP_PLAYBACK_"; public static final String PLAY_BLACK_PREFIX = "VMP_PLAYBACK_";
public static final String PLAY_INFO_PREFIX = "VMP_PLAY_INFO_";
public static final String DOWNLOAD_PREFIX = "VMP_DOWNLOAD_"; public static final String DOWNLOAD_PREFIX = "VMP_DOWNLOAD_";

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

@ -4,11 +4,12 @@ public class SsrcTransaction {
private String deviceId; private String deviceId;
private String channelId; private String channelId;
private String ssrc; private String callId;
private String streamId; private String stream;
private byte[] transaction; private byte[] transaction;
private byte[] dialog; private byte[] dialog;
private String mediaServerId; private String mediaServerId;
private String ssrc;
public String getDeviceId() { public String getDeviceId() {
return deviceId; return deviceId;
@ -26,20 +27,20 @@ public class SsrcTransaction {
this.channelId = channelId; this.channelId = channelId;
} }
public String getSsrc() { public String getCallId() {
return ssrc; return callId;
} }
public void setSsrc(String ssrc) { public void setCallId(String callId) {
this.ssrc = ssrc; this.callId = callId;
} }
public String getStreamId() { public String getStream() {
return streamId; return stream;
} }
public void setStreamId(String streamId) { public void setStream(String stream) {
this.streamId = streamId; this.stream = stream;
} }
public byte[] getTransaction() { public byte[] getTransaction() {
@ -65,4 +66,12 @@ public class SsrcTransaction {
public void setMediaServerId(String mediaServerId) { public void setMediaServerId(String mediaServerId) {
this.mediaServerId = mediaServerId; this.mediaServerId = mediaServerId;
} }
public String getSsrc() {
return ssrc;
}
public void setSsrc(String ssrc) {
this.ssrc = ssrc;
}
} }

36
src/main/java/com/genersoft/iot/vmp/gb28181/event/SipSubscribe.java
View File

@ -23,24 +23,36 @@ public class SipSubscribe {
private Map<String, SipSubscribe.Event> okSubscribes = new ConcurrentHashMap<>(); private Map<String, SipSubscribe.Event> okSubscribes = new ConcurrentHashMap<>();
private Map<String, Date> timeSubscribes = new ConcurrentHashMap<>(); private Map<String, Date> okTimeSubscribes = new ConcurrentHashMap<>();
private Map<String, Date> errorTimeSubscribes = new ConcurrentHashMap<>();
// @Scheduled(cron="*/5 * * * * ?") //每五秒执行一次 // @Scheduled(cron="*/5 * * * * ?") //每五秒执行一次
// @Scheduled(fixedRate= 100 * 60 * 60 ) // @Scheduled(fixedRate= 100 * 60 * 60 )
@Scheduled(cron="0 0 * * * ?") //每小时执行一次， 每个整点 @Scheduled(cron="0 0/5 * * * ?") //每5分钟执行一次
public void execute(){ public void execute(){
logger.info("[定时任务] 清理过期的订阅信息"); logger.info("[定时任务] 清理过期的订阅信息");
Calendar calendar = Calendar.getInstance(); Calendar calendar = Calendar.getInstance();
calendar.setTime(new Date()); calendar.setTime(new Date());
calendar.set(Calendar.HOUR, calendar.get(Calendar.HOUR) - 1); calendar.set(Calendar.MINUTE, calendar.get(Calendar.MINUTE) - 5);
for (String key : timeSubscribes.keySet()) {
if (timeSubscribes.get(key).before(calendar.getTime())){ for (String key : okTimeSubscribes.keySet()) {
logger.info("[定时任务] 清理过期的订阅信息： {}", key); if (okTimeSubscribes.get(key).before(calendar.getTime())){
errorSubscribes.remove(key); // logger.info("[定时任务] 清理过期的订阅信息： {}", key);
okSubscribes.remove(key); okSubscribes.remove(key);
timeSubscribes.remove(key); okTimeSubscribes.remove(key);
} }
} }
for (String key : errorTimeSubscribes.keySet()) {
if (errorTimeSubscribes.get(key).before(calendar.getTime())){
// logger.info("[定时任务] 清理过期的订阅信息： {}", key);
errorSubscribes.remove(key);
errorTimeSubscribes.remove(key);
}
}
logger.info("okTimeSubscribes.size:{}",okTimeSubscribes.size());
logger.info("okSubscribes.size:{}",okSubscribes.size());
logger.info("errorTimeSubscribes.size:{}",errorTimeSubscribes.size());
logger.info("errorSubscribes.size:{}",errorSubscribes.size());
} }
public interface Event { public interface Event {
@ -105,12 +117,12 @@ public class SipSubscribe {
public void addErrorSubscribe(String key, SipSubscribe.Event event) { public void addErrorSubscribe(String key, SipSubscribe.Event event) {
errorSubscribes.put(key, event); errorSubscribes.put(key, event);
timeSubscribes.put(key, new Date()); errorTimeSubscribes.put(key, new Date());
} }
public void addOkSubscribe(String key, SipSubscribe.Event event) { public void addOkSubscribe(String key, SipSubscribe.Event event) {
okSubscribes.put(key, event); okSubscribes.put(key, event);
timeSubscribes.put(key, new Date()); okTimeSubscribes.put(key, new Date());
} }
public SipSubscribe.Event getErrorSubscribe(String key) { public SipSubscribe.Event getErrorSubscribe(String key) {
@ -119,7 +131,7 @@ public class SipSubscribe {
public void removeErrorSubscribe(String key) { public void removeErrorSubscribe(String key) {
errorSubscribes.remove(key); errorSubscribes.remove(key);
timeSubscribes.remove(key); errorTimeSubscribes.remove(key);
} }
public SipSubscribe.Event getOkSubscribe(String key) { public SipSubscribe.Event getOkSubscribe(String key) {
@ -128,7 +140,7 @@ public class SipSubscribe {
public void removeOkSubscribe(String key) { public void removeOkSubscribe(String key) {
okSubscribes.remove(key); okSubscribes.remove(key);
timeSubscribes.remove(key); okTimeSubscribes.remove(key);
} }
public int getErrorSubscribesSize(){ public int getErrorSubscribesSize(){
return errorSubscribes.size(); return errorSubscribes.size();

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

@ -14,6 +14,7 @@ import com.genersoft.iot.vmp.utils.redis.RedisUtil;
import gov.nist.javax.sip.stack.SIPDialog; import gov.nist.javax.sip.stack.SIPDialog;
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 org.springframework.util.StringUtils;
/** /**
* @description:视频流session管理器，管理视频预览、预览回放的通信句柄 * @description:视频流session管理器，管理视频预览、预览回放的通信句柄
@ -29,39 +30,55 @@ public class VideoStreamSessionManager {
@Autowired @Autowired
private UserSetup userSetup; private UserSetup userSetup;
public void put(String deviceId, String channelId ,String ssrc, String streamId, String mediaServerId, ClientTransaction transaction){ /**
* 添加一个点播/回放的事务信息
* 后续可以通过流Id/callID
* @param deviceId 设备ID
* @param channelId 通道ID
* @param callId 一次请求的CallID
* @param stream 流名称
* @param mediaServerId 所使用的流媒体ID
* @param transaction 事务
*/
public void put(String deviceId, String channelId, String callId, String stream, String ssrc, String mediaServerId, ClientTransaction transaction){
SsrcTransaction ssrcTransaction = new SsrcTransaction(); SsrcTransaction ssrcTransaction = new SsrcTransaction();
ssrcTransaction.setDeviceId(deviceId); ssrcTransaction.setDeviceId(deviceId);
ssrcTransaction.setChannelId(channelId); ssrcTransaction.setChannelId(channelId);
ssrcTransaction.setStreamId(streamId); ssrcTransaction.setStream(stream);
byte[] transactionByteArray = SerializeUtils.serialize(transaction); byte[] transactionByteArray = SerializeUtils.serialize(transaction);
ssrcTransaction.setTransaction(transactionByteArray); ssrcTransaction.setTransaction(transactionByteArray);
ssrcTransaction.setCallId(callId);
ssrcTransaction.setSsrc(ssrc); ssrcTransaction.setSsrc(ssrc);
ssrcTransaction.setMediaServerId(mediaServerId); ssrcTransaction.setMediaServerId(mediaServerId);
redisUtil.set(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetup.getServerId() + "_" + deviceId + "_" + channelId, ssrcTransaction); redisUtil.set(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetup.getServerId()
+ "_" + deviceId + "_" + channelId + "_" + callId + "_" + stream, ssrcTransaction);
redisUtil.set(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetup.getServerId()
+ "_" + deviceId + "_" + channelId + "_" + callId + "_" + stream, ssrcTransaction);
} }
public void put(String deviceId, String channelId , Dialog dialog){ public void put(String deviceId, String channelId, String callId, Dialog dialog){
SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId); SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId, callId, null);
if (ssrcTransaction != null) { if (ssrcTransaction != null) {
byte[] dialogByteArray = SerializeUtils.serialize(dialog); byte[] dialogByteArray = SerializeUtils.serialize(dialog);
ssrcTransaction.setDialog(dialogByteArray); ssrcTransaction.setDialog(dialogByteArray);
} }
redisUtil.set(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetup.getServerId() + "_" + deviceId + "_" + channelId, ssrcTransaction); redisUtil.set(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetup.getServerId()
+ "_" + deviceId + "_" + channelId + "_" + ssrcTransaction.getCallId() + "_"
+ ssrcTransaction.getStream(), ssrcTransaction);
} }
public ClientTransaction getTransaction(String deviceId, String channelId){ public ClientTransaction getTransactionByStream(String deviceId, String channelId, String stream){
SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId); SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId, null, stream);
if (ssrcTransaction == null) return null; if (ssrcTransaction == null) return null;
byte[] transactionByteArray = ssrcTransaction.getTransaction(); byte[] transactionByteArray = ssrcTransaction.getTransaction();
ClientTransaction clientTransaction = (ClientTransaction)SerializeUtils.deSerialize(transactionByteArray); ClientTransaction clientTransaction = (ClientTransaction)SerializeUtils.deSerialize(transactionByteArray);
return clientTransaction; return clientTransaction;
} }
public SIPDialog getDialog(String deviceId, String channelId){ public SIPDialog getDialogByStream(String deviceId, String channelId, String stream){
SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId); SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId, null, stream);
if (ssrcTransaction == null) return null; if (ssrcTransaction == null) return null;
byte[] dialogByteArray = ssrcTransaction.getDialog(); byte[] dialogByteArray = ssrcTransaction.getDialog();
if (dialogByteArray == null) return null; if (dialogByteArray == null) return null;
@ -69,36 +86,37 @@ public class VideoStreamSessionManager {
return dialog; return dialog;
} }
public SsrcTransaction getSsrcTransaction(String deviceId, String channelId){ public SsrcTransaction getSsrcTransaction(String deviceId, String channelId, String callId, String stream){
SsrcTransaction ssrcTransaction = (SsrcTransaction)redisUtil.get(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetup.getServerId() + "_" + deviceId + "_" + channelId); if (StringUtils.isEmpty(callId)) callId ="*";
return ssrcTransaction; if (StringUtils.isEmpty(stream)) stream ="*";
String key = VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetup.getServerId() + "_" + deviceId + "_" + channelId + "_" + callId+ "_" + stream;
List<Object> scanResult = redisUtil.scan(key);
if (scanResult.size() == 0) return null;
return (SsrcTransaction)redisUtil.get((String) scanResult.get(0));
} }
public String getStreamId(String deviceId, String channelId){ public String getMediaServerId(String deviceId, String channelId, String stream){
SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId); SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId, null, stream);
if (ssrcTransaction == null) return null;
return ssrcTransaction.getStreamId();
}
public String getMediaServerId(String deviceId, String channelId){
SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId);
if (ssrcTransaction == null) return null; if (ssrcTransaction == null) return null;
return ssrcTransaction.getMediaServerId(); return ssrcTransaction.getMediaServerId();
} }
public String getSSRC(String deviceId, String channelId){ public String getSSRC(String deviceId, String channelId, String stream){
SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId); SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId, null, stream);
if (ssrcTransaction == null) return null; if (ssrcTransaction == null) return null;
return ssrcTransaction.getSsrc(); return ssrcTransaction.getSsrc();
} }
public void remove(String deviceId, String channelId) { public void remove(String deviceId, String channelId, String stream) {
SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId); SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId, null, stream);
if (ssrcTransaction == null) return; if (ssrcTransaction == null) return;
redisUtil.del(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetup.getServerId() + "_" + deviceId + "_" + channelId); redisUtil.del(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetup.getServerId() + "_"
+ deviceId + "_" + channelId + "_" + ssrcTransaction.getCallId() + "_" + ssrcTransaction.getStream());
} }
public List<SsrcTransaction> getAllSsrc() { public List<SsrcTransaction> getAllSsrc() {
List<Object> ssrcTransactionKeys = redisUtil.scan(String.format("%s_*_*", VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX+ userSetup.getServerId() + "_" )); List<Object> ssrcTransactionKeys = redisUtil.scan(String.format("%s_*_*_*_*", VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX+ userSetup.getServerId() + "_" ));
List<SsrcTransaction> result= new ArrayList<>(); List<SsrcTransaction> result= new ArrayList<>();
for (int i = 0; i < ssrcTransactionKeys.size(); i++) { for (int i = 0; i < ssrcTransactionKeys.size(); i++) {
String key = (String)ssrcTransactionKeys.get(i); String key = (String)ssrcTransactionKeys.get(i);

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

@ -119,8 +119,8 @@ public interface ISIPCommander {
/** /**
* 视频流停止 * 视频流停止
*/ */
void streamByeCmd(String deviceId, String channelId, SipSubscribe.Event okEvent); void streamByeCmd(String deviceId, String channelId, String ssrc, SipSubscribe.Event okEvent);
void streamByeCmd(String deviceId, String channelId); void streamByeCmd(String deviceId, String channelId, String ssrc);
/** /**
* 回放暂停 * 回放暂停

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

@ -18,7 +18,7 @@ public interface ISIPCommanderForPlatform {
* @return * @return
*/ */
boolean register(ParentPlatform parentPlatform, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent); boolean register(ParentPlatform parentPlatform, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent);
boolean register(ParentPlatform parentPlatform, String callId, WWWAuthenticateHeader www, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent); boolean register(ParentPlatform parentPlatform, String callId, WWWAuthenticateHeader www, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent, boolean registerAgain);
/** /**
* 向上级平台注销 * 向上级平台注销

11
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderPlarformProvider.java
View File

@ -128,7 +128,15 @@ public class SIPRequestHeaderPlarformProvider {
Request registerRequest = createRegisterRequest(parentPlatform, redisCatchStorage.getCSEQ(Request.REGISTER), fromTag, viaTag, callIdHeader); Request registerRequest = createRegisterRequest(parentPlatform, redisCatchStorage.getCSEQ(Request.REGISTER), fromTag, viaTag, callIdHeader);
SipURI requestURI = sipFactory.createAddressFactory().createSipURI(parentPlatform.getServerGBId(), parentPlatform.getServerIP() + ":" + parentPlatform.getServerPort());
if (www == null) {
AuthorizationHeader authorizationHeader = sipFactory.createHeaderFactory().createAuthorizationHeader("Digest");
authorizationHeader.setUsername(parentPlatform.getDeviceGBId());
authorizationHeader.setURI(requestURI);
authorizationHeader.setAlgorithm("MD5");
registerRequest.addHeader(authorizationHeader);
return registerRequest;
}
String realm = www.getRealm(); String realm = www.getRealm();
String nonce = www.getNonce(); String nonce = www.getNonce();
String scheme = www.getScheme(); String scheme = www.getScheme();
@ -139,7 +147,6 @@ public class SIPRequestHeaderPlarformProvider {
callIdHeader.setCallId(callId); callIdHeader.setCallId(callId);
SipURI requestURI = sipFactory.createAddressFactory().createSipURI(parentPlatform.getServerGBId(), parentPlatform.getServerIP() + ":" + parentPlatform.getServerPort());
String cNonce = null; String cNonce = null;
String nc = "00000001"; String nc = "00000001";
if (qop != null) { if (qop != null) {

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

@ -226,7 +226,7 @@ public class SIPRequestHeaderProvider {
throws PeerUnavailableException, ParseException, InvalidArgumentException { throws PeerUnavailableException, ParseException, InvalidArgumentException {
Request request = null; Request request = null;
if (streamInfo == null) return null; if (streamInfo == null) return null;
Dialog dialog = streamSession.getDialog(streamInfo.getDeviceID(), streamInfo.getChannelId()); Dialog dialog = streamSession.getDialogByStream(streamInfo.getDeviceID(), streamInfo.getChannelId(), streamInfo.getStream());
SipURI requestLine = sipFactory.createAddressFactory().createSipURI(device.getDeviceId(), SipURI requestLine = sipFactory.createAddressFactory().createSipURI(device.getDeviceId(),
device.getHostAddress()); device.getHostAddress());

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

@ -331,7 +331,7 @@ public class SIPCommander implements ISIPCommander {
*/ */
@Override @Override
public void playStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, ZLMHttpHookSubscribe.Event event, SipSubscribe.Event errorEvent) { public void playStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, ZLMHttpHookSubscribe.Event event, SipSubscribe.Event errorEvent) {
String streamId = ssrcInfo.getStreamId(); String streamId = ssrcInfo.getStream();
try { try {
if (device == null) return; if (device == null) return;
String streamMode = device.getStreamMode().toUpperCase(); String streamMode = device.getStreamMode().toUpperCase();
@ -407,6 +407,8 @@ public class SIPCommander implements ISIPCommander {
} }
content.append("y="+ssrcInfo.getSsrc()+"\r\n");//ssrc content.append("y="+ssrcInfo.getSsrc()+"\r\n");//ssrc
// f字段:f= v/编码格式/分辨率/帧率/码率类型/码率大小a/编码格式/码率大小/采样率
// content.append("f=v/2/5/25/1/4000a/1/8/1" + "\r\n"); // 未发现支持此特性的设备
String tm = Long.toString(System.currentTimeMillis()); String tm = Long.toString(System.currentTimeMillis());
@ -415,14 +417,14 @@ public class SIPCommander implements ISIPCommander {
Request request = headerProvider.createInviteRequest(device, channelId, content.toString(), null, "FromInvt" + tm, null, ssrcInfo.getSsrc(), callIdHeader); Request request = headerProvider.createInviteRequest(device, channelId, content.toString(), null, "FromInvt" + tm, null, ssrcInfo.getSsrc(), callIdHeader);
String finalStreamId = streamId;
transmitRequest(device, request, (e -> { transmitRequest(device, request, (e -> {
streamSession.remove(device.getDeviceId(), channelId); streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
mediaServerService.releaseSsrc(mediaServerItem, ssrcInfo.getSsrc()); mediaServerService.releaseSsrc(mediaServerItem, ssrcInfo.getSsrc());
errorEvent.response(e); errorEvent.response(e);
}), e ->{ }), e ->{
streamSession.put(device.getDeviceId(), channelId ,ssrcInfo.getSsrc(), finalStreamId, mediaServerItem.getId(), ((ResponseEvent)e.event).getClientTransaction()); // 这里为例避免一个通道的点播只有一个callID这个参数使用一个固定值
streamSession.put(device.getDeviceId(), channelId , e.dialog); streamSession.put(device.getDeviceId(), channelId ,"play", streamId, ssrcInfo.getSsrc(), mediaServerItem.getId(), ((ResponseEvent)e.event).getClientTransaction());
streamSession.put(device.getDeviceId(), channelId ,"play", e.dialog);
}); });
@ -444,12 +446,12 @@ public class SIPCommander implements ISIPCommander {
, SipSubscribe.Event errorEvent) { , SipSubscribe.Event errorEvent) {
try { try {
logger.info("{} 分配的ZLM为: {} [{}:{}]", ssrcInfo.getStreamId(), mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort()); logger.info("{} 分配的ZLM为: {} [{}:{}]", ssrcInfo.getStream(), mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort());
// 添加订阅 // 添加订阅
JSONObject subscribeKey = new JSONObject(); JSONObject subscribeKey = new JSONObject();
subscribeKey.put("app", "rtp"); subscribeKey.put("app", "rtp");
subscribeKey.put("stream", ssrcInfo.getStreamId()); subscribeKey.put("stream", ssrcInfo.getStream());
subscribeKey.put("regist", true); subscribeKey.put("regist", true);
subscribeKey.put("mediaServerId", mediaServerItem.getId()); subscribeKey.put("mediaServerId", mediaServerItem.getId());
logger.debug("录像回放添加订阅，订阅内容：" + subscribeKey.toString()); logger.debug("录像回放添加订阅，订阅内容：" + subscribeKey.toString());
@ -530,8 +532,8 @@ public class SIPCommander implements ISIPCommander {
transmitRequest(device, request, errorEvent, okEvent -> { transmitRequest(device, request, errorEvent, okEvent -> {
ResponseEvent responseEvent = (ResponseEvent) okEvent.event; ResponseEvent responseEvent = (ResponseEvent) okEvent.event;
streamSession.put(device.getDeviceId(), channelId, ssrcInfo.getSsrc(), ssrcInfo.getStreamId(), mediaServerItem.getId(), responseEvent.getClientTransaction()); streamSession.put(device.getDeviceId(), channelId, callIdHeader.getCallId(), ssrcInfo.getStream(), ssrcInfo.getSsrc(), mediaServerItem.getId(), responseEvent.getClientTransaction());
streamSession.put(device.getDeviceId(), channelId, okEvent.dialog); streamSession.put(device.getDeviceId(), channelId, callIdHeader.getCallId(), okEvent.dialog);
}); });
} catch ( SipException | ParseException | InvalidArgumentException e) { } catch ( SipException | ParseException | InvalidArgumentException e) {
e.printStackTrace(); e.printStackTrace();
@ -551,12 +553,12 @@ public class SIPCommander implements ISIPCommander {
public void downloadStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, String startTime, String endTime, String downloadSpeed, ZLMHttpHookSubscribe.Event event public void downloadStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, String startTime, String endTime, String downloadSpeed, ZLMHttpHookSubscribe.Event event
, SipSubscribe.Event errorEvent) { , SipSubscribe.Event errorEvent) {
try { try {
logger.info("{} 分配的ZLM为: {} [{}:{}]", ssrcInfo.getStreamId(), mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort()); logger.info("{} 分配的ZLM为: {} [{}:{}]", ssrcInfo.getStream(), mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort());
// 添加订阅 // 添加订阅
JSONObject subscribeKey = new JSONObject(); JSONObject subscribeKey = new JSONObject();
subscribeKey.put("app", "rtp"); subscribeKey.put("app", "rtp");
subscribeKey.put("stream", ssrcInfo.getStreamId()); subscribeKey.put("stream", ssrcInfo.getStream());
subscribeKey.put("regist", true); subscribeKey.put("regist", true);
subscribeKey.put("mediaServerId", mediaServerItem.getId()); subscribeKey.put("mediaServerId", mediaServerItem.getId());
logger.debug("录像回放添加订阅，订阅内容：" + subscribeKey.toString()); logger.debug("录像回放添加订阅，订阅内容：" + subscribeKey.toString());
@ -637,7 +639,8 @@ public class SIPCommander implements ISIPCommander {
Request request = headerProvider.createPlaybackInviteRequest(device, channelId, content.toString(), null, "fromplybck" + tm, null, callIdHeader, ssrcInfo.getSsrc()); Request request = headerProvider.createPlaybackInviteRequest(device, channelId, content.toString(), null, "fromplybck" + tm, null, callIdHeader, ssrcInfo.getSsrc());
ClientTransaction transaction = transmitRequest(device, request, errorEvent); ClientTransaction transaction = transmitRequest(device, request, errorEvent);
streamSession.put(device.getDeviceId(), channelId, ssrcInfo.getSsrc(), ssrcInfo.getStreamId(), mediaServerItem.getId(), transaction); streamSession.put(device.getDeviceId(), channelId, callIdHeader.getCallId(), ssrcInfo.getStream(), ssrcInfo.getSsrc(), mediaServerItem.getId(), transaction);
streamSession.put(device.getDeviceId(), channelId, callIdHeader.getCallId(), ssrcInfo.getStream(), ssrcInfo.getSsrc(), mediaServerItem.getId(), transaction);
} catch ( SipException | ParseException | InvalidArgumentException e) { } catch ( SipException | ParseException | InvalidArgumentException e) {
e.printStackTrace(); e.printStackTrace();
@ -648,17 +651,17 @@ public class SIPCommander implements ISIPCommander {
* 视频流停止, 不使用回调 * 视频流停止, 不使用回调
*/ */
@Override @Override
public void streamByeCmd(String deviceId, String channelId) { public void streamByeCmd(String deviceId, String channelId, String stream) {
streamByeCmd(deviceId, channelId, null); streamByeCmd(deviceId, channelId, stream, null);
} }
/** /**
* 视频流停止 * 视频流停止
*/ */
@Override @Override
public void streamByeCmd(String deviceId, String channelId, SipSubscribe.Event okEvent) { public void streamByeCmd(String deviceId, String channelId, String stream, SipSubscribe.Event okEvent) {
try { try {
ClientTransaction transaction = streamSession.getTransaction(deviceId, channelId); ClientTransaction transaction = streamSession.getTransactionByStream(deviceId, channelId, stream);
if (transaction == null) { if (transaction == null) {
logger.warn("[ {} -> {}]停止视频流的时候发现事务已丢失", deviceId, channelId); logger.warn("[ {} -> {}]停止视频流的时候发现事务已丢失", deviceId, channelId);
SipSubscribe.EventResult<Object> eventResult = new SipSubscribe.EventResult<>(); SipSubscribe.EventResult<Object> eventResult = new SipSubscribe.EventResult<>();
@ -667,7 +670,7 @@ public class SIPCommander implements ISIPCommander {
} }
return; return;
} }
SIPDialog dialog = streamSession.getDialog(deviceId, channelId); SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, stream);
if (dialog == null) { if (dialog == null) {
logger.warn("[ {} -> {}]停止视频流的时候发现对话已丢失", deviceId, channelId); logger.warn("[ {} -> {}]停止视频流的时候发现对话已丢失", deviceId, channelId);
return; return;
@ -711,11 +714,11 @@ public class SIPCommander implements ISIPCommander {
dialog.sendRequest(clientTransaction); dialog.sendRequest(clientTransaction);
SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(deviceId, channelId); SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(deviceId, channelId, callIdHeader.getCallId(), null);
if (ssrcTransaction != null) { if (ssrcTransaction != null) {
MediaServerItem mediaServerItem = mediaServerService.getOne(ssrcTransaction.getMediaServerId()); MediaServerItem mediaServerItem = mediaServerService.getOne(ssrcTransaction.getMediaServerId());
mediaServerService.releaseSsrc(mediaServerItem, ssrcTransaction.getSsrc()); mediaServerService.releaseSsrc(mediaServerItem, ssrcTransaction.getSsrc());
streamSession.remove(deviceId, channelId); streamSession.remove(deviceId, channelId, ssrcTransaction.getStream());
} }
} catch (SipException | ParseException e) { } catch (SipException | ParseException e) {
e.printStackTrace(); e.printStackTrace();

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

@ -52,7 +52,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
@Override @Override
public boolean register(ParentPlatform parentPlatform, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent) { public boolean register(ParentPlatform parentPlatform, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent) {
return register(parentPlatform, null, null, errorEvent, okEvent); return register(parentPlatform, null, null, errorEvent, okEvent, false);
} }
@Override @Override
@ -64,15 +64,16 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch); redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
} }
return register(parentPlatform, null, null, errorEvent, okEvent); return register(parentPlatform, null, null, errorEvent, okEvent, false);
} }
@Override @Override
public boolean register(ParentPlatform parentPlatform, @Nullable String callId, @Nullable WWWAuthenticateHeader www, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent) { public boolean register(ParentPlatform parentPlatform, @Nullable String callId, @Nullable WWWAuthenticateHeader www,
SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent, boolean registerAgain) {
try { try {
Request request = null; Request request = null;
String tm = Long.toString(System.currentTimeMillis()); String tm = Long.toString(System.currentTimeMillis());
if (www == null ) { if (!registerAgain ) {
// //callid // //callid
CallIdHeader callIdHeader = null; CallIdHeader callIdHeader = null;
if(parentPlatform.getTransport().equals("TCP")) { if(parentPlatform.getTransport().equals("TCP")) {

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

@ -87,7 +87,11 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In
if (streamInfo == null) { if (streamInfo == null) {
streamInfo = new StreamInfo(); streamInfo = new StreamInfo();
streamInfo.setApp(sendRtpItem.getApp()); streamInfo.setApp(sendRtpItem.getApp());
streamInfo.setStreamId(sendRtpItem.getStreamId()); streamInfo.setStream(sendRtpItem.getStreamId());
}else {
streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
sendRtpItem.setStreamId(streamInfo.getStream());
streamInfo.setApp("rtp");
} }
redisCatchStorage.updateSendRTPSever(sendRtpItem); redisCatchStorage.updateSendRTPSever(sendRtpItem);
logger.info(platformGbId); logger.info(platformGbId);
@ -95,7 +99,7 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In
Map<String, Object> param = new HashMap<>(); Map<String, Object> param = new HashMap<>();
param.put("vhost","__defaultVhost__"); param.put("vhost","__defaultVhost__");
param.put("app",streamInfo.getApp()); param.put("app",streamInfo.getApp());
param.put("stream",streamInfo.getStreamId()); param.put("stream",streamInfo.getStream());
param.put("ssrc", sendRtpItem.getSsrc()); param.put("ssrc", sendRtpItem.getSsrc());
param.put("dst_url",sendRtpItem.getIp()); param.put("dst_url",sendRtpItem.getIp());
param.put("dst_port", sendRtpItem.getPort()); param.put("dst_port", sendRtpItem.getPort());

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

@ -111,7 +111,7 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
} }
storager.stopPlay(device.getDeviceId(), channelId); storager.stopPlay(device.getDeviceId(), channelId);
mediaServerService.closeRTPServer(device, channelId); mediaServerService.closeRTPServer(device, channelId, streamInfo.getStream());
} }
} }
} catch (SipException e) { } catch (SipException e) {

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

@ -68,6 +68,7 @@ public class KeepaliveNotifyMessageHandler extends SIPRequestProcessorParent imp
} }
if (device.getPort() != rPort) { if (device.getPort() != rPort) {
device.setPort(rPort); device.setPort(rPort);
device.setHostAddress(received.concat(":").concat(String.valueOf(rPort)));
videoManagerStorager.updateDevice(device); videoManagerStorager.updateDevice(device);
redisCatchStorage.updateDevice(device); redisCatchStorage.updateDevice(device);
} }

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

@ -62,7 +62,7 @@ public class MediaStatusNotifyMessageHandler extends SIPRequestProcessorParent i
StreamInfo streamInfo = redisCatchStorage.queryPlaybackByDevice(device.getDeviceId(), "*"); StreamInfo streamInfo = redisCatchStorage.queryPlaybackByDevice(device.getDeviceId(), "*");
if (streamInfo != null) { if (streamInfo != null) {
redisCatchStorage.stopPlayback(streamInfo); redisCatchStorage.stopPlayback(streamInfo);
cmder.streamByeCmd(streamInfo.getDeviceID(), streamInfo.getChannelId()); cmder.streamByeCmd(streamInfo.getDeviceID(), streamInfo.getChannelId(), streamInfo.getStream());
} }
} }
} }

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

@ -78,7 +78,7 @@ public class RegisterResponseProcessor extends SIPResponseProcessorAbstract {
if (response.getStatusCode() == 401) { if (response.getStatusCode() == 401) {
WWWAuthenticateHeader www = (WWWAuthenticateHeader)response.getHeader(WWWAuthenticateHeader.NAME); WWWAuthenticateHeader www = (WWWAuthenticateHeader)response.getHeader(WWWAuthenticateHeader.NAME);
sipCommanderForPlatform.register(parentPlatform, callId, www, null, null); sipCommanderForPlatform.register(parentPlatform, callId, www, null, null, true);
}else if (response.getStatusCode() == 200){ }else if (response.getStatusCode() == 200){
// 注册/注销成功 // 注册/注销成功
logger.info(String.format("%s %s成功", platformGBId, action)); logger.info(String.format("%s %s成功", platformGBId, action));

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

@ -360,6 +360,7 @@ public class ZLMHttpHookListener {
StreamPushItem streamPushItem = null; StreamPushItem streamPushItem = null;
StreamInfo streamInfoByAppAndStream = mediaService.getStreamInfoByAppAndStream(mediaServerItem, app, streamId, tracks); StreamInfo streamInfoByAppAndStream = mediaService.getStreamInfoByAppAndStream(mediaServerItem, app, streamId, tracks);
item.setStreamInfo(streamInfoByAppAndStream); item.setStreamInfo(streamInfoByAppAndStream);
redisCatchStorage.addStream(mediaServerItem, type, app, streamId, item); redisCatchStorage.addStream(mediaServerItem, type, app, streamId, item);
if (item.getOriginType() == OriginType.RTSP_PUSH.ordinal() if (item.getOriginType() == OriginType.RTSP_PUSH.ordinal()
|| item.getOriginType() == OriginType.RTMP_PUSH.ordinal() || item.getOriginType() == OriginType.RTMP_PUSH.ordinal()
@ -438,14 +439,16 @@ public class ZLMHttpHookListener {
if (redisCatchStorage.isChannelSendingRTP(streamInfoForPlayCatch.getChannelId())) { if (redisCatchStorage.isChannelSendingRTP(streamInfoForPlayCatch.getChannelId())) {
ret.put("close", false); ret.put("close", false);
} else { } else {
cmder.streamByeCmd(streamInfoForPlayCatch.getDeviceID(), streamInfoForPlayCatch.getChannelId()); cmder.streamByeCmd(streamInfoForPlayCatch.getDeviceID(), streamInfoForPlayCatch.getChannelId(),
streamInfoForPlayCatch.getStream());
redisCatchStorage.stopPlay(streamInfoForPlayCatch); redisCatchStorage.stopPlay(streamInfoForPlayCatch);
storager.stopPlay(streamInfoForPlayCatch.getDeviceID(), streamInfoForPlayCatch.getChannelId()); storager.stopPlay(streamInfoForPlayCatch.getDeviceID(), streamInfoForPlayCatch.getChannelId());
} }
}else{ }else{
StreamInfo streamInfoForPlayBackCatch = redisCatchStorage.queryPlaybackByStreamId(streamId); StreamInfo streamInfoForPlayBackCatch = redisCatchStorage.queryPlaybackByStreamId(streamId);
if (streamInfoForPlayBackCatch != null) { if (streamInfoForPlayBackCatch != null) {
cmder.streamByeCmd(streamInfoForPlayBackCatch.getDeviceID(), streamInfoForPlayBackCatch.getChannelId()); cmder.streamByeCmd(streamInfoForPlayBackCatch.getDeviceID(),
streamInfoForPlayBackCatch.getChannelId(), streamInfoForPlayBackCatch.getStream());
redisCatchStorage.stopPlayback(streamInfoForPlayBackCatch); redisCatchStorage.stopPlayback(streamInfoForPlayBackCatch);
}else { }else {
StreamInfo streamInfoForDownload = redisCatchStorage.queryDownloadByStreamId(streamId); StreamInfo streamInfoForDownload = redisCatchStorage.queryDownloadByStreamId(streamId);

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

@ -46,7 +46,7 @@ public interface IMediaServerService {
SSRCInfo openRTPServer(MediaServerItem mediaServerItem, String streamId, boolean isPlayback); SSRCInfo openRTPServer(MediaServerItem mediaServerItem, String streamId, boolean isPlayback);
void closeRTPServer(Device device, String channelId); void closeRTPServer(Device device, String channelId, String ssrc);
void clearRTPServer(MediaServerItem mediaServerItem); void clearRTPServer(MediaServerItem mediaServerItem);

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

@ -5,14 +5,16 @@ import com.genersoft.iot.vmp.gb28181.bean.Device;
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;
import com.genersoft.iot.vmp.service.bean.PlayBackCallback;
import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.PlayResult; import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.PlayResult;
import org.springframework.http.ResponseEntity;
import org.springframework.web.context.request.async.DeferredResult;
/** /**
* 点播处理 * 点播处理
*/ */
public interface IPlayService { public interface IPlayService {
void onPublishHandlerForPlayBack(MediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId, String uuid);
void onPublishHandlerForPlay(MediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId, String uuid); void onPublishHandlerForPlay(MediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId, String uuid);
PlayResult play(MediaServerItem mediaServerItem, String deviceId, String channelId, ZLMHttpHookSubscribe.Event event, SipSubscribe.Event errorEvent); PlayResult play(MediaServerItem mediaServerItem, String deviceId, String channelId, ZLMHttpHookSubscribe.Event event, SipSubscribe.Event errorEvent);
@ -20,4 +22,6 @@ public interface IPlayService {
MediaServerItem getNewMediaServerItem(Device device); MediaServerItem getNewMediaServerItem(Device device);
void onPublishHandlerForDownload(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, String toString); void onPublishHandlerForDownload(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, String toString);
DeferredResult<ResponseEntity<String>> playBack(String deviceId, String channelId, String startTime, String endTime, PlayBackCallback errorCallBack);
} }

9
src/main/java/com/genersoft/iot/vmp/service/bean/PlayBackCallback.java Normal file
View File

@ -0,0 +1,9 @@
package com.genersoft.iot.vmp.service.bean;
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
public interface PlayBackCallback {
void call(RequestMessage msg);
}

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

@ -4,12 +4,12 @@ public class SSRCInfo {
private int port; private int port;
private String ssrc; private String ssrc;
private String StreamId; private String Stream;
public SSRCInfo(int port, String ssrc, String streamId) { public SSRCInfo(int port, String ssrc, String stream) {
this.port = port; this.port = port;
this.ssrc = ssrc; this.ssrc = ssrc;
StreamId = streamId; Stream = stream;
} }
public int getPort() { public int getPort() {
@ -28,11 +28,11 @@ public class SSRCInfo {
this.ssrc = ssrc; this.ssrc = ssrc;
} }
public String getStreamId() { public String getStream() {
return StreamId; return Stream;
} }
public void setStreamId(String streamId) { public void setStream(String stream) {
StreamId = streamId; Stream = stream;
} }
} }

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

@ -162,15 +162,16 @@ public class MediaServerServiceImpl implements IMediaServerService, CommandLineR
} }
@Override @Override
public void closeRTPServer(Device device, String channelId) { public void closeRTPServer(Device device, String channelId, String stream) {
String mediaServerId = streamSession.getMediaServerId(device.getDeviceId(), channelId); String mediaServerId = streamSession.getMediaServerId(device.getDeviceId(), channelId, stream);
String ssrc = streamSession.getSSRC(device.getDeviceId(), channelId, stream);
MediaServerItem mediaServerItem = this.getOne(mediaServerId); MediaServerItem mediaServerItem = this.getOne(mediaServerId);
if (mediaServerItem != null) { if (mediaServerItem != null) {
String streamId = String.format("%s_%s", device.getDeviceId(), channelId); String streamId = String.format("%s_%s", device.getDeviceId(), channelId);
zlmrtpServerFactory.closeRTPServer(mediaServerItem, streamId); zlmrtpServerFactory.closeRTPServer(mediaServerItem, streamId);
releaseSsrc(mediaServerItem, streamSession.getSSRC(device.getDeviceId(), channelId)); releaseSsrc(mediaServerItem, ssrc);
} }
streamSession.remove(device.getDeviceId(), channelId); streamSession.remove(device.getDeviceId(), channelId, stream);
} }
@Override @Override

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

@ -74,7 +74,7 @@ public class MediaServiceImpl implements IMediaService {
@Override @Override
public StreamInfo getStreamInfoByAppAndStream(MediaServerItem mediaInfo, String app, String stream, Object tracks, String addr) { public StreamInfo getStreamInfoByAppAndStream(MediaServerItem mediaInfo, String app, String stream, Object tracks, String addr) {
StreamInfo streamInfoResult = new StreamInfo(); StreamInfo streamInfoResult = new StreamInfo();
streamInfoResult.setStreamId(stream); streamInfoResult.setStream(stream);
streamInfoResult.setApp(app); streamInfoResult.setApp(app);
if (addr == null) { if (addr == null) {
addr = mediaInfo.getStreamIp(); addr = mediaInfo.getStreamIp();

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

@ -16,6 +16,7 @@ import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
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.IMediaServerService; import com.genersoft.iot.vmp.service.IMediaServerService;
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 com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager; import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
@ -104,19 +105,21 @@ public class PlayServiceImpl implements IPlayService {
logger.warn(String.format("设备点播超时，deviceId：%s ，channelId：%s", deviceId, channelId)); logger.warn(String.format("设备点播超时，deviceId：%s ，channelId：%s", deviceId, channelId));
WVPResult wvpResult = new WVPResult(); WVPResult wvpResult = new WVPResult();
wvpResult.setCode(-1); wvpResult.setCode(-1);
SIPDialog dialog = streamSession.getDialog(deviceId, channelId); SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, streamInfo.getStream());
if (dialog != null) { if (dialog != null) {
wvpResult.setMsg("收流超时，请稍候重试"); wvpResult.setMsg("收流超时，请稍候重试");
}else { }else {
wvpResult.setMsg("点播超时，请稍候重试"); wvpResult.setMsg("点播超时，请稍候重试");
} }
msg.setData(wvpResult); msg.setData(wvpResult);
// 点播超时回复BYE // 点播超时回复BYE
cmder.streamByeCmd(device.getDeviceId(), channelId); cmder.streamByeCmd(device.getDeviceId(), channelId, streamInfo.getStream());
// 释放rtpserver // 释放rtpserver
mediaServerService.closeRTPServer(playResult.getDevice(), channelId); mediaServerService.closeRTPServer(playResult.getDevice(), channelId, streamInfo.getStream());
// 回复之前所有的点播请求 // 回复之前所有的点播请求
resultHolder.invokeAllResult(msg); resultHolder.invokeAllResult(msg);
// TODO 释放ssrc
}); });
result.onCompletion(()->{ result.onCompletion(()->{
// 点播结束时调用截图接口 // 点播结束时调用截图接口
@ -153,14 +156,12 @@ public class PlayServiceImpl implements IPlayService {
} }
}); });
if (streamInfo == null) { if (streamInfo == null) {
SSRCInfo ssrcInfo;
String streamId = null; String streamId = null;
if (mediaServerItem.isRtpEnable()) { if (mediaServerItem.isRtpEnable()) {
streamId = String.format("%s_%s", device.getDeviceId(), channelId); streamId = String.format("%s_%s", device.getDeviceId(), channelId);
} }
ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId); SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId);
// 发送点播消息 // 发送点播消息
cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channelId, (MediaServerItem mediaServerItemInUse, JSONObject response) -> { cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channelId, (MediaServerItem mediaServerItemInUse, JSONObject response) -> {
logger.info("收到订阅消息： " + response.toJSONString()); logger.info("收到订阅消息： " + response.toJSONString());
@ -172,7 +173,7 @@ public class PlayServiceImpl implements IPlayService {
WVPResult wvpResult = new WVPResult(); WVPResult wvpResult = new WVPResult();
wvpResult.setCode(-1); wvpResult.setCode(-1);
// 点播返回sip错误 // 点播返回sip错误
mediaServerService.closeRTPServer(playResult.getDevice(), channelId); mediaServerService.closeRTPServer(playResult.getDevice(), channelId, ssrcInfo.getStream());
wvpResult.setMsg(String.format("点播失败， 错误码： %s, %s", event.statusCode, event.msg)); wvpResult.setMsg(String.format("点播失败， 错误码： %s, %s", event.statusCode, event.msg));
msg.setData(wvpResult); msg.setData(wvpResult);
resultHolder.invokeAllResult(msg); resultHolder.invokeAllResult(msg);
@ -183,7 +184,7 @@ public class PlayServiceImpl implements IPlayService {
}); });
} else { } else {
String streamId = streamInfo.getStreamId(); String streamId = streamInfo.getStream();
if (streamId == null) { if (streamId == null) {
WVPResult wvpResult = new WVPResult(); WVPResult wvpResult = new WVPResult();
wvpResult.setCode(-1); wvpResult.setCode(-1);
@ -212,18 +213,16 @@ public class PlayServiceImpl implements IPlayService {
// TODO 点播前是否重置状态 // TODO 点播前是否重置状态
redisCatchStorage.stopPlay(streamInfo); redisCatchStorage.stopPlay(streamInfo);
storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
SSRCInfo ssrcInfo;
String streamId2 = null; String streamId2 = null;
if (mediaServerItem.isRtpEnable()) { if (mediaServerItem.isRtpEnable()) {
streamId2 = String.format("%s_%s", device.getDeviceId(), channelId); streamId2 = String.format("%s_%s", device.getDeviceId(), channelId);
} }
ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId2); SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId2);
cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channelId, (MediaServerItem mediaServerItemInuse, JSONObject response) -> { cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channelId, (MediaServerItem mediaServerItemInuse, JSONObject response) -> {
logger.info("收到订阅消息： " + response.toJSONString()); logger.info("收到订阅消息： " + response.toJSONString());
onPublishHandlerForPlay(mediaServerItemInuse, response, deviceId, channelId, uuid); onPublishHandlerForPlay(mediaServerItemInuse, response, deviceId, channelId, uuid);
}, (event) -> { }, (event) -> {
mediaServerService.closeRTPServer(playResult.getDevice(), channelId); mediaServerService.closeRTPServer(playResult.getDevice(), channelId, ssrcInfo.getStream());
WVPResult wvpResult = new WVPResult(); WVPResult wvpResult = new WVPResult();
wvpResult.setCode(-1); wvpResult.setCode(-1);
wvpResult.setMsg(String.format("点播失败， 错误码： %s, %s", event.statusCode, event.msg)); wvpResult.setMsg(String.format("点播失败， 错误码： %s, %s", event.statusCode, event.msg));
@ -241,12 +240,12 @@ public class PlayServiceImpl implements IPlayService {
RequestMessage msg = new RequestMessage(); RequestMessage msg = new RequestMessage();
msg.setId(uuid); msg.setId(uuid);
msg.setKey(DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId); msg.setKey(DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId);
StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId, uuid); StreamInfo streamInfo = onPublishHandler(mediaServerItem, resonse, deviceId, channelId);
if (streamInfo != null) { if (streamInfo != null) {
DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId); DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
if (deviceChannel != null) { if (deviceChannel != null) {
deviceChannel.setStreamId(streamInfo.getStreamId()); deviceChannel.setStreamId(streamInfo.getStream());
storager.startPlay(deviceId, channelId, streamInfo.getStreamId()); storager.startPlay(deviceId, channelId, streamInfo.getStream());
} }
redisCatchStorage.startPlay(streamInfo); redisCatchStorage.startPlay(streamInfo);
msg.setData(JSON.toJSONString(streamInfo)); msg.setData(JSON.toJSONString(streamInfo));
@ -283,29 +282,53 @@ public class PlayServiceImpl implements IPlayService {
@Override @Override
public void onPublishHandlerForPlayBack(MediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId, String uuid) { public DeferredResult<ResponseEntity<String>> playBack(String deviceId, String channelId, String startTime, String endTime, PlayBackCallback callback) {
String uuid = UUID.randomUUID().toString();
String key = DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId;
DeferredResult<ResponseEntity<String>> result = new DeferredResult<>(30000L);
Device device = storager.queryVideoDevice(deviceId);
if (device == null) {
result.setResult(new ResponseEntity<>(HttpStatus.BAD_REQUEST));
return result;
}
MediaServerItem newMediaServerItem = getNewMediaServerItem(device);
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, true);
resultHolder.put(DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId, uuid, result);
RequestMessage msg = new RequestMessage(); RequestMessage msg = new RequestMessage();
msg.setKey(DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId);
msg.setId(uuid); msg.setId(uuid);
StreamInfo streamInfo = onPublishHandler(mediaServerItem, resonse, deviceId, channelId, uuid); msg.setKey(key);
if (streamInfo != null) { result.onTimeout(()->{
redisCatchStorage.startPlayback(streamInfo); msg.setData("回放超时");
msg.setData(JSON.toJSONString(streamInfo)); callback.call(msg);
resultHolder.invokeResult(msg); });
} else { cmder.playbackStreamCmd(newMediaServerItem, ssrcInfo, device, channelId, startTime, endTime, (MediaServerItem mediaServerItem, JSONObject response) -> {
logger.info("收到订阅消息： " + response.toJSONString());
StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId);
if (streamInfo == null) {
logger.warn("设备回放API调用失败！"); logger.warn("设备回放API调用失败！");
msg.setData("设备回放API调用失败！"); msg.setData("设备回放API调用失败！");
resultHolder.invokeResult(msg); callback.call(msg);
return;
} }
redisCatchStorage.startPlayback(streamInfo);
msg.setData(JSON.toJSONString(streamInfo));
callback.call(msg);
}, event -> {
msg.setData(String.format("回放失败， 错误码： %s, %s", event.statusCode, event.msg));
callback.call(msg);
});
return result;
} }
@Override @Override
public void onPublishHandlerForDownload(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, String uuid) { public void onPublishHandlerForDownload(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, String uuid) {
RequestMessage msg = new RequestMessage(); RequestMessage msg = new RequestMessage();
msg.setKey(DeferredResultHolder.CALLBACK_CMD_DOWNLOAD + deviceId + channelId); msg.setKey(DeferredResultHolder.CALLBACK_CMD_DOWNLOAD + deviceId + channelId);
msg.setId(uuid); msg.setId(uuid);
StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId, uuid); StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId);
if (streamInfo != null) { if (streamInfo != null) {
redisCatchStorage.startDownload(streamInfo); redisCatchStorage.startDownload(streamInfo);
msg.setData(JSON.toJSONString(streamInfo)); msg.setData(JSON.toJSONString(streamInfo));
@ -318,7 +341,7 @@ public class PlayServiceImpl implements IPlayService {
} }
public StreamInfo onPublishHandler(MediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId, String uuid) { public StreamInfo onPublishHandler(MediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId) {
String streamId = resonse.getString("stream"); String streamId = resonse.getString("stream");
JSONArray tracks = resonse.getJSONArray("tracks"); JSONArray tracks = resonse.getJSONArray("tracks");
StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(mediaServerItem,"rtp", streamId, tracks); StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(mediaServerItem,"rtp", streamId, tracks);

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

@ -132,7 +132,7 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
}else { }else {
streamLive = true; streamLive = true;
StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream( StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(
mediaInfo, param.getApp(), param.getStream(), null); mediaInfo, param.getApp(), param.getStream(), null, null);
wvpResult.setData(streamInfo); wvpResult.setData(streamInfo);
} }

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

@ -7,6 +7,7 @@ import com.genersoft.iot.vmp.media.zlm.dto.MediaItem;
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.StreamPushItem; import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo; import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.service.bean.SSRCInfo;
import com.genersoft.iot.vmp.service.bean.ThirdPartyGB; import com.genersoft.iot.vmp.service.bean.ThirdPartyGB;
import java.util.List; import java.util.List;
@ -220,4 +221,5 @@ public interface IRedisCatchStorage {
void addMemInfo(double memInfo); void addMemInfo(double memInfo);
void addNetInfo(Map<String, String> networkInterfaces); void addNetInfo(Map<String, String> networkInterfaces);
} }

5
src/main/java/com/genersoft/iot/vmp/storager/dao/StreamProxyMapper.java
View File

@ -10,15 +10,16 @@ import java.util.List;
@Repository @Repository
public interface StreamProxyMapper { public interface StreamProxyMapper {
@Insert("INSERT INTO stream_proxy (type, app, stream,mediaServerId, url, src_url, dst_url, " + @Insert("INSERT INTO stream_proxy (type, name, app, stream,mediaServerId, url, src_url, dst_url, " +
"timeout_ms, ffmpeg_cmd_key, rtp_type, enable_hls, enable_mp4, enable, status, enable_remove_none_reader, createTime) VALUES" + "timeout_ms, ffmpeg_cmd_key, rtp_type, enable_hls, enable_mp4, enable, status, enable_remove_none_reader, createTime) VALUES" +
"('${type}','${app}', '${stream}', '${mediaServerId}','${url}', '${src_url}', '${dst_url}', " + "('${type}','${name}', '${app}', '${stream}', '${mediaServerId}','${url}', '${src_url}', '${dst_url}', " +
"'${timeout_ms}', '${ffmpeg_cmd_key}', '${rtp_type}', ${enable_hls}, ${enable_mp4}, ${enable}, ${status}, " + "'${timeout_ms}', '${ffmpeg_cmd_key}', '${rtp_type}', ${enable_hls}, ${enable_mp4}, ${enable}, ${status}, " +
"${enable_remove_none_reader}, '${createTime}' )") "${enable_remove_none_reader}, '${createTime}' )")
int add(StreamProxyItem streamProxyDto); int add(StreamProxyItem streamProxyDto);
@Update("UPDATE stream_proxy " + @Update("UPDATE stream_proxy " +
"SET type=#{type}, " + "SET type=#{type}, " +
"name=#{name}," +
"app=#{app}," + "app=#{app}," +
"stream=#{stream}," + "stream=#{stream}," +
"url=#{url}, " + "url=#{url}, " +

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

@ -10,6 +10,7 @@ import com.genersoft.iot.vmp.media.zlm.dto.MediaItem;
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.StreamPushItem; import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo; import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.service.bean.SSRCInfo;
import com.genersoft.iot.vmp.service.bean.ThirdPartyGB; 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;
@ -91,7 +92,8 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
*/ */
@Override @Override
public boolean startPlay(StreamInfo stream) { public boolean startPlay(StreamInfo stream) {
return redis.set(String.format("%S_%S_%s_%s_%s", VideoManagerConstants.PLAYER_PREFIX, userSetup.getServerId(), stream.getStreamId(),stream.getDeviceID(), stream.getChannelId()), return redis.set(String.format("%S_%S_%s_%s_%s", VideoManagerConstants.PLAYER_PREFIX, userSetup.getServerId(),
stream.getStream(), stream.getDeviceID(), stream.getChannelId()),
stream); stream);
} }
@ -105,7 +107,7 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
if (streamInfo == null) return false; if (streamInfo == null) return false;
return redis.del(String.format("%S_%s_%s_%s_%s", VideoManagerConstants.PLAYER_PREFIX, return redis.del(String.format("%S_%s_%s_%s_%s", VideoManagerConstants.PLAYER_PREFIX,
userSetup.getServerId(), userSetup.getServerId(),
streamInfo.getStreamId(), streamInfo.getStream(),
streamInfo.getDeviceID(), streamInfo.getDeviceID(),
streamInfo.getChannelId())); streamInfo.getChannelId()));
} }
@ -119,7 +121,7 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
return (StreamInfo)redis.get(String.format("%S_%s_%s_%s_%s", return (StreamInfo)redis.get(String.format("%S_%s_%s_%s_%s",
VideoManagerConstants.PLAYER_PREFIX, VideoManagerConstants.PLAYER_PREFIX,
userSetup.getServerId(), userSetup.getServerId(),
streamInfo.getStreamId(), streamInfo.getStream(),
streamInfo.getDeviceID(), streamInfo.getDeviceID(),
streamInfo.getChannelId())); streamInfo.getChannelId()));
} }
@ -164,14 +166,14 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
@Override @Override
public boolean startPlayback(StreamInfo stream) { public boolean startPlayback(StreamInfo stream) {
return redis.set(String.format("%S_%s_%s_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX, userSetup.getServerId(),stream.getStreamId(), return redis.set(String.format("%S_%s_%s_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX,
stream.getDeviceID(), stream.getChannelId()), stream); userSetup.getServerId(), stream.getStream(), stream.getDeviceID(), stream.getChannelId()), stream);
} }
@Override @Override
public boolean startDownload(StreamInfo streamInfo) { public boolean startDownload(StreamInfo streamInfo) {
return redis.set(String.format("%S_%s_%s_%s_%s", VideoManagerConstants.DOWNLOAD_PREFIX, userSetup.getServerId(),streamInfo.getStreamId(), return redis.set(String.format("%S_%s_%s_%s_%s", VideoManagerConstants.DOWNLOAD_PREFIX, userSetup.getServerId(),
streamInfo.getDeviceID(), streamInfo.getChannelId()), streamInfo); streamInfo.getStream(), streamInfo.getDeviceID(), streamInfo.getChannelId()), streamInfo);
} }
@Override @Override
@ -185,7 +187,7 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
} }
return redis.del(String.format("%S_%s_%s_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX, return redis.del(String.format("%S_%s_%s_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX,
userSetup.getServerId(), userSetup.getServerId(),
streamInfo.getStreamId(), streamInfo.getStream(),
streamInfo.getDeviceID(), streamInfo.getDeviceID(),
streamInfo.getChannelId())); streamInfo.getChannelId()));
} }

17
src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStoragerImpl.java
View File

@ -1,5 +1,6 @@
package com.genersoft.iot.vmp.storager.impl; package com.genersoft.iot.vmp.storager.impl;
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.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher; import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
@ -157,7 +158,10 @@ public class VideoManagerStoragerImpl implements IVideoManagerStorager {
public synchronized void updateChannel(String deviceId, DeviceChannel channel) { public synchronized void updateChannel(String deviceId, DeviceChannel channel) {
String channelId = channel.getChannelId(); String channelId = channel.getChannelId();
channel.setDeviceId(deviceId); channel.setDeviceId(deviceId);
channel.setStreamId(streamSession.getStreamId(deviceId, channel.getChannelId())); StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
if (streamInfo != null) {
channel.setStreamId(streamInfo.getStream());
}
String now = this.format.format(System.currentTimeMillis()); String now = this.format.format(System.currentTimeMillis());
channel.setUpdateTime(now); channel.setUpdateTime(now);
DeviceChannel deviceChannel = deviceChannelMapper.queryChannel(deviceId, channelId); DeviceChannel deviceChannel = deviceChannelMapper.queryChannel(deviceId, channelId);
@ -179,7 +183,10 @@ public class VideoManagerStoragerImpl implements IVideoManagerStorager {
if (channelList.size() == 0) { if (channelList.size() == 0) {
for (DeviceChannel channel : channels) { for (DeviceChannel channel : channels) {
channel.setDeviceId(deviceId); channel.setDeviceId(deviceId);
channel.setStreamId(streamSession.getStreamId(deviceId, channel.getChannelId())); StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channel.getChannelId());
if (streamInfo != null) {
channel.setStreamId(streamInfo.getStream());
}
String now = this.format.format(System.currentTimeMillis()); String now = this.format.format(System.currentTimeMillis());
channel.setUpdateTime(now); channel.setUpdateTime(now);
channel.setCreateTime(now); channel.setCreateTime(now);
@ -190,9 +197,11 @@ public class VideoManagerStoragerImpl implements IVideoManagerStorager {
channelsInStore.put(deviceChannel.getChannelId(), deviceChannel); channelsInStore.put(deviceChannel.getChannelId(), deviceChannel);
} }
for (DeviceChannel channel : channels) { for (DeviceChannel channel : channels) {
String channelId = channel.getChannelId();
channel.setDeviceId(deviceId); channel.setDeviceId(deviceId);
channel.setStreamId(streamSession.getStreamId(deviceId, channel.getChannelId())); StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channel.getChannelId());
if (streamInfo != null) {
channel.setStreamId(streamInfo.getStream());
}
String now = this.format.format(System.currentTimeMillis()); String now = this.format.format(System.currentTimeMillis());
channel.setUpdateTime(now); channel.setUpdateTime(now);
if (channelsInStore.get(channel.getChannelId()) != null) { if (channelsInStore.get(channel.getChannelId()) != null) {

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

@ -110,7 +110,6 @@ public class PlayController {
String key = DeferredResultHolder.CALLBACK_CMD_STOP + deviceId + channelId; String key = DeferredResultHolder.CALLBACK_CMD_STOP + deviceId + channelId;
resultHolder.put(key, uuid, result); resultHolder.put(key, uuid, result);
Device device = storager.queryVideoDevice(deviceId); Device device = storager.queryVideoDevice(deviceId);
cmder.streamByeCmd(deviceId, channelId, (event) -> {
StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId); StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
if (streamInfo == null) { if (streamInfo == null) {
RequestMessage msg = new RequestMessage(); RequestMessage msg = new RequestMessage();
@ -119,7 +118,9 @@ public class PlayController {
msg.setData("点播未找到"); msg.setData("点播未找到");
resultHolder.invokeAllResult(msg); resultHolder.invokeAllResult(msg);
storager.stopPlay(deviceId, channelId); storager.stopPlay(deviceId, channelId);
}else { return result;
}
cmder.streamByeCmd(deviceId, channelId, streamInfo.getStream(), (event) -> {
redisCatchStorage.stopPlay(streamInfo); redisCatchStorage.stopPlay(streamInfo);
storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
RequestMessage msg = new RequestMessage(); RequestMessage msg = new RequestMessage();
@ -128,8 +129,7 @@ public class PlayController {
//Response response = event.getResponse(); //Response response = event.getResponse();
msg.setData(String.format("success")); msg.setData(String.format("success"));
resultHolder.invokeAllResult(msg); resultHolder.invokeAllResult(msg);
} mediaServerService.closeRTPServer(device, channelId, streamInfo.getStream());
mediaServerService.closeRTPServer(device, channelId);
}); });
if (deviceId != null || channelId != null) { if (deviceId != null || channelId != null) {
@ -329,7 +329,7 @@ public class PlayController {
jsonObject.put("deviceId", transaction.getDeviceId()); jsonObject.put("deviceId", transaction.getDeviceId());
jsonObject.put("channelId", transaction.getChannelId()); jsonObject.put("channelId", transaction.getChannelId());
jsonObject.put("ssrc", transaction.getSsrc()); jsonObject.put("ssrc", transaction.getSsrc());
jsonObject.put("streamId", transaction.getStreamId()); jsonObject.put("streamId", transaction.getStream());
objects.add(jsonObject); objects.add(jsonObject);
} }

11
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/playback/DownloadController.java
View File

@ -96,7 +96,7 @@ public class DownloadController {
StreamInfo streamInfo = redisCatchStorage.queryPlaybackByDevice(deviceId, channelId); StreamInfo streamInfo = redisCatchStorage.queryPlaybackByDevice(deviceId, channelId);
if (streamInfo != null) { if (streamInfo != null) {
// 停止之前的下载 // 停止之前的下载
cmder.streamByeCmd(deviceId, channelId); cmder.streamByeCmd(deviceId, channelId, streamInfo.getStream());
} }
MediaServerItem newMediaServerItem = playService.getNewMediaServerItem(device); MediaServerItem newMediaServerItem = playService.getNewMediaServerItem(device);
@ -114,7 +114,7 @@ public class DownloadController {
cmder.downloadStreamCmd(newMediaServerItem, ssrcInfo, device, channelId, startTime, endTime, downloadSpeed, (MediaServerItem mediaServerItem, JSONObject response) -> { cmder.downloadStreamCmd(newMediaServerItem, ssrcInfo, device, channelId, startTime, endTime, downloadSpeed, (MediaServerItem mediaServerItem, JSONObject response) -> {
logger.info("收到订阅消息： " + response.toJSONString()); logger.info("收到订阅消息： " + response.toJSONString());
playService.onPublishHandlerForDownload(mediaServerItem, response, deviceId, channelId, uuid.toString()); playService.onPublishHandlerForDownload(mediaServerItem, response, deviceId, channelId, uuid);
}, event -> { }, event -> {
RequestMessage msg = new RequestMessage(); RequestMessage msg = new RequestMessage();
msg.setId(uuid); msg.setId(uuid);
@ -130,11 +130,12 @@ public class DownloadController {
@ApiImplicitParams({ @ApiImplicitParams({
@ApiImplicitParam(name = "deviceId", value = "设备ID", dataTypeClass = String.class), @ApiImplicitParam(name = "deviceId", value = "设备ID", dataTypeClass = String.class),
@ApiImplicitParam(name = "channelId", value = "通道ID", dataTypeClass = String.class), @ApiImplicitParam(name = "channelId", value = "通道ID", dataTypeClass = String.class),
@ApiImplicitParam(name = "stream", value = "流ID", dataTypeClass = String.class),
}) })
@GetMapping("/stop/{deviceId}/{channelId}") @GetMapping("/stop/{deviceId}/{channelId}/{stream}")
public ResponseEntity<String> playStop(@PathVariable String deviceId, @PathVariable String channelId) { public ResponseEntity<String> playStop(@PathVariable String deviceId, @PathVariable String channelId, @PathVariable String stream) {
cmder.streamByeCmd(deviceId, channelId); cmder.streamByeCmd(deviceId, channelId, stream);
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
logger.debug(String.format("设备历史媒体下载停止 API调用，deviceId/channelId：%s_%s", deviceId, channelId)); logger.debug(String.format("设备历史媒体下载停止 API调用，deviceId/channelId：%s_%s", deviceId, channelId));

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

@ -18,6 +18,7 @@ import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus; import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity; import org.springframework.http.ResponseEntity;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.CrossOrigin; 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;
@ -75,52 +76,8 @@ public class PlaybackController {
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));
} }
String uuid = UUID.randomUUID().toString();
String key = DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId;
DeferredResult<ResponseEntity<String>> result = new DeferredResult<ResponseEntity<String>>(30000L);
Device device = storager.queryVideoDevice(deviceId);
if (device == null) {
result.setResult(new ResponseEntity<>(HttpStatus.BAD_REQUEST));
return result;
}
MediaServerItem newMediaServerItem = playService.getNewMediaServerItem(device);
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, true);
// 超时处理 DeferredResult<ResponseEntity<String>> result = playService.playBack(deviceId, channelId, startTime, endTime, msg->{
result.onTimeout(()->{
logger.warn(String.format("设备回放超时，deviceId：%s ，channelId：%s", deviceId, channelId));
RequestMessage msg = new RequestMessage();
msg.setId(uuid);
msg.setKey(key);
msg.setData("Timeout");
resultHolder.invokeResult(msg);
});
StreamInfo streamInfo = redisCatchStorage.queryPlaybackByDevice(deviceId, channelId);
if (streamInfo != null) {
// 停止之前的回放
cmder.streamByeCmd(deviceId, channelId);
}
resultHolder.put(DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId, uuid, result);
if (newMediaServerItem == null) {
logger.warn(String.format("设备回放超时，deviceId：%s ，channelId：%s", deviceId, channelId));
RequestMessage msg = new RequestMessage();
msg.setId(uuid);
msg.setKey(key);
msg.setData("Timeout");
resultHolder.invokeResult(msg);
return result;
}
cmder.playbackStreamCmd(newMediaServerItem, ssrcInfo, device, channelId, startTime, endTime, (MediaServerItem mediaServerItem, JSONObject response) -> {
logger.info("收到订阅消息： " + response.toJSONString());
playService.onPublishHandlerForPlayBack(mediaServerItem, response, deviceId, channelId, uuid.toString());
}, event -> {
RequestMessage msg = new RequestMessage();
msg.setId(uuid);
msg.setKey(key);
msg.setData(String.format("回放失败， 错误码： %s, %s", event.statusCode, event.msg));
resultHolder.invokeResult(msg); resultHolder.invokeResult(msg);
}); });
@ -131,24 +88,31 @@ public class PlaybackController {
@ApiImplicitParams({ @ApiImplicitParams({
@ApiImplicitParam(name = "deviceId", value = "设备ID", dataTypeClass = String.class), @ApiImplicitParam(name = "deviceId", value = "设备ID", dataTypeClass = String.class),
@ApiImplicitParam(name = "channelId", value = "通道ID", dataTypeClass = String.class), @ApiImplicitParam(name = "channelId", value = "通道ID", dataTypeClass = String.class),
@ApiImplicitParam(name = "stream", value = "流ID", dataTypeClass = String.class),
}) })
@GetMapping("/stop/{deviceId}/{channelId}") @GetMapping("/stop/{deviceId}/{channelId}/{stream}")
public ResponseEntity<String> playStop(@PathVariable String deviceId, @PathVariable String channelId) { public ResponseEntity<String> playStop(
@PathVariable String deviceId,
@PathVariable String channelId,
@PathVariable String stream) {
cmder.streamByeCmd(deviceId, channelId); cmder.streamByeCmd(deviceId, channelId, stream);
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
logger.debug(String.format("设备录像回放停止 API调用，deviceId/channelId：%s/%s", deviceId, channelId)); logger.debug(String.format("设备录像回放停止 API调用，deviceId/channelId：%s/%s", deviceId, channelId));
} }
if (StringUtils.isEmpty(deviceId) || StringUtils.isEmpty(channelId) || StringUtils.isEmpty(stream)) {
return new ResponseEntity<>(HttpStatus.BAD_REQUEST);
}
if (deviceId != null && channelId != null) { if (deviceId != null && channelId != null) {
JSONObject json = new JSONObject(); JSONObject json = new JSONObject();
json.put("deviceId", deviceId); json.put("deviceId", deviceId);
json.put("channelId", channelId); json.put("channelId", channelId);
return new ResponseEntity<String>(json.toString(), HttpStatus.OK); return new ResponseEntity<>(json.toString(), HttpStatus.OK);
} else { } else {
logger.warn("设备录像回放停止API调用失败！"); logger.warn("设备录像回放停止API调用失败！");
return new ResponseEntity<String>(HttpStatus.INTERNAL_SERVER_ERROR); return new ResponseEntity<>(HttpStatus.INTERNAL_SERVER_ERROR);
} }
} }

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

@ -103,7 +103,7 @@ public class ApiStreamController {
PlayResult play = playService.play(newMediaServerItem, serial, code, (mediaServerItem, response)->{ PlayResult play = playService.play(newMediaServerItem, serial, code, (mediaServerItem, response)->{
StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(serial, code); StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(serial, code);
JSONObject result = new JSONObject(); JSONObject result = new JSONObject();
result.put("StreamID", streamInfo.getStreamId()); result.put("StreamID", streamInfo.getStream());
result.put("DeviceID", device.getDeviceId()); result.put("DeviceID", device.getDeviceId());
result.put("ChannelID", code); result.put("ChannelID", code);
result.put("ChannelName", deviceChannel.getName()); result.put("ChannelName", deviceChannel.getName());
@ -177,7 +177,7 @@ public class ApiStreamController {
result.put("error","未找到流信息"); result.put("error","未找到流信息");
return result; return result;
} }
cmder.streamByeCmd(serial, code); cmder.streamByeCmd(serial, code, streamInfo.getStream());
redisCatchStorage.stopPlay(streamInfo); redisCatchStorage.stopPlay(streamInfo);
storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
return null; return null;

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

@ -186,7 +186,7 @@ user-settings:
# 是否将日志存储进数据库 # 是否将日志存储进数据库
logInDatebase: true logInDatebase: true
# 第三方匹配，用于从stream钟获取有效信息 # 第三方匹配，用于从stream钟获取有效信息
thirdPartyGBIdReg: [\s\S]* thirdPartyGBIdReg: "[\\s\\S]*"
# 在线文档： swagger-ui（生产环境建议关闭） # 在线文档： swagger-ui（生产环境建议关闭）
swagger-ui: swagger-ui:

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

@ -83,7 +83,7 @@
<logger name="com.genersoft.iot.vmp.storager.dao" level="INFO"> <logger name="com.genersoft.iot.vmp.storager.dao" level="INFO">
<appender-ref ref="STDOUT"/> <appender-ref ref="STDOUT"/>
</logger> </logger>
<logger name="com.genersoft.iot.vmp.gb28181" level="DEBUG"> <logger name="com.genersoft.iot.vmp.gb28181" level="INFO">
<appender-ref ref="STDOUT"/> <appender-ref ref="STDOUT"/>
</logger> </logger>

5
web_src/src/components/dialog/changePassword.vue
View File

@ -75,7 +75,10 @@ export default {
isLoging: false, isLoging: false,
rules: { rules: {
oldPassword: [{ required: true, validator: validatePass0, trigger: "blur" }], oldPassword: [{ required: true, validator: validatePass0, trigger: "blur" }],
newPassword: [{ required: true, validator: validatePass1, trigger: "blur" }], newPassword: [{ required: true, validator: validatePass1, trigger: "blur" }, {
pattern: /^(?=.*[a-zA-Z])(?=.*\d)(?=.*[~!@#$%^&*()_+`\-={}:";'<>?,.\/]).{8,20}$/,
message: "密码长度在8-20位之间,由字母+数字+特殊字符组成",
},],
confirmPassword: [{ required: true, validator: validatePass2, trigger: "blur" }], confirmPassword: [{ required: true, validator: validatePass2, trigger: "blur" }],
}, },
}; };

13
web_src/src/components/dialog/devicePlayer.vue
View File

@ -307,7 +307,7 @@ export default {
this.isLoging = false; this.isLoging = false;
// this.videoUrl = streamInfo.rtc; // this.videoUrl = streamInfo.rtc;
this.videoUrl = this.getUrlByStreamInfo(streamInfo); this.videoUrl = this.getUrlByStreamInfo(streamInfo);
this.streamId = streamInfo.streamId; this.streamId = streamInfo.stream;
this.app = streamInfo.app; this.app = streamInfo.app;
this.mediaServerId = streamInfo.mediaServerId; this.mediaServerId = streamInfo.mediaServerId;
this.playFromStreamInfo(false, streamInfo) this.playFromStreamInfo(false, streamInfo)
@ -485,8 +485,9 @@ export default {
}).then(function (res) { }).then(function (res) {
var streamInfo = res.data; var streamInfo = res.data;
that.app = streamInfo.app; that.app = streamInfo.app;
that.streamId = streamInfo.streamId; that.streamId = streamInfo.stream;
that.mediaServerId = streamInfo.mediaServerId; that.mediaServerId = streamInfo.mediaServerId;
that.ssrc = streamInfo.ssrc;
that.videoUrl = that.getUrlByStreamInfo(streamInfo); that.videoUrl = that.getUrlByStreamInfo(streamInfo);
that.recordPlay = true; that.recordPlay = true;
}); });
@ -497,7 +498,7 @@ export default {
this.videoUrl = ''; this.videoUrl = '';
this.$axios({ this.$axios({
method: 'get', method: 'get',
url: '/api/playback/stop/' + this.deviceId + "/" + this.channelId url: '/api/playback/stop/' + this.deviceId + "/" + this.channelId + "/" + this.streamId
}).then(function (res) { }).then(function (res) {
if (callback) callback() if (callback) callback()
}); });
@ -517,7 +518,7 @@ export default {
}).then(function (res) { }).then(function (res) {
var streamInfo = res.data; var streamInfo = res.data;
that.app = streamInfo.app; that.app = streamInfo.app;
that.streamId = streamInfo.streamId; that.streamId = streamInfo.stream;
that.mediaServerId = streamInfo.mediaServerId; that.mediaServerId = streamInfo.mediaServerId;
that.videoUrl = that.getUrlByStreamInfo(streamInfo); that.videoUrl = that.getUrlByStreamInfo(streamInfo);
that.recordPlay = true; that.recordPlay = true;
@ -529,7 +530,7 @@ export default {
this.videoUrl = ''; this.videoUrl = '';
this.$axios({ this.$axios({
method: 'get', method: 'get',
url: '/api/download/stop/' + this.deviceId + "/" + this.channelId url: '/api/download/stop/' + this.deviceId + "/" + this.channelId+ "/" + this.streamId
}).then(function (res) { }).then(function (res) {
if (callback) callback() if (callback) callback()
}); });
@ -539,8 +540,6 @@ export default {
let that = this; let that = this;
this.$axios({ this.$axios({
method: 'post', method: 'post',
// url: '/api/ptz/' + this.deviceId + '/' + this.channelId + '?leftRight=' + leftRight + '&upDown=' + upDown +
// '&inOut=' + zoom + '&moveSpeed=50&zoomSpeed=50'
url: '/api/ptz/control/' + this.deviceId + '/' + this.channelId + '?command=' + command + '&horizonSpeed=' + this.controSpeed + '&verticalSpeed=' + this.controSpeed + '&zoomSpeed=' + this.controSpeed url: '/api/ptz/control/' + this.deviceId + '/' + this.channelId + '?command=' + command + '&horizonSpeed=' + this.controSpeed + '&verticalSpeed=' + this.controSpeed + '&zoomSpeed=' + this.controSpeed
}).then(function (res) {}); }).then(function (res) {});
}, },