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-28181-2.0

pull/353/head
mrjackwang 2022-03-15 10:28:23 +08:00 committed by GitHub
parent 1cb2b2cc5f 28e5c1ae05
commit 42901d0374
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
47 changed files with 759 additions and 548 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
DOCKERFILE
View File

@ -85,7 +85,7 @@ RUN echo '#!/bin/bash' > run.sh && \
echo 'nohup /opt/media/MediaServer -d -m 3 &' >> run.sh && \
echo 'cd /opt/wvp' >> run.sh && \
echo 'if [${WVP_CONFIG}]; then' >> run.sh && \
echo ' java -jar *.jar --spring.confi g.location=/opt/wvp/config/application.yml --media.record-assist-port=18081 ${WVP_CONFIG}' >> run.sh && \
echo ' java -jar *.jar --spring.config.location=/opt/wvp/config/application.yml --media.record-assist-port=18081 ${WVP_CONFIG}' >> run.sh && \
echo 'else' >> run.sh && \
echo ' java -jar *.jar --spring.config.location=/opt/wvp/config/application.yml --media.record-assist-port=18081 --media.ip=127.0.0.1 --media.sdp-ip=${WVP_IP} --sip.ip=${WVP_IP} --media.stream-ip=${WVP_IP}' >> run.sh && \
echo 'fi' >> run.sh

10
pom.xml
View File

@ -169,13 +169,6 @@
<version>1.2.73</version>
</dependency>
<!--Guava是一种基于开源的Java库-->
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>30.0-jre</version>
</dependency>
<!-- okhttp -->
<dependency>
<groupId>com.squareup.okhttp3</groupId>
@ -279,6 +272,9 @@
<plugin>
<groupId>pl.project13.maven</groupId>
<artifactId>git-commit-id-plugin</artifactId>
<configuration>
<offline>true</offline>
</configuration>
</plugin>
<plugin>

7
src/main/java/com/genersoft/iot/vmp/conf/SipPlatformRunner.java
View File

@ -60,12 +60,9 @@ public class SipPlatformRunner implements CommandLineRunner {
// 取消订阅
sipCommanderForPlatform.unregister(parentPlatform, null, (eventResult)->{
ParentPlatform platform = storager.queryParentPlatByServerGBId(parentPlatform.getServerGBId());
sipCommanderForPlatform.register(platform, null, null);
// 发送平台未注册消息
publisher.platformNotRegisterEventPublish(parentPlatform.getServerGBId());
});
// 发送平台未注册消息
publisher.platformNotRegisterEventPublish(parentPlatform.getServerGBId());
}
}
}

5
src/main/java/com/genersoft/iot/vmp/gb28181/bean/InviteStreamCallback.java Normal file
View File

@ -0,0 +1,5 @@
package com.genersoft.iot.vmp.gb28181.bean;
public interface InviteStreamCallback {
void call(InviteStreamInfo inviteStreamInfo);
}

61
src/main/java/com/genersoft/iot/vmp/gb28181/bean/InviteStreamInfo.java Normal file
View File

@ -0,0 +1,61 @@
package com.genersoft.iot.vmp.gb28181.bean;
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
public class InviteStreamInfo {
public InviteStreamInfo(MediaServerItem mediaServerItem, JSONObject response, String callId, String app, String stream) {
this.mediaServerItem = mediaServerItem;
this.response = response;
this.callId = callId;
this.app = app;
this.stream = stream;
}
private MediaServerItem mediaServerItem;
private JSONObject response;
private String callId;
private String app;
private String stream;
public MediaServerItem getMediaServerItem() {
return mediaServerItem;
}
public void setMediaServerItem(MediaServerItem mediaServerItem) {
this.mediaServerItem = mediaServerItem;
}
public JSONObject getResponse() {
return response;
}
public void setResponse(JSONObject response) {
this.response = response;
}
public String getCallId() {
return callId;
}
public void setCallId(String callId) {
this.callId = callId;
}
public String getApp() {
return app;
}
public void setApp(String app) {
this.app = app;
}
public String getStream() {
return stream;
}
public void setStream(String stream) {
this.stream = stream;
}
}

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

@ -114,6 +114,21 @@ public class ParentPlatform {
*/
private String catalogId;
/**
* 已被订阅目录信息
*/
private boolean catalogSubscribe;
/**
* 已被订阅报警信息
*/
private boolean alarmSubscribe;
/**
* 已被订阅GPS信息
*/
private boolean gpsSubscribe;
public Integer getId() {
return id;
}
@ -290,4 +305,28 @@ public class ParentPlatform {
public void setCatalogId(String catalogId) {
this.catalogId = catalogId;
}
public boolean isCatalogSubscribe() {
return catalogSubscribe;
}
public void setCatalogSubscribe(boolean catalogSubscribe) {
this.catalogSubscribe = catalogSubscribe;
}
public boolean isAlarmSubscribe() {
return alarmSubscribe;
}
public void setAlarmSubscribe(boolean alarmSubscribe) {
this.alarmSubscribe = alarmSubscribe;
}
public boolean isGpsSubscribe() {
return gpsSubscribe;
}
public void setGpsSubscribe(boolean gpsSubscribe) {
this.gpsSubscribe = gpsSubscribe;
}
}

12
src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeHolder.java
View File

@ -2,6 +2,8 @@ package com.genersoft.iot.vmp.gb28181.bean;
import org.springframework.stereotype.Component;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
@Component
@ -34,4 +36,14 @@ public class SubscribeHolder {
public void removeMobilePositionSubscribe(String platformId) {
mobilePositionMap.remove(platformId);
}
public List<String> getAllCatalogSubscribePlatform() {
List<String> platforms = new ArrayList<>();
if(catalogMap.size() > 0) {
for (String key : catalogMap.keySet()) {
platforms.add(catalogMap.get(key).getId());
}
}
return platforms;
}
}

35
src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeInfo.java
View File

@ -14,19 +14,15 @@ public class SubscribeInfo {
public SubscribeInfo(RequestEvent evt, String id) {
this.id = id;
Request request = evt.getRequest();
CallIdHeader callIdHeader = (CallIdHeader)request.getHeader(CallIdHeader.NAME);
this.callId = callIdHeader.getCallId();
FromHeader fromHeader = (FromHeader)request.getHeader(FromHeader.NAME);
this.fromTag = fromHeader.getTag();
ExpiresHeader expiresHeader = (ExpiresHeader)request.getHeader(ExpiresHeader.NAME);
this.expires = expiresHeader.getExpires();
EventHeader eventHeader = (EventHeader)request.getHeader(EventHeader.NAME);
this.eventId = eventHeader.getEventId();
this.eventType = eventHeader.getEventType();
ViaHeader viaHeader = (ViaHeader)request.getHeader(ViaHeader.NAME);
this.branch = viaHeader.getBranch();
this.transaction = evt.getServerTransaction();
this.dialog = evt.getDialog();
CallIdHeader callIdHeader = (CallIdHeader)evt.getRequest().getHeader(CallIdHeader.NAME);
this.callId = callIdHeader.getCallId();
}
private String id;
@ -34,9 +30,6 @@ public class SubscribeInfo {
private String callId;
private String eventId;
private String eventType;
private String fromTag;
private String toTag;
private String branch;
private ServerTransaction transaction;
private Dialog dialog;
@ -52,18 +45,6 @@ public class SubscribeInfo {
return callId;
}
public String getFromTag() {
return fromTag;
}
public void setToTag(String toTag) {
this.toTag = toTag;
}
public String getToTag() {
return toTag;
}
public void setId(String id) {
this.id = id;
}
@ -76,10 +57,6 @@ public class SubscribeInfo {
this.callId = callId;
}
public void setFromTag(String fromTag) {
this.fromTag = fromTag;
}
public String getEventId() {
return eventId;
}
@ -96,14 +73,6 @@ public class SubscribeInfo {
this.eventType = eventType;
}
public String getBranch() {
return branch;
}
public void setBranch(String branch) {
this.branch = branch;
}
public ServerTransaction getTransaction() {
return transaction;
}

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

@ -30,7 +30,7 @@ public class SipSubscribe {
// @Scheduled(fixedRate= 100 * 60 * 60 )
@Scheduled(cron="0 0/5 * * * ?") //每5分钟执行一次
public void execute(){
logger.info("[定时任务] 清理过期的订阅信息");
logger.info("[定时任务] 清理过期的SIP订阅信息");
Calendar calendar = Calendar.getInstance();
calendar.setTime(new Date());
calendar.set(Calendar.MINUTE, calendar.get(Calendar.MINUTE) - 5);
@ -49,10 +49,10 @@ public class SipSubscribe {
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());
logger.debug("okTimeSubscribes.size:{}",okTimeSubscribes.size());
logger.debug("okSubscribes.size:{}",okSubscribes.size());
logger.debug("errorTimeSubscribes.size:{}",errorTimeSubscribes.size());
logger.debug("errorSubscribes.size:{}",errorSubscribes.size());
}
public interface Event {

5
src/main/java/com/genersoft/iot/vmp/gb28181/event/online/OnlineEventListener.java
View File

@ -68,8 +68,6 @@ public class OnlineEventListener implements ApplicationListener<OnlineEvent> {
String key = VideoManagerConstants.KEEPLIVEKEY_PREFIX + userSetup.getServerId() + "_" + event.getDevice().getDeviceId();
Device deviceInStore = storager.queryVideoDevice(device.getDeviceId());
device.setOnline(1);
// 处理上线监听
storager.updateDevice(device);
switch (event.getFrom()) {
// 注册时触发的在线事件，先在redis中增加超时超时监听
case VideoManagerConstants.EVENT_ONLINE_REGISTER:
@ -98,7 +96,8 @@ public class OnlineEventListener implements ApplicationListener<OnlineEvent> {
break;
}
// 处理上线监听
storager.updateDevice(device);
List<DeviceChannel> deviceChannelList = storager.queryOnlineChannelsByDeviceId(device.getDeviceId());
eventPublisher.catalogEventPublish(null, deviceChannelList, CatalogEvent.ON);
// 上线添加订阅

4
src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java
View File

@ -74,7 +74,7 @@ public class CatalogEventLister implements ApplicationListener<CatalogEvent> {
}
}else {
// 获取所用订阅
List<String> platforms = redisCatchStorage.getAllSubscribePlatform();
List<String> platforms = subscribeHolder.getAllCatalogSubscribePlatform();
if (event.getDeviceChannels() != null) {
if (platforms.size() > 0) {
for (DeviceChannel deviceChannel : event.getDeviceChannels()) {
@ -117,8 +117,6 @@ public class CatalogEventLister implements ApplicationListener<CatalogEvent> {
List<ParentPlatform> parentPlatforms = parentPlatformMap.get(gbId);
if (parentPlatforms != null && parentPlatforms.size() > 0) {
for (ParentPlatform platform : parentPlatforms) {
String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetup.getServerId() + "_Catalog_" + platform.getServerGBId();
// SubscribeInfo subscribeInfo = redisCatchStorage.getSubscribe(key);
SubscribeInfo subscribeInfo = subscribeHolder.getCatalogSubscribe(platform.getServerGBId());
if (subscribeInfo == null) continue;
logger.info("[Catalog事件: {}]平台：{}，影响通道{}", event.getType(), platform.getServerGBId(), gbId);

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

@ -95,14 +95,14 @@ public class SIPProcessorObserver implements ISIPProcessorObserver {
logger.debug("\n收到响应：\n{}", responseEvent.getResponse());
int status = response.getStatusCode();
if (((status >= 200) && (status < 300)) || status == 401) { // Success!
if (((status >= 200) && (status < 300)) || status == Response.UNAUTHORIZED) { // Success!
CSeqHeader cseqHeader = (CSeqHeader) responseEvent.getResponse().getHeader(CSeqHeader.NAME);
String method = cseqHeader.getMethod();
ISIPResponseProcessor sipRequestProcessor = responseProcessorMap.get(method);
if (sipRequestProcessor != null) {
sipRequestProcessor.process(responseEvent);
}
if (responseEvent.getResponse() != null && sipSubscribe.getOkSubscribesSize() > 0 ) {
if (status != Response.UNAUTHORIZED && responseEvent.getResponse() != null && sipSubscribe.getOkSubscribesSize() > 0 ) {
CallIdHeader callIdHeader = (CallIdHeader)responseEvent.getResponse().getHeader(CallIdHeader.NAME);
if (callIdHeader != null) {
SipSubscribe.Event subscribe = sipSubscribe.getOkSubscribe(callIdHeader.getCallId());

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

@ -2,6 +2,7 @@ package com.genersoft.iot.vmp.gb28181.transmit.cmd;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.InviteStreamCallback;
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;
@ -103,7 +104,7 @@ public interface ISIPCommander {
* @param startTime 开始时间,格式要求：yyyy-MM-dd HH:mm:ss
* @param endTime 结束时间,格式要求：yyyy-MM-dd HH:mm:ss
*/
void playbackStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInf, Device device, String channelId, String startTime, String endTime, ZLMHttpHookSubscribe.Event event, SipSubscribe.Event errorEvent);
void playbackStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInf, Device device, String channelId, String startTime, String endTime,InviteStreamCallback inviteStreamCallback, InviteStreamCallback event, SipSubscribe.Event errorEvent);
/**
* 请求历史媒体下载
@ -114,13 +115,13 @@ public interface ISIPCommander {
* @param endTime 结束时间,格式要求：yyyy-MM-dd HH:mm:ss
* @param downloadSpeed 下载倍速参数
*/
void downloadStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, String startTime, String endTime, String downloadSpeed, ZLMHttpHookSubscribe.Event event, SipSubscribe.Event errorEvent);
void downloadStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, String startTime, String endTime, String downloadSpeed, InviteStreamCallback event, SipSubscribe.Event errorEvent);
/**
* 视频流停止
*/
void streamByeCmd(String deviceId, String channelId, String stream, SipSubscribe.Event okEvent);
void streamByeCmd(String deviceId, String channelId, String stream);
void streamByeCmd(String deviceId, String channelId, String stream, String callId, SipSubscribe.Event okEvent);
void streamByeCmd(String deviceId, String channelId, String stream, String callId);
/**
* 回放暂停

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

@ -6,6 +6,8 @@ import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetup;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.InviteStreamCallback;
import com.genersoft.iot.vmp.gb28181.bean.InviteStreamInfo;
import com.genersoft.iot.vmp.gb28181.bean.SsrcTransaction;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
@ -445,27 +447,13 @@ public class SIPCommander implements ISIPCommander {
* @param endTime 结束时间,格式要求：yyyy-MM-dd HH:mm:ss
*/
@Override
public void playbackStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, String startTime, String endTime, ZLMHttpHookSubscribe.Event event
, SipSubscribe.Event errorEvent) {
public void playbackStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId,
String startTime, String endTime, InviteStreamCallback inviteStreamCallback, InviteStreamCallback hookEvent,
SipSubscribe.Event errorEvent) {
try {
logger.info("{} 分配的ZLM为: {} [{}:{}]", ssrcInfo.getStream(), mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort());
// 添加订阅
JSONObject subscribeKey = new JSONObject();
subscribeKey.put("app", "rtp");
subscribeKey.put("stream", ssrcInfo.getStream());
subscribeKey.put("regist", true);
subscribeKey.put("schema", "rtmp");
subscribeKey.put("mediaServerId", mediaServerItem.getId());
logger.debug("录像回放添加订阅，订阅内容：" + subscribeKey.toString());
subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
(MediaServerItem mediaServerItemInUse, JSONObject json)->{
if (event != null) {
event.response(mediaServerItemInUse, json);
}
});
StringBuffer content = new StringBuffer(200);
content.append("v=0\r\n");
content.append("o="+sipConfig.getId()+" 0 0 IN IP4 " + mediaServerItem.getSdpIp() + "\r\n");
@ -530,6 +518,21 @@ public class SIPCommander implements ISIPCommander {
CallIdHeader callIdHeader = device.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
: udpSipProvider.getNewCallId();
// 添加订阅
JSONObject subscribeKey = new JSONObject();
subscribeKey.put("app", "rtp");
subscribeKey.put("stream", ssrcInfo.getStream());
subscribeKey.put("regist", true);
subscribeKey.put("schema", "rtmp");
subscribeKey.put("mediaServerId", mediaServerItem.getId());
logger.debug("录像回放添加订阅，订阅内容：" + subscribeKey);
subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
(MediaServerItem mediaServerItemInUse, JSONObject json)->{
if (hookEvent != null) {
InviteStreamInfo inviteStreamInfo = new InviteStreamInfo(mediaServerItemInUse, json, callIdHeader.getCallId(), "rtp", ssrcInfo.getStream());
hookEvent.call(inviteStreamInfo);
}
});
Request request = headerProvider.createPlaybackInviteRequest(device, channelId, content.toString(), null, "fromplybck" + tm, null, callIdHeader, ssrcInfo.getSsrc());
transmitRequest(device, request, errorEvent, okEvent -> {
@ -537,6 +540,9 @@ public class SIPCommander implements ISIPCommander {
streamSession.put(device.getDeviceId(), channelId, callIdHeader.getCallId(), ssrcInfo.getStream(), ssrcInfo.getSsrc(), mediaServerItem.getId(), responseEvent.getClientTransaction());
streamSession.put(device.getDeviceId(), channelId, callIdHeader.getCallId(), okEvent.dialog);
});
if (inviteStreamCallback != null) {
inviteStreamCallback.call(new InviteStreamInfo(mediaServerItem, null, callIdHeader.getCallId(), "rtp", ssrcInfo.getStream()));
}
} catch ( SipException | ParseException | InvalidArgumentException e) {
e.printStackTrace();
}
@ -552,24 +558,11 @@ public class SIPCommander implements ISIPCommander {
* @param downloadSpeed 下载倍速参数
*/
@Override
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, InviteStreamCallback event
, SipSubscribe.Event errorEvent) {
try {
logger.info("{} 分配的ZLM为: {} [{}:{}]", ssrcInfo.getStream(), mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort());
// 添加订阅
JSONObject subscribeKey = new JSONObject();
subscribeKey.put("app", "rtp");
subscribeKey.put("stream", ssrcInfo.getStream());
subscribeKey.put("regist", true);
subscribeKey.put("mediaServerId", mediaServerItem.getId());
logger.debug("录像回放添加订阅，订阅内容：" + subscribeKey.toString());
subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
(MediaServerItem mediaServerItemInUse, JSONObject json)->{
event.response(mediaServerItemInUse, json);
subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey);
});
StringBuffer content = new StringBuffer(200);
content.append("v=0\r\n");
content.append("o="+sipConfig.getId()+" 0 0 IN IP4 " + mediaServerItem.getSdpIp() + "\r\n");
@ -637,6 +630,19 @@ public class SIPCommander implements ISIPCommander {
CallIdHeader callIdHeader = device.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
: udpSipProvider.getNewCallId();
// 添加订阅
JSONObject subscribeKey = new JSONObject();
subscribeKey.put("app", "rtp");
subscribeKey.put("stream", ssrcInfo.getStream());
subscribeKey.put("regist", true);
subscribeKey.put("mediaServerId", mediaServerItem.getId());
logger.debug("录像回放添加订阅，订阅内容：" + subscribeKey.toString());
subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
(MediaServerItem mediaServerItemInUse, JSONObject json)->{
event.call(new InviteStreamInfo(mediaServerItem, json, callIdHeader.getCallId(), "rtp", ssrcInfo.getStream()));
subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey);
});
Request request = headerProvider.createPlaybackInviteRequest(device, channelId, content.toString(), null, "fromplybck" + tm, null, callIdHeader, ssrcInfo.getSsrc());
ClientTransaction transaction = transmitRequest(device, request, errorEvent);
@ -652,15 +658,15 @@ public class SIPCommander implements ISIPCommander {
* 视频流停止, 不使用回调
*/
@Override
public void streamByeCmd(String deviceId, String channelId, String stream) {
streamByeCmd(deviceId, channelId, stream, null);
public void streamByeCmd(String deviceId, String channelId, String stream, String callId) {
streamByeCmd(deviceId, channelId, stream, callId, null);
}
/**
* 视频流停止
*/
@Override
public void streamByeCmd(String deviceId, String channelId, String stream, SipSubscribe.Event okEvent) {
public void streamByeCmd(String deviceId, String channelId, String stream, String callId, SipSubscribe.Event okEvent) {
try {
SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(deviceId, channelId, null, stream);
ClientTransaction transaction = streamSession.getTransactionByStream(deviceId, channelId, stream);
@ -672,7 +678,15 @@ public class SIPCommander implements ISIPCommander {
}
return;
}
SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, stream);
SIPDialog dialog;
if (callId != null) {
dialog = streamSession.getDialogByCallId(deviceId, channelId, callId);
}else {
if (stream == null) return;
dialog = streamSession.getDialogByStream(deviceId, channelId, stream);
}
if (dialog == null) {
logger.warn("[ {} -> {}]停止视频流的时候发现对话已丢失", deviceId, channelId);
return;

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

@ -13,6 +13,7 @@ import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.utils.SerializeUtils;
import gov.nist.javax.sip.SipProviderImpl;
import gov.nist.javax.sip.SipStackImpl;
import gov.nist.javax.sip.message.MessageFactoryImpl;
import gov.nist.javax.sip.message.SIPRequest;
import gov.nist.javax.sip.stack.SIPDialog;
import org.slf4j.Logger;
@ -77,11 +78,11 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
@Override
public boolean unregister(ParentPlatform parentPlatform, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent) {
ParentPlatformCatch parentPlatformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId());
parentPlatform.setExpires("0");
if (parentPlatformCatch != null) {
parentPlatformCatch.setParentPlatform(parentPlatform);
redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
}
parentPlatform.setExpires("0");
return register(parentPlatform, null, null, errorEvent, okEvent, false);
}
@ -101,7 +102,9 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
callIdHeader = udpSipProvider.getNewCallId();
}
request = headerProviderPlarformProvider.createRegisterRequest(parentPlatform, redisCatchStorage.getCSEQ(Request.REGISTER), "FromRegister" + tm, null, callIdHeader);
request = headerProviderPlarformProvider.createRegisterRequest(parentPlatform,
redisCatchStorage.getCSEQ(Request.REGISTER), "FromRegister" + tm,
"z9hG4bK-" + UUID.randomUUID().toString().replace("-", ""), callIdHeader);
// 将 callid 写入缓存， 等注册成功可以更新状态
String callIdFromHeader = callIdHeader.getCallId();
redisCatchStorage.updatePlatformRegisterInfo(callIdFromHeader, parentPlatform.getServerGBId());
@ -414,11 +417,13 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
private void sendNotify(ParentPlatform parentPlatform, String catalogXmlContent,
SubscribeInfo subscribeInfo, SipSubscribe.Event errorEvent, SipSubscribe.Event okEvent )
throws NoSuchFieldException, IllegalAccessException, SipException, ParseException {
MessageFactoryImpl messageFactory = (MessageFactoryImpl) sipFactory.createMessageFactory();
// 设置编码， 防止中文乱码
messageFactory.setDefaultContentEncodingCharset("gb2312");
Dialog dialog = subscribeInfo.getDialog();
Request notifyRequest = dialog.createRequest(Request.NOTIFY);
if (dialog == null) return;
SIPRequest notifyRequest = (SIPRequest)dialog.createRequest(Request.NOTIFY);
ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("Application", "MANSCDP+xml");
notifyRequest.setContent(catalogXmlContent, contentTypeHeader);
SubscriptionStateHeader subscriptionState = sipFactory.createHeaderFactory()
@ -509,7 +514,8 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
}
@Override
public boolean sendNotifyForCatalogOther(String type, ParentPlatform parentPlatform, List<DeviceChannel> deviceChannels, SubscribeInfo subscribeInfo, Integer index) {
public boolean sendNotifyForCatalogOther(String type, ParentPlatform parentPlatform, List<DeviceChannel> deviceChannels,
SubscribeInfo subscribeInfo, Integer index) {
if (parentPlatform == null
|| deviceChannels == null
|| deviceChannels.size() == 0
@ -577,24 +583,30 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
recordXml.append("<SN>" +recordInfo.getSn() + "</SN>\r\n");
recordXml.append("<DeviceID>" + recordInfo.getDeviceId() + "</DeviceID>\r\n");
recordXml.append("<SumNum>" + recordInfo.getSumNum() + "</SumNum>\r\n");
recordXml.append("<RecordList Num=\"" + recordInfo.getRecordList().size()+"\">\r\n");
for (RecordItem recordItem : recordInfo.getRecordList()) {
recordXml.append("<Item>\r\n");
if (deviceChannel != null) {
recordXml.append("<DeviceID>" + recordItem.getDeviceId() + "</DeviceID>\r\n");
recordXml.append("<Name>" + recordItem.getName() + "</Name>\r\n");
recordXml.append("<StartTime>" + DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(recordItem.getStartTime()) + "</StartTime>\r\n");
recordXml.append("<EndTime>" + DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(recordItem.getEndTime()) + "</EndTime>\r\n");
recordXml.append("<Secrecy>" + recordItem.getSecrecy() + "</Secrecy>\r\n");
recordXml.append("<Type>" + recordItem.getType() + "</Type>\r\n");
if (!StringUtils.isEmpty(recordItem.getFileSize())) {
recordXml.append("<FileSize>" + recordItem.getFileSize() + "</FileSize>\r\n");
}
if (!StringUtils.isEmpty(recordItem.getFilePath())) {
recordXml.append("<FilePath>" + recordItem.getFilePath() + "</FilePath>\r\n");
if (recordInfo.getRecordList() == null ) {
recordXml.append("<RecordList Num=\"0\">\r\n");
}else {
recordXml.append("<RecordList Num=\"" + recordInfo.getRecordList().size()+"\">\r\n");
if (recordInfo.getRecordList().size() > 0) {
for (RecordItem recordItem : recordInfo.getRecordList()) {
recordXml.append("<Item>\r\n");
if (deviceChannel != null) {
recordXml.append("<DeviceID>" + recordItem.getDeviceId() + "</DeviceID>\r\n");
recordXml.append("<Name>" + recordItem.getName() + "</Name>\r\n");
recordXml.append("<StartTime>" + DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(recordItem.getStartTime()) + "</StartTime>\r\n");
recordXml.append("<EndTime>" + DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(recordItem.getEndTime()) + "</EndTime>\r\n");
recordXml.append("<Secrecy>" + recordItem.getSecrecy() + "</Secrecy>\r\n");
recordXml.append("<Type>" + recordItem.getType() + "</Type>\r\n");
if (!StringUtils.isEmpty(recordItem.getFileSize())) {
recordXml.append("<FileSize>" + recordItem.getFileSize() + "</FileSize>\r\n");
}
if (!StringUtils.isEmpty(recordItem.getFilePath())) {
recordXml.append("<FilePath>" + recordItem.getFilePath() + "</FilePath>\r\n");
}
}
recordXml.append("</Item>\r\n");
}
}
recordXml.append("</Item>\r\n");
}
recordXml.append("</RecordList>\r\n");

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

@ -27,10 +27,7 @@ import javax.sip.header.CallIdHeader;
import javax.sip.header.FromHeader;
import javax.sip.header.HeaderAddress;
import javax.sip.header.ToHeader;
import java.util.HashMap;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.*;
/**
* SIP命令类型： ACK请求
@ -84,44 +81,72 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In
String channelId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(ToHeader.NAME)).getAddress().getURI()).getUser();
SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(platformGbId, channelId, null, callIdHeader.getCallId());
String is_Udp = sendRtpItem.isTcp() ? "0" : "1";
String deviceId = sendRtpItem.getDeviceId();
StreamInfo streamInfo = null;
if (sendRtpItem.isPlay()) {
streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
}else {
streamInfo = redisCatchStorage.queryPlaybackByDevice(deviceId, channelId);
}
if (streamInfo == null) {
streamInfo = new StreamInfo();
streamInfo.setApp(sendRtpItem.getApp());
streamInfo.setStream(sendRtpItem.getStreamId());
}
redisCatchStorage.updateSendRTPSever(sendRtpItem);
MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
logger.info("收到ACK，开始向上级推流 rtp/{}", sendRtpItem.getStreamId());
Map<String, Object> param = new HashMap<>();
param.put("vhost","__defaultVhost__");
param.put("app",streamInfo.getApp());
param.put("stream",streamInfo.getStream());
param.put("app",sendRtpItem.getApp());
param.put("stream",sendRtpItem.getStreamId());
param.put("ssrc", sendRtpItem.getSsrc());
param.put("dst_url",sendRtpItem.getIp());
param.put("dst_port", sendRtpItem.getPort());
param.put("is_udp", is_Udp);
MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
JSONObject jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
if (jsonObject.getInteger("code") != 0) {
logger.info("监听流以等待流上线{}/{}", streamInfo.getApp(), streamInfo.getStream());
// 监听流上线
// 添加订阅
JSONObject subscribeKey = new JSONObject();
subscribeKey.put("app", "rtp");
subscribeKey.put("stream", streamInfo.getStream());
subscribeKey.put("regist", true);
subscribeKey.put("schema", "rtmp");
subscribeKey.put("mediaServerId", sendRtpItem.getMediaServerId());
subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
(MediaServerItem mediaServerItemInUse, JSONObject json)->{
zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
});
}
param.put("src_port", sendRtpItem.getLocalPort());
zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
// if (streamInfo == null) { // 流还没上来，对方就回复ack
// logger.info("监听流以等待流上线1 rtp/{}", sendRtpItem.getStreamId());
// // 监听流上线
// // 添加订阅
// JSONObject subscribeKey = new JSONObject();
// subscribeKey.put("app", "rtp");
// subscribeKey.put("stream", sendRtpItem.getStreamId());
// subscribeKey.put("regist", true);
// subscribeKey.put("schema", "rtmp");
// subscribeKey.put("mediaServerId", sendRtpItem.getMediaServerId());
// subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
// (MediaServerItem mediaServerItemInUse, JSONObject json)->{
// Map<String, Object> param = new HashMap<>();
// param.put("vhost","__defaultVhost__");
// param.put("app",json.getString("app"));
// param.put("stream",json.getString("stream"));
// param.put("ssrc", sendRtpItem.getSsrc());
// param.put("dst_url",sendRtpItem.getIp());
// param.put("dst_port", sendRtpItem.getPort());
// param.put("is_udp", is_Udp);
// param.put("src_port", sendRtpItem.getLocalPort());
// zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
// });
// }else {
// Map<String, Object> param = new HashMap<>();
// param.put("vhost","__defaultVhost__");
// param.put("app",streamInfo.getApp());
// param.put("stream",streamInfo.getStream());
// param.put("ssrc", sendRtpItem.getSsrc());
// param.put("dst_url",sendRtpItem.getIp());
// param.put("dst_port", sendRtpItem.getPort());
// param.put("is_udp", is_Udp);
// param.put("src_port", sendRtpItem.getLocalPort());
//
// JSONObject jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
// if (jsonObject.getInteger("code") != 0) {
// logger.info("监听流以等待流上线2 {}/{}", streamInfo.getApp(), streamInfo.getStream());
// // 监听流上线
// // 添加订阅
// JSONObject subscribeKey = new JSONObject();
// subscribeKey.put("app", "rtp");
// subscribeKey.put("stream", streamInfo.getStream());
// subscribeKey.put("regist", true);
// subscribeKey.put("schema", "rtmp");
// subscribeKey.put("mediaServerId", sendRtpItem.getMediaServerId());
// subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
// (MediaServerItem mediaServerItemInUse, JSONObject json)->{
// zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
// });
// }
// }
}
}
}

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

@ -93,14 +93,16 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
param.put("app",sendRtpItem.getApp());
param.put("stream",streamId);
param.put("ssrc",sendRtpItem.getSsrc());
logger.info("停止向上级推流：" + streamId);
logger.info("收到bye:停止向上级推流：" + streamId);
MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param);
redisCatchStorage.deleteSendRTPServer(platformGbId, channelId, callIdHeader.getCallId(), null);
int totalReaderCount = zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId);
if (totalReaderCount <= 0) {
logger.info(streamId + "无其它观看者，通知设备停止推流");
cmder.streamByeCmd(sendRtpItem.getDeviceId(), channelId, streamId);
logger.info("收到bye: {}无其它观看者，通知设备停止推流", streamId);
if (sendRtpItem.isPlay()) {
cmder.streamByeCmd(sendRtpItem.getDeviceId(), channelId, streamId, null);
}
}
}
// 可能是设备主动停止

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

@ -6,6 +6,7 @@ import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
@ -91,6 +92,9 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
@Autowired
private SIPProcessorObserver sipProcessorObserver;
@Autowired
private VideoStreamSessionManager sessionManager;
@Override
public void afterPropertiesSet() throws Exception {
@ -233,6 +237,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
}
String username = sdp.getOrigin().getUsername();
String addressStr = sdp.getOrigin().getAddress();
logger.info("[上级点播]用户：{}， 地址：{}:{}， ssrc：{}", username, addressStr, port, ssrc);
Device device = null;
// 通过 channel 和 gbStream 是否为null 值判断来源是直播流合适国标
@ -266,13 +271,14 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
sendRtpItem.setDialog(dialogByteArray);
byte[] transactionByteArray = SerializeUtils.serialize(evt.getServerTransaction());
sendRtpItem.setTransaction(transactionByteArray);
// 写入redis， 超时时回复
redisCatchStorage.updateSendRTPSever(sendRtpItem);
Long finalStartTime = startTime;
Long finalStopTime = stopTime;
ZLMHttpHookSubscribe.Event hookEvent = (mediaServerItemInUSe, responseJSON)->{
logger.info("[上级点播]下级已经开始推流。 回复200OK(SDP)， {}/{}", sendRtpItem.getApp(), sendRtpItem.getStreamId());
String app = responseJSON.getString("app");
String stream = responseJSON.getString("stream");
logger.info("[上级点播]下级已经开始推流。 回复200OK(SDP)， {}/{}", app, stream);
// * 0 等待设备推流上来
// * 1 下级已经推流，等待上级平台回复ack
// * 2 推流中
@ -325,46 +331,66 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
e.printStackTrace();
}
});
sendRtpItem.setApp("rtp");
if ("Playback".equals(sessionName)) {
sendRtpItem.setPlay(false);
sendRtpItem.setStreamId(ssrc);
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, true);
sendRtpItem.setStreamId(ssrcInfo.getStream());
// 写入redis， 超时时回复
redisCatchStorage.updateSendRTPSever(sendRtpItem);
SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
playService.playBack(device.getDeviceId(), channelId, format.format(start), format.format(end),result -> {
if (result.getCode() != 0){
logger.warn("录像回放失败");
if (result.getEvent() != null) {
errorEvent.response(result.getEvent());
playService.playBack(mediaServerItem, ssrcInfo, device.getDeviceId(), channelId, format.format(start),
format.format(end), null, result -> {
if (result.getCode() != 0){
logger.warn("录像回放失败");
if (result.getEvent() != null) {
errorEvent.response(result.getEvent());
}
redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), channelId, callIdHeader.getCallId(), null);
try {
responseAck(evt, Response.REQUEST_TIMEOUT);
} catch (SipException e) {
e.printStackTrace();
} catch (InvalidArgumentException e) {
e.printStackTrace();
} catch (ParseException e) {
e.printStackTrace();
}
}else {
if (result.getMediaServerItem() != null) {
hookEvent.response(result.getMediaServerItem(), result.getResponse());
}
}
redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), channelId, callIdHeader.getCallId(), null);
try {
responseAck(evt, Response.REQUEST_TIMEOUT);
} catch (SipException e) {
e.printStackTrace();
} catch (InvalidArgumentException e) {
e.printStackTrace();
} catch (ParseException e) {
e.printStackTrace();
}
}else {
if (result.getMediaServerItem() != null) {
hookEvent.response(result.getMediaServerItem(), result.getResponse());
}
}
});
});
}else {
sendRtpItem.setPlay(true);
StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(device.getDeviceId(), channelId);
if (streamInfo == null) {
SsrcTransaction playTransaction = sessionManager.getSsrcTransaction(device.getDeviceId(), channelId, "play", null);
if (playTransaction != null) {
Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, "rtp", playTransaction.getStream());
if (!streamReady) {
playTransaction = null;
}
}
if (playTransaction == null) {
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, true);
if (mediaServerItem.isRtpEnable()) {
sendRtpItem.setStreamId(String.format("%s_%s", device.getDeviceId(), channelId));
}else {
sendRtpItem.setStreamId(ssrcInfo.getStream());
}
sendRtpItem.setPlay(false);
playService.play(mediaServerItem,device.getDeviceId(), channelId, hookEvent, errorEvent, ()->{
// 写入redis， 超时时回复
redisCatchStorage.updateSendRTPSever(sendRtpItem);
playService.play(mediaServerItem, ssrcInfo, device, channelId, hookEvent, errorEvent, (code, msg)->{
redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), channelId, callIdHeader.getCallId(), null);
});
}, null);
}else {
sendRtpItem.setStreamId(streamInfo.getStream());
hookEvent.response(mediaServerItem, null);
sendRtpItem.setStreamId(playTransaction.getStream());
// 写入redis， 超时时回复
redisCatchStorage.updateSendRTPSever(sendRtpItem);
JSONObject jsonObject = new JSONObject();
jsonObject.put("app", sendRtpItem.getApp());
jsonObject.put("stream", sendRtpItem.getStreamId());
hookEvent.response(mediaServerItem, jsonObject);
}
}
}else if (gbStream != null) {

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

@ -233,7 +233,6 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
*/
private void processNotifyCatalogList(RequestEvent evt) {
try {
System.out.println(343434);
FromHeader fromHeader = (FromHeader) evt.getRequest().getHeader(FromHeader.NAME);
String deviceId = SipUtils.getUserIdFromFromHeader(fromHeader);

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

@ -158,20 +158,14 @@ public class SubscribeRequestProcessor extends SIPRequestProcessorParent impleme
String interval = XmlUtil.getText(rootElement, "Interval"); // GPS上报时间间隔
dynamicTask.startCron(key, new GPSSubscribeTask(redisCatchStorage, sipCommanderForPlatform, storager, platformId, sn, key, subscribeHolder), Integer.parseInt(interval));
subscribeHolder.putMobilePositionSubscribe(platformId, subscribeInfo);
// redisCatchStorage.updateSubscribe(key, subscribeInfo);
}else if (subscribeInfo.getExpires() == 0) {
dynamicTask.stop(key);
// redisCatchStorage.delSubscribe(key);
subscribeHolder.removeMobilePositionSubscribe(platformId);
}
try {
ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformId);
Response response = responseXmlAck(evt, resultXml.toString(), parentPlatform);
ToHeader toHeader = (ToHeader)response.getHeader(ToHeader.NAME);
subscribeInfo.setToTag(toHeader.getTag());
redisCatchStorage.updateSubscribe(key, subscribeInfo);
responseXmlAck(evt, resultXml.toString(), parentPlatform);
} catch (SipException e) {
e.printStackTrace();
} catch (InvalidArgumentException e) {
@ -211,21 +205,14 @@ public class SubscribeRequestProcessor extends SIPRequestProcessorParent impleme
.append("</Response>\r\n");
if (subscribeInfo.getExpires() > 0) {
// redisCatchStorage.updateSubscribe(key, subscribeInfo);
subscribeHolder.putCatalogSubscribe(platformId, subscribeInfo);
}else if (subscribeInfo.getExpires() == 0) {
// redisCatchStorage.delSubscribe(key);
subscribeHolder.removeCatalogSubscribe(platformId);
}
try {
ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformId);
Response response = responseXmlAck(evt, resultXml.toString(), parentPlatform);
ToHeader toHeader = (ToHeader)response.getHeader(ToHeader.NAME);
subscribeInfo.setToTag(toHeader.getTag());
// redisCatchStorage.updateSubscribe(key, subscribeInfo);
subscribeHolder.putCatalogSubscribe(platformId, subscribeInfo);
responseXmlAck(evt, resultXml.toString(), parentPlatform);
} catch (SipException e) {
e.printStackTrace();
} catch (InvalidArgumentException e) {

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

@ -67,9 +67,6 @@ public class MessageRequestProcessor extends SIPRequestProcessorParent implement
// 查询设备是否存在
CSeqHeader cseqHeader = (CSeqHeader) evt.getRequest().getHeader(CSeqHeader.NAME);
String method = cseqHeader.getMethod();
if (method.equals("MESSAGE")) {
System.out.println();
}
Device device = redisCatchStorage.getDevice(deviceId);
// 查询上级平台是否存在
ParentPlatform parentPlatform = storage.queryParentPlatByServerGBId(deviceId);

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

@ -18,6 +18,7 @@ import org.springframework.stereotype.Component;
import javax.sip.InvalidArgumentException;
import javax.sip.RequestEvent;
import javax.sip.SipException;
import javax.sip.header.CallIdHeader;
import javax.sip.message.Response;
import java.text.ParseException;
@ -56,14 +57,15 @@ public class MediaStatusNotifyMessageHandler extends SIPRequestProcessorParent i
} catch (ParseException e) {
e.printStackTrace();
}
CallIdHeader callIdHeader = (CallIdHeader)evt.getRequest().getHeader(CallIdHeader.NAME);
String NotifyType =getText(rootElement, "NotifyType");
if (NotifyType.equals("121")){
logger.info("媒体播放完毕，通知关流");
StreamInfo streamInfo = redisCatchStorage.queryPlaybackByDevice(device.getDeviceId(), "*");
if (streamInfo != null) {
redisCatchStorage.stopPlayback(streamInfo);
cmder.streamByeCmd(streamInfo.getDeviceID(), streamInfo.getChannelId(), streamInfo.getStream());
}
String channelId =getText(rootElement, "DeviceID");
redisCatchStorage.stopPlayback(device.getDeviceId(), channelId, null, callIdHeader.getCallId());
cmder.streamByeCmd(device.getDeviceId(), channelId, null, callIdHeader.getCallId());
// TODO 如果级联播放，需要给上级发送此通知
}
}

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

@ -88,7 +88,7 @@ public class RecordInfoQueryMessageHandler extends SIPRequestProcessorParent imp
Element secrecyElement = rootElement.element("Secrecy");
int secrecy = 0;
if (secrecyElement != null) {
secrecy = Integer.parseInt(secrecyElement.getText());
secrecy = Integer.parseInt(secrecyElement.getText().trim());
}
String type = "all";
Element typeElement = rootElement.element("Type");

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

@ -2,6 +2,7 @@ package com.genersoft.iot.vmp.gb28181.transmit.event.response.impl;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch;
import com.genersoft.iot.vmp.gb28181.bean.SubscribeHolder;
import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.event.response.SIPResponseProcessorAbstract;
@ -40,6 +41,9 @@ public class RegisterResponseProcessor extends SIPResponseProcessorAbstract {
@Autowired
private SIPProcessorObserver sipProcessorObserver;
@Autowired
private SubscribeHolder subscribeHolder;
@Override
public void afterPropertiesSet() throws Exception {
// 添加消息处理的订阅
@ -83,19 +87,19 @@ public class RegisterResponseProcessor extends SIPResponseProcessorAbstract {
// 注册/注销成功
logger.info(String.format("%s %s成功", platformGBId, action));
redisCatchStorage.delPlatformRegisterInfo(callId);
parentPlatform.setStatus("注册".equals(action));
redisCatchStorage.delPlatformCatchInfo(platformGBId);
// 取回Expires设置，避免注销过程中被置为0
if (!parentPlatformCatch.getParentPlatform().getExpires().equals("0")) {
ParentPlatform parentPlatformTmp = storager.queryParentPlatByServerGBId(platformGBId);
String expires = parentPlatformTmp.getExpires();
parentPlatform.setExpires(expires);
parentPlatform.setId(parentPlatformTmp.getId());
redisCatchStorage.updatePlatformRegister(parentPlatform);
redisCatchStorage.updatePlatformKeepalive(parentPlatform);
parentPlatformCatch.setParentPlatform(parentPlatform);
redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
}
ParentPlatform parentPlatformTmp = storager.queryParentPlatByServerGBId(platformGBId);
parentPlatformTmp.setStatus("注册".equals(action));
redisCatchStorage.updatePlatformRegister(parentPlatformTmp);
redisCatchStorage.updatePlatformKeepalive(parentPlatformTmp);
parentPlatformCatch.setParentPlatform(parentPlatformTmp);
redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
storager.updateParentPlatformStatus(platformGBId, "注册".equals(action));
if ("注销".equals(action)) {
subscribeHolder.removeCatalogSubscribe(platformGBId);
subscribeHolder.removeMobilePositionSubscribe(platformGBId);
}
}
}

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

@ -9,9 +9,12 @@ import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.MediaConfig;
import com.genersoft.iot.vmp.conf.UserSetup;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import com.genersoft.iot.vmp.gb28181.bean.SsrcTransaction;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.media.zlm.dto.*;
import com.genersoft.iot.vmp.service.*;
import com.genersoft.iot.vmp.service.bean.SSRCInfo;
@ -81,7 +84,7 @@ public class ZLMHttpHookListener {
private UserSetup userSetup;
@Autowired
private MediaConfig mediaConfig;
private VideoStreamSessionManager sessionManager;
@Autowired
private ZLMRESTfulUtils zlmresTfulUtils;
@ -207,15 +210,15 @@ public class ZLMHttpHookListener {
}else {
ret.put("enableMP4", userSetup.isRecordPushLive());
}
StreamInfo streamInfo = redisCatchStorage.queryPlaybackByStreamId(stream);
// 录像回放时不进行录像下载
if (streamInfo != null) {
ret.put("enableMP4", false);
}else {
ret.put("enableMP4", userSetup.isRecordPushLive());
List<SsrcTransaction> ssrcTransactionForAll = sessionManager.getSsrcTransactionForAll(null, null, null, stream);
if (ssrcTransactionForAll != null && ssrcTransactionForAll.size() == 1) {
String deviceId = ssrcTransactionForAll.get(0).getDeviceId();
String channelId = ssrcTransactionForAll.get(0).getChannelId();
DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
if (deviceChannel != null) {
ret.put("enable_audio", deviceChannel.isHasAudio());
}
}
return new ResponseEntity<String>(ret.toString(), HttpStatus.OK);
}
@ -350,8 +353,12 @@ public class ZLMHttpHookListener {
redisCatchStorage.stopPlay(streamInfo);
storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
}else{
streamInfo = redisCatchStorage.queryPlaybackByStreamId(streamId);
redisCatchStorage.stopPlayback(streamInfo);
streamInfo = redisCatchStorage.queryPlayback(null, null, streamId, null);
if (streamInfo != null) {
redisCatchStorage.stopPlayback(streamInfo.getDeviceID(), streamInfo.getChannelId(),
streamInfo.getStream(), null);
}
}
}else {
if (!"rtp".equals(app)){
@ -443,18 +450,19 @@ public class ZLMHttpHookListener {
ret.put("close", false);
} else {
cmder.streamByeCmd(streamInfoForPlayCatch.getDeviceID(), streamInfoForPlayCatch.getChannelId(),
streamInfoForPlayCatch.getStream());
streamInfoForPlayCatch.getStream(), null);
redisCatchStorage.stopPlay(streamInfoForPlayCatch);
storager.stopPlay(streamInfoForPlayCatch.getDeviceID(), streamInfoForPlayCatch.getChannelId());
}
}else{
StreamInfo streamInfoForPlayBackCatch = redisCatchStorage.queryPlaybackByStreamId(streamId);
StreamInfo streamInfoForPlayBackCatch = redisCatchStorage.queryPlayback(null, null, streamId, null);
if (streamInfoForPlayBackCatch != null) {
cmder.streamByeCmd(streamInfoForPlayBackCatch.getDeviceID(),
streamInfoForPlayBackCatch.getChannelId(), streamInfoForPlayBackCatch.getStream());
redisCatchStorage.stopPlayback(streamInfoForPlayBackCatch);
streamInfoForPlayBackCatch.getChannelId(), streamInfoForPlayBackCatch.getStream(), null);
redisCatchStorage.stopPlayback(streamInfoForPlayBackCatch.getDeviceID(),
streamInfoForPlayBackCatch.getChannelId(), streamInfoForPlayBackCatch.getStream(), null);
}else {
StreamInfo streamInfoForDownload = redisCatchStorage.queryDownloadByStreamId(streamId);
StreamInfo streamInfoForDownload = redisCatchStorage.queryDownload(null, null, streamId, null);
// 进行录像下载时无人观看不断流
if (streamInfoForDownload != null) {
ret.put("close", false);
@ -462,7 +470,7 @@ public class ZLMHttpHookListener {
}
}
MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
if (mediaServerItem != null && "-1".equals(mediaServerItem.getStreamNoneReaderDelayMS())) {
if (mediaServerItem != null && mediaServerItem.getStreamNoneReaderDelayMS() == -1) {
ret.put("close", false);
}
return new ResponseEntity<String>(ret.toString(),HttpStatus.OK);

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

@ -45,12 +45,8 @@ public class ZLMRTPServerFactory {
Map<String, Object> param = new HashMap<>();
int result = -1;
/**
* 不设置推流端口端则使用随机端口
*/
if (StringUtils.isEmpty(mediaServerItem.getSendRtpPortRange())){
param.put("port", 0);
}else {
// 不设置推流端口端则使用随机端口
if (!StringUtils.isEmpty(mediaServerItem.getSendRtpPortRange())){
int newPort = getPortFromportRange(mediaServerItem);
param.put("port", newPort);
}

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

@ -2,10 +2,14 @@ package com.genersoft.iot.vmp.service;
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.InviteStreamCallback;
import com.genersoft.iot.vmp.gb28181.bean.InviteStreamInfo;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.bean.InviteTimeOutCallback;
import com.genersoft.iot.vmp.service.bean.PlayBackCallback;
import com.genersoft.iot.vmp.service.bean.SSRCInfo;
import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.PlayResult;
import org.springframework.http.ResponseEntity;
import org.springframework.web.context.request.async.DeferredResult;
@ -17,13 +21,17 @@ public interface IPlayService {
void onPublishHandlerForPlay(MediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId, String uuid);
void play(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId,
ZLMHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent,
InviteTimeOutCallback timeoutCallback, String uuid);
PlayResult play(MediaServerItem mediaServerItem, String deviceId, String channelId, ZLMHttpHookSubscribe.Event event, SipSubscribe.Event errorEvent, Runnable timeoutCallback);
MediaServerItem getNewMediaServerItem(Device device);
void onPublishHandlerForDownload(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, String toString);
void onPublishHandlerForDownload(InviteStreamInfo inviteStreamInfo, String deviceId, String channelId, String toString);
DeferredResult<ResponseEntity<String>> playBack(String deviceId, String channelId, String startTime, String endTime, PlayBackCallback errorCallBack);
DeferredResult<ResponseEntity<String>> playBack(String deviceId, String channelId, String startTime, String endTime, InviteStreamCallback infoCallBack, PlayBackCallback hookCallBack);
DeferredResult<ResponseEntity<String>> playBack(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo,String deviceId, String channelId, String startTime, String endTime, InviteStreamCallback infoCallBack, PlayBackCallback hookCallBack);
void zlmServerOffline(String mediaServerId);
}

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

@ -0,0 +1,6 @@
package com.genersoft.iot.vmp.service.bean;
public interface InviteTimeOutCallback {
void run(int code, String msg); // code: 0 sip超时, 1 收流超时
}

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

@ -7,9 +7,9 @@ import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import javax.sip.RequestEvent;
public class PlayBackResult<T> {
private int code;
private T data;
private MediaServerItem mediaServerItem;
private int code;
private T data;
private MediaServerItem mediaServerItem;
private JSONObject response;
private SipSubscribe.EventResult event;

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

@ -512,7 +512,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
param.put("hook.on_stream_not_found",String.format("%s/on_stream_not_found", hookPrex));
param.put("hook.on_server_keepalive",String.format("%s/on_server_keepalive", hookPrex));
param.put("hook.timeoutSec","20");
param.put("general.streamNoneReaderDelayMS","-1".equals(mediaServerItem.getStreamNoneReaderDelayMS())?"3600000":mediaServerItem.getStreamNoneReaderDelayMS() );
param.put("general.streamNoneReaderDelayMS",mediaServerItem.getStreamNoneReaderDelayMS()==-1?"3600000":mediaServerItem.getStreamNoneReaderDelayMS() );
// 推流断开后可以在超时时间内重新连接上继续推流，这样播放器会接着播放。
// 置0关闭此特性(推流断开会导致立即断开播放器)
// 此参数不应大于播放器超时时间

311
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.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.bean.InviteTimeOutCallback;
import com.genersoft.iot.vmp.service.bean.PlayBackCallback;
import com.genersoft.iot.vmp.service.bean.PlayBackResult;
import com.genersoft.iot.vmp.service.bean.SSRCInfo;
@ -27,6 +28,7 @@ import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.PlayResult;
import com.genersoft.iot.vmp.service.IMediaService;
import com.genersoft.iot.vmp.service.IPlayService;
import gov.nist.javax.sip.stack.SIPDialog;
import jdk.nashorn.internal.ir.RuntimeNode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@ -36,6 +38,9 @@ import org.springframework.stereotype.Service;
import org.springframework.util.ResourceUtils;
import org.springframework.web.context.request.async.DeferredResult;
import javax.sip.header.CallIdHeader;
import javax.sip.header.Header;
import javax.sip.message.Request;
import java.io.FileNotFoundException;
import java.util.*;
@ -79,6 +84,8 @@ public class PlayServiceImpl implements IPlayService {
private UserSetup userSetup;
@Override
public PlayResult play(MediaServerItem mediaServerItem, String deviceId, String channelId,
ZLMHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent,
@ -141,67 +148,7 @@ public class PlayServiceImpl implements IPlayService {
e.printStackTrace();
}
});
if (streamInfo == null) {
String streamId = null;
if (mediaServerItem.isRtpEnable()) {
streamId = String.format("%s_%s", device.getDeviceId(), channelId);
}
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId);
// 超时处理
Timer timer = new Timer();
timer.schedule(new TimerTask() {
@Override
public void run() {
logger.warn(String.format("设备点播超时，deviceId：%s ，channelId：%s", deviceId, channelId));
if (timeoutCallback != null) {
timeoutCallback.run();
}
WVPResult wvpResult = new WVPResult();
wvpResult.setCode(-1);
SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, ssrcInfo.getStream());
if (dialog != null) {
wvpResult.setMsg("收流超时，请稍候重试");
// 点播超时回复BYE 同时释放ssrc以及此次点播的资源
cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream());
}else {
wvpResult.setMsg("点播超时，请稍候重试");
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
mediaServerService.closeRTPServer(deviceId, channelId, ssrcInfo.getStream());
streamSession.remove(deviceId, channelId, ssrcInfo.getStream());
}
msg.setData(wvpResult);
// 回复之前所有的点播请求
resultHolder.invokeAllResult(msg);
}
}, userSetup.getPlayTimeout());
// 发送点播消息
cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channelId, (MediaServerItem mediaServerItemInUse, JSONObject response) -> {
logger.info("收到订阅消息： " + response.toJSONString());
timer.cancel();
onPublishHandlerForPlay(mediaServerItemInUse, response, deviceId, channelId, uuid);
if (hookEvent != null) {
hookEvent.response(mediaServerItem, response);
}
}, (event) -> {
timer.cancel();
WVPResult wvpResult = new WVPResult();
wvpResult.setCode(-1);
// 点播返回sip错误
mediaServerService.closeRTPServer(playResult.getDevice().getDeviceId(), channelId, ssrcInfo.getStream());
// 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
streamSession.remove(deviceId, channelId, ssrcInfo.getStream());
wvpResult.setMsg(String.format("点播失败， 错误码： %s, %s", event.statusCode, event.msg));
msg.setData(wvpResult);
resultHolder.invokeAllResult(msg);
if (errorEvent != null) {
errorEvent.response(event);
}
});
} else {
if (streamInfo != null) {
String streamId = streamInfo.getStream();
if (streamId == null) {
WVPResult wvpResult = new WVPResult();
@ -227,67 +174,109 @@ public class PlayServiceImpl implements IPlayService {
if (hookEvent != null) {
hookEvent.response(mediaServerItem, JSONObject.parseObject(JSON.toJSONString(streamInfo)));
}
} else {
// TODO 点播前是否重置状态
}else {
redisCatchStorage.stopPlay(streamInfo);
storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
String streamId2 = null;
if (mediaServerItem.isRtpEnable()) {
streamId2 = String.format("%s_%s", device.getDeviceId(), channelId);
}
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId2);
// 超时处理
Timer timer = new Timer();
timer.schedule(new TimerTask() {
@Override
public void run() {
logger.warn(String.format("设备点播超时，deviceId：%s ，channelId：%s", deviceId, channelId));
if (timeoutCallback != null) {
timeoutCallback.run();
}
WVPResult wvpResult = new WVPResult();
wvpResult.setCode(-1);
SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, ssrcInfo.getStream());
if (dialog != null) {
wvpResult.setMsg("收流超时，请稍候重试");
// 点播超时回复BYE 同时释放ssrc以及此次点播的资源
cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream());
}else {
wvpResult.setMsg("点播超时，请稍候重试");
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
mediaServerService.closeRTPServer(deviceId, channelId, ssrcInfo.getStream());
streamSession.remove(deviceId, channelId, ssrcInfo.getStream());
}
msg.setData(wvpResult);
// 回复之前所有的点播请求
resultHolder.invokeAllResult(msg);
}
}, userSetup.getPlayTimeout());
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().getDeviceId(), channelId, ssrcInfo.getStream());
// 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
streamSession.remove(deviceId, channelId, ssrcInfo.getStream());
WVPResult wvpResult = new WVPResult();
wvpResult.setCode(-1);
wvpResult.setMsg(String.format("点播失败， 错误码： %s, %s", event.statusCode, event.msg));
msg.setData(wvpResult);
resultHolder.invokeAllResult(msg);
});
streamInfo = null;
}
}
if (streamInfo == null) {
String streamId = null;
if (mediaServerItem.isRtpEnable()) {
streamId = String.format("%s_%s", device.getDeviceId(), channelId);
}
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId);
play(mediaServerItem, ssrcInfo, device, channelId, (mediaServerItemInUse, response)->{
if (hookEvent != null) {
hookEvent.response(mediaServerItem, response);
}
}, event -> {
// sip error错误
WVPResult wvpResult = new WVPResult();
wvpResult.setCode(-1);
wvpResult.setMsg(String.format("点播失败， 错误码： %s, %s", event.statusCode, event.msg));
msg.setData(wvpResult);
resultHolder.invokeAllResult(msg);
if (errorEvent != null) {
errorEvent.response(event);
}
}, (code, msgStr)->{
// invite点播超时
WVPResult wvpResult = new WVPResult();
wvpResult.setCode(-1);
if (code == 0) {
wvpResult.setMsg("点播超时，请稍候重试");
}else if (code == 1) {
wvpResult.setMsg("收流超时，请稍候重试");
}
msg.setData(wvpResult);
// 回复之前所有的点播请求
resultHolder.invokeAllResult(msg);
}, uuid);
}
return playResult;
}
@Override
public void play(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId,
ZLMHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent,
InviteTimeOutCallback timeoutCallback, String uuid) {
String streamId = null;
if (mediaServerItem.isRtpEnable()) {
streamId = String.format("%s_%s", device.getDeviceId(), channelId);
}
if (ssrcInfo == null) {
ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId);
}
return playResult;
// 超时处理
Timer timer = new Timer();
SSRCInfo finalSsrcInfo = ssrcInfo;
timer.schedule(new TimerTask() {
@Override
public void run() {
logger.warn(String.format("设备点播超时，deviceId：%s ，channelId：%s", device.getDeviceId(), channelId));
SIPDialog dialog = streamSession.getDialogByStream(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
if (dialog != null) {
timeoutCallback.run(1, "收流超时");
// 点播超时回复BYE 同时释放ssrc以及此次点播的资源
cmder.streamByeCmd(device.getDeviceId(), channelId, finalSsrcInfo.getStream(), null);
}else {
timeoutCallback.run(0, "点播超时");
mediaServerService.releaseSsrc(mediaServerItem.getId(), finalSsrcInfo.getSsrc());
mediaServerService.closeRTPServer(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
streamSession.remove(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
}
}
}, userSetup.getPlayTimeout());
cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channelId, (MediaServerItem mediaServerItemInuse, JSONObject response) -> {
logger.info("收到订阅消息： " + response.toJSONString());
timer.cancel();
// hook响应
onPublishHandlerForPlay(mediaServerItemInuse, response, device.getDeviceId(), channelId, uuid);
hookEvent.response(mediaServerItemInuse, response);
}, (event) -> {
timer.cancel();
mediaServerService.closeRTPServer(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
// 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), finalSsrcInfo.getSsrc());
streamSession.remove(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
errorEvent.response(event);
});
}
@Override
public void onPublishHandlerForPlay(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, String uuid) {
RequestMessage msg = new RequestMessage();
msg.setId(uuid);
if (uuid != null) {
msg.setId(uuid);
}
msg.setKey(DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId);
StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId);
if (streamInfo != null) {
@ -297,7 +286,6 @@ public class PlayServiceImpl implements IPlayService {
storager.startPlay(deviceId, channelId, streamInfo.getStream());
}
redisCatchStorage.startPlay(streamInfo);
msg.setData(JSON.toJSONString(streamInfo));
WVPResult wvpResult = new WVPResult();
wvpResult.setCode(0);
@ -329,9 +317,24 @@ public class PlayServiceImpl implements IPlayService {
return mediaServerItem;
}
@Override
public DeferredResult<ResponseEntity<String>> playBack(String deviceId, String channelId, String startTime,
String endTime,InviteStreamCallback inviteStreamCallback,
PlayBackCallback callback) {
Device device = storager.queryVideoDevice(deviceId);
if (device == null) return null;
MediaServerItem newMediaServerItem = getNewMediaServerItem(device);
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, true);
return playBack(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, inviteStreamCallback, callback);
}
@Override
public DeferredResult<ResponseEntity<String>> playBack(String deviceId, String channelId, String startTime, String endTime, PlayBackCallback callback) {
public DeferredResult<ResponseEntity<String>> playBack(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo,
String deviceId, String channelId, String startTime,
String endTime, InviteStreamCallback infoCallBack,
PlayBackCallback playBackCallback) {
if (mediaServerItem == null || ssrcInfo == null) return null;
String uuid = UUID.randomUUID().toString();
String key = DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId;
DeferredResult<ResponseEntity<String>> result = new DeferredResult<>(30000L);
@ -341,8 +344,6 @@ public class PlayServiceImpl implements IPlayService {
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.setId(uuid);
@ -356,63 +357,62 @@ public class PlayServiceImpl implements IPlayService {
logger.warn(String.format("设备回放超时，deviceId：%s ，channelId：%s", deviceId, channelId));
playBackResult.setCode(-1);
playBackResult.setData(msg);
callback.call(playBackResult);
playBackCallback.call(playBackResult);
SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, ssrcInfo.getStream());
// 点播超时回复BYE 同时释放ssrc以及此次点播的资源
if (dialog != null) {
// 点播超时回复BYE 同时释放ssrc以及此次点播的资源
cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream());
cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream(), null);
}else {
mediaServerService.releaseSsrc(newMediaServerItem.getId(), ssrcInfo.getSsrc());
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
mediaServerService.closeRTPServer(deviceId, channelId, ssrcInfo.getStream());
streamSession.remove(deviceId, channelId, ssrcInfo.getStream());
}
cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream());
cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream(), null);
// 回复之前所有的点播请求
callback.call(playBackResult);
playBackCallback.call(playBackResult);
}
}, userSetup.getPlayTimeout());
cmder.playbackStreamCmd(newMediaServerItem, ssrcInfo, device, channelId, startTime, endTime, (MediaServerItem mediaServerItem, JSONObject response) -> {
logger.info("收到订阅消息： " + response.toJSONString());
timer.cancel();
StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId);
if (streamInfo == null) {
logger.warn("设备回放API调用失败！");
msg.setData("设备回放API调用失败！");
playBackResult.setCode(-1);
playBackResult.setData(msg);
callback.call(playBackResult);
return;
}
redisCatchStorage.startPlayback(streamInfo);
msg.setData(JSON.toJSONString(streamInfo));
playBackResult.setCode(0);
playBackResult.setData(msg);
playBackResult.setMediaServerItem(mediaServerItem);
playBackResult.setResponse(response);
callback.call(playBackResult);
}, event -> {
timer.cancel();
msg.setData(String.format("回放失败， 错误码： %s, %s", event.statusCode, event.msg));
playBackResult.setCode(-1);
playBackResult.setData(msg);
playBackResult.setEvent(event);
callback.call(playBackResult);
streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
});
cmder.playbackStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, infoCallBack,
(InviteStreamInfo inviteStreamInfo) -> {
logger.info("收到订阅消息： " + inviteStreamInfo.getResponse().toJSONString());
timer.cancel();
StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId);
if (streamInfo == null) {
logger.warn("设备回放API调用失败！");
msg.setData("设备回放API调用失败！");
playBackResult.setCode(-1);
playBackResult.setData(msg);
playBackCallback.call(playBackResult);
return;
}
redisCatchStorage.startPlayback(streamInfo, inviteStreamInfo.getCallId());
msg.setData(JSON.toJSONString(streamInfo));
playBackResult.setCode(0);
playBackResult.setData(msg);
playBackResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem());
playBackResult.setResponse(inviteStreamInfo.getResponse());
playBackCallback.call(playBackResult);
}, event -> {
timer.cancel();
msg.setData(String.format("回放失败， 错误码： %s, %s", event.statusCode, event.msg));
playBackResult.setCode(-1);
playBackResult.setData(msg);
playBackResult.setEvent(event);
playBackCallback.call(playBackResult);
streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
});
return result;
}
@Override
public void onPublishHandlerForDownload(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, String uuid) {
public void onPublishHandlerForDownload(InviteStreamInfo inviteStreamInfo, 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);
StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId);
if (streamInfo != null) {
redisCatchStorage.startDownload(streamInfo);
redisCatchStorage.startDownload(streamInfo, inviteStreamInfo.getCallId());
msg.setData(JSON.toJSONString(streamInfo));
resultHolder.invokeResult(msg);
} else {
@ -449,7 +449,8 @@ public class PlayServiceImpl implements IPlayService {
if (allSsrc.size() > 0) {
for (SsrcTransaction ssrcTransaction : allSsrc) {
if(ssrcTransaction.getMediaServerId().equals(mediaServerId)) {
cmder.streamByeCmd(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId(), ssrcTransaction.getStream());
cmder.streamByeCmd(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId(),
ssrcTransaction.getStream(), null);
}
}
}

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

@ -47,17 +47,15 @@ public interface IRedisCatchStorage {
StreamInfo queryPlayByStreamId(String steamId);
StreamInfo queryPlaybackByStreamId(String steamId);
StreamInfo queryPlayByDevice(String deviceId, String channelId);
Map<String, StreamInfo> queryPlayByDeviceId(String deviceId);
boolean startPlayback(StreamInfo stream);
boolean startPlayback(StreamInfo stream, String callId);
boolean stopPlayback(StreamInfo streamInfo);
boolean stopPlayback(String deviceId, String channelId, String stream, String callId);
StreamInfo queryPlaybackByDevice(String deviceId, String code);
StreamInfo queryPlayback(String deviceId, String channelID, String stream, String callId);
void updatePlatformCatchInfo(ParentPlatformCatch parentPlatformCatch);
@ -167,9 +165,9 @@ public interface IRedisCatchStorage {
* 开始下载录像时存入
* @param streamInfo
*/
boolean startDownload(StreamInfo streamInfo);
boolean startDownload(StreamInfo streamInfo, String callId);
StreamInfo queryDownloadByStreamId(String streamId);
StreamInfo queryDownload(String deviceId, String channelId, String stream, String callId);
/**
* 查找第三方系统留下的国标预设值
@ -204,18 +202,8 @@ public interface IRedisCatchStorage {
void resetAllSN();
void updateSubscribe(String key, SubscribeInfo subscribeInfo);
SubscribeInfo getSubscribe(String key);
void delSubscribe(String key);
MediaItem getStreamInfo(String app, String streamId, String mediaServerId);
List<SubscribeInfo> getAllSubscribe();
List<String> getAllSubscribePlatform();
void addCpuInfo(double cpuInfo);
void addMemInfo(double memInfo);

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

@ -231,7 +231,6 @@ public interface DeviceChannelMapper {
" name as title,\n" +
" channelId as \"value\",\n" +
" channelId as \"key\",\n" +
" channelId,\n" +
" longitude,\n" +
" latitude\n" +
" from device_channel\n" +
@ -248,4 +247,13 @@ public interface DeviceChannelMapper {
"<foreach collection='channels' item='item' open='(' separator=',' close=')' > #{item.channelId}</foreach>" +
" </script>"})
int cleanChannelsNotInList(String deviceId, List<DeviceChannel> channels);
@Update(" update device_channel" +
" set subCount = (select *" +
" from (select count(0)" +
" from device_channel" +
" where deviceId = #{deviceId} and parentId = #{channelId}) as temp)" +
" where deviceId = #{deviceId} " +
" and channelId = #{channelId}")
int updateChannelSubCount(String deviceId, String channelId);
}

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

@ -55,7 +55,7 @@ public interface PlatformChannelMapper {
int cleanChannelForGB(String platformId);
@Select("SELECT dc.* FROM platform_gb_channel pgc left join device_channel dc on dc.id = pgc.deviceChannelId WHERE dc.channelId='${channelId}' and pgc.platformId='${platformId}'")
DeviceChannel queryChannelInParentPlatform(String platformId, String channelId);
List<DeviceChannel> queryChannelInParentPlatform(String platformId, String channelId);
@Select(" select dc.channelId as id, dc.name as name, pgc.platformId as platformId, pgc.catalogId as parentId, 0 as childrenCount, 1 as type " +
" from device_channel dc left join platform_gb_channel pgc on dc.id = pgc.deviceChannelId " +
@ -67,7 +67,7 @@ public interface PlatformChannelMapper {
" left join device_channel dc on dc.id = pgc.deviceChannelId\n" +
" left join device d on dc.deviceId = d.deviceId\n" +
"where dc.channelId = #{channelId} and pgc.platformId=#{platformId}")
Device queryVideoDeviceByPlatformIdAndChannelId(String platformId, String channelId);
List<Device> queryVideoDeviceByPlatformIdAndChannelId(String platformId, String channelId);
@Delete("<script> "+
"DELETE FROM platform_gb_channel WHERE catalogId=#{id}" +

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

@ -133,13 +133,6 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
return (StreamInfo)redis.get(playLeys.get(0).toString());
}
@Override
public StreamInfo queryPlaybackByStreamId(String streamId) {
List<Object> playLeys = redis.scan(String.format("%S_%s_%s_*", VideoManagerConstants.PLAY_BLACK_PREFIX, userSetup.getServerId(), streamId));
if (playLeys == null || playLeys.size() == 0) return null;
return (StreamInfo)redis.get(playLeys.get(0).toString());
}
@Override
public StreamInfo queryPlayByDevice(String deviceId, String channelId) {
List<Object> playLeys = redis.scan(String.format("%S_%s_*_%s_%s", VideoManagerConstants.PLAYER_PREFIX,
@ -166,49 +159,67 @@ 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.getStream(), stream.getDeviceID(), stream.getChannelId()), stream);
public boolean startPlayback(StreamInfo stream, String callId) {
return redis.set(String.format("%S_%s_%s_%s_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX,
userSetup.getServerId(), stream.getDeviceID(), stream.getChannelId(), stream.getStream(), callId), stream);
}
@Override
public boolean startDownload(StreamInfo streamInfo) {
return redis.set(String.format("%S_%s_%s_%s_%s", VideoManagerConstants.DOWNLOAD_PREFIX, userSetup.getServerId(),
streamInfo.getStream(), streamInfo.getDeviceID(), streamInfo.getChannelId()), streamInfo);
public boolean startDownload(StreamInfo stream, String callId) {
return redis.set(String.format("%S_%s_%s_%s_%s_%s", VideoManagerConstants.DOWNLOAD_PREFIX,
userSetup.getServerId(), stream.getDeviceID(), stream.getChannelId(), stream.getStream(), callId), stream);
}
@Override
public boolean stopPlayback(StreamInfo streamInfo) {
if (streamInfo == null) return false;
DeviceChannel deviceChannel = deviceChannelMapper.queryChannel(streamInfo.getDeviceID(), streamInfo.getChannelId());
public boolean stopPlayback(String deviceId, String channelId, String stream, String callId) {
DeviceChannel deviceChannel = deviceChannelMapper.queryChannel(deviceId, channelId);
if (deviceChannel != null) {
deviceChannel.setStreamId(null);
deviceChannel.setDeviceId(streamInfo.getDeviceID());
deviceChannel.setDeviceId(deviceId);
deviceChannelMapper.update(deviceChannel);
}
return redis.del(String.format("%S_%s_%s_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX,
if (deviceId == null) deviceId = "*";
if (channelId == null) channelId = "*";
if (stream == null) stream = "*";
if (callId == null) callId = "*";
String key = String.format("%S_%s_%s_%s_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX,
userSetup.getServerId(),
streamInfo.getStream(),
streamInfo.getDeviceID(),
streamInfo.getChannelId()));
deviceId,
channelId,
stream,
callId
);
List<Object> scan = redis.scan(key);
if (scan.size() > 0) {
for (Object keyObj : scan) {
redis.del((String) keyObj);
}
}
return true;
}
@Override
public StreamInfo queryPlaybackByDevice(String deviceId, String code) {
// String format = String.format("%S_*_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX,
// deviceId,
// code);
List<Object> playLeys = redis.scan(String.format("%S_%s_*_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX,
public StreamInfo queryPlayback(String deviceId, String channelId, String stream, String callId) {
if (stream == null && callId == null) {
return null;
}
if (deviceId == null) deviceId = "*";
if (channelId == null) channelId = "*";
if (stream == null) stream = "*";
if (callId == null) callId = "*";
String key = String.format("%S_%s_%s_%s_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX,
userSetup.getServerId(),
deviceId,
code));
if (playLeys == null || playLeys.size() == 0) {
playLeys = redis.scan(String.format("%S_%s_*_*_%s", VideoManagerConstants.PLAY_BLACK_PREFIX,
userSetup.getServerId(),
deviceId));
channelId,
stream,
callId
);
List<Object> streamInfoScan = redis.scan(key);
if (streamInfoScan.size() > 0) {
return (StreamInfo) redis.get((String) streamInfoScan.get(0));
}else {
return null;
}
if (playLeys == null || playLeys.size() == 0) return null;
return (StreamInfo)redis.get(playLeys.get(0).toString());
}
@Override
@ -361,7 +372,7 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
}
}
List<Object> playBackers = redis.scan(String.format("%S_%s_*_%s_*", VideoManagerConstants.PLAY_BLACK_PREFIX,
List<Object> playBackers = redis.scan(String.format("%S_%s_%s_*_*_*", VideoManagerConstants.PLAY_BLACK_PREFIX,
userSetup.getServerId(),
deviceId));
if (playBackers.size() > 0) {
@ -426,10 +437,27 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
}
@Override
public StreamInfo queryDownloadByStreamId(String streamId) {
List<Object> playLeys = redis.scan(String.format("%S_%s_%s_*", VideoManagerConstants.DOWNLOAD_PREFIX, userSetup.getServerId(), streamId));
if (playLeys == null || playLeys.size() == 0) return null;
return (StreamInfo)redis.get(playLeys.get(0).toString());
public StreamInfo queryDownload(String deviceId, String channelId, String stream, String callId) {
if (stream == null && callId == null) {
return null;
}
if (deviceId == null) deviceId = "*";
if (channelId == null) channelId = "*";
if (stream == null) stream = "*";
if (callId == null) callId = "*";
String key = String.format("%S_%s_%s_%s_%s_%s", VideoManagerConstants.DOWNLOAD_PREFIX,
userSetup.getServerId(),
deviceId,
channelId,
stream,
callId
);
List<Object> streamInfoScan = redis.scan(key);
if (streamInfoScan.size() > 0) {
return (StreamInfo) redis.get((String) streamInfoScan.get(0));
}else {
return null;
}
}
@Override
@ -490,21 +518,6 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
return (GPSMsgInfo)redis.get(key);
}
@Override
public void updateSubscribe(String key, SubscribeInfo subscribeInfo) {
redis.set(key, subscribeInfo, subscribeInfo.getExpires());
}
@Override
public SubscribeInfo getSubscribe(String key) {
return (SubscribeInfo)redis.get(key);
}
@Override
public void delSubscribe(String key) {
redis.del(key);
}
@Override
public List<GPSMsgInfo> getAllGpsMsgInfo() {
String scanKey = VideoManagerConstants.WVP_STREAM_GPS_MSG_PREFIX + userSetup.getServerId() + "_*";
@ -535,32 +548,6 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
return result;
}
@Override
public List<SubscribeInfo> getAllSubscribe() {
String scanKey = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetup.getServerId() + "_Catalog_*";
List<SubscribeInfo> result = new ArrayList<>();
List<Object> keys = redis.scan(scanKey);
for (int i = 0; i < keys.size(); i++) {
String key = (String) keys.get(i);
SubscribeInfo subscribeInfo = (SubscribeInfo) redis.get(key);
result.add(subscribeInfo);
}
return result;
}
@Override
public List<String> getAllSubscribePlatform() {
String scanKey = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetup.getServerId() + "_Catalog_*";
List<String> result = new ArrayList<>();
List<Object> keys = redis.scan(scanKey);
for (int i = 0; i < keys.size(); i++) {
String key = (String) keys.get(i);
String platformId = key.substring(scanKey.length() - 1);
result.add(platformId);
}
return result;
}
@Override
public void addCpuInfo(double cpuInfo) {
String key = VideoManagerConstants.SYSTEM_INFO_CPU_PREFIX + userSetup.getServerId();

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

@ -42,7 +42,7 @@ import java.util.*;
@Component
public class VideoManagerStoragerImpl implements IVideoManagerStorager {
private Logger logger = LoggerFactory.getLogger(VideoManagerStoragerImpl.class);
private final Logger logger = LoggerFactory.getLogger(VideoManagerStoragerImpl.class);
@Autowired
EventPublisher eventPublisher;
@ -171,6 +171,7 @@ public class VideoManagerStoragerImpl implements IVideoManagerStorager {
}else {
deviceChannelMapper.update(channel);
}
deviceChannelMapper.updateChannelSubCount(deviceId,channel.getParentId());
}
@Override
@ -542,7 +543,7 @@ public class VideoManagerStoragerImpl implements IVideoManagerStorager {
if (parentPlatformCatch == null) { // serverGBId 已变化
ParentPlatform parentPlatById = platformMapper.getParentPlatById(parentPlatform.getId());
// 使用旧的查出缓存ID
parentPlatformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatById.getServerGBId());
parentPlatformCatch = new ParentPlatformCatch();
parentPlatformCatch.setId(parentPlatform.getServerGBId());
redisCatchStorage.delPlatformCatchInfo(parentPlatById.getServerGBId());
}
@ -662,8 +663,16 @@ public class VideoManagerStoragerImpl implements IVideoManagerStorager {
@Override
public DeviceChannel queryChannelInParentPlatform(String platformId, String channelId) {
DeviceChannel channel = platformChannelMapper.queryChannelInParentPlatform(platformId, channelId);
return channel;
List<DeviceChannel> channels = platformChannelMapper.queryChannelInParentPlatform(platformId, channelId);
if (channels.size() > 1) {
// 出现长度大于0的时候肯定是国标通道的ID重复了
logger.warn("国标ID存在重复：{}", channelId);
}
if (channels.size() == 0) {
return null;
}else {
return channels.get(0);
}
}
@Override
@ -680,8 +689,18 @@ public class VideoManagerStoragerImpl implements IVideoManagerStorager {
@Override
public Device queryVideoDeviceByPlatformIdAndChannelId(String platformId, String channelId) {
Device device = platformChannelMapper.queryVideoDeviceByPlatformIdAndChannelId(platformId, channelId);
return device;
List<Device> devices = platformChannelMapper.queryVideoDeviceByPlatformIdAndChannelId(platformId, channelId);
if (devices.size() > 1) {
// 出现长度大于0的时候肯定是国标通道的ID重复了
logger.warn("国标ID存在重复：{}", channelId);
}
if (devices.size() == 0) {
return null;
}else {
return devices.get(0);
}
}
/**
@ -1083,6 +1102,9 @@ public class VideoManagerStoragerImpl implements IVideoManagerStorager {
@Override
public List<ParentPlatform> queryPlatFormListForStreamWithGBId(String app, String stream, List<String> platforms) {
if (platforms == null || platforms.size() == 0) {
return new ArrayList<>();
}
return platformGbStreamMapper.queryPlatFormListForGBWithGBId(app, stream, platforms);
}

16
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/platform/PlatformController.java
View File

@ -7,6 +7,7 @@ import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetup;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.PlatformCatalog;
import com.genersoft.iot.vmp.gb28181.bean.SubscribeHolder;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
@ -49,6 +50,9 @@ public class PlatformController {
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Autowired
private SubscribeHolder subscribeHolder;
@Autowired
private ISIPCommanderForPlatform commanderForPlatform;
@ -110,10 +114,14 @@ public class PlatformController {
})
public PageInfo<ParentPlatform> platforms(@PathVariable int page, @PathVariable int count) {
// if (logger.isDebugEnabled()) {
// logger.debug("查询所有上级设备API调用");
// }
return storager.queryParentPlatformList(page, count);
PageInfo<ParentPlatform> parentPlatformPageInfo = storager.queryParentPlatformList(page, count);
if (parentPlatformPageInfo.getList().size() > 0) {
for (ParentPlatform platform : parentPlatformPageInfo.getList()) {
platform.setGpsSubscribe(subscribeHolder.getMobilePositionSubscribe(platform.getServerGBId()) != null);
platform.setCatalogSubscribe(subscribeHolder.getCatalogSubscribe(platform.getServerGBId()) != null);
}
}
return parentPlatformPageInfo;
}
/**

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

@ -120,7 +120,7 @@ public class PlayController {
storager.stopPlay(deviceId, channelId);
return result;
}
cmder.streamByeCmd(deviceId, channelId, streamInfo.getStream(), (event) -> {
cmder.streamByeCmd(deviceId, channelId, streamInfo.getStream(), null, (event) -> {
redisCatchStorage.stopPlay(streamInfo);
storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
RequestMessage msg = new RequestMessage();
@ -174,7 +174,7 @@ public class PlayController {
public ResponseEntity<String> playConvert(@PathVariable String streamId) {
StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(streamId);
if (streamInfo == null) {
streamInfo = redisCatchStorage.queryPlaybackByStreamId(streamId);
streamInfo = redisCatchStorage.queryPlayback(null, null, streamId, null);
}
if (streamInfo == null) {
logger.warn("视频转码API调用失败！, 视频流已经停止!");

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

@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.vmanager.gb28181.playback;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.gb28181.bean.InviteStreamInfo;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
@ -93,11 +94,6 @@ public class DownloadController {
}
resultHolder.put(key, uuid, result);
Device device = storager.queryVideoDevice(deviceId);
StreamInfo streamInfo = redisCatchStorage.queryPlaybackByDevice(deviceId, channelId);
if (streamInfo != null) {
// 停止之前的下载
cmder.streamByeCmd(deviceId, channelId, streamInfo.getStream());
}
MediaServerItem newMediaServerItem = playService.getNewMediaServerItem(device);
if (newMediaServerItem == null) {
@ -112,9 +108,9 @@ public class DownloadController {
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, true);
cmder.downloadStreamCmd(newMediaServerItem, ssrcInfo, device, channelId, startTime, endTime, downloadSpeed, (MediaServerItem mediaServerItem, JSONObject response) -> {
logger.info("收到订阅消息： " + response.toJSONString());
playService.onPublishHandlerForDownload(mediaServerItem, response, deviceId, channelId, uuid);
cmder.downloadStreamCmd(newMediaServerItem, ssrcInfo, device, channelId, startTime, endTime, downloadSpeed, (InviteStreamInfo inviteStreamInfo) -> {
logger.info("收到订阅消息： " + inviteStreamInfo.getResponse().toJSONString());
playService.onPublishHandlerForDownload(inviteStreamInfo, deviceId, channelId, uuid);
}, event -> {
RequestMessage msg = new RequestMessage();
msg.setId(uuid);
@ -135,7 +131,7 @@ public class DownloadController {
@GetMapping("/stop/{deviceId}/{channelId}/{stream}")
public ResponseEntity<String> playStop(@PathVariable String deviceId, @PathVariable String channelId, @PathVariable String stream) {
cmder.streamByeCmd(deviceId, channelId, stream);
cmder.streamByeCmd(deviceId, channelId, stream, null);
if (logger.isDebugEnabled()) {
logger.debug(String.format("设备历史媒体下载停止 API调用，deviceId/channelId：%s_%s", deviceId, channelId));

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

@ -77,7 +77,7 @@ public class PlaybackController {
logger.debug(String.format("设备回放 API调用，deviceId：%s ，channelId：%s", deviceId, channelId));
}
DeferredResult<ResponseEntity<String>> result = playService.playBack(deviceId, channelId, startTime, endTime, wvpResult->{
DeferredResult<ResponseEntity<String>> result = playService.playBack(deviceId, channelId, startTime, endTime, null, wvpResult->{
resultHolder.invokeResult(wvpResult.getData());
});
@ -96,7 +96,7 @@ public class PlaybackController {
@PathVariable String channelId,
@PathVariable String stream) {
cmder.streamByeCmd(deviceId, channelId, stream);
cmder.streamByeCmd(deviceId, channelId, stream, null);
if (logger.isDebugEnabled()) {
logger.debug(String.format("设备录像回放停止 API调用，deviceId/channelId：%s/%s", deviceId, channelId));
@ -124,7 +124,7 @@ public class PlaybackController {
public ResponseEntity<String> playPause(@PathVariable String streamId) {
logger.info("playPause: "+streamId);
JSONObject json = new JSONObject();
StreamInfo streamInfo = redisCatchStorage.queryPlaybackByStreamId(streamId);
StreamInfo streamInfo = redisCatchStorage.queryPlayback(null, null, streamId, null);
if (null == streamInfo) {
json.put("msg", "streamId不存在");
logger.warn("streamId不存在!");
@ -144,7 +144,7 @@ public class PlaybackController {
public ResponseEntity<String> playResume(@PathVariable String streamId) {
logger.info("playResume: "+streamId);
JSONObject json = new JSONObject();
StreamInfo streamInfo = redisCatchStorage.queryPlaybackByStreamId(streamId);
StreamInfo streamInfo = redisCatchStorage.queryPlayback(null, null, streamId, null);
if (null == streamInfo) {
json.put("msg", "streamId不存在");
logger.warn("streamId不存在!");
@ -165,7 +165,7 @@ public class PlaybackController {
public ResponseEntity<String> playSeek(@PathVariable String streamId, @PathVariable long seekTime) {
logger.info("playSeek: "+streamId+", "+seekTime);
JSONObject json = new JSONObject();
StreamInfo streamInfo = redisCatchStorage.queryPlaybackByStreamId(streamId);
StreamInfo streamInfo = redisCatchStorage.queryPlayback(null, null, streamId, null);
if (null == streamInfo) {
json.put("msg", "streamId不存在");
logger.warn("streamId不存在!");
@ -186,7 +186,7 @@ public class PlaybackController {
public ResponseEntity<String> playSpeed(@PathVariable String streamId, @PathVariable Double speed) {
logger.info("playSpeed: "+streamId+", "+speed);
JSONObject json = new JSONObject();
StreamInfo streamInfo = redisCatchStorage.queryPlaybackByStreamId(streamId);
StreamInfo streamInfo = redisCatchStorage.queryPlayback(null, null, streamId, null);
if (null == streamInfo) {
json.put("msg", "streamId不存在");
logger.warn("streamId不存在!");

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

@ -177,7 +177,7 @@ public class ApiStreamController {
result.put("error","未找到流信息");
return result;
}
cmder.streamByeCmd(serial, code, streamInfo.getStream());
cmder.streamByeCmd(serial, code, streamInfo.getStream(), null);
redisCatchStorage.stopPlay(streamInfo);
storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
return null;

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

@ -13,7 +13,7 @@
</div>
<!--设备列表-->
<el-table :data="platformList" border style="width: 100%" :height="winHeight">
<el-table-column prop="name" label="名称" width="240" align="center"></el-table-column>
<el-table-column prop="name" label="名称" align="center"></el-table-column>
<el-table-column prop="serverGBId" label="平台编号" width="180" align="center"></el-table-column>
<el-table-column label="是否启用" width="120" align="center">
<template slot-scope="scope">
@ -38,9 +38,19 @@
</div>
</template>
</el-table-column>
<el-table-column prop="deviceGBId" label="设备国标编号" width="240" align="center"></el-table-column>
<el-table-column prop="deviceGBId" label="设备国标编号" width="200" align="center"></el-table-column>
<el-table-column prop="transport" label="信令传输模式" width="120" align="center"></el-table-column>
<el-table-column prop="channelCount" label="通道数" align="center"></el-table-column>
<el-table-column prop="channelCount" label="通道数" width="120" align="center"></el-table-column>
<el-table-column label="订阅信息" width="240" align="center" fixed="right">
<template slot-scope="scope">
<i v-if="scope.row.alarmSubscribe" style="font-size: 1.5rem;" title="报警订阅" class="subscribe-on iconfont icon-gbaojings" ></i>
<i v-if="!scope.row.alarmSubscribe" style="font-size: 1.5rem;" title="报警订阅" class="subscribe-off iconfont icon-gbaojings" ></i>
<i v-if="scope.row.catalogSubscribe" title="目录订阅" class="subscribe-on iconfont icon-gjichus" ></i>
<i v-if="!scope.row.catalogSubscribe" title="目录订阅" class="subscribe-off iconfont icon-gjichus" ></i>
<i v-if="scope.row.gpsSubscribe" title="位置订阅" class="subscribe-on iconfont icon-gxunjians" ></i>
<i v-if="!scope.row.gpsSubscribe" title="位置订阅" class="subscribe-off iconfont icon-gxunjians" ></i>
</template>
</el-table-column>
<el-table-column label="操作" width="300" align="center" fixed="right">
<template slot-scope="scope">
@ -169,3 +179,13 @@ export default {
}
};
</script>
<style>
.subscribe-on{
color: #409EFF;
font-size: 1.3rem;
}
.subscribe-off{
color: #afafb3;
font-size: 1.3rem;
}
</style>

2
web_src/src/components/dialog/StreamProxyEdit.vue
View File

@ -193,6 +193,7 @@ export default {
this.mediaServer.getOnlineMediaServerList((data)=>{
this.mediaServerList = data.data;
this.proxyParam.mediaServerId = this.mediaServerList[0].id
this.mediaServerIdChange()
})
},
mediaServerIdChange:function (){
@ -206,6 +207,7 @@ export default {
}
}).then(function (res) {
that.ffmpegCmdList = res.data.data;
that.proxyParam.ffmpeg_cmd_key = Object.keys(res.data.data)[0];
}).catch(function (error) {
console.log(error);
});

4
web_src/src/components/dialog/chooseChannelForGb.vue
View File

@ -1,8 +1,8 @@
<template>
<div id="chooseChannelForGb" >
<div style="font-size: 17px; color: #606060; white-space: nowrap; line-height: 30px; font-family: monospace;">
<span v-if="catalogId == null">{{catalogName}}的直播流</span>
<span v-if="catalogId != null">{{catalogName}}({{catalogId}})的直播流</span>
<span v-if="catalogId == null">{{catalogName}}的国标通道</span>
<span v-if="catalogId != null">{{catalogName}}({{catalogId}})的国标通道</span>
</div>
<div style="background-color: #FFFFFF; position: relative; padding: 0.5rem; text-align: left;font-size: 14px;">
搜索: <el-input @input="search" style="margin-right: 1rem; width: auto;" size="mini" placeholder="关键字" prefix-icon="el-icon-search" v-model="searchSrt" clearable> </el-input>

4
web_src/src/components/dialog/chooseChannelForStream.vue
View File

@ -1,8 +1,8 @@
<template>
<div id="chooseChannelFoStream" >
<div style="font-size: 17px; color: #606060; white-space: nowrap; line-height: 30px; font-family: monospace;">
<span v-if="catalogId == null">{{catalogName}}的直播流</span>
<span v-if="catalogId != null">{{catalogName}}({{catalogId}})的直播流</span>
<span v-if="catalogId == null">{{catalogName}}的直播通道</span>
<span v-if="catalogId != null">{{catalogName}}({{catalogId}})的直播通道</span>
</div>
<div style="background-color: #FFFFFF; margin-bottom: 1rem; position: relative; padding: 0.5rem; text-align: left;font-size: 14px;">

22
web_src/static/css/iconfont.css
View File

@ -1,8 +1,8 @@
@font-face {
font-family: "iconfont"; /* Project id 1291092 */
src: url('iconfont.woff2?t=1644809302709') format('woff2'),
url('iconfont.woff?t=1644809302709') format('woff'),
url('iconfont.ttf?t=1644809302709') format('truetype');
src: url('iconfont.woff2?t=1647245982270') format('woff2'),
url('iconfont.woff?t=1647245982270') format('woff'),
url('iconfont.ttf?t=1647245982270') format('truetype');
}
.iconfont {
@ -13,6 +13,22 @@
-moz-osx-font-smoothing: grayscale;
}
.icon-xitongxinxi:before {
content: "\e7d6";
}
.icon-gbaojings:before {
content: "\e7d7";
}
.icon-gjichus:before {
content: "\e7d8";
}
.icon-gxunjians:before {
content: "\e7d9";
}
.icon-ziyuan:before {
content: "\e7d5";
}

BIN
web_src/static/css/iconfont.woff2
View File

Binary file not shown.