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

去除对redis key过期事件的使用；重构国标级联的注册保活

Browse Source
This commit is contained in:
648540858 2022-08-31 11:29:13 +08:00
parent 03c43f84ce
commit e4bd61860d
45 changed files with 513 additions and 693 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -4,6 +4,7 @@ import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.service.IPlatformService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import org.springframework.beans.factory.annotation.Autowired;
@ -15,6 +16,7 @@ import java.util.List;
/**
* 系统启动时控制上级平台重新注册
* @author lin
*/
@Component
@Order(value=3)
@ -27,7 +29,7 @@ public class SipPlatformRunner implements CommandLineRunner {
private IRedisCatchStorage redisCatchStorage;
@Autowired
private EventPublisher publisher;
private IPlatformService platformService;
@Autowired
private ISIPCommanderForPlatform sipCommanderForPlatform;
@ -35,33 +37,26 @@ public class SipPlatformRunner implements CommandLineRunner {
@Override
public void run(String... args) throws Exception {
// 设置所有平台离线
storager.outlineForAllParentPlatform();
// 清理所有平台注册缓存
redisCatchStorage.cleanPlatformRegisterInfos();
// 停止所有推流
// zlmrtpServerFactory.closeAllSendRtpStream();
// 获取所有启用的平台
List<ParentPlatform> parentPlatforms = storager.queryEnableParentPlatformList(true);
for (ParentPlatform parentPlatform : parentPlatforms) {
redisCatchStorage.updatePlatformRegister(parentPlatform);
redisCatchStorage.updatePlatformKeepalive(parentPlatform);
// 更新缓存
ParentPlatformCatch parentPlatformCatch = new ParentPlatformCatch();
parentPlatformCatch.setParentPlatform(parentPlatform);
parentPlatformCatch.setId(parentPlatform.getServerGBId());
redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
if (parentPlatform.isStatus()) {
// 设置所有平台离线
platformService.offline(parentPlatform);
// 取消订阅
sipCommanderForPlatform.unregister(parentPlatform, null, (eventResult)->{
platformService.login(parentPlatform);
});
}else {
platformService.login(parentPlatform);
}
// 取消订阅
sipCommanderForPlatform.unregister(parentPlatform, null, (eventResult)->{
// 发送平台未注册消息
publisher.platformNotRegisterEventPublish(parentPlatform.getServerGBId());
});
}
}
}

10
src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java
View File

@ -31,8 +31,6 @@ public class UserSetting {
private Boolean logInDatebase = Boolean.TRUE;
private Boolean redisConfig = Boolean.TRUE;
private String serverId = "000000";
private String thirdPartyGBIdReg = "[\\s\\S]*";
@ -123,14 +121,6 @@ public class UserSetting {
this.thirdPartyGBIdReg = thirdPartyGBIdReg;
}
public Boolean getRedisConfig() {
return redisConfig;
}
public void setRedisConfig(Boolean redisConfig) {
this.redisConfig = redisConfig;
}
public Boolean getRecordSip() {
return recordSip;
}

41
src/main/java/com/genersoft/iot/vmp/conf/redis/RedisKeyExpirationEventMessageListener.java
View File

@ -1,41 +0,0 @@
package com.genersoft.iot.vmp.conf.redis;
import com.genersoft.iot.vmp.conf.UserSetting;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.listener.KeyExpirationEventMessageListener;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import java.util.Properties;
public class RedisKeyExpirationEventMessageListener extends KeyExpirationEventMessageListener {
private UserSetting userSetting;
private RedisMessageListenerContainer listenerContainer;
private String keyspaceNotificationsConfigParameter = "EA";
public RedisKeyExpirationEventMessageListener(RedisMessageListenerContainer listenerContainer, UserSetting userSetting) {
super(listenerContainer);
this.listenerContainer = listenerContainer;
this.userSetting = userSetting;
}
@Override
public void init() {
if (!userSetting.getRedisConfig()) {
// 配置springboot默认Config为空，即不让应用去修改redis的默认配置，因为Redis服务出于安全会禁用CONFIG命令给远程用户使用
setKeyspaceNotificationsConfigParameter("");
}else {
RedisConnection connection = this.listenerContainer.getConnectionFactory().getConnection();
Properties config = connection.getConfig("notify-keyspace-events");
try {
if (!keyspaceNotificationsConfigParameter.equals(config.getProperty("notify-keyspace-events"))) {
connection.setConfig("notify-keyspace-events", keyspaceNotificationsConfigParameter);
}
} finally {
connection.close();
}
}
super.init();
}
}

15
src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java
View File

@ -56,7 +56,7 @@ public class SipLayer{
* gov/nist/javax/sip/SipStackImpl.class
*/
if (logger.isDebugEnabled()) {
properties.setProperty("gov.nist.javax.sip.LOG_MESSAGE_CONTENT", "true");
properties.setProperty("gov.nist.javax.sip.LOG_MESSAGE_CONTENT", "false");
}
// 接收所有notify请求，即使没有订阅
properties.setProperty("gov.nist.javax.sip.DELIVER_UNSOLICITED_NOTIFY", "true");
@ -68,14 +68,13 @@ public class SipLayer{
properties.setProperty("gov.nist.javax.sip.RELIABLE_CONNECTION_KEEP_ALIVE_TIMEOUT", "60");
/**
* sip_server_log.log 和 sip_debug_log.log public static final int TRACE_NONE =
* 0; public static final int TRACE_MESSAGES = 16; public static final int
* TRACE_EXCEPTION = 17; public static final int TRACE_DEBUG = 32;
* sip_server_log.log 和 sip_debug_log.log ERROR, INFO, WARNING, OFF, DEBUG, TRACE
*/
if (logger.isDebugEnabled()) {
properties.setProperty("gov.nist.javax.sip.TRACE_LEVEL", "DEBUG");
}
properties.setProperty("gov.nist.javax.sip.TRACE_LEVEL", "INFO");
properties.setProperty("gov.nist.javax.sip.TRACE_LEVEL", "ERROR");
// if (logger.isDebugEnabled()) {
// properties.setProperty("gov.nist.javax.sip.TRACE_LEVEL", "DEBUG");
// }
sipStack = (SipStackImpl) sipFactory.createSipStack(properties);
return sipStack;

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

@ -84,7 +84,7 @@ public class ParentPlatform {
* 注册周期 (秒)
*/
@Schema(description = "注册周期 (秒)")
private String expires;
private int expires;
/**
* 心跳周期(秒)
@ -286,11 +286,11 @@ public class ParentPlatform {
this.password = password;
}
public String getExpires() {
public int getExpires() {
return expires;
}
public void setExpires(String expires) {
public void setExpires(int expires) {
this.expires = expires;
}

4
src/main/java/com/genersoft/iot/vmp/gb28181/bean/ParentPlatformCatch.java
View File

@ -4,7 +4,9 @@ public class ParentPlatformCatch {
private String id;
// 心跳未回复次数
/**
* 心跳未回复次数
*/
private int keepAliveReply;
// 注册未回复次数

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

@ -14,6 +14,9 @@ import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
/**
* @author lin
*/
@Component
public class SubscribeHolder {

33
src/main/java/com/genersoft/iot/vmp/gb28181/event/EventPublisher.java
View File

@ -2,9 +2,6 @@ package com.genersoft.iot.vmp.gb28181.event;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.device.RequestTimeoutEvent;
import com.genersoft.iot.vmp.gb28181.event.platformKeepaliveExpire.PlatformKeepaliveExpireEvent;
import com.genersoft.iot.vmp.gb28181.event.platformNotRegister.PlatformCycleRegisterEvent;
import com.genersoft.iot.vmp.gb28181.event.platformNotRegister.PlatformNotRegisterEvent;
import com.genersoft.iot.vmp.gb28181.event.record.RecordEndEvent;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.media.zlm.event.ZLMOfflineEvent;
@ -31,36 +28,6 @@ public class EventPublisher {
@Autowired
private ApplicationEventPublisher applicationEventPublisher;
/**
* 平台心跳到期事件
* @param platformGbId
*/
public void platformKeepaliveExpireEventPublish(String platformGbId){
PlatformKeepaliveExpireEvent platformKeepaliveExpireEvent = new PlatformKeepaliveExpireEvent(this);
platformKeepaliveExpireEvent.setPlatformGbID(platformGbId);
applicationEventPublisher.publishEvent(platformKeepaliveExpireEvent);
}
/**
* 平台未注册事件
* @param platformGbId
*/
public void platformNotRegisterEventPublish(String platformGbId){
PlatformNotRegisterEvent platformNotRegisterEvent = new PlatformNotRegisterEvent(this);
platformNotRegisterEvent.setPlatformGbID(platformGbId);
applicationEventPublisher.publishEvent(platformNotRegisterEvent);
}
/**
* 平台周期注册事件
* @param paltformGbId
*/
public void platformRegisterCycleEventPublish(String paltformGbId) {
PlatformCycleRegisterEvent platformCycleRegisterEvent = new PlatformCycleRegisterEvent(this);
platformCycleRegisterEvent.setPlatformGbID(paltformGbId);
applicationEventPublisher.publishEvent(platformCycleRegisterEvent);
}
/**
* 设备报警事件

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

@ -59,9 +59,25 @@ public class SipSubscribe {
void response(EventResult eventResult);
}
/**
*
*/
public enum EventResultType{
// 超时
timeout,
// 回复
response,
// 事务已结束
transactionTerminated,
// 会话已结束
dialogTerminated,
// 设备未找到
deviceNotFoundEvent
}
public static class EventResult<EventObject>{
public int statusCode;
public String type;
public EventResultType type;
public String msg;
public String callId;
public Dialog dialog;
@ -76,7 +92,7 @@ public class SipSubscribe {
ResponseEvent responseEvent = (ResponseEvent)event;
Response response = responseEvent.getResponse();
this.dialog = responseEvent.getDialog();
this.type = "response";
this.type = EventResultType.response;
if (response != null) {
this.msg = response.getReasonPhrase();
this.statusCode = response.getStatusCode();
@ -85,28 +101,28 @@ public class SipSubscribe {
}else if (event instanceof TimeoutEvent) {
TimeoutEvent timeoutEvent = (TimeoutEvent)event;
this.type = "timeout";
this.type = EventResultType.timeout;
this.msg = "消息超时未回复";
this.statusCode = -1024;
this.dialog = timeoutEvent.getClientTransaction().getDialog();
this.callId = this.dialog != null?timeoutEvent.getClientTransaction().getDialog().getCallId().getCallId(): null;
}else if (event instanceof TransactionTerminatedEvent) {
TransactionTerminatedEvent transactionTerminatedEvent = (TransactionTerminatedEvent)event;
this.type = "transactionTerminated";
this.type = EventResultType.transactionTerminated;
this.msg = "事务已结束";
this.statusCode = -1024;
this.callId = transactionTerminatedEvent.getClientTransaction().getDialog().getCallId().getCallId();
this.dialog = transactionTerminatedEvent.getClientTransaction().getDialog();
}else if (event instanceof DialogTerminatedEvent) {
DialogTerminatedEvent dialogTerminatedEvent = (DialogTerminatedEvent)event;
this.type = "dialogTerminated";
this.type = EventResultType.dialogTerminated;
this.msg = "会话已结束";
this.statusCode = -1024;
this.callId = dialogTerminatedEvent.getDialog().getCallId().getCallId();
this.dialog = dialogTerminatedEvent.getDialog();
}else if (event instanceof DeviceNotFoundEvent) {
DeviceNotFoundEvent deviceNotFoundEvent = (DeviceNotFoundEvent)event;
this.type = "deviceNotFoundEvent";
this.type = EventResultType.deviceNotFoundEvent;
this.msg = "设备未找到";
this.statusCode = -1024;
this.dialog = deviceNotFoundEvent.getDialog();

81
src/main/java/com/genersoft/iot/vmp/gb28181/event/offline/KeepaliveTimeoutListenerForPlatform.java
View File

@ -1,81 +0,0 @@
package com.genersoft.iot.vmp.gb28181.event.offline;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.redis.RedisKeyExpirationEventMessageListener;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.stereotype.Component;
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
/**
* 设备心跳超时监听,借助redis过期特性，进行监听，监听到说明设备心跳超时，发送离线事件
* @author swwheihei
*/
@Component
public class KeepaliveTimeoutListenerForPlatform extends RedisKeyExpirationEventMessageListener {
private Logger logger = LoggerFactory.getLogger(KeepaliveTimeoutListenerForPlatform.class);
@Autowired
private EventPublisher publisher;
@Autowired
private UserSetting userSetting;
@Autowired
private SipSubscribe sipSubscribe;
@Autowired
private IVideoManagerStorage storager;
public KeepaliveTimeoutListenerForPlatform(RedisMessageListenerContainer listenerContainer, UserSetting userSetting) {
super(listenerContainer, userSetting);
}
/**
* 监听失效的key
* @param message
* @param pattern
*/
@Override
public void onMessage(Message message, byte[] pattern) {
// 获取失效的key
String expiredKey = message.toString();
// 平台心跳到期,需要重发, 判断是否已经多次未收到心跳回复, 多次未收到,则重新发起注册, 注册尝试多次未得到回复,则认为平台离线
String PLATFORM_KEEPLIVEKEY_PREFIX = VideoManagerConstants.PLATFORM_KEEPALIVE_PREFIX + userSetting.getServerId() + "_";
String PLATFORM_REGISTER_PREFIX = VideoManagerConstants.PLATFORM_REGISTER_PREFIX + userSetting.getServerId() + "_";
String REGISTER_INFO_PREFIX = VideoManagerConstants.PLATFORM_REGISTER_INFO_PREFIX + userSetting.getServerId() + "_";
if (expiredKey.startsWith(PLATFORM_KEEPLIVEKEY_PREFIX)) {
String platformGbId = expiredKey.substring(PLATFORM_KEEPLIVEKEY_PREFIX.length());
ParentPlatform platform = storager.queryParentPlatByServerGBId(platformGbId);
if (platform != null) {
publisher.platformKeepaliveExpireEventPublish(platformGbId);
}
}else if (expiredKey.startsWith(PLATFORM_REGISTER_PREFIX)) {
String platformGbId = expiredKey.substring(PLATFORM_REGISTER_PREFIX.length(),expiredKey.length());
ParentPlatform platform = storager.queryParentPlatByServerGBId(platformGbId);
if (platform != null) {
publisher.platformRegisterCycleEventPublish(platformGbId);
}
}else if (expiredKey.startsWith(REGISTER_INFO_PREFIX)) {
String callId = expiredKey.substring(REGISTER_INFO_PREFIX.length());
if (sipSubscribe.getErrorSubscribe(callId) != null) {
SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult();
eventResult.callId = callId;
eventResult.msg = "注册超时";
eventResult.type = "register timeout";
sipSubscribe.getErrorSubscribe(callId).response(eventResult);
}
}
}
}

28
src/main/java/com/genersoft/iot/vmp/gb28181/event/platformKeepaliveExpire/PlatformKeepaliveExpireEvent.java
View File

@ -1,28 +0,0 @@
package com.genersoft.iot.vmp.gb28181.event.platformKeepaliveExpire;
import org.springframework.context.ApplicationEvent;
/**
* 平台心跳超时事件
*/
public class PlatformKeepaliveExpireEvent extends ApplicationEvent {
/**
* Add default serial version ID
*/
private static final long serialVersionUID = 1L;
private String platformGbID;
public PlatformKeepaliveExpireEvent(Object source) {
super(source);
}
public String getPlatformGbID() {
return platformGbID;
}
public void setPlatformGbID(String platformGbID) {
this.platformGbID = platformGbID;
}
}

87
src/main/java/com/genersoft/iot/vmp/gb28181/event/platformKeepaliveExpire/PlatformKeepaliveExpireEventLister.java
View File

@ -1,87 +0,0 @@
package com.genersoft.iot.vmp.gb28181.event.platformKeepaliveExpire;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;
import javax.sip.message.Response;
/**
* @description: 平台心跳超时事件
* @author: panll
* @date: 2020年11月5日 10:00
*/
@Component
public class PlatformKeepaliveExpireEventLister implements ApplicationListener<PlatformKeepaliveExpireEvent> {
private final static Logger logger = LoggerFactory.getLogger(PlatformKeepaliveExpireEventLister.class);
@Autowired
private IVideoManagerStorage storager;
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Autowired
private ISIPCommanderForPlatform sipCommanderForPlatform;
@Autowired
private SipSubscribe sipSubscribe;
@Autowired
private EventPublisher publisher;
@Override
public void onApplicationEvent(@NotNull PlatformKeepaliveExpireEvent event) {
if (logger.isDebugEnabled()) {
logger.debug("平台心跳到期事件事件触发，平台国标ID：" + event.getPlatformGbID());
}
ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(event.getPlatformGbID());
ParentPlatformCatch parentPlatformCatch = redisCatchStorage.queryPlatformCatchInfo(event.getPlatformGbID());
if (parentPlatformCatch == null) {
return;
}
if (parentPlatform == null) {
logger.debug("平台心跳到期事件事件触发，但平台已经删除!!! 平台国标ID：" + event.getPlatformGbID());
return;
}
parentPlatformCatch.setParentPlatform(parentPlatform);
// 发送心跳
if (parentPlatformCatch.getKeepAliveReply() >= 3) {
// 有3次未收到心跳回复, 设置平台状态为离线, 开始重新注册
logger.warn("有3次未收到心跳回复,标记设置平台状态为离线, 并重新注册 平台国标ID：" + event.getPlatformGbID());
storager.updateParentPlatformStatus(event.getPlatformGbID(), false);
publisher.platformNotRegisterEventPublish(event.getPlatformGbID());
parentPlatformCatch.setKeepAliveReply(0);
redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
}else {
// 再次发送心跳
String callId = sipCommanderForPlatform.keepalive(parentPlatform);
parentPlatformCatch.setKeepAliveReply( parentPlatformCatch.getKeepAliveReply() + 1);
// 存储心跳信息, 并设置状态为未回复, 如果多次过期仍未收到回复,则认为上级平台已经离线
redisCatchStorage.updatePlatformKeepalive(parentPlatform);
redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
sipSubscribe.addOkSubscribe(callId, (SipSubscribe.EventResult eventResult) ->{
if (eventResult.statusCode == Response.OK) {
// 收到心跳响应信息,
parentPlatformCatch.setKeepAliveReply(0);
redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
}
} );
}
}
}

24
src/main/java/com/genersoft/iot/vmp/gb28181/event/platformNotRegister/PlatformCycleRegisterEvent.java
View File

@ -1,24 +0,0 @@
package com.genersoft.iot.vmp.gb28181.event.platformNotRegister;
import org.springframework.context.ApplicationEvent;
public class PlatformCycleRegisterEvent extends ApplicationEvent {
/**
* Add default serial version ID
*/
private static final long serialVersionUID = 1L;
private String platformGbID;
public String getPlatformGbID() {
return platformGbID;
}
public void setPlatformGbID(String platformGbID) {
this.platformGbID = platformGbID;
}
public PlatformCycleRegisterEvent(Object source) {
super(source);
}
}

46
src/main/java/com/genersoft/iot/vmp/gb28181/event/platformNotRegister/PlatformCycleRegisterEventLister.java
View File

@ -1,46 +0,0 @@
package com.genersoft.iot.vmp.gb28181.event.platformNotRegister;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;
import java.util.Timer;
import java.util.TimerTask;
@Component
public class PlatformCycleRegisterEventLister implements ApplicationListener<PlatformCycleRegisterEvent> {
private final static Logger logger = LoggerFactory.getLogger(PlatformCycleRegisterEventLister.class);
@Autowired
private IVideoManagerStorage storager;
@Autowired
private ISIPCommanderForPlatform sipCommanderFroPlatform;
@Autowired
private DynamicTask dynamicTask;
@Override
public void onApplicationEvent(PlatformCycleRegisterEvent event) {
logger.info("上级平台周期注册事件");
ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(event.getPlatformGbID());
if (parentPlatform == null) {
logger.info("[ 平台未注册事件 ] 平台已经删除!!! 平台国标ID：" + event.getPlatformGbID());
return;
}
String taskKey = "platform-cycle-register" + parentPlatform.getServerGBId();;
SipSubscribe.Event okEvent = (responseEvent)->{
dynamicTask.stop(taskKey);
};
dynamicTask.startCron(taskKey, ()->{
logger.info("[平台注册]再次向平台注册，平台国标ID：" + event.getPlatformGbID());
sipCommanderFroPlatform.register(parentPlatform, null, okEvent);
}, Integer.parseInt(parentPlatform.getExpires())* 1000);
}
}

25
src/main/java/com/genersoft/iot/vmp/gb28181/event/platformNotRegister/PlatformNotRegisterEvent.java
View File

@ -1,25 +0,0 @@
package com.genersoft.iot.vmp.gb28181.event.platformNotRegister;
import org.springframework.context.ApplicationEvent;
public class PlatformNotRegisterEvent extends ApplicationEvent {
/**
* Add default serial version ID
*/
private static final long serialVersionUID = 1L;
private String platformGbID;
public PlatformNotRegisterEvent(Object source) {
super(source);
}
public String getPlatformGbID() {
return platformGbID;
}
public void setPlatformGbID(String platformGbID) {
this.platformGbID = platformGbID;
}
}

90
src/main/java/com/genersoft/iot/vmp/gb28181/event/platformNotRegister/PlatformNotRegisterEventLister.java
View File

@ -1,90 +0,0 @@
package com.genersoft.iot.vmp.gb28181.event.platformNotRegister;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;
import java.util.*;
/**
* @description: 平台未注册事件,来源有二:
* 1、平台新添加
* 2、平台心跳超时
* @author: panll
* @date: 2020年11月24日 10:00
*/
@Component
public class PlatformNotRegisterEventLister implements ApplicationListener<PlatformNotRegisterEvent> {
private final static Logger logger = LoggerFactory.getLogger(PlatformNotRegisterEventLister.class);
@Autowired
private IVideoManagerStorage storager;
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Autowired
private IMediaServerService mediaServerService;
@Autowired
private SIPCommanderFroPlatform sipCommanderFroPlatform;
@Autowired
private ZLMRTPServerFactory zlmrtpServerFactory;
@Autowired
private SipConfig config;
@Autowired
private DynamicTask dynamicTask;
// @Autowired
// private RedisUtil redis;
@Override
public void onApplicationEvent(PlatformNotRegisterEvent event) {
logger.info("[ 平台未注册事件 ]平台国标ID：" + event.getPlatformGbID());
ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(event.getPlatformGbID());
if (parentPlatform == null) {
logger.info("[ 平台未注册事件 ] 平台已经删除!!! 平台国标ID：" + event.getPlatformGbID());
return;
}
// 查询是否有推流， 如果有则都停止
List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServer(event.getPlatformGbID());
if (sendRtpItems != null && sendRtpItems.size() > 0) {
logger.info("[ 平台未注册事件 ] 停止[ {} ]的所有推流", event.getPlatformGbID());
for (SendRtpItem sendRtpItem : sendRtpItems) {
redisCatchStorage.deleteSendRTPServer(event.getPlatformGbID(), sendRtpItem.getChannelId(), null, null);
MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
Map<String, Object> param = new HashMap<>();
param.put("vhost", "__defaultVhost__");
param.put("app", sendRtpItem.getApp());
param.put("stream", sendRtpItem.getStreamId());
zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param);
}
}
String taskKey = "platform-not-register-" + parentPlatform.getServerGBId();
SipSubscribe.Event okEvent = (responseEvent)->{
dynamicTask.stop(taskKey);
};
dynamicTask.startCron(taskKey, ()->{
logger.info("[平台注册]再次向平台注册，平台国标ID：" + event.getPlatformGbID());
sipCommanderFroPlatform.register(parentPlatform, null, okEvent);
}, config.getRegisterTimeInterval()* 1000);
}
}

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

@ -30,23 +30,10 @@ public class CatalogEventLister implements ApplicationListener<CatalogEvent> {
@Autowired
private IVideoManagerStorage storager;
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Autowired
private IMediaServerService mediaServerService;
@Autowired
private SIPCommanderFroPlatform sipCommanderFroPlatform;
@Autowired
private ZLMRTPServerFactory zlmrtpServerFactory;
@Autowired
private SipConfig config;
@Autowired
private UserSetting userSetting;
@Autowired
private IGbStreamService gbStreamService;

1
src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/MobilePositionSubscribeHandlerTask.java
View File

@ -60,7 +60,6 @@ public class MobilePositionSubscribeHandlerTask implements ISubscribeTask {
// TODO 暂时只处理视频流的回复,后续增加对国标设备的支持
List<DeviceChannel> gbStreams = storager.queryGbStreamListInPlatform(platform.getServerGBId());
if (gbStreams.size() == 0) {
logger.info("发送订阅时发现平台已经没有关联的直播流：{}", platform.getServerGBId());
return;
}
for (DeviceChannel deviceChannel : gbStreams) {

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

@ -15,7 +15,7 @@ public interface ISIPCommanderForPlatform {
* @return
*/
boolean register(ParentPlatform parentPlatform, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent);
boolean register(ParentPlatform parentPlatform, String callId, WWWAuthenticateHeader www, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent, boolean registerAgain);
boolean register(ParentPlatform parentPlatform, String callId, WWWAuthenticateHeader www, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent, boolean registerAgain, boolean isRegister);
/**
* 向上级平台注销
@ -30,7 +30,7 @@ public interface ISIPCommanderForPlatform {
* @param parentPlatform
* @return callId(作为接受回复的判定)
*/
String keepalive(ParentPlatform parentPlatform);
String keepalive(ParentPlatform parentPlatform,SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent);
/**

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

@ -3,6 +3,7 @@ package com.genersoft.iot.vmp.gb28181.transmit.cmd;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.SubscribeInfo;
import com.genersoft.iot.vmp.gb28181.utils.HeaderUtils;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import gov.nist.javax.sip.message.MessageFactoryImpl;
import org.springframework.beans.factory.annotation.Autowired;
@ -75,7 +76,7 @@ public class SIPRequestHeaderPlarformProvider {
}
public Request createRegisterRequest(@NotNull ParentPlatform platform, long CSeq, String fromTag, String viaTag, CallIdHeader callIdHeader) throws ParseException, InvalidArgumentException, PeerUnavailableException {
public Request createRegisterRequest(@NotNull ParentPlatform platform, long CSeq, String fromTag, String viaTag, CallIdHeader callIdHeader, boolean isRegister) throws ParseException, InvalidArgumentException, PeerUnavailableException {
Request request = null;
String sipAddress = sipConfig.getIp() + ":" + sipConfig.getPort();
//请求行
@ -109,18 +110,20 @@ public class SIPRequestHeaderPlarformProvider {
.createSipURI(platform.getDeviceGBId(), sipAddress));
request.addHeader(sipFactory.createHeaderFactory().createContactHeader(concatAddress));
ExpiresHeader expires = sipFactory.createHeaderFactory().createExpiresHeader(Integer.parseInt(platform.getExpires()));
ExpiresHeader expires = sipFactory.createHeaderFactory().createExpiresHeader(isRegister ? platform.getExpires() : 0);
request.addHeader(expires);
UserAgentHeader userAgentHeader = HeaderUtils.createUserAgentHeader(sipFactory);
request.addHeader(userAgentHeader);
return request;
}
public Request createRegisterRequest(@NotNull ParentPlatform parentPlatform, String fromTag, String viaTag,
String callId, WWWAuthenticateHeader www , CallIdHeader callIdHeader) throws ParseException, PeerUnavailableException, InvalidArgumentException {
String callId, WWWAuthenticateHeader www , CallIdHeader callIdHeader, boolean isRegister) throws ParseException, PeerUnavailableException, InvalidArgumentException {
Request registerRequest = createRegisterRequest(parentPlatform, redisCatchStorage.getCSEQ(), fromTag, viaTag, callIdHeader);
Request registerRequest = createRegisterRequest(parentPlatform, redisCatchStorage.getCSEQ(), fromTag, viaTag, callIdHeader, isRegister);
SipURI requestURI = sipFactory.createAddressFactory().createSipURI(parentPlatform.getServerGBId(), parentPlatform.getServerIP() + ":" + parentPlatform.getServerPort());
if (www == null) {
AuthorizationHeader authorizationHeader = sipFactory.createHeaderFactory().createAuthorizationHeader("Digest");

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

@ -12,6 +12,7 @@ import javax.sip.message.Request;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.gb28181.utils.HeaderUtils;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import gov.nist.javax.sip.SipProviderImpl;
import gov.nist.javax.sip.SipStackImpl;
@ -266,15 +267,7 @@ public class SIPRequestHeaderProvider {
Address concatAddress = sipFactory.createAddressFactory().createAddress(sipFactory.createAddressFactory()
.createSipURI(sipConfig.getId(), sipConfig.getIp() + ":" + sipConfig.getPort()));
infoRequest.addHeader(sipFactory.createHeaderFactory().createContactHeader(concatAddress));
List<String> agentParam = new ArrayList<>();
agentParam.add("wvp-pro");
// TODO 添加版本信息以及日期
UserAgentHeader userAgentHeader = null;
try {
userAgentHeader = sipFactory.createHeaderFactory().createUserAgentHeader(agentParam);
} catch (ParseException e) {
throw new RuntimeException(e);
}
UserAgentHeader userAgentHeader = HeaderUtils.createUserAgentHeader(sipFactory);
infoRequest.addHeader(userAgentHeader);
ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("Application",

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

@ -10,6 +10,7 @@ import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.SIPRequestHeaderProvider;
import com.genersoft.iot.vmp.gb28181.utils.HeaderUtils;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.media.zlm.dto.HookType;
@ -740,15 +741,7 @@ public class SIPCommander implements ISIPCommander {
// 增加Contact header
Address concatAddress = sipFactory.createAddressFactory().createAddress(sipFactory.createAddressFactory().createSipURI(sipConfig.getId(), sipConfig.getIp()+":"+sipConfig.getPort()));
byeRequest.addHeader(sipFactory.createHeaderFactory().createContactHeader(concatAddress));
List<String> agentParam = new ArrayList<>();
agentParam.add("wvp-pro");
// TODO 添加版本信息以及日期
UserAgentHeader userAgentHeader = null;
try {
userAgentHeader = sipFactory.createHeaderFactory().createUserAgentHeader(agentParam);
} catch (ParseException e) {
throw new RuntimeException(e);
}
UserAgentHeader userAgentHeader = HeaderUtils.createUserAgentHeader(sipFactory);
byeRequest.addHeader(userAgentHeader);
ClientTransaction clientTransaction = null;
if("TCP".equals(protocol)) {
@ -1677,14 +1670,11 @@ public class SIPCommander implements ISIPCommander {
clientTransaction = udpSipProvider.getNewClientTransaction(request);
}
if (request.getHeader(UserAgentHeader.NAME) == null) {
List<String> agentParam = new ArrayList<>();
agentParam.add("wvp-pro");
// TODO 添加版本信息以及日期
UserAgentHeader userAgentHeader = null;
try {
userAgentHeader = sipFactory.createHeaderFactory().createUserAgentHeader(agentParam);
userAgentHeader = HeaderUtils.createUserAgentHeader(sipFactory);
} catch (ParseException e) {
throw new RuntimeException(e);
logger.error("添加UserAgentHeader失败", e);
}
request.addHeader(userAgentHeader);
}

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

@ -4,6 +4,7 @@ import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.SIPRequestHeaderPlarformProvider;
import com.genersoft.iot.vmp.storager.dao.dto.PlatformRegisterInfo;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
@ -75,28 +76,21 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
@Override
public boolean register(ParentPlatform parentPlatform, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent) {
return register(parentPlatform, null, null, errorEvent, okEvent, false);
return register(parentPlatform, null, null, errorEvent, okEvent, false, true);
}
@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);
}
return register(parentPlatform, null, null, errorEvent, okEvent, false);
return register(parentPlatform, null, null, errorEvent, okEvent, false, false);
}
@Override
public boolean register(ParentPlatform parentPlatform, @Nullable String callId, @Nullable WWWAuthenticateHeader www,
SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent, boolean registerAgain) {
SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent, boolean registerAgain, boolean isRegister) {
try {
Request request;
String tm = Long.toString(System.currentTimeMillis());
if (!registerAgain ) {
// //callid
CallIdHeader callIdHeader = null;
if(parentPlatform.getTransport().equals("TCP")) {
callIdHeader = tcpSipProvider.getNewCallId();
@ -107,10 +101,10 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
request = headerProviderPlarformProvider.createRegisterRequest(parentPlatform,
redisCatchStorage.getCSEQ(), "FromRegister" + tm,
"z9hG4bK-" + UUID.randomUUID().toString().replace("-", ""), callIdHeader);
"z9hG4bK-" + UUID.randomUUID().toString().replace("-", ""), callIdHeader, isRegister);
// 将 callid 写入缓存， 等注册成功可以更新状态
String callIdFromHeader = callIdHeader.getCallId();
redisCatchStorage.updatePlatformRegisterInfo(callIdFromHeader, parentPlatform.getServerGBId());
redisCatchStorage.updatePlatformRegisterInfo(callIdFromHeader, PlatformRegisterInfo.getInstance(parentPlatform.getServerGBId(), isRegister));
sipSubscribe.addErrorSubscribe(callIdHeader.getCallId(), (event)->{
if (event != null) {
@ -127,7 +121,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
}else {
CallIdHeader callIdHeader = parentPlatform.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
: udpSipProvider.getNewCallId();
request = headerProviderPlarformProvider.createRegisterRequest(parentPlatform, "FromRegister" + tm, null, callId, www, callIdHeader);
request = headerProviderPlarformProvider.createRegisterRequest(parentPlatform, "FromRegister" + tm, null, callId, www, callIdHeader, isRegister);
}
transmitRequest(parentPlatform, request, null, okEvent);
@ -145,7 +139,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
}
@Override
public String keepalive(ParentPlatform parentPlatform) {
public String keepalive(ParentPlatform parentPlatform,SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent) {
String callId = null;
try {
String characterSet = parentPlatform.getCharacterSet();
@ -168,7 +162,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
UUID.randomUUID().toString().replace("-", ""),
null,
callIdHeader);
transmitRequest(parentPlatform, request);
transmitRequest(parentPlatform, request, errorEvent, okEvent);
callId = callIdHeader.getCallId();
} catch (ParseException | InvalidArgumentException | SipException e) {
e.printStackTrace();

9
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/SIPRequestProcessorParent.java
View File

@ -59,6 +59,9 @@ public abstract class SIPRequestProcessorParent {
public ServerTransaction getServerTransaction(RequestEvent evt) {
Request request = evt.getRequest();
ServerTransaction serverTransaction = evt.getServerTransaction();
if (serverTransaction != null) {
System.out.println(serverTransaction.getState().toString());
}
// 判断TCP还是UDP
boolean isTcp = false;
ViaHeader reqViaHeader = (ViaHeader) request.getHeader(ViaHeader.NAME);
@ -86,6 +89,8 @@ public abstract class SIPRequestProcessorParent {
logger.error(e.getMessage());
} catch (TransactionUnavailableException e) {
logger.error(e.getMessage());
}finally {
}
}
return serverTransaction;
@ -182,6 +187,10 @@ public abstract class SIPRequestProcessorParent {
sipFactory.createAddressFactory().createSipURI(sipURI.getUser(), sipURI.getHost()+":"+sipURI.getPort()
));
response.addHeader(sipFactory.createHeaderFactory().createContactHeader(concatAddress));
ServerTransaction serverTransaction = getServerTransaction(evt);
if (serverTransaction == null) {
}
getServerTransaction(evt).sendResponse(response);
}

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

@ -6,6 +6,7 @@ import com.genersoft.iot.vmp.gb28181.bean.SsrcTransaction;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
import com.genersoft.iot.vmp.gb28181.transmit.event.response.SIPResponseProcessorAbstract;
import com.genersoft.iot.vmp.gb28181.utils.HeaderUtils;
import gov.nist.javax.sip.ResponseEventExt;
import gov.nist.javax.sip.message.SIPResponse;
import gov.nist.javax.sip.stack.SIPDialog;
@ -103,15 +104,7 @@ public class InviteResponseProcessor extends SIPResponseProcessorAbstract {
}
requestURI.setPort(event.getRemotePort());
reqAck.setRequestURI(requestURI);
List<String> agentParam = new ArrayList<>();
agentParam.add("wvp-pro");
// TODO 添加版本信息以及日期
UserAgentHeader userAgentHeader = null;
try {
userAgentHeader = sipFactory.createHeaderFactory().createUserAgentHeader(agentParam);
} catch (ParseException e) {
throw new RuntimeException(e);
}
UserAgentHeader userAgentHeader = HeaderUtils.createUserAgentHeader(sipFactory);
reqAck.addHeader(userAgentHeader);
Address concatAddress = sipFactory.createAddressFactory().createAddress(sipFactory.createAddressFactory().createSipURI(sipConfig.getId(), sipConfig.getIp()+":"+sipConfig.getPort()));
reqAck.addHeader(sipFactory.createHeaderFactory().createContactHeader(concatAddress));

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

@ -6,8 +6,10 @@ 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;
import com.genersoft.iot.vmp.service.IPlatformService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.storager.dao.dto.PlatformRegisterInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@ -44,6 +46,9 @@ public class RegisterResponseProcessor extends SIPResponseProcessorAbstract {
@Autowired
private SubscribeHolder subscribeHolder;
@Autowired
private IPlatformService platformService;
@Override
public void afterPropertiesSet() throws Exception {
// 添加消息处理的订阅
@ -60,48 +65,39 @@ public class RegisterResponseProcessor extends SIPResponseProcessorAbstract {
Response response = evt.getResponse();
CallIdHeader callIdHeader = (CallIdHeader) response.getHeader(CallIdHeader.NAME);
String callId = callIdHeader.getCallId();
String platformGBId = redisCatchStorage.queryPlatformRegisterInfo(callId);
if (platformGBId == null) {
logger.info(String.format("未找到callId： %s 的注册/注销平台id", callId ));
PlatformRegisterInfo platformRegisterInfo = redisCatchStorage.queryPlatformRegisterInfo(callId);
if (platformRegisterInfo == null) {
logger.info(String.format("[国标级联]未找到callId： %s 的注册/注销平台id", callId ));
return;
}
ParentPlatformCatch parentPlatformCatch = redisCatchStorage.queryPlatformCatchInfo(platformGBId);
ParentPlatformCatch parentPlatformCatch = redisCatchStorage.queryPlatformCatchInfo(platformRegisterInfo.getPlatformId());
if (parentPlatformCatch == null) {
logger.warn(String.format("[收到注册/注销%S请求]平台：%s，但是平台缓存信息未查询到!!!", response.getStatusCode(),platformGBId));
logger.warn(String.format("[国标级联]收到注册/注销%S请求，平台：%s，但是平台缓存信息未查询到!!!", response.getStatusCode(),platformRegisterInfo.getPlatformId()));
return;
}
String action = parentPlatformCatch.getParentPlatform().getExpires().equals("0") ? "注销" : "注册";
logger.info(String.format("[%s %S响应]%s ", action, response.getStatusCode(), platformGBId ));
String action = platformRegisterInfo.isRegister() ? "注册" : "注销";
logger.info(String.format("[国标级联]%s %S响应,%s ", action, response.getStatusCode(), platformRegisterInfo.getPlatformId() ));
ParentPlatform parentPlatform = parentPlatformCatch.getParentPlatform();
if (parentPlatform == null) {
logger.warn(String.format("收到 %s %s的%S请求, 但是平台信息未查询到!!!", platformGBId, action, response.getStatusCode()));
logger.warn(String.format("[国标级联]收到 %s %s的%S请求, 但是平台信息未查询到!!!", platformRegisterInfo.getPlatformId(), action, response.getStatusCode()));
return;
}
if (response.getStatusCode() == 401) {
if (response.getStatusCode() == Response.UNAUTHORIZED) {
WWWAuthenticateHeader www = (WWWAuthenticateHeader)response.getHeader(WWWAuthenticateHeader.NAME);
sipCommanderForPlatform.register(parentPlatform, callId, www, null, null, true);
}else if (response.getStatusCode() == 200){
// 注册/注销成功
logger.info(String.format("%s %s成功", platformGBId, action));
sipCommanderForPlatform.register(parentPlatform, callId, www, null, null, true, platformRegisterInfo.isRegister());
}else if (response.getStatusCode() == Response.OK){
if (platformRegisterInfo.isRegister()) {
platformService.online(parentPlatform);
}else {
platformService.offline(parentPlatform);
}
// 注册/注销成功移除缓存的信息
redisCatchStorage.delPlatformRegisterInfo(callId);
redisCatchStorage.delPlatformCatchInfo(platformGBId);
// 取回Expires设置，避免注销过程中被置为0
ParentPlatform parentPlatformTmp = storager.queryParentPlatByServerGBId(platformGBId);
if (parentPlatformTmp != null) {
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);
}
}
}

22
src/main/java/com/genersoft/iot/vmp/gb28181/utils/HeaderUtils.java Normal file
View File

@ -0,0 +1,22 @@
package com.genersoft.iot.vmp.gb28181.utils;
import javax.sip.PeerUnavailableException;
import javax.sip.SipFactory;
import javax.sip.header.UserAgentHeader;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.List;
/**
* 生成header的工具类
* @author lin
*/
public class HeaderUtils {
public static UserAgentHeader createUserAgentHeader(SipFactory sipFactory) throws PeerUnavailableException, ParseException {
List<String> agentParam = new ArrayList<>();
agentParam.add("WVP PRO");
// TODO 添加版本信息以及日期
return sipFactory.createHeaderFactory().createUserAgentHeader(agentParam);
}
}

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

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

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

@ -18,8 +18,11 @@ import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.PostMapping;
@ -92,6 +95,10 @@ public class ZLMHttpHookListener {
@Autowired
private AssistRESTfulUtils assistRESTfulUtils;
@Qualifier("taskExecutor")
@Autowired
private ThreadPoolTaskExecutor taskExecutor;
/**
* 服务器定时上报时间，上报间隔可配置，默认10s上报一次
*
@ -238,9 +245,12 @@ public class ZLMHttpHookListener {
// 鉴权通过
redisCatchStorage.updateStreamAuthorityInfo(param.getApp(), param.getStream(), streamAuthorityInfo);
// 通知assist新的callId
if (mediaInfo != null && mediaInfo.getRecordAssistPort() > 0) {
assistRESTfulUtils.addStreamCallInfo(mediaInfo, param.getApp(), param.getStream(), callId, null);
}
taskExecutor.execute(()->{
if (mediaInfo != null && mediaInfo.getRecordAssistPort() > 0) {
assistRESTfulUtils.addStreamCallInfo(mediaInfo, param.getApp(), param.getStream(), callId, null);
}
});
}else {
zlmMediaListManager.sendStreamEvent(param.getApp(),param.getStream(), param.getMediaServerId());
}
@ -416,18 +426,23 @@ public class ZLMHttpHookListener {
String schema = item.getSchema();
List<MediaItem.MediaTrack> tracks = item.getTracks();
boolean regist = item.isRegist();
if (regist) {
StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(app, stream);
if (streamAuthorityInfo == null) {
streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(item);
if (item.getOriginType() == OriginType.RTMP_PUSH.ordinal()
|| item.getOriginType() == OriginType.RTSP_PUSH.ordinal()
|| item.getOriginType() == OriginType.RTC_PUSH.ordinal()) {
if (regist) {
StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(app, stream);
if (streamAuthorityInfo == null) {
streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(item);
}else {
streamAuthorityInfo.setOriginType(item.getOriginType());
streamAuthorityInfo.setOriginTypeStr(item.getOriginTypeStr());
}
redisCatchStorage.updateStreamAuthorityInfo(app, stream, streamAuthorityInfo);
}else {
streamAuthorityInfo.setOriginType(item.getOriginType());
streamAuthorityInfo.setOriginTypeStr(item.getOriginTypeStr());
redisCatchStorage.removeStreamAuthorityInfo(app, stream);
}
redisCatchStorage.updateStreamAuthorityInfo(app, stream, streamAuthorityInfo);
}else {
redisCatchStorage.removeStreamAuthorityInfo(app, stream);
}
if ("rtsp".equals(schema)){
logger.info("on_stream_changed：注册->{}, app->{}, stream->{}", regist, app, stream);
if (regist) {

45
src/main/java/com/genersoft/iot/vmp/service/IPlatformService.java Normal file
View File

@ -0,0 +1,45 @@
package com.genersoft.iot.vmp.service;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.github.pagehelper.PageInfo;
/**
* 国标平台的业务类
* @author lin
*/
public interface IPlatformService {
ParentPlatform queryPlatformByServerGBId(String platformGbId);
/**
* 分页获取上级平台
* @param page
* @param count
* @return
*/
PageInfo<ParentPlatform> queryParentPlatformList(int page, int count);
/**
* 添加级联平台
* @param parentPlatform 级联平台
*/
boolean add(ParentPlatform parentPlatform);
/**
* 平台上线
* @param parentPlatform 平台信息
*/
void online(ParentPlatform parentPlatform);
/**
* 平台离线
* @param parentPlatform 平台信息
*/
void offline(ParentPlatform parentPlatform);
/**
* 向上级平台发起注册
* @param parentPlatform
*/
void login(ParentPlatform parentPlatform);
}

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

@ -513,10 +513,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
String protocol = sslEnabled ? "https" : "http";
String hookPrex = String.format("%s://%s:%s/index/hook", protocol, mediaServerItem.getHookIp(), serverPort);
String recordHookPrex = null;
if (mediaServerItem.getRecordAssistPort() != 0) {
recordHookPrex = String.format("http://127.0.0.1:%s/api/record", mediaServerItem.getRecordAssistPort());
}
Map<String, Object> param = new HashMap<>();
param.put("api.secret",mediaServerItem.getSecret()); // -profile:v Baseline
param.put("ffmpeg.cmd","%s -fflags nobuffer -i %s -c:a aac -strict -2 -ar 44100 -ab 48k -c:v libx264 -f flv %s");
@ -525,7 +522,6 @@ public class MediaServerServiceImpl implements IMediaServerService {
param.put("hook.on_play",String.format("%s/on_play", hookPrex));
param.put("hook.on_http_access",String.format("%s/on_http_access", hookPrex));
param.put("hook.on_publish", String.format("%s/on_publish", hookPrex));
param.put("hook.on_record_mp4",recordHookPrex != null? String.format("%s/on_record_mp4", recordHookPrex): "");
param.put("hook.on_record_ts",String.format("%s/on_record_ts", hookPrex));
param.put("hook.on_rtsp_auth",String.format("%s/on_rtsp_auth", hookPrex));
param.put("hook.on_rtsp_realm",String.format("%s/on_rtsp_realm", hookPrex));
@ -535,6 +531,11 @@ public class MediaServerServiceImpl implements IMediaServerService {
param.put("hook.on_stream_none_reader",String.format("%s/on_stream_none_reader", hookPrex));
param.put("hook.on_stream_not_found",String.format("%s/on_stream_not_found", hookPrex));
param.put("hook.on_server_keepalive",String.format("%s/on_server_keepalive", hookPrex));
if (mediaServerItem.getRecordAssistPort() > 0) {
param.put("hook.on_record_mp4",String.format("http://127.0.0.1:%s/api/record/on_record_mp4", mediaServerItem.getRecordAssistPort()));
}else {
param.put("hook.on_record_mp4","");
}
param.put("hook.timeoutSec","20");
param.put("general.streamNoneReaderDelayMS",mediaServerItem.getStreamNoneReaderDelayMS()==-1?"3600000":mediaServerItem.getStreamNoneReaderDelayMS() );
// 推流断开后可以在超时时间内重新连接上继续推流，这样播放器会接着播放。
@ -544,7 +545,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
param.put("general.continue_push_ms", "3000" );
// 最多等待未初始化的Track时间，单位毫秒，超时之后会忽略未初始化的Track, 设置此选项优化那些音频错误的不规范流，
// 等zlm支持给每个rtpServer设置关闭音频的时候可以不设置此选项
param.put("general.wait_track_ready_ms", "3000" );
// param.put("general.wait_track_ready_ms", "3000" );
if (mediaServerItem.isRtpEnable() && !ObjectUtils.isEmpty(mediaServerItem.getRtpPortRange())) {
param.put("rtp_proxy.port_range", mediaServerItem.getRtpPortRange().replace(",", "-"));
}

232
src/main/java/com/genersoft/iot/vmp/service/impl/PlatformServiceImpl.java Normal file
View File

@ -0,0 +1,232 @@
package com.genersoft.iot.vmp.service.impl;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.gb28181.bean.SubscribeHolder;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IPlatformService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.dao.ParentPlatformMapper;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.sip.TimeoutEvent;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* @author lin
*/
@Service
public class PlatformServiceImpl implements IPlatformService {
private final static String REGISTER_KEY_PREFIX = "platform_register_";
private final static String KEEPALIVE_KEY_PREFIX = "platform_keepalive_";
private final static Logger logger = LoggerFactory.getLogger(PlatformServiceImpl.class);
@Autowired
private ParentPlatformMapper platformMapper;
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Autowired
private IMediaServerService mediaServerService;
@Autowired
private SIPCommanderFroPlatform commanderForPlatform;
@Autowired
private DynamicTask dynamicTask;
@Autowired
private ZLMRTPServerFactory zlmrtpServerFactory;
@Autowired
private SubscribeHolder subscribeHolder;
@Override
public ParentPlatform queryPlatformByServerGBId(String platformGbId) {
return platformMapper.getParentPlatByServerGBId(platformGbId);
}
@Override
public PageInfo<ParentPlatform> queryParentPlatformList(int page, int count) {
PageHelper.startPage(page, count);
List<ParentPlatform> all = platformMapper.getParentPlatformList();
return new PageInfo<>(all);
}
@Override
public boolean add(ParentPlatform parentPlatform) {
if (parentPlatform.getCatalogGroup() == 0) {
// 每次发送目录的数量默认为1
parentPlatform.setCatalogGroup(1);
}
if (parentPlatform.getAdministrativeDivision() == null) {
// 行政区划默认去编号的前6位
parentPlatform.setAdministrativeDivision(parentPlatform.getServerGBId().substring(0,6));
}
parentPlatform.setCatalogId(parentPlatform.getDeviceGBId());
int result = platformMapper.addParentPlatform(parentPlatform);
// 添加缓存
ParentPlatformCatch parentPlatformCatch = new ParentPlatformCatch();
parentPlatformCatch.setParentPlatform(parentPlatform);
parentPlatformCatch.setId(parentPlatform.getServerGBId());
parentPlatformCatch.setParentPlatform(parentPlatform);
redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
if (parentPlatform.isEnable()) {
// 保存时启用就发送注册
// 注册成功时由程序直接调用了online方法
commanderForPlatform.register(parentPlatform, eventResult -> {
logger.info("[国标级联] {},添加向上级注册失败，请确定上级平台可用时重新保存", parentPlatform.getServerGBId());
}, null);
}
return result > 0;
}
@Override
public void online(ParentPlatform parentPlatform) {
logger.info("[国标级联]：{}, 平台上线", parentPlatform.getServerGBId());
platformMapper.updateParentPlatformStatus(parentPlatform.getServerGBId(), true);
ParentPlatformCatch parentPlatformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId());
if (parentPlatformCatch != null) {
parentPlatformCatch.getParentPlatform().setStatus(true);
redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
}else {
parentPlatformCatch = new ParentPlatformCatch();
parentPlatformCatch.setParentPlatform(parentPlatform);
parentPlatformCatch.setId(parentPlatform.getServerGBId());
parentPlatform.setStatus(true);
parentPlatformCatch.setParentPlatform(parentPlatform);
redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
}
final String registerTaskKey = REGISTER_KEY_PREFIX + parentPlatform.getServerGBId();
if (dynamicTask.contains(registerTaskKey)) {
dynamicTask.stop(registerTaskKey);
}
// 添加注册任务
dynamicTask.startDelay(registerTaskKey,
// 注册失败（注册成功时由程序直接调用了online方法）
()->commanderForPlatform.register(parentPlatform, eventResult -> offline(parentPlatform),null),
parentPlatform.getExpires()*1000);
final String keepaliveTaskKey = KEEPALIVE_KEY_PREFIX + parentPlatform.getServerGBId();
if (!dynamicTask.contains(keepaliveTaskKey)) {
// 添加心跳任务
dynamicTask.startCron(keepaliveTaskKey,
()-> commanderForPlatform.keepalive(parentPlatform, eventResult -> {
// 心跳失败
if (eventResult.type == SipSubscribe.EventResultType.timeout) {
// 心跳超时
ParentPlatformCatch platformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId());
// 此时是第三次心跳超时， 平台离线
if (platformCatch.getKeepAliveReply() == 2) {
// 设置平台离线，并重新注册
offline(parentPlatform);
logger.info("[国标级联] {}，三次心跳超时后再次发起注册", parentPlatform.getServerGBId());
commanderForPlatform.register(parentPlatform, eventResult1 -> {
logger.info("[国标级联] {}，三次心跳超时后再次发起注册仍然失败，开始定时发起注册，间隔为1分钟", parentPlatform.getServerGBId());
// 添加注册任务
dynamicTask.startCron(registerTaskKey,
// 注册失败（注册成功时由程序直接调用了online方法）
()->logger.info("[国标级联] {},平台离线后持续发起注册，失败", parentPlatform.getServerGBId()),
60*1000);
}, null);
}
}else {
logger.warn("[国标级联]发送心跳收到错误，code： {}, msg: {}", eventResult.statusCode, eventResult.msg);
}
}, eventResult -> {
// 心跳成功
// 清空之前的心跳超时计数
ParentPlatformCatch platformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId());
if (platformCatch.getKeepAliveReply() > 0) {
platformCatch.setKeepAliveReply(0);
redisCatchStorage.updatePlatformCatchInfo(platformCatch);
}
}),
parentPlatform.getExpires()*1000);
}
}
@Override
public void offline(ParentPlatform parentPlatform) {
logger.info("[平台离线]：{}", parentPlatform.getServerGBId());
ParentPlatformCatch parentPlatformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId());
parentPlatformCatch.setKeepAliveReply(0);
parentPlatformCatch.setRegisterAliveReply(0);
ParentPlatform parentPlatformInCatch = parentPlatformCatch.getParentPlatform();
parentPlatformInCatch.setStatus(false);
parentPlatformCatch.setParentPlatform(parentPlatformInCatch);
redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
platformMapper.updateParentPlatformStatus(parentPlatform.getServerGBId(), false);
// 停止所有推流
logger.info("[平台离线] {}, 停止所有推流", parentPlatform.getServerGBId());
stopAllPush(parentPlatform.getServerGBId());
// 清除注册定时
logger.info("[平台离线] {}, 停止定时注册任务", parentPlatform.getServerGBId());
final String registerTaskKey = REGISTER_KEY_PREFIX + parentPlatform.getServerGBId();
if (dynamicTask.contains(registerTaskKey)) {
dynamicTask.stop(registerTaskKey);
}
// 清除心跳定时
logger.info("[平台离线] {}, 停止定时发送心跳任务", parentPlatform.getServerGBId());
final String keepaliveTaskKey = KEEPALIVE_KEY_PREFIX + parentPlatform.getServerGBId();
if (dynamicTask.contains(keepaliveTaskKey)) {
// 添加心跳任务
dynamicTask.stop(keepaliveTaskKey);
}
// 停止目录订阅回复
logger.info("[平台离线] {}, 停止订阅回复", parentPlatform.getServerGBId());
subscribeHolder.removeAllSubscribe(parentPlatform.getServerGBId());
}
private void stopAllPush(String platformId) {
List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServer(platformId);
if (sendRtpItems != null && sendRtpItems.size() > 0) {
for (SendRtpItem sendRtpItem : sendRtpItems) {
redisCatchStorage.deleteSendRTPServer(platformId, sendRtpItem.getChannelId(), null, null);
MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
Map<String, Object> param = new HashMap<>(3);
param.put("vhost", "__defaultVhost__");
param.put("app", sendRtpItem.getApp());
param.put("stream", sendRtpItem.getStreamId());
zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param);
}
}
}
@Override
public void login(ParentPlatform parentPlatform) {
final String registerTaskKey = REGISTER_KEY_PREFIX + parentPlatform.getServerGBId();
commanderForPlatform.register(parentPlatform, eventResult1 -> {
logger.info("[国标级联] {}，开始定时发起注册，间隔为1分钟", parentPlatform.getServerGBId());
// 添加注册任务
dynamicTask.startCron(registerTaskKey,
// 注册失败（注册成功时由程序直接调用了online方法）
()->logger.info("[国标级联] {},平台离线后持续发起注册，失败", parentPlatform.getServerGBId()),
60*1000);
}, null);
}
}

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

@ -575,7 +575,7 @@ public class PlayServiceImpl implements IPlayService {
logger.warn("查询录像信息时发现节点已离线");
return null;
}
if (mediaServerItem.getRecordAssistPort() != 0) {
if (mediaServerItem.getRecordAssistPort() > 0) {
JSONObject jsonObject = assistRESTfulUtils.fileDuration(mediaServerItem, streamInfo.getApp(), streamInfo.getStream(), null);
if (jsonObject != null && jsonObject.getInteger("code") == 0) {
long duration = jsonObject.getLong("data");
@ -691,7 +691,7 @@ public class PlayServiceImpl implements IPlayService {
// for (SendRtpItem sendRtpItem : sendRtpItems) {
// if (sendRtpItem.getMediaServerId().equals(mediaServerId)) {
// if (mediaListMap.get(sendRtpItem.getStreamId()) == null) {
// ParentPlatform platform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
// ParentPlatform platform = storager.queryPlatformByServerGBId(sendRtpItem.getPlatformId());
// sipCommanderFroPlatform.streamByeCmd(platform, sendRtpItem.getCallId());
// }
// }

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

@ -8,6 +8,7 @@ import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
import com.genersoft.iot.vmp.service.bean.SSRCInfo;
import com.genersoft.iot.vmp.service.bean.ThirdPartyGB;
import com.genersoft.iot.vmp.storager.dao.dto.PlatformRegisterInfo;
import java.util.List;
import java.util.Map;
@ -61,17 +62,13 @@ public interface IRedisCatchStorage {
void delPlatformCatchInfo(String platformGbId);
void updatePlatformKeepalive(ParentPlatform parentPlatform);
void delPlatformKeepalive(String platformGbId);
void updatePlatformRegister(ParentPlatform parentPlatform);
void delPlatformRegister(String platformGbId);
void updatePlatformRegisterInfo(String callId, String platformGbId);
void updatePlatformRegisterInfo(String callId, PlatformRegisterInfo platformRegisterInfo);
String queryPlatformRegisterInfo(String callId);
PlatformRegisterInfo queryPlatformRegisterInfo(String callId);
void delPlatformRegisterInfo(String callId);

9
src/main/java/com/genersoft/iot/vmp/storager/IVideoManagerStorage.java
View File

@ -170,15 +170,6 @@ public interface IVideoManagerStorage {
*/
boolean deleteParentPlatform(ParentPlatform parentPlatform);
/**
* 分页获取上级平台
* @param page
* @param count
* @return
*/
PageInfo<ParentPlatform> queryParentPlatformList(int page, int count);
/**
* 获取所有已启用的平台
* @return

41
src/main/java/com/genersoft/iot/vmp/storager/dao/dto/PlatformRegisterInfo.java Normal file
View File

@ -0,0 +1,41 @@
package com.genersoft.iot.vmp.storager.dao.dto;
/**
* 平台发送注册/注销消息时缓存此消息
* @author lin
*/
public class PlatformRegisterInfo {
/**
* 平台Id
*/
private String platformId;
/**
* 是否时注册，false为注销
*/
private boolean register;
public static PlatformRegisterInfo getInstance(String platformId, boolean register) {
PlatformRegisterInfo platformRegisterInfo = new PlatformRegisterInfo();
platformRegisterInfo.setPlatformId(platformId);
platformRegisterInfo.setRegister(register);
return platformRegisterInfo;
}
public String getPlatformId() {
return platformId;
}
public void setPlatformId(String platformId) {
this.platformId = platformId;
}
public boolean isRegister() {
return register;
}
public void setRegister(boolean register) {
this.register = register;
}
}

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

@ -16,6 +16,7 @@ import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
import com.genersoft.iot.vmp.service.bean.ThirdPartyGB;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.dao.DeviceChannelMapper;
import com.genersoft.iot.vmp.storager.dao.dto.PlatformRegisterInfo;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.utils.redis.RedisUtil;
import org.slf4j.Logger;
@ -290,18 +291,6 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
RedisUtil.set(key, parentPlatformCatch);
}
@Override
public void updatePlatformKeepalive(ParentPlatform parentPlatform) {
String key = VideoManagerConstants.PLATFORM_KEEPALIVE_PREFIX + userSetting.getServerId() + "_" + parentPlatform.getServerGBId();
RedisUtil.set(key, "", Integer.parseInt(parentPlatform.getKeepTimeout()));
}
@Override
public void updatePlatformRegister(ParentPlatform parentPlatform) {
String key = VideoManagerConstants.PLATFORM_REGISTER_PREFIX + userSetting.getServerId() + "_" + parentPlatform.getServerGBId();
RedisUtil.set(key, "", Integer.parseInt(parentPlatform.getExpires()));
}
@Override
public ParentPlatformCatch queryPlatformCatchInfo(String platformGbId) {
return (ParentPlatformCatch)RedisUtil.get(VideoManagerConstants.PLATFORM_CATCH_PREFIX + userSetting.getServerId() + "_" + platformGbId);
@ -324,15 +313,15 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
@Override
public void updatePlatformRegisterInfo(String callId, String platformGbId) {
public void updatePlatformRegisterInfo(String callId, PlatformRegisterInfo platformRegisterInfo) {
String key = VideoManagerConstants.PLATFORM_REGISTER_INFO_PREFIX + userSetting.getServerId() + "_" + callId;
RedisUtil.set(key, platformGbId, 30);
RedisUtil.set(key, platformRegisterInfo, 30);
}
@Override
public String queryPlatformRegisterInfo(String callId) {
return (String)RedisUtil.get(VideoManagerConstants.PLATFORM_REGISTER_INFO_PREFIX + userSetting.getServerId() + "_" + callId);
public PlatformRegisterInfo queryPlatformRegisterInfo(String callId) {
return (PlatformRegisterInfo)RedisUtil.get(VideoManagerConstants.PLATFORM_REGISTER_INFO_PREFIX + userSetting.getServerId() + "_" + callId);
}
@Override

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

@ -457,13 +457,6 @@ public class VideoManagerStorageImpl implements IVideoManagerStorage {
return result > 0;
}
@Override
public PageInfo<ParentPlatform> queryParentPlatformList(int page, int count) {
PageHelper.startPage(page, count);
List<ParentPlatform> all = platformMapper.getParentPlatformList();
return new PageInfo<>(all);
}
@Override
public ParentPlatform queryParentPlatByServerGBId(String platformGbId) {
return platformMapper.getParentPlatByServerGBId(platformGbId);

1
src/main/java/com/genersoft/iot/vmp/utils/redis/RedisUtil.java
View File

@ -5,6 +5,7 @@ import java.util.concurrent.TimeUnit;
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.utils.SpringBeanFactory;
import gov.nist.javax.sip.stack.UDPMessageChannel;
import org.springframework.data.redis.core.*;
import org.springframework.util.CollectionUtils;

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

@ -9,14 +9,13 @@ import com.genersoft.iot.vmp.conf.exception.ControllerException;
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.bean.TreeType;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.service.IPlatformChannelService;
import com.genersoft.iot.vmp.service.IPlatformService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
import com.genersoft.iot.vmp.vmanager.gb28181.platform.bean.ChannelReduce;
import com.genersoft.iot.vmp.vmanager.gb28181.platform.bean.UpdateChannelParam;
import com.github.pagehelper.PageInfo;
@ -26,10 +25,7 @@ import io.swagger.v3.oas.annotations.tags.Tag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.*;
import com.genersoft.iot.vmp.conf.SipConfig;
@ -70,6 +66,9 @@ public class PlatformController {
@Autowired
private DynamicTask dynamicTask;
@Autowired
private IPlatformService platformService;
/**
* 获取国标服务的配置
*
@ -95,8 +94,7 @@ public class PlatformController {
@Parameter(name = "id", description = "平台国标编号", required = true)
@GetMapping("/info/{id}")
public ParentPlatform getPlatform(@PathVariable String id) {
ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(id);
WVPResult<ParentPlatform> wvpResult = new WVPResult<>();
ParentPlatform parentPlatform = platformService.queryPlatformByServerGBId(id);
if (parentPlatform != null) {
return parentPlatform;
} else {
@ -117,7 +115,7 @@ public class PlatformController {
@Parameter(name = "count", description = "每页条数", required = true)
public PageInfo<ParentPlatform> platforms(@PathVariable int page, @PathVariable int count) {
PageInfo<ParentPlatform> parentPlatformPageInfo = storager.queryParentPlatformList(page, count);
PageInfo<ParentPlatform> parentPlatformPageInfo = platformService.queryParentPlatformList(page, count);
if (parentPlatformPageInfo.getList().size() > 0) {
for (ParentPlatform platform : parentPlatformPageInfo.getList()) {
platform.setMobilePositionSubscribe(subscribeHolder.getMobilePositionSubscribe(platform.getServerGBId()) != null);
@ -136,7 +134,7 @@ public class PlatformController {
@Operation(summary = "添加上级平台信息")
@PostMapping("/add")
@ResponseBody
public String addPlatform(@RequestBody ParentPlatform parentPlatform) {
public void addPlatform(@RequestBody ParentPlatform parentPlatform) {
if (logger.isDebugEnabled()) {
logger.debug("保存上级平台信息API调用");
@ -158,33 +156,16 @@ public class PlatformController {
throw new ControllerException(ErrorCode.ERROR400.getCode(), "error severPort");
}
ParentPlatform parentPlatformOld = storager.queryParentPlatByServerGBId(parentPlatform.getServerGBId());
if (parentPlatformOld != null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "平台 " + parentPlatform.getServerGBId() + " 已存在");
}
parentPlatform.setCreateTime(DateUtil.getNow());
parentPlatform.setUpdateTime(DateUtil.getNow());
boolean updateResult = storager.updateParentPlatform(parentPlatform);
boolean updateResult = platformService.add(parentPlatform);
if (updateResult) {
// 保存时启用就发送注册
if (parentPlatform.isEnable()) {
if (parentPlatformOld != null && parentPlatformOld.isStatus()) {
commanderForPlatform.unregister(parentPlatformOld, null, eventResult -> {
// 只要保存就发送注册
commanderForPlatform.register(parentPlatform, null, null);
});
} else {
// 只要保存就发送注册
commanderForPlatform.register(parentPlatform, null, null);
}
} else if (parentPlatformOld != null && parentPlatformOld.isEnable()) {
// 关闭启用时注销
commanderForPlatform.unregister(parentPlatform, null, null);
}
return null;
} else {
if (!updateResult) {
throw new ControllerException(ErrorCode.ERROR100.getCode(),"写入数据库失败");
}
}

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

@ -179,8 +179,6 @@ user-settings:
platform-play-timeout: 60000
# 是否开启接口鉴权
interface-authentication: true
# 自动配置redis 可以过期事件
redis-config: true
# 接口鉴权例外的接口, 即不进行接口鉴权的接口,尽量详细书写，尽量不用/**，至少两级目录
interface-authentication-excludes:
- /api/v1/**

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

@ -357,7 +357,7 @@ export default {
var result = false;
var that = this;
await that.$axios({
method: 'post',
method: 'get',
url:`/api/platform/exit/${deviceGbId}`
}).then(function (res) {
result = res.data;

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

@ -263,7 +263,7 @@ export default {
var result = false;
var that = this;
await that.$axios({
method: 'post',
method: 'get',
url:`/api/platform/exit/${deviceGbId}`
}).then(function (res) {
result = res.data;

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

@ -327,7 +327,7 @@ export default {
var result = false;
var that = this;
await that.$axios({
method: 'post',
method: 'get',
url:`/api/platform/exit/${deviceGbId}`})
.then(function (res) {
if (res.data.code === 0) {

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

@ -158,7 +158,7 @@ export default {
var result = false;
var that = this;
await that.$axios({
method:"post",
method:"get",
url:`/api/platform/exit/${deviceGbId}`
}).then(function (res) {
result = res.data;