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

国标录像支持多端同时播放

Browse Source
This commit is contained in:
648540858 2022-03-03 15:57:28 +08:00
parent ed9e8ef456
commit 1527b0e839
33 changed files with 282 additions and 230 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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 {
private String app;
private String streamId;
private String stream;
private String deviceID;
private String channelId;
private String flv;
@ -153,12 +153,12 @@ public class StreamInfo {
this.ws_ts = ws_ts;
}
public String getStreamId() {
return streamId;
public String getStream() {
return stream;
}
public void setStreamId(String streamId) {
this.streamId = streamId;
public void setStream(String stream) {
this.stream = stream;
}
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 PLAY_BLACK_PREFIX = "VMP_PLAYBACK_";
public static final String PLAY_INFO_PREFIX = "VMP_PLAY_INFO_";
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 channelId;
private String ssrc;
private String streamId;
private String callId;
private String stream;
private byte[] transaction;
private byte[] dialog;
private String mediaServerId;
private String ssrc;
public String getDeviceId() {
return deviceId;
@ -26,20 +27,20 @@ public class SsrcTransaction {
this.channelId = channelId;
}
public String getSsrc() {
return ssrc;
public String getCallId() {
return callId;
}
public void setSsrc(String ssrc) {
this.ssrc = ssrc;
public void setCallId(String callId) {
this.callId = callId;
}
public String getStreamId() {
return streamId;
public String getStream() {
return stream;
}
public void setStreamId(String streamId) {
this.streamId = streamId;
public void setStream(String stream) {
this.stream = stream;
}
public byte[] getTransaction() {
@ -65,4 +66,12 @@ public class SsrcTransaction {
public void setMediaServerId(String mediaServerId) {
this.mediaServerId = mediaServerId;
}
public String getSsrc() {
return ssrc;
}
public void setSsrc(String ssrc) {
this.ssrc = ssrc;
}
}

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 org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
/**
* @description:视频流session管理器，管理视频预览、预览回放的通信句柄
@ -29,39 +30,55 @@ public class VideoStreamSessionManager {
@Autowired
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.setDeviceId(deviceId);
ssrcTransaction.setChannelId(channelId);
ssrcTransaction.setStreamId(streamId);
ssrcTransaction.setStream(stream);
byte[] transactionByteArray = SerializeUtils.serialize(transaction);
ssrcTransaction.setTransaction(transactionByteArray);
ssrcTransaction.setCallId(callId);
ssrcTransaction.setSsrc(ssrc);
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){
SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId);
public void put(String deviceId, String channelId, String callId, Dialog dialog){
SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId, callId, null);
if (ssrcTransaction != null) {
byte[] dialogByteArray = SerializeUtils.serialize(dialog);
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){
SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId);
public ClientTransaction getTransactionByStream(String deviceId, String channelId, String stream){
SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId, null, stream);
if (ssrcTransaction == null) return null;
byte[] transactionByteArray = ssrcTransaction.getTransaction();
ClientTransaction clientTransaction = (ClientTransaction)SerializeUtils.deSerialize(transactionByteArray);
return clientTransaction;
}
public SIPDialog getDialog(String deviceId, String channelId){
SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId);
public SIPDialog getDialogByStream(String deviceId, String channelId, String stream){
SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId, null, stream);
if (ssrcTransaction == null) return null;
byte[] dialogByteArray = ssrcTransaction.getDialog();
if (dialogByteArray == null) return null;
@ -69,36 +86,37 @@ public class VideoStreamSessionManager {
return dialog;
}
public SsrcTransaction getSsrcTransaction(String deviceId, String channelId){
SsrcTransaction ssrcTransaction = (SsrcTransaction)redisUtil.get(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetup.getServerId() + "_" + deviceId + "_" + channelId);
return ssrcTransaction;
public SsrcTransaction getSsrcTransaction(String deviceId, String channelId, String callId, String stream){
if (StringUtils.isEmpty(callId)) callId ="*";
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){
SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId);
if (ssrcTransaction == null) return null;
return ssrcTransaction.getStreamId();
}
public String getMediaServerId(String deviceId, String channelId){
SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId);
public String getMediaServerId(String deviceId, String channelId, String stream){
SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId, null, stream);
if (ssrcTransaction == null) return null;
return ssrcTransaction.getMediaServerId();
}
public String getSSRC(String deviceId, String channelId){
SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId);
public String getSSRC(String deviceId, String channelId, String stream){
SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId, null, stream);
if (ssrcTransaction == null) return null;
return ssrcTransaction.getSsrc();
}
public void remove(String deviceId, String channelId) {
SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId);
public void remove(String deviceId, String channelId, String stream) {
SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId, null, stream);
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() {
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<>();
for (int i = 0; i < ssrcTransactionKeys.size(); 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);
void streamByeCmd(String deviceId, String channelId, String ssrc, SipSubscribe.Event okEvent);
void streamByeCmd(String deviceId, String channelId, String ssrc);
/**
* 回放暂停

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

@ -17,7 +17,7 @@ public interface ISIPCommanderForPlatform {
* @return
*/
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);
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 nonce = www.getNonce();
String scheme = www.getScheme();
@ -139,7 +147,6 @@ public class SIPRequestHeaderPlarformProvider {
callIdHeader.setCallId(callId);
SipURI requestURI = sipFactory.createAddressFactory().createSipURI(parentPlatform.getServerGBId(), parentPlatform.getServerIP() + ":" + parentPlatform.getServerPort());
String cNonce = null;
String nc = "00000001";
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 {
Request request = 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(),
device.getHostAddress());

43
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
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 {
if (device == null) return;
String streamMode = device.getStreamMode().toUpperCase();
@ -404,6 +404,8 @@ public class SIPCommander implements ISIPCommander {
}
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());
@ -412,14 +414,14 @@ public class SIPCommander implements ISIPCommander {
Request request = headerProvider.createInviteRequest(device, channelId, content.toString(), null, "FromInvt" + tm, null, ssrcInfo.getSsrc(), callIdHeader);
String finalStreamId = streamId;
transmitRequest(device, request, (e -> {
streamSession.remove(device.getDeviceId(), channelId);
streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
mediaServerService.releaseSsrc(mediaServerItem, ssrcInfo.getSsrc());
errorEvent.response(e);
}), e ->{
streamSession.put(device.getDeviceId(), channelId ,ssrcInfo.getSsrc(), finalStreamId, mediaServerItem.getId(), ((ResponseEvent)e.event).getClientTransaction());
streamSession.put(device.getDeviceId(), channelId , e.dialog);
// 这里为例避免一个通道的点播只有一个callID这个参数使用一个固定值
streamSession.put(device.getDeviceId(), channelId ,"play", streamId, ssrcInfo.getSsrc(), mediaServerItem.getId(), ((ResponseEvent)e.event).getClientTransaction());
streamSession.put(device.getDeviceId(), channelId ,"play", e.dialog);
});
@ -441,12 +443,12 @@ public class SIPCommander implements ISIPCommander {
, SipSubscribe.Event errorEvent) {
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();
subscribeKey.put("app", "rtp");
subscribeKey.put("stream", ssrcInfo.getStreamId());
subscribeKey.put("stream", ssrcInfo.getStream());
subscribeKey.put("regist", true);
subscribeKey.put("mediaServerId", mediaServerItem.getId());
logger.debug("录像回放添加订阅，订阅内容：" + subscribeKey.toString());
@ -466,8 +468,6 @@ public class SIPCommander implements ISIPCommander {
content.append("t="+DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(startTime)+" "
+DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(endTime) +"\r\n");
String streamMode = device.getStreamMode().toUpperCase();
if (userSetup.isSeniorSdp()) {
@ -527,8 +527,8 @@ public class SIPCommander implements ISIPCommander {
transmitRequest(device, request, errorEvent, okEvent -> {
ResponseEvent responseEvent = (ResponseEvent) okEvent.event;
streamSession.put(device.getDeviceId(), channelId, ssrcInfo.getSsrc(), ssrcInfo.getStreamId(), mediaServerItem.getId(), responseEvent.getClientTransaction());
streamSession.put(device.getDeviceId(), channelId, okEvent.dialog);
streamSession.put(device.getDeviceId(), channelId, callIdHeader.getCallId(), ssrcInfo.getStream(), ssrcInfo.getSsrc(), mediaServerItem.getId(), responseEvent.getClientTransaction());
streamSession.put(device.getDeviceId(), channelId, callIdHeader.getCallId(), okEvent.dialog);
});
} catch ( SipException | ParseException | InvalidArgumentException e) {
e.printStackTrace();
@ -548,12 +548,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
, SipSubscribe.Event errorEvent) {
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();
subscribeKey.put("app", "rtp");
subscribeKey.put("stream", ssrcInfo.getStreamId());
subscribeKey.put("stream", ssrcInfo.getStream());
subscribeKey.put("regist", true);
subscribeKey.put("mediaServerId", mediaServerItem.getId());
logger.debug("录像回放添加订阅，订阅内容：" + subscribeKey.toString());
@ -634,7 +634,8 @@ public class SIPCommander implements ISIPCommander {
Request request = headerProvider.createPlaybackInviteRequest(device, channelId, content.toString(), null, "fromplybck" + tm, null, callIdHeader, ssrcInfo.getSsrc());
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) {
e.printStackTrace();
@ -645,17 +646,17 @@ public class SIPCommander implements ISIPCommander {
* 视频流停止, 不使用回调
*/
@Override
public void streamByeCmd(String deviceId, String channelId) {
streamByeCmd(deviceId, channelId, null);
public void streamByeCmd(String deviceId, String channelId, String stream) {
streamByeCmd(deviceId, channelId, stream, null);
}
/**
* 视频流停止
*/
@Override
public void streamByeCmd(String deviceId, String channelId, SipSubscribe.Event okEvent) {
public void streamByeCmd(String deviceId, String channelId, String stream, SipSubscribe.Event okEvent) {
try {
ClientTransaction transaction = streamSession.getTransaction(deviceId, channelId);
ClientTransaction transaction = streamSession.getTransactionByStream(deviceId, channelId, stream);
if (transaction == null) {
logger.warn("[ {} -> {}]停止视频流的时候发现事务已丢失", deviceId, channelId);
SipSubscribe.EventResult<Object> eventResult = new SipSubscribe.EventResult<>();
@ -664,7 +665,7 @@ public class SIPCommander implements ISIPCommander {
}
return;
}
SIPDialog dialog = streamSession.getDialog(deviceId, channelId);
SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, stream);
if (dialog == null) {
logger.warn("[ {} -> {}]停止视频流的时候发现对话已丢失", deviceId, channelId);
return;
@ -708,11 +709,11 @@ public class SIPCommander implements ISIPCommander {
dialog.sendRequest(clientTransaction);
SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(deviceId, channelId);
SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(deviceId, channelId, callIdHeader.getCallId(), null);
if (ssrcTransaction != null) {
MediaServerItem mediaServerItem = mediaServerService.getOne(ssrcTransaction.getMediaServerId());
mediaServerService.releaseSsrc(mediaServerItem, ssrcTransaction.getSsrc());
streamSession.remove(deviceId, channelId);
streamSession.remove(deviceId, channelId, ssrcTransaction.getStream());
}
} catch (SipException | ParseException e) {
e.printStackTrace();

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

@ -53,7 +53,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
@Override
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
@ -65,15 +65,16 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
}
return register(parentPlatform, null, null, errorEvent, okEvent);
return register(parentPlatform, null, null, errorEvent, okEvent, false);
}
@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 {
Request request = null;
String tm = Long.toString(System.currentTimeMillis());
if (www == null ) {
if (!registerAgain ) {
// //callid
CallIdHeader callIdHeader = null;
if(parentPlatform.getTransport().equals("TCP")) {

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

@ -72,10 +72,10 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In
if (deviceId == null) {
streamInfo = new StreamInfo();
streamInfo.setApp(sendRtpItem.getApp());
streamInfo.setStreamId(sendRtpItem.getStreamId());
streamInfo.setStream(sendRtpItem.getStreamId());
}else {
streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
sendRtpItem.setStreamId(streamInfo.getStreamId());
sendRtpItem.setStreamId(streamInfo.getStream());
streamInfo.setApp("rtp");
}
@ -85,7 +85,7 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In
Map<String, Object> param = new HashMap<>();
param.put("vhost","__defaultVhost__");
param.put("app",streamInfo.getApp());
param.put("stream",streamInfo.getStreamId());
param.put("stream",streamInfo.getStream());
param.put("ssrc", sendRtpItem.getSsrc());
param.put("dst_url",sendRtpItem.getIp());
param.put("dst_port", sendRtpItem.getPort());
@ -98,21 +98,21 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In
try {
if (System.currentTimeMillis() - startTime < 30 * 1000) {
MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
if (zlmrtpServerFactory.isStreamReady(mediaInfo, streamInfo.getApp(), streamInfo.getStreamId())) {
if (zlmrtpServerFactory.isStreamReady(mediaInfo, streamInfo.getApp(), streamInfo.getStream())) {
rtpPushed = true;
logger.info("已获取设备推流[{}/{}]，开始向上级推流[{}:{}]",
streamInfo.getApp() ,streamInfo.getStreamId(), sendRtpItem.getIp(), sendRtpItem.getPort());
streamInfo.getApp() ,streamInfo.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort());
zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
} else {
logger.info("等待设备推流[{}/{}].......",
streamInfo.getApp() ,streamInfo.getStreamId());
streamInfo.getApp() ,streamInfo.getStream());
Thread.sleep(1000);
continue;
}
} else {
rtpPushed = true;
logger.info("设备推流[{}/{}]超时，终止向上级推流",
streamInfo.getApp() ,streamInfo.getStreamId());
streamInfo.getApp() ,streamInfo.getStream());
}
} catch (InterruptedException e) {
e.printStackTrace();

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

@ -89,18 +89,19 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
redisCatchStorage.deleteSendRTPServer(platformGbId, channelId);
if (zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId) == 0) {
logger.info(streamId + "无其它观看者，通知设备停止推流");
cmder.streamByeCmd(sendRtpItem.getDeviceId(), channelId);
cmder.streamByeCmd(sendRtpItem.getDeviceId(), channelId, streamId);
}
}
// 可能是设备主动停止
Device device = storager.queryVideoDeviceByChannelId(platformGbId);
if (device != null) {
StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(device.getDeviceId(), channelId);
if (streamInfo != null) {
redisCatchStorage.stopPlay(streamInfo);
}
storager.stopPlay(device.getDeviceId(), channelId);
mediaServerService.closeRTPServer(device, channelId);
mediaServerService.closeRTPServer(device, channelId, streamInfo.getStream());
}
}
} catch (SipException e) {

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(), "*");
if (streamInfo != null) {
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) {
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){
// 注册/注销成功
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;
StreamInfo streamInfoByAppAndStream = mediaService.getStreamInfoByAppAndStream(mediaServerItem, app, streamId, tracks);
item.setStreamInfo(streamInfoByAppAndStream);
redisCatchStorage.addStream(mediaServerItem, type, app, streamId, item);
if (item.getOriginType() == OriginType.RTSP_PUSH.ordinal()
|| item.getOriginType() == OriginType.RTMP_PUSH.ordinal()
@ -438,14 +439,16 @@ public class ZLMHttpHookListener {
if (redisCatchStorage.isChannelSendingRTP(streamInfoForPlayCatch.getChannelId())) {
ret.put("close", false);
} else {
cmder.streamByeCmd(streamInfoForPlayCatch.getDeviceID(), streamInfoForPlayCatch.getChannelId());
cmder.streamByeCmd(streamInfoForPlayCatch.getDeviceID(), streamInfoForPlayCatch.getChannelId(),
streamInfoForPlayCatch.getStream());
redisCatchStorage.stopPlay(streamInfoForPlayCatch);
storager.stopPlay(streamInfoForPlayCatch.getDeviceID(), streamInfoForPlayCatch.getChannelId());
}
}else{
StreamInfo streamInfoForPlayBackCatch = redisCatchStorage.queryPlaybackByStreamId(streamId);
if (streamInfoForPlayBackCatch != null) {
cmder.streamByeCmd(streamInfoForPlayBackCatch.getDeviceID(), streamInfoForPlayBackCatch.getChannelId());
cmder.streamByeCmd(streamInfoForPlayBackCatch.getDeviceID(),
streamInfoForPlayBackCatch.getChannelId(), streamInfoForPlayBackCatch.getStream());
redisCatchStorage.stopPlayback(streamInfoForPlayBackCatch);
}else {
StreamInfo streamInfoForDownload = redisCatchStorage.queryDownloadByStreamId(streamId);

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

@ -91,7 +91,8 @@ public class ZLMHttpHookSubscribe {
}
}
if (null != result && result){
eventMap.remove(key);
// TODO 报错未处理
iterator.remove();
}
}
}

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);
void closeRTPServer(Device device, String channelId);
void closeRTPServer(Device device, String channelId, String ssrc);
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.media.zlm.ZLMHttpHookSubscribe;
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 org.springframework.http.ResponseEntity;
import org.springframework.web.context.request.async.DeferredResult;
/**
* 点播处理
*/
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);
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);
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 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.ssrc = ssrc;
StreamId = streamId;
Stream = stream;
}
public int getPort() {
@ -28,11 +28,11 @@ public class SSRCInfo {
this.ssrc = ssrc;
}
public String getStreamId() {
return StreamId;
public String getStream() {
return Stream;
}
public void setStreamId(String streamId) {
StreamId = streamId;
public void setStream(String stream) {
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
public void closeRTPServer(Device device, String channelId) {
String mediaServerId = streamSession.getMediaServerId(device.getDeviceId(), channelId);
public void closeRTPServer(Device device, String channelId, String stream) {
String mediaServerId = streamSession.getMediaServerId(device.getDeviceId(), channelId, stream);
String ssrc = streamSession.getSSRC(device.getDeviceId(), channelId, stream);
MediaServerItem mediaServerItem = this.getOne(mediaServerId);
if (mediaServerItem != null) {
String streamId = String.format("%s_%s", device.getDeviceId(), channelId);
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

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

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

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

@ -16,10 +16,10 @@ import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
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.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
import com.genersoft.iot.vmp.utils.redis.RedisUtil;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.PlayResult;
import com.genersoft.iot.vmp.service.IMediaService;
@ -53,9 +53,6 @@ public class PlayServiceImpl implements IPlayService {
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Autowired
private RedisUtil redis;
@Autowired
private DeferredResultHolder resultHolder;
@ -104,19 +101,21 @@ public class PlayServiceImpl implements IPlayService {
logger.warn(String.format("设备点播超时，deviceId：%s ，channelId：%s", deviceId, channelId));
WVPResult wvpResult = new WVPResult();
wvpResult.setCode(-1);
SIPDialog dialog = streamSession.getDialog(deviceId, channelId);
SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, streamInfo.getStream());
if (dialog != null) {
wvpResult.setMsg("收流超时，请稍候重试");
}else {
wvpResult.setMsg("点播超时，请稍候重试");
}
msg.setData(wvpResult);
// 点播超时回复BYE
cmder.streamByeCmd(device.getDeviceId(), channelId);
cmder.streamByeCmd(device.getDeviceId(), channelId, streamInfo.getStream());
// 释放rtpserver
mediaServerService.closeRTPServer(playResult.getDevice(), channelId);
mediaServerService.closeRTPServer(playResult.getDevice(), channelId, streamInfo.getStream());
// 回复之前所有的点播请求
resultHolder.invokeAllResult(msg);
// TODO 释放ssrc
});
result.onCompletion(()->{
// 点播结束时调用截图接口
@ -154,14 +153,12 @@ public class PlayServiceImpl implements IPlayService {
}
});
if (streamInfo == null) {
SSRCInfo ssrcInfo;
String streamId = null;
if (mediaServerItem.isRtpEnable()) {
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) -> {
logger.info("收到订阅消息： " + response.toJSONString());
@ -173,7 +170,7 @@ public class PlayServiceImpl implements IPlayService {
WVPResult wvpResult = new WVPResult();
wvpResult.setCode(-1);
// 点播返回sip错误
mediaServerService.closeRTPServer(playResult.getDevice(), channelId);
mediaServerService.closeRTPServer(playResult.getDevice(), channelId, ssrcInfo.getStream());
wvpResult.setMsg(String.format("点播失败， 错误码： %s, %s", event.statusCode, event.msg));
msg.setData(wvpResult);
resultHolder.invokeAllResult(msg);
@ -184,7 +181,7 @@ public class PlayServiceImpl implements IPlayService {
});
} else {
String streamId = streamInfo.getStreamId();
String streamId = streamInfo.getStream();
if (streamId == null) {
WVPResult wvpResult = new WVPResult();
wvpResult.setCode(-1);
@ -213,18 +210,16 @@ public class PlayServiceImpl implements IPlayService {
// TODO 点播前是否重置状态
redisCatchStorage.stopPlay(streamInfo);
storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
SSRCInfo ssrcInfo;
String streamId2 = null;
if (mediaServerItem.isRtpEnable()) {
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) -> {
logger.info("收到订阅消息： " + response.toJSONString());
onPublishHandlerForPlay(mediaServerItemInuse, response, deviceId, channelId, uuid);
}, (event) -> {
mediaServerService.closeRTPServer(playResult.getDevice(), channelId);
mediaServerService.closeRTPServer(playResult.getDevice(), channelId, ssrcInfo.getStream());
WVPResult wvpResult = new WVPResult();
wvpResult.setCode(-1);
wvpResult.setMsg(String.format("点播失败， 错误码： %s, %s", event.statusCode, event.msg));
@ -242,12 +237,12 @@ public class PlayServiceImpl implements IPlayService {
RequestMessage msg = new RequestMessage();
msg.setId(uuid);
msg.setKey(DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId);
StreamInfo streamInfo = onPublishHandler(mediaServerItem, resonse, deviceId, channelId, uuid);
StreamInfo streamInfo = onPublishHandler(mediaServerItem, resonse, deviceId, channelId);
if (streamInfo != null) {
DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
if (deviceChannel != null) {
deviceChannel.setStreamId(streamInfo.getStreamId());
storager.startPlay(deviceId, channelId, streamInfo.getStreamId());
deviceChannel.setStreamId(streamInfo.getStream());
storager.startPlay(deviceId, channelId, streamInfo.getStream());
}
redisCatchStorage.startPlay(streamInfo);
msg.setData(JSON.toJSONString(streamInfo));
@ -284,29 +279,53 @@ public class PlayServiceImpl implements IPlayService {
@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();
msg.setKey(DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId);
msg.setId(uuid);
StreamInfo streamInfo = onPublishHandler(mediaServerItem, resonse, deviceId, channelId, uuid);
if (streamInfo != null) {
msg.setKey(key);
result.onTimeout(()->{
msg.setData("回放超时");
callback.call(msg);
});
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调用失败！");
msg.setData("设备回放API调用失败！");
callback.call(msg);
return;
}
redisCatchStorage.startPlayback(streamInfo);
msg.setData(JSON.toJSONString(streamInfo));
resultHolder.invokeResult(msg);
} else {
logger.warn("设备回放API调用失败！");
msg.setData("设备回放API调用失败！");
resultHolder.invokeResult(msg);
}
callback.call(msg);
}, event -> {
msg.setData(String.format("回放失败， 错误码： %s, %s", event.statusCode, event.msg));
callback.call(msg);
});
return result;
}
@Override
public void onPublishHandlerForDownload(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, String uuid) {
RequestMessage msg = new RequestMessage();
msg.setKey(DeferredResultHolder.CALLBACK_CMD_DOWNLOAD + deviceId + channelId);
msg.setId(uuid);
StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId, uuid);
StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId);
if (streamInfo != null) {
redisCatchStorage.startDownload(streamInfo);
msg.setData(JSON.toJSONString(streamInfo));
@ -319,7 +338,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");
JSONArray tracks = resonse.getJSONArray("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 {
streamLive = true;
StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(
mediaInfo, param.getApp(), param.getStream(), null);
mediaInfo, param.getApp(), param.getStream(), null, null);
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.StreamPushItem;
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 java.util.List;
@ -220,4 +221,5 @@ public interface IRedisCatchStorage {
void addMemInfo(double memInfo);
void addNetInfo(Map<String, String> networkInterfaces);
}

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.StreamPushItem;
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.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.dao.DeviceChannelMapper;
@ -91,7 +92,8 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
*/
@Override
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);
}
@ -105,7 +107,7 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
if (streamInfo == null) return false;
return redis.del(String.format("%S_%s_%s_%s_%s", VideoManagerConstants.PLAYER_PREFIX,
userSetup.getServerId(),
streamInfo.getStreamId(),
streamInfo.getStream(),
streamInfo.getDeviceID(),
streamInfo.getChannelId()));
}
@ -119,7 +121,7 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
return (StreamInfo)redis.get(String.format("%S_%s_%s_%s_%s",
VideoManagerConstants.PLAYER_PREFIX,
userSetup.getServerId(),
streamInfo.getStreamId(),
streamInfo.getStream(),
streamInfo.getDeviceID(),
streamInfo.getChannelId()));
}
@ -165,14 +167,14 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
@Override
public boolean startPlayback(StreamInfo stream) {
return redis.set(String.format("%S_%s_%s_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX, userSetup.getServerId(),stream.getStreamId(),
stream.getDeviceID(), stream.getChannelId()), stream);
return redis.set(String.format("%S_%s_%s_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX,
userSetup.getServerId(), stream.getStream(), stream.getDeviceID(), stream.getChannelId()), stream);
}
@Override
public boolean startDownload(StreamInfo streamInfo) {
return redis.set(String.format("%S_%s_%s_%s_%s", VideoManagerConstants.DOWNLOAD_PREFIX, userSetup.getServerId(),streamInfo.getStreamId(),
streamInfo.getDeviceID(), streamInfo.getChannelId()), streamInfo);
return redis.set(String.format("%S_%s_%s_%s_%s", VideoManagerConstants.DOWNLOAD_PREFIX, userSetup.getServerId(),
streamInfo.getStream(), streamInfo.getDeviceID(), streamInfo.getChannelId()), streamInfo);
}
@Override
@ -186,7 +188,7 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
}
return redis.del(String.format("%S_%s_%s_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX,
userSetup.getServerId(),
streamInfo.getStreamId(),
streamInfo.getStream(),
streamInfo.getDeviceID(),
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;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
@ -156,7 +157,10 @@ public class VideoManagerStoragerImpl implements IVideoManagerStorager {
public synchronized void updateChannel(String deviceId, DeviceChannel channel) {
String channelId = channel.getChannelId();
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());
channel.setUpdateTime(now);
DeviceChannel deviceChannel = deviceChannelMapper.queryChannel(deviceId, channelId);
@ -178,7 +182,10 @@ public class VideoManagerStoragerImpl implements IVideoManagerStorager {
if (channelList.size() == 0) {
for (DeviceChannel channel : channels) {
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());
channel.setUpdateTime(now);
channel.setCreateTime(now);
@ -189,9 +196,11 @@ public class VideoManagerStoragerImpl implements IVideoManagerStorager {
channelsInStore.put(deviceChannel.getChannelId(), deviceChannel);
}
for (DeviceChannel channel : channels) {
String channelId = channel.getChannelId();
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());
channel.setUpdateTime(now);
if (channelsInStore.get(channel.getChannelId()) != null) {

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

@ -110,26 +110,26 @@ public class PlayController {
String key = DeferredResultHolder.CALLBACK_CMD_STOP + deviceId + channelId;
resultHolder.put(key, uuid, result);
Device device = storager.queryVideoDevice(deviceId);
cmder.streamByeCmd(deviceId, channelId, (event) -> {
StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
if (streamInfo == null) {
RequestMessage msg = new RequestMessage();
msg.setId(uuid);
msg.setKey(key);
msg.setData("点播未找到");
resultHolder.invokeAllResult(msg);
storager.stopPlay(deviceId, channelId);
}else {
redisCatchStorage.stopPlay(streamInfo);
storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
RequestMessage msg = new RequestMessage();
msg.setId(uuid);
msg.setKey(key);
//Response response = event.getResponse();
msg.setData(String.format("success"));
resultHolder.invokeAllResult(msg);
}
mediaServerService.closeRTPServer(device, channelId);
StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
if (streamInfo == null) {
RequestMessage msg = new RequestMessage();
msg.setId(uuid);
msg.setKey(key);
msg.setData("点播未找到");
resultHolder.invokeAllResult(msg);
storager.stopPlay(deviceId, channelId);
return result;
}
cmder.streamByeCmd(deviceId, channelId, streamInfo.getStream(), (event) -> {
redisCatchStorage.stopPlay(streamInfo);
storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
RequestMessage msg = new RequestMessage();
msg.setId(uuid);
msg.setKey(key);
//Response response = event.getResponse();
msg.setData(String.format("success"));
resultHolder.invokeAllResult(msg);
mediaServerService.closeRTPServer(device, channelId, streamInfo.getStream());
});
if (deviceId != null || channelId != null) {
@ -329,7 +329,7 @@ public class PlayController {
jsonObject.put("deviceId", transaction.getDeviceId());
jsonObject.put("channelId", transaction.getChannelId());
jsonObject.put("ssrc", transaction.getSsrc());
jsonObject.put("streamId", transaction.getStreamId());
jsonObject.put("streamId", transaction.getStream());
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);
if (streamInfo != null) {
// 停止之前的下载
cmder.streamByeCmd(deviceId, channelId);
cmder.streamByeCmd(deviceId, channelId, streamInfo.getStream());
}
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) -> {
logger.info("收到订阅消息： " + response.toJSONString());
playService.onPublishHandlerForDownload(mediaServerItem, response, deviceId, channelId, uuid.toString());
playService.onPublishHandlerForDownload(mediaServerItem, response, deviceId, channelId, uuid);
}, event -> {
RequestMessage msg = new RequestMessage();
msg.setId(uuid);
@ -130,11 +130,12 @@ public class DownloadController {
@ApiImplicitParams({
@ApiImplicitParam(name = "deviceId", 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}")
public ResponseEntity<String> playStop(@PathVariable String deviceId, @PathVariable String channelId) {
@GetMapping("/stop/{deviceId}/{channelId}/{stream}")
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()) {
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.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.CrossOrigin;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
@ -75,52 +76,8 @@ public class PlaybackController {
if (logger.isDebugEnabled()) {
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);
// 超时处理
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));
DeferredResult<ResponseEntity<String>> result = playService.playBack(deviceId, channelId, startTime, endTime, msg->{
resultHolder.invokeResult(msg);
});
@ -131,24 +88,31 @@ public class PlaybackController {
@ApiImplicitParams({
@ApiImplicitParam(name = "deviceId", 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}")
public ResponseEntity<String> playStop(@PathVariable String deviceId, @PathVariable String channelId) {
@GetMapping("/stop/{deviceId}/{channelId}/{stream}")
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()) {
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) {
JSONObject json = new JSONObject();
json.put("deviceId", deviceId);
json.put("channelId", channelId);
return new ResponseEntity<String>(json.toString(), HttpStatus.OK);
return new ResponseEntity<>(json.toString(), HttpStatus.OK);
} else {
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)->{
StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(serial, code);
JSONObject result = new JSONObject();
result.put("StreamID", streamInfo.getStreamId());
result.put("StreamID", streamInfo.getStream());
result.put("DeviceID", device.getDeviceId());
result.put("ChannelID", code);
result.put("ChannelName", deviceChannel.getName());
@ -177,7 +177,7 @@ public class ApiStreamController {
result.put("error","未找到流信息");
return result;
}
cmder.streamByeCmd(serial, code);
cmder.streamByeCmd(serial, code, streamInfo.getStream());
redisCatchStorage.stopPlay(streamInfo);
storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
return null;

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

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

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

@ -307,7 +307,7 @@ export default {
this.isLoging = false;
// this.videoUrl = streamInfo.rtc;
this.videoUrl = this.getUrlByStreamInfo(streamInfo);
this.streamId = streamInfo.streamId;
this.streamId = streamInfo.stream;
this.app = streamInfo.app;
this.mediaServerId = streamInfo.mediaServerId;
this.playFromStreamInfo(false, streamInfo)
@ -485,8 +485,9 @@ export default {
}).then(function (res) {
var streamInfo = res.data;
that.app = streamInfo.app;
that.streamId = streamInfo.streamId;
that.streamId = streamInfo.stream;
that.mediaServerId = streamInfo.mediaServerId;
that.ssrc = streamInfo.ssrc;
that.videoUrl = that.getUrlByStreamInfo(streamInfo);
that.recordPlay = true;
});
@ -497,7 +498,7 @@ export default {
this.videoUrl = '';
this.$axios({
method: 'get',
url: '/api/playback/stop/' + this.deviceId + "/" + this.channelId
url: '/api/playback/stop/' + this.deviceId + "/" + this.channelId + "/" + this.streamId
}).then(function (res) {
if (callback) callback()
});
@ -517,7 +518,7 @@ export default {
}).then(function (res) {
var streamInfo = res.data;
that.app = streamInfo.app;
that.streamId = streamInfo.streamId;
that.streamId = streamInfo.stream;
that.mediaServerId = streamInfo.mediaServerId;
that.videoUrl = that.getUrlByStreamInfo(streamInfo);
that.recordPlay = true;
@ -529,7 +530,7 @@ export default {
this.videoUrl = '';
this.$axios({
method: 'get',
url: '/api/download/stop/' + this.deviceId + "/" + this.channelId
url: '/api/download/stop/' + this.deviceId + "/" + this.channelId+ "/" + this.streamId
}).then(function (res) {
if (callback) callback()
});
@ -539,8 +540,6 @@ export default {
let that = this;
this.$axios({
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
}).then(function (res) {});
},