XW-AIOT/wvp-platform
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化设备在线状态

Browse Source
This commit is contained in:
648540858
2022-05-11 18:37:24 +08:00
parent c395cf42d1
commit f6893cf95b
52 changed files with 571 additions and 802 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -48,8 +48,15 @@ public class SipLayer{
Properties properties = new Properties();
properties.setProperty("javax.sip.STACK_NAME", "GB28181_SIP");
properties.setProperty("javax.sip.IP_ADDRESS", sipConfig.getMonitorIp());
/**
* 完整配置参考 gov.nist.javax.sip.SipStackImpl，需要下载源码
* gov/nist/javax/sip/SipStackImpl.class
*/
properties.setProperty("gov.nist.javax.sip.LOG_MESSAGE_CONTENT", "true");
properties.setProperty("gov.nist.javax.sip.DELIVER_UNSOLICITED_NOTIFY", "true"); // 接收所有notify请求，即使没有订阅
properties.setProperty("gov.nist.javax.sip.DELIVER_TERMINATED_EVENT_FOR_NULL_DIALOG", "true"); // 为_NULL _对话框传递_终止的_事件
properties.setProperty("gov.nist.javax.sip.RELEASE_REFERENCES_STRATEGY", "Normal"); // 会话清理策略
properties.setProperty("gov.nist.javax.sip.RELIABLE_CONNECTION_KEEP_ALIVE_TIMEOUT", "10");
/**
* 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

13
src/main/java/com/genersoft/iot/vmp/gb28181/bean/Device.java
View File

@@ -99,11 +99,6 @@ public class Device {
*/
private String mediaServerId;
/**
* 首次注册
*/
private boolean firsRegister;
/**
* 字符集, 支持 UTF-8 与 GB2312
*/
@@ -279,14 +274,6 @@ public class Device {
this.mediaServerId = mediaServerId;
}
public boolean isFirsRegister() {
return firsRegister;
}
public void setFirsRegister(boolean firsRegister) {
this.firsRegister = firsRegister;
}
public String getCharset() {
return charset;
}

7
src/main/java/com/genersoft/iot/vmp/gb28181/bean/RecordItem.java
View File

@@ -1,10 +1,10 @@
package com.genersoft.iot.vmp.gb28181.bean;
import com.genersoft.iot.vmp.utils.DateUtil;
import org.jetbrains.annotations.NotNull;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
/**
@@ -116,10 +116,9 @@ public class RecordItem implements Comparable<RecordItem>{
@Override
public int compareTo(@NotNull RecordItem recordItem) {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
try {
Date startTime_now = sdf.parse(startTime);
Date startTime_param = sdf.parse(recordItem.getStartTime());
Date startTime_now = DateUtil.format.parse(startTime);
Date startTime_param = DateUtil.format.parse(recordItem.getStartTime());
if (startTime_param.compareTo(startTime_now) > 0) {
return -1;
}else {

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

@@ -61,7 +61,7 @@ public class SubscribeHolder {
// 添加任务处理GPS定时推送
dynamicTask.startCron(key, new MobilePositionSubscribeHandlerTask(redisCatchStorage, sipCommanderForPlatform,
storager, platformId, subscribeInfo.getSn(), key, this, dynamicTask),
subscribeInfo.getGpsInterval());
subscribeInfo.getGpsInterval() * 1000);
String taskOverdueKey = taskOverduePrefix + "MobilePosition_" + platformId;
dynamicTask.stop(taskOverdueKey);
// 添加任务处理订阅过期

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

@@ -1,7 +1,7 @@
package com.genersoft.iot.vmp.gb28181.event;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.offline.OfflineEvent;
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;
@@ -11,12 +11,11 @@ import com.genersoft.iot.vmp.media.zlm.event.ZLMOfflineEvent;
import com.genersoft.iot.vmp.media.zlm.event.ZLMOnlineEvent;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import com.genersoft.iot.vmp.gb28181.event.alarm.AlarmEvent;
import com.genersoft.iot.vmp.gb28181.event.online.OnlineEvent;
import javax.sip.TimeoutEvent;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
@@ -32,28 +31,6 @@ public class EventPublisher {
@Autowired
private ApplicationEventPublisher applicationEventPublisher;
public void onlineEventPublish(Device device, String from, int expires) {
OnlineEvent onEvent = new OnlineEvent(this);
onEvent.setDevice(device);
onEvent.setFrom(from);
onEvent.setExpires(expires);
applicationEventPublisher.publishEvent(onEvent);
}
public void onlineEventPublish(Device device, String from) {
OnlineEvent onEvent = new OnlineEvent(this);
onEvent.setDevice(device);
onEvent.setFrom(from);
applicationEventPublisher.publishEvent(onEvent);
}
public void outlineEventPublish(String deviceId, String from){
OfflineEvent outEvent = new OfflineEvent(this);
outEvent.setDeviceId(deviceId);
outEvent.setFrom(from);
applicationEventPublisher.publishEvent(outEvent);
}
/**
* 平台心跳到期事件
@@ -115,6 +92,13 @@ public class EventPublisher {
}
public void requestTimeOut(TimeoutEvent timeoutEvent) {
RequestTimeoutEvent requestTimeoutEvent = new RequestTimeoutEvent(this);
requestTimeoutEvent.setTimeoutEvent(timeoutEvent);
applicationEventPublisher.publishEvent(requestTimeoutEvent);
}
/**
*
* @param platformId

25
src/main/java/com/genersoft/iot/vmp/gb28181/event/device/RequestTimeoutEvent.java Normal file
View File

@@ -0,0 +1,25 @@
package com.genersoft.iot.vmp.gb28181.event.device;
import org.springframework.context.ApplicationEvent;
import javax.sip.TimeoutEvent;
/**
* @author lin
*/
public class RequestTimeoutEvent extends ApplicationEvent {
public RequestTimeoutEvent(Object source) {
super(source);
}
private TimeoutEvent timeoutEvent;
public TimeoutEvent getTimeoutEvent() {
return timeoutEvent;
}
public void setTimeoutEvent(TimeoutEvent timeoutEvent) {
this.timeoutEvent = timeoutEvent;
}
}

41
src/main/java/com/genersoft/iot/vmp/gb28181/event/device/RequestTimeoutEventImpl.java Normal file
View File

@@ -0,0 +1,41 @@
package com.genersoft.iot.vmp.gb28181.event.device;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.service.IDeviceService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;
import javax.sip.ClientTransaction;
import javax.sip.address.SipURI;
import javax.sip.header.CallIdHeader;
import javax.sip.header.ToHeader;
import javax.sip.message.Request;
/**
* @author lin
*/
@Component
public class RequestTimeoutEventImpl implements ApplicationListener<RequestTimeoutEvent> {
@Autowired
private IDeviceService deviceService;
@Override
public void onApplicationEvent(RequestTimeoutEvent event) {
ClientTransaction clientTransaction = event.getTimeoutEvent().getClientTransaction();
if (clientTransaction != null) {
Request request = clientTransaction.getRequest();
if (request != null) {
String host = ((SipURI) request.getRequestURI()).getHost();
int port = ((SipURI) request.getRequestURI()).getPort();
Device device = deviceService.getDeviceByHostAndPort(host, port);
if (device == null) {
return;
}
deviceService.offline(device.getDeviceId());
}
}
}
}

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

@@ -17,9 +17,8 @@ import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
/**
* @description:设备心跳超时监听,借助redis过期特性，进行监听，监听到说明设备心跳超时，发送离线事件
* @author: swwheihei
* @date: 2020年5月6日 上午11:35:46
* 设备心跳超时监听,借助redis过期特性，进行监听，监听到说明设备心跳超时，发送离线事件
* @author swwheihei
*/
@Component
public class KeepaliveTimeoutListenerForPlatform extends RedisKeyExpirationEventMessageListener {
@@ -55,25 +54,18 @@ public class KeepaliveTimeoutListenerForPlatform extends RedisKeyExpirationEvent
// 平台心跳到期,需要重发, 判断是否已经多次未收到心跳回复, 多次未收到,则重新发起注册, 注册尝试多次未得到回复,则认为平台离线
String PLATFORM_KEEPLIVEKEY_PREFIX = VideoManagerConstants.PLATFORM_KEEPALIVE_PREFIX + userSetting.getServerId() + "_";
String PLATFORM_REGISTER_PREFIX = VideoManagerConstants.PLATFORM_REGISTER_PREFIX + userSetting.getServerId() + "_";
String KEEPLIVEKEY_PREFIX = VideoManagerConstants.KEEPLIVEKEY_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(),expiredKey.length());
ParentPlatform platform = storager.queryParentPlatByServerGBId(platformGBId);
String platformGbId = expiredKey.substring(PLATFORM_KEEPLIVEKEY_PREFIX.length());
ParentPlatform platform = storager.queryParentPlatByServerGBId(platformGbId);
if (platform != null) {
publisher.platformKeepaliveExpireEventPublish(platformGBId);
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);
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(KEEPLIVEKEY_PREFIX)){
String deviceId = expiredKey.substring(KEEPLIVEKEY_PREFIX.length(),expiredKey.length());
Device device = storager.queryVideoDevice(deviceId);
if (device != null) {
publisher.outlineEventPublish(deviceId, KEEPLIVEKEY_PREFIX);
publisher.platformRegisterCycleEventPublish(platformGbId);
}
}else if (expiredKey.startsWith(REGISTER_INFO_PREFIX)) {
String callId = expiredKey.substring(REGISTER_INFO_PREFIX.length());
@@ -85,6 +77,5 @@ public class KeepaliveTimeoutListenerForPlatform extends RedisKeyExpirationEvent
sipSubscribe.getErrorSubscribe(callId).response(eventResult);
}
}
}
}

63
src/main/java/com/genersoft/iot/vmp/gb28181/event/offline/KeepliveTimeoutListener.java
View File

@@ -1,63 +0,0 @@
package com.genersoft.iot.vmp.gb28181.event.offline;
import com.genersoft.iot.vmp.conf.RedisKeyExpirationEventMessageListener;
import com.genersoft.iot.vmp.conf.UserSetting;
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;
/**
* @description:设备心跳超时监听,借助redis过期特性，进行监听，监听到说明设备心跳超时，发送离线事件
* @author: swwheihei
* @date: 2020年5月6日 上午11:35:46
*/
@Component
public class KeepliveTimeoutListener extends RedisKeyExpirationEventMessageListener {
private Logger logger = LoggerFactory.getLogger(KeepliveTimeoutListener.class);
@Autowired
private EventPublisher publisher;
@Autowired
private UserSetting userSetting;
public KeepliveTimeoutListener(RedisMessageListenerContainer listenerContainer, UserSetting userSetting) {
super(listenerContainer, userSetting);
}
@Override
public void init() {
if (!userSetting.getRedisConfig()) {
// 配置springboot默认Config为空，即不让应用去修改redis的默认配置，因为Redis服务出于安全会禁用CONFIG命令给远程用户使用
setKeyspaceNotificationsConfigParameter("");
}
super.init();
}
/**
* 监听失效的key，key格式为keeplive_deviceId
* @param message
* @param pattern
*/
@Override
public void onMessage(Message message, byte[] pattern) {
// 获取失效的key
String expiredKey = message.toString();
String KEEPLIVEKEY_PREFIX = VideoManagerConstants.KEEPLIVEKEY_PREFIX + userSetting.getServerId() + "_";
if(!expiredKey.startsWith(KEEPLIVEKEY_PREFIX)){
logger.debug("收到redis过期监听，但开头不是"+KEEPLIVEKEY_PREFIX+"，忽略");
return;
}
String deviceId = expiredKey.substring(KEEPLIVEKEY_PREFIX.length(),expiredKey.length());
publisher.outlineEventPublish(deviceId, VideoManagerConstants.EVENT_OUTLINE_TIMEOUT);
}
}

40
src/main/java/com/genersoft/iot/vmp/gb28181/event/offline/OfflineEvent.java
View File

@@ -1,40 +0,0 @@
package com.genersoft.iot.vmp.gb28181.event.offline;
import org.springframework.context.ApplicationEvent;
/**
* @description: 离线事件类
* @author: swwheihei
* @date: 2020年5月6日 上午11:33:13
*/
public class OfflineEvent extends ApplicationEvent {
/**
*
*/
private static final long serialVersionUID = 1L;
public OfflineEvent(Object source) {
super(source);
}
private String deviceId;
private String from;
public String getDeviceId() {
return deviceId;
}
public void setDeviceId(String deviceId) {
this.deviceId = deviceId;
}
public String getFrom() {
return from;
}
public void setFrom(String from) {
this.from = from;
}
}

98
src/main/java/com/genersoft/iot/vmp/gb28181/event/offline/OfflineEventListener.java
View File

@@ -1,98 +0,0 @@
package com.genersoft.iot.vmp.gb28181.event.offline;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
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.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.service.IMediaServerService;
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 com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.utils.redis.RedisUtil;
import java.util.List;
/**
* @description: 离线事件监听器，监听到离线后，修改设备离在线状态。 设备离线有两个来源：
* 1、设备主动注销，发送注销指令，{@link com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.RegisterRequestProcessor}
* 2、设备未知原因离线，心跳超时,{@link com.genersoft.iot.vmp.gb28181.event.offline.OfflineEventListener}
* @author: swwheihei
* @date: 2020年5月6日 下午1:51:23
*/
@Component
public class OfflineEventListener implements ApplicationListener<OfflineEvent> {
private final static Logger logger = LoggerFactory.getLogger(OfflineEventListener.class);
@Autowired
private IVideoManagerStorage storager;
@Autowired
private VideoStreamSessionManager streamSession;
@Autowired
private RedisUtil redis;
@Autowired
private UserSetting userSetting;
@Autowired
private EventPublisher eventPublisher;
@Autowired
private IMediaServerService mediaServerService;
@Autowired
private ZLMRTPServerFactory zlmrtpServerFactory;
@Override
public void onApplicationEvent(OfflineEvent event) {
logger.info("设备离线事件触发，deviceId：" + event.getDeviceId() + ",from:" + event.getFrom());
String key = VideoManagerConstants.KEEPLIVEKEY_PREFIX + userSetting.getServerId() + "_" + event.getDeviceId();
switch (event.getFrom()) {
// 心跳超时触发的离线事件，说明redis中已删除，无需处理
case VideoManagerConstants.EVENT_OUTLINE_TIMEOUT:
break;
// 设备主动注销触发的离线事件，需要删除redis中的超时监听
case VideoManagerConstants.EVENT_OUTLINE_UNREGISTER:
redis.del(key);
break;
default:
boolean exist = redis.hasKey(key);
if (exist) {
redis.del(key);
}
}
List<DeviceChannel> deviceChannelList = storager.queryOnlineChannelsByDeviceId(event.getDeviceId());
eventPublisher.catalogEventPublish(null, deviceChannelList, CatalogEvent.OFF);
// 处理离线监听
storager.outline(event.getDeviceId());
// TODO 离线取消订阅
// 离线释放所有ssrc
List<SsrcTransaction> ssrcTransactions = streamSession.getSsrcTransactionForAll(event.getDeviceId(), null, null, null);
if (ssrcTransactions != null && ssrcTransactions.size() > 0) {
for (SsrcTransaction ssrcTransaction : ssrcTransactions) {
mediaServerService.releaseSsrc(ssrcTransaction.getMediaServerId(), ssrcTransaction.getSsrc());
mediaServerService.closeRTPServer(event.getDeviceId(), ssrcTransaction.getChannelId(), ssrcTransaction.getStream());
streamSession.remove(event.getDeviceId(), ssrcTransaction.getChannelId(), ssrcTransaction.getStream());
}
}
}
}

51
src/main/java/com/genersoft/iot/vmp/gb28181/event/online/OnlineEvent.java
View File

@@ -1,51 +0,0 @@
package com.genersoft.iot.vmp.gb28181.event.online;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import org.springframework.context.ApplicationEvent;
/**
* @description: 在线事件类
* @author: swwheihei
* @date: 2020年5月6日 上午11:32:56
*/
public class OnlineEvent extends ApplicationEvent {
/**
*
*/
private static final long serialVersionUID = 1L;
public OnlineEvent(Object source) {
super(source);
}
private Device device;
private String from;
private int expires;
public Device getDevice() {
return device;
}
public void setDevice(Device device) {
this.device = device;
}
public String getFrom() {
return from;
}
public void setFrom(String from) {
this.from = from;
}
public int getExpires() {
return expires;
}
public void setExpires(int expires) {
this.expires = expires;
}
}

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

@@ -1,110 +0,0 @@
package com.genersoft.iot.vmp.gb28181.event.online;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.service.IDeviceService;
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 com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.utils.redis.RedisUtil;
import java.text.SimpleDateFormat;
import java.util.List;
/**
* @description: 在线事件监听器，监听到离线后，修改设备离在线状态。 设备在线有两个来源：
* 1、设备主动注销，发送注销指令
* 2、设备未知原因离线，心跳超时
* @author: swwheihei
* @date: 2020年5月6日 下午1:51:23
*/
@Component
public class OnlineEventListener implements ApplicationListener<OnlineEvent> {
private final static Logger logger = LoggerFactory.getLogger(OnlineEventListener.class);
@Autowired
private IVideoManagerStorage storager;
@Autowired
private IDeviceService deviceService;
@Autowired
private RedisUtil redis;
@Autowired
private SipConfig sipConfig;
@Autowired
private UserSetting userSetting;
@Autowired
private EventPublisher eventPublisher;
@Autowired
private SIPCommander cmder;
private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
@Override
public void onApplicationEvent(OnlineEvent event) {
logger.info("设备上线事件触发，deviceId：" + event.getDevice().getDeviceId() + ",from:" + event.getFrom());
Device device = event.getDevice();
if (device == null) {
return;
}
String key = VideoManagerConstants.KEEPLIVEKEY_PREFIX + userSetting.getServerId() + "_" + event.getDevice().getDeviceId();
Device deviceInStore = storager.queryVideoDevice(device.getDeviceId());
device.setOnline(1);
switch (event.getFrom()) {
// 注册时触发的在线事件，先在redis中增加超时超时监听
case VideoManagerConstants.EVENT_ONLINE_REGISTER:
// 超时时间
redis.set(key, event.getDevice().getDeviceId(), sipConfig.getKeepaliveTimeOut());
device.setRegisterTime(format.format(System.currentTimeMillis()));
if (deviceInStore == null) { //第一次上线
logger.info("[{}] 首次注册，查询设备信息以及通道信息", device.getDeviceId());
cmder.deviceInfoQuery(device);
deviceService.sync(device);
}
break;
// 设备主动发送心跳触发的在线事件
case VideoManagerConstants.EVENT_ONLINE_KEEPLIVE:
boolean exist = redis.hasKey(key);
// 先判断是否还存在，当设备先心跳超时后又发送心跳时，redis没有监听，需要增加
if (!exist) {
redis.set(key, event.getDevice().getDeviceId(), sipConfig.getKeepaliveTimeOut());
} else {
redis.expire(key, sipConfig.getKeepaliveTimeOut());
}
device.setKeepaliveTime(format.format(System.currentTimeMillis()));
break;
// 设备主动发送消息触发的在线事件
case VideoManagerConstants.EVENT_ONLINE_MESSAGE:
break;
}
// 处理上线监听
storager.updateDevice(device);
// 上线添加订阅
if (device.getSubscribeCycleForCatalog() > 0) {
// 查询在线设备那些开启了订阅，为设备开启定时的目录订阅
deviceService.addCatalogSubscribe(device);
}
if (device.getSubscribeCycleForMobilePosition() > 0) {
deviceService.addMobilePositionSubscribe(device);
}
}
}

51
src/main/java/com/genersoft/iot/vmp/gb28181/task/SipDeviceRunner.java Normal file
View File

@@ -0,0 +1,51 @@
package com.genersoft.iot.vmp.gb28181.task;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.service.IDeviceService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
import java.util.ArrayList;
import java.util.List;
/**
* 系统启动时控制设备
* @author lin
*/
@Component
@Order(value=4)
public class SipDeviceRunner implements CommandLineRunner {
@Autowired
private IVideoManagerStorage storager;
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Autowired
private UserSetting userSetting;
@Autowired
private IDeviceService deviceService;
@Override
public void run(String... args) throws Exception {
List<Device> deviceList = deviceService.getAllOnlineDevice();
for (Device device : deviceList) {
if (deviceService.expire(device)){
deviceService.offline(device.getDeviceId());
}else {
deviceService.online(device);
}
}
// 重置cseq计数
redisCatchStorage.resetAllCSEQ();
}
}

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

@@ -1,7 +1,10 @@
package com.genersoft.iot.vmp.gb28181.transmit;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.RegisterRequestProcessor;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.notify.cmd.KeepaliveNotifyMessageHandler;
import com.genersoft.iot.vmp.gb28181.transmit.event.response.ISIPResponseProcessor;
import com.genersoft.iot.vmp.gb28181.transmit.event.timeout.ITimeoutProcessor;
import org.slf4j.Logger;
@@ -11,10 +14,13 @@ import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import javax.sip.*;
import javax.sip.header.CSeqHeader;
import javax.sip.header.CallIdHeader;
import javax.sip.address.SipURI;
import javax.sip.address.URI;
import javax.sip.header.*;
import javax.sip.message.Request;
import javax.sip.message.Response;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
/**
@@ -34,10 +40,11 @@ public class SIPProcessorObserver implements ISIPProcessorObserver {
@Autowired
private SipSubscribe sipSubscribe;
@Autowired
private EventPublisher eventPublisher;
// @Autowired
// @Qualifier(value = "taskExecutor")
// private ThreadPoolTaskExecutor poolTaskExecutor;
/**
* 添加 request订阅
@@ -141,9 +148,32 @@ public class SIPProcessorObserver implements ISIPProcessorObserver {
*/
@Override
public void processTimeout(TimeoutEvent timeoutEvent) {
if(timeoutProcessor != null) {
timeoutProcessor.process(timeoutEvent);
logger.info("[消息发送超时]");
ClientTransaction clientTransaction = timeoutEvent.getClientTransaction();
eventPublisher.requestTimeOut(timeoutEvent);
if (clientTransaction != null) {
Request request = clientTransaction.getRequest();
if (request != null) {
CallIdHeader callIdHeader = (CallIdHeader) request.getHeader(CallIdHeader.NAME);
if (callIdHeader != null) {
SipSubscribe.Event subscribe = sipSubscribe.getErrorSubscribe(callIdHeader.getCallId());
SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(timeoutEvent);
subscribe.response(eventResult);
sipSubscribe.removeErrorSubscribe(callIdHeader.getCallId());
}
}
}
// Timeout timeout = timeoutEvent.getTimeout();
// ServerTransaction serverTransaction = timeoutEvent.getServerTransaction();
// if (serverTransaction != null) {
// Request request = serverTransaction.getRequest();
// URI requestURI = request.getRequestURI();
// Header header = request.getHeader(FromHeader.NAME);
// }
// if(timeoutProcessor != null) {
// timeoutProcessor.process(timeoutEvent);
// }
}
@Override

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

@@ -10,7 +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.DateUtil;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.gb28181.utils.NumericUtil;
import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
@@ -27,7 +27,6 @@ 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.boot.SpringBootVersion;
import org.springframework.context.annotation.DependsOn;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

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

@@ -4,7 +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.gb28181.utils.DateUtil;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;

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

@@ -23,6 +23,7 @@ import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
import com.genersoft.iot.vmp.service.bean.SSRCInfo;
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.utils.SerializeUtils;
import gov.nist.javax.sdp.TimeDescriptionImpl;
import gov.nist.javax.sdp.fields.TimeField;
@@ -39,7 +40,6 @@ import javax.sip.header.CallIdHeader;
import javax.sip.message.Request;
import javax.sip.message.Response;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Vector;
@@ -335,9 +335,8 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
sendRtpItem.setStreamId(ssrcInfo.getStream());
// 写入redis， 超时时回复
redisCatchStorage.updateSendRTPSever(sendRtpItem);
SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
playService.playBack(mediaServerItem, ssrcInfo, device.getDeviceId(), channelId, format.format(start),
format.format(end), null, result -> {
playService.playBack(mediaServerItem, ssrcInfo, device.getDeviceId(), channelId, DateUtil.format.format(start),
DateUtil.format.format(end), null, result -> {
if (result.getCode() != 0){
logger.warn("录像回放失败");
if (result.getEvent() != null) {

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

@@ -252,14 +252,12 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
FromHeader fromHeader = (FromHeader) evt.getRequest().getHeader(FromHeader.NAME);
String deviceId = SipUtils.getUserIdFromFromHeader(fromHeader);
Element rootElement = getRootElement(evt);
Device device = redisCatchStorage.getDevice(deviceId);
if (device == null) {
if (device == null || device.getOnline() == 0) {
logger.warn("[收到 目录订阅]：{}, 但是设备已经离线", (device != null ? device.getDeviceId():"" ));
return;
}
if (device != null ) {
rootElement = getRootElement(evt, device.getCharset());
}
Element rootElement = getRootElement(evt, device.getCharset());
Element deviceListElement = rootElement.element("DeviceList");
if (deviceListElement == null) {
return;
@@ -279,39 +277,46 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
channel.setDeviceId(device.getDeviceId());
logger.info("[收到 目录订阅]：{}/{}", device.getDeviceId(), channel.getChannelId());
switch (eventElement.getText().toUpperCase()) {
case CatalogEvent.ON: // 上线
case CatalogEvent.ON:
// 上线
logger.info("收到来自设备【{}】的通道【{}】上线通知", device.getDeviceId(), channel.getChannelId());
storager.deviceChannelOnline(deviceId, channel.getChannelId());
// 回复200 OK
responseAck(evt, Response.OK);
break;
case CatalogEvent.OFF : // 离线
case CatalogEvent.OFF :
// 离线
logger.info("收到来自设备【{}】的通道【{}】离线通知", device.getDeviceId(), channel.getChannelId());
storager.deviceChannelOffline(deviceId, channel.getChannelId());
// 回复200 OK
responseAck(evt, Response.OK);
break;
case CatalogEvent.VLOST: // 视频丢失
case CatalogEvent.VLOST:
// 视频丢失
logger.info("收到来自设备【{}】的通道【{}】视频丢失通知", device.getDeviceId(), channel.getChannelId());
storager.deviceChannelOffline(deviceId, channel.getChannelId());
// 回复200 OK
responseAck(evt, Response.OK);
break;
case CatalogEvent.DEFECT: // 故障
case CatalogEvent.DEFECT:
// 故障
// 回复200 OK
responseAck(evt, Response.OK);
break;
case CatalogEvent.ADD: // 增加
case CatalogEvent.ADD:
// 增加
logger.info("收到来自设备【{}】的增加通道【{}】通知", device.getDeviceId(), channel.getChannelId());
storager.updateChannel(deviceId, channel);
responseAck(evt, Response.OK);
break;
case CatalogEvent.DEL: // 删除
case CatalogEvent.DEL:
// 删除
logger.info("收到来自设备【{}】的删除通道【{}】通知", device.getDeviceId(), channel.getChannelId());
storager.delChannel(deviceId, channel.getChannelId());
responseAck(evt, Response.OK);
break;
case CatalogEvent.UPDATE: // 更新
case CatalogEvent.UPDATE:
// 更新
logger.info("收到来自设备【{}】的更新通道【{}】通知", device.getDeviceId(), channel.getChannelId());
storager.updateChannel(deviceId, channel);
responseAck(evt, Response.OK);
@@ -324,10 +329,6 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
eventPublisher.catalogEventPublish(null, channel, eventElement.getText().toUpperCase());
}
if (!redisCatchStorage.deviceIsOnline(deviceId)) {
publisher.onlineEventPublish(device, VideoManagerConstants.EVENT_ONLINE_MESSAGE);
}
}
} catch (DocumentException | SipException | InvalidArgumentException | ParseException e) {
e.printStackTrace();

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

@@ -9,6 +9,7 @@ import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.service.IDeviceService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import gov.nist.javax.sip.RequestEventExt;
@@ -60,6 +61,9 @@ public class RegisterRequestProcessor extends SIPRequestProcessorParent implemen
@Autowired
private SIPProcessorObserver sipProcessorObserver;
@Autowired
private IDeviceService deviceService;
@Override
public void afterPropertiesSet() throws Exception {
// 添加消息处理的订阅
@@ -82,7 +86,7 @@ public class RegisterRequestProcessor extends SIPRequestProcessorParent implemen
Response response = null;
boolean passwordCorrect = false;
// 注册标志 0：未携带授权头或者密码错误 1：注册成功 2：注销成功
int registerFlag = 0;
boolean registerFlag = false;
FromHeader fromHeader = (FromHeader) request.getHeader(FromHeader.NAME);
AddressImpl address = (AddressImpl) fromHeader.getAddress();
SipUri uri = (SipUri) address.getURI();
@@ -111,12 +115,8 @@ public class RegisterRequestProcessor extends SIPRequestProcessorParent implemen
return;
}
Device deviceInRedis = redisCatchStorage.getDevice(deviceId);
Device device = storager.queryVideoDevice(deviceId);
if (deviceInRedis != null && device == null) {
// redis 存在脏数据
redisCatchStorage.clearCatchByDeviceId(deviceId);
}
Device device = deviceService.queryDevice(deviceId);
// 携带授权头并且密码正确
response = getMessageFactory().createResponse(Response.OK, request);
// 添加date头
@@ -154,20 +154,17 @@ public class RegisterRequestProcessor extends SIPRequestProcessorParent implemen
device.setStreamMode("UDP");
device.setCharset("GB2312");
device.setDeviceId(deviceId);
device.setFirsRegister(true);
} else {
device.setFirsRegister(device.getOnline() == 0);
}
device.setIp(received);
device.setPort(rPort);
device.setHostAddress(received.concat(":").concat(String.valueOf(rPort)));
if (expiresHeader.getExpires() == 0) {
// 注销成功
registerFlag = 2;
registerFlag = false;
} else {
// 注册成功
device.setExpires(expiresHeader.getExpires());
registerFlag = 1;
registerFlag = true;
// 判断TCP还是UDP
ViaHeader reqViaHeader = (ViaHeader) request.getHeader(ViaHeader.NAME);
String transport = reqViaHeader.getTransport();
@@ -177,12 +174,12 @@ public class RegisterRequestProcessor extends SIPRequestProcessorParent implemen
sendResponse(evt, response);
// 注册成功
// 保存到redis
if (registerFlag == 1) {
if (registerFlag) {
logger.info("[{}] 注册成功! deviceId:" + deviceId, requestAddress);
publisher.onlineEventPublish(device, VideoManagerConstants.EVENT_ONLINE_REGISTER, expiresHeader.getExpires());
} else if (registerFlag == 2) {
deviceService.online(device);
} else {
logger.info("[{}] 注销成功! deviceId:" + deviceId, requestAddress);
publisher.outlineEventPublish(deviceId, VideoManagerConstants.EVENT_OUTLINE_UNREGISTER);
deviceService.offline(deviceId);
}
} catch (SipException | InvalidArgumentException | NoSuchAlgorithmException | ParseException e) {
e.printStackTrace();

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

@@ -1,14 +1,15 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.notify.cmd;
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.IMessageHandler;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.notify.NotifyMessageHandler;
import com.genersoft.iot.vmp.service.IDeviceService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.utils.DateUtil;
import org.dom4j.Element;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -23,24 +24,20 @@ import javax.sip.SipException;
import javax.sip.header.ViaHeader;
import javax.sip.message.Response;
import java.text.ParseException;
import java.util.Calendar;
import java.util.Date;
@Component
public class KeepaliveNotifyMessageHandler extends SIPRequestProcessorParent implements InitializingBean, IMessageHandler {
private Logger logger = LoggerFactory.getLogger(KeepaliveNotifyMessageHandler.class);
private final String cmdType = "Keepalive";
private final static String cmdType = "Keepalive";
@Autowired
private NotifyMessageHandler notifyMessageHandler;
@Autowired
private EventPublisher publisher;
@Autowired
private IVideoManagerStorage videoManagerStorager;
@Autowired
private IRedisCatchStorage redisCatchStorage;
private IDeviceService deviceService;
@Override
public void afterPropertiesSet() throws Exception {
@@ -49,29 +46,35 @@ public class KeepaliveNotifyMessageHandler extends SIPRequestProcessorParent imp
@Override
public void handForDevice(RequestEvent evt, Device device, Element element) {
// 检查设备是否存在并在线， 不在线则设置为在线
if (device == null) {
// 未注册的设备不做处理
return;
}
try {
if (device != null ) {
if (device.getOnline() == 1) {
// 回复200 OK
responseAck(evt, Response.OK);
// 判断RPort是否改变，改变则说明路由nat信息变化，修改设备信息
// 获取到通信地址等信息
ViaHeader viaHeader = (ViaHeader) evt.getRequest().getHeader(ViaHeader.NAME);
String received = viaHeader.getReceived();
int rPort = viaHeader.getRPort();
// 解析本地地址替代
if (StringUtils.isEmpty(received) || rPort == -1) {
received = viaHeader.getHost();
rPort = viaHeader.getPort();
}
if (device.getPort() != rPort) {
device.setPort(rPort);
device.setHostAddress(received.concat(":").concat(String.valueOf(rPort)));
videoManagerStorager.updateDevice(device);
redisCatchStorage.updateDevice(device);
}
if (!redisCatchStorage.deviceIsOnline(device.getDeviceId())) {
publisher.onlineEventPublish(device, VideoManagerConstants.EVENT_ONLINE_KEEPLIVE);
}else {
// 对于已经离线的设备判断他的注册是否已经过期
if (!deviceService.expire(device)){
device.setKeepaliveTime(DateUtil.getNow());
// 判断RPort是否改变，改变则说明路由nat信息变化，修改设备信息
// 获取到通信地址等信息
ViaHeader viaHeader = (ViaHeader) evt.getRequest().getHeader(ViaHeader.NAME);
String received = viaHeader.getReceived();
int rPort = viaHeader.getRPort();
// 解析本地地址替代
if (StringUtils.isEmpty(received) || rPort == -1) {
received = viaHeader.getHost();
rPort = viaHeader.getPort();
}
if (device.getPort() != rPort) {
device.setPort(rPort);
device.setHostAddress(received.concat(":").concat(String.valueOf(rPort)));
}
deviceService.online(device);
// 回复200 OK
responseAck(evt, Response.OK);
}
}
} catch (SipException e) {

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

@@ -9,7 +9,7 @@ import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.IMessageHandler;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.query.QueryMessageHandler;
import com.genersoft.iot.vmp.gb28181.utils.DateUtil;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.storager.dao.dto.ChannelSourceInfo;
import org.dom4j.Element;

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

@@ -28,7 +28,6 @@ import javax.sip.RequestEvent;
import javax.sip.SipException;
import javax.sip.message.Response;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
@@ -39,8 +38,6 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp
private Logger logger = LoggerFactory.getLogger(CatalogResponseMessageHandler.class);
private final String cmdType = "Catalog";
private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
@Autowired
private ResponseMessageHandler responseMessageHandler;

11
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/DeviceInfoResponseMessageHandler.java
View File

@@ -29,6 +29,9 @@ import java.text.ParseException;
import static com.genersoft.iot.vmp.gb28181.utils.XmlUtil.getText;
/**
* @author lin
*/
@Component
public class DeviceInfoResponseMessageHandler extends SIPRequestProcessorParent implements InitializingBean, IMessageHandler {
@@ -61,6 +64,11 @@ public class DeviceInfoResponseMessageHandler extends SIPRequestProcessorParent
@Override
public void handForDevice(RequestEvent evt, Device device, Element rootElement) {
logger.debug("接收到DeviceInfo应答消息");
// 检查设备是否存在， 不存在则不回复
if (device == null || device.getOnline() == 0) {
logger.warn("[接收到DeviceInfo应答消息,但是设备已经离线]：" + (device != null ? device.getDeviceId():"" ));
return;
}
try {
rootElement = getRootElement(evt, device.getCharset());
Element deviceIdElement = rootElement.element("DeviceID");
@@ -82,9 +90,6 @@ public class DeviceInfoResponseMessageHandler extends SIPRequestProcessorParent
deferredResultHolder.invokeAllResult(msg);
// 回复200 OK
responseAck(evt, Response.OK);
if (redisCatchStorage.deviceIsOnline(device.getDeviceId())) {
publisher.onlineEventPublish(device, VideoManagerConstants.EVENT_ONLINE_MESSAGE);
}
} catch (DocumentException e) {
e.printStackTrace();
} catch (InvalidArgumentException e) {

19
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/DeviceStatusResponseMessageHandler.java
View File

@@ -11,6 +11,7 @@ import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorP
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.IMessageHandler;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.ResponseMessageHandler;
import com.genersoft.iot.vmp.gb28181.utils.XmlUtil;
import com.genersoft.iot.vmp.service.IDeviceService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import org.dom4j.Element;
import org.slf4j.Logger;
@@ -24,6 +25,7 @@ import javax.sip.RequestEvent;
import javax.sip.SipException;
import javax.sip.message.Response;
import java.text.ParseException;
import java.util.Objects;
@Component
public class DeviceStatusResponseMessageHandler extends SIPRequestProcessorParent implements InitializingBean, IMessageHandler {
@@ -34,12 +36,11 @@ public class DeviceStatusResponseMessageHandler extends SIPRequestProcessorParen
@Autowired
private ResponseMessageHandler responseMessageHandler;
@Autowired
private DeferredResultHolder deferredResultHolder;
@Autowired
private EventPublisher publisher;
private IDeviceService deviceService;
@Autowired
private IRedisCatchStorage redisCatchStorage;
@@ -53,6 +54,9 @@ public class DeviceStatusResponseMessageHandler extends SIPRequestProcessorParen
public void handForDevice(RequestEvent evt, Device device, Element element) {
logger.info("接收到DeviceStatus应答消息");
// 检查设备是否存在， 不存在则不回复
if (device == null) {
return;
}
// 回复200 OK
try {
responseAck(evt, Response.OK);
@@ -64,20 +68,23 @@ public class DeviceStatusResponseMessageHandler extends SIPRequestProcessorParen
e.printStackTrace();
}
Element deviceIdElement = element.element("DeviceID");
Element onlineElement = element.element("Online");
String channelId = deviceIdElement.getText();
JSONObject json = new JSONObject();
XmlUtil.node2Json(element, json);
if (logger.isDebugEnabled()) {
logger.debug(json.toJSONString());
}
String text = onlineElement.getText();
if (Objects.equals(text.trim().toUpperCase(), "ONLINE")) {
deviceService.online(device);
}else {
deviceService.offline(device.getDeviceId());
}
RequestMessage msg = new RequestMessage();
msg.setKey(DeferredResultHolder.CALLBACK_CMD_DEVICESTATUS + device.getDeviceId() + channelId);
msg.setData(json);
deferredResultHolder.invokeAllResult(msg);
if (redisCatchStorage.deviceIsOnline(device.getDeviceId())) {
publisher.onlineEventPublish(device, VideoManagerConstants.EVENT_ONLINE_MESSAGE);
}
}
@Override

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

@@ -20,7 +20,6 @@ import javax.sip.RequestEvent;
import javax.sip.SipException;
import javax.sip.message.Response;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
@@ -33,8 +32,6 @@ public class PresetQueryResponseMessageHandler extends SIPRequestProcessorParent
private Logger logger = LoggerFactory.getLogger(PresetQueryResponseMessageHandler.class);
private final String cmdType = "PresetQuery";
private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
@Autowired
private ResponseMessageHandler responseMessageHandler;

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

@@ -11,7 +11,7 @@ import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.IMessageHandler;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.ResponseMessageHandler;
import com.genersoft.iot.vmp.gb28181.utils.DateUtil;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.utils.redis.RedisUtil;
import org.dom4j.DocumentException;
import org.dom4j.Element;

57
src/main/java/com/genersoft/iot/vmp/gb28181/utils/DateUtil.java
View File

@@ -1,57 +0,0 @@
package com.genersoft.iot.vmp.gb28181.utils;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Locale;
/**
* @description:时间工具类，主要处理ISO 8601格式转换
* @author: swwheihei
* @date: 2020年5月8日 下午3:24:42
*/
public class DateUtil {
//private static final String yyyy_MM_dd_T_HH_mm_ss_SSSXXX = "yyyy-MM-dd'T'HH:mm:ss.SSSXXX";
private static final String yyyy_MM_dd_T_HH_mm_ss_SSSXXX = "yyyy-MM-dd'T'HH:mm:ss";
private static final String yyyy_MM_dd_HH_mm_ss = "yyyy-MM-dd HH:mm:ss";
public static String yyyy_MM_dd_HH_mm_ssToISO8601(String formatTime) {
SimpleDateFormat oldsdf = new SimpleDateFormat(yyyy_MM_dd_HH_mm_ss, Locale.getDefault());
SimpleDateFormat newsdf = new SimpleDateFormat(yyyy_MM_dd_T_HH_mm_ss_SSSXXX, Locale.getDefault());
try {
return newsdf.format(oldsdf.parse(formatTime));
} catch (ParseException e) {
e.printStackTrace();
}
return "";
}
public static String ISO8601Toyyyy_MM_dd_HH_mm_ss(String formatTime) {
SimpleDateFormat oldsdf = new SimpleDateFormat(yyyy_MM_dd_T_HH_mm_ss_SSSXXX, Locale.getDefault());
SimpleDateFormat newsdf = new SimpleDateFormat(yyyy_MM_dd_HH_mm_ss, Locale.getDefault());
try {
return newsdf.format(oldsdf.parse(formatTime));
} catch (ParseException e) {
e.printStackTrace();
}
return "";
}
public static long yyyy_MM_dd_HH_mm_ssToTimestamp(String formatTime) {
SimpleDateFormat format=new SimpleDateFormat(yyyy_MM_dd_HH_mm_ss);
//设置要读取的时间字符串格式
Date date;
try {
date = format.parse(formatTime);
Long timestamp=date.getTime()/1000;
//转换为Date类
return timestamp;
} catch (ParseException e) {
e.printStackTrace();
}
return 0;
}
}