roouzenn/wvp
Template
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

临时提交

Browse Source
This commit is contained in:
648540858
2024-06-25 16:45:31 +08:00
parent 306c42b4b7
commit 68fce9177f
14 changed files with 654 additions and 675 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -3,6 +3,7 @@ package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.CatalogChannelEvent;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.HandlerCatchData;
@@ -10,14 +11,12 @@ 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.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.gb28181.utils.XmlUtil;
import com.genersoft.iot.vmp.service.IDeviceChannelService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.utils.DateUtil;
import lombok.extern.slf4j.Slf4j;
import org.dom4j.DocumentException;
import org.dom4j.Element;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
@@ -25,6 +24,8 @@ import org.springframework.transaction.annotation.Transactional;
import javax.sip.RequestEvent;
import javax.sip.header.FromHeader;
import java.lang.reflect.InvocationTargetException;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
@@ -36,12 +37,10 @@ import java.util.concurrent.CopyOnWriteArrayList;
/**
* SIP命令类型： NOTIFY请求中的目录请求处理
*/
@Slf4j
@Component
public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent {
private final static Logger logger = LoggerFactory.getLogger(NotifyRequestForCatalogProcessor.class);
private final List<DeviceChannel> updateChannelOnlineList = new CopyOnWriteArrayList<>();
private final List<DeviceChannel> updateChannelOfflineList = new CopyOnWriteArrayList<>();
private final Map<String, DeviceChannel> updateChannelMap = new ConcurrentHashMap<>();
@@ -72,7 +71,7 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
@Transactional
public void process(RequestEvent evt) {
if (taskQueue.size() >= userSetting.getMaxNotifyCountQueue()) {
logger.error("[notify-目录订阅] 待处理消息队列已满 {}，返回486 BUSY_HERE，消息不做处理", userSetting.getMaxNotifyCountQueue());
log.error("[notify-目录订阅] 待处理消息队列已满 {}，返回486 BUSY_HERE，消息不做处理", userSetting.getMaxNotifyCountQueue());
return;
}
taskQueue.offer(new HandlerCatchData(evt, null, null));
@@ -95,12 +94,12 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
Device device = redisCatchStorage.getDevice(deviceId);
if (device == null || !device.isOnLine()) {
logger.warn("[收到目录订阅]：{}, 但是设备已经离线", (device != null ? device.getDeviceId() : ""));
log.warn("[收到目录订阅]：{}, 但是设备已经离线", (device != null ? device.getDeviceId() : ""));
return;
}
Element rootElement = getRootElement(evt, device.getCharset());
if (rootElement == null) {
logger.warn("[ 收到目录订阅 ] content cannot be null, {}", evt.getRequest());
log.warn("[ 收到目录订阅 ] content cannot be null, {}", evt.getRequest());
return;
}
Element deviceListElement = rootElement.element("DeviceList");
@@ -113,132 +112,134 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
// 遍历DeviceList
while (deviceListIterator.hasNext()) {
Element itemDevice = deviceListIterator.next();
Element eventElement = itemDevice.element("Event");
String event;
if (eventElement == null) {
logger.warn("[收到目录订阅]：{}, 但是Event为空, 设为默认值 ADD", (device != null ? device.getDeviceId() : ""));
event = CatalogEvent.ADD;
} else {
event = eventElement.getText().toUpperCase();
}
DeviceChannel channel = XmlUtil.channelContentHandler(itemDevice, device, event);
if (channel == null) {
logger.info("[收到目录订阅]：但是解析失败 {}", new String(evt.getRequest().getRawContent()));
CatalogChannelEvent catalogChannelEvent = null;
try {
catalogChannelEvent = CatalogChannelEvent.decode(itemDevice);
if (catalogChannelEvent.getChannel() == null) {
log.info("[解析CatalogChannelEvent]成功：但是解析通道信息失败， 原文如下： \n{}", new String(evt.getRequest().getRawContent()));
continue;
}
} catch (InvocationTargetException | NoSuchMethodException | InstantiationException |
IllegalAccessException e) {
log.error("[解析CatalogChannelEvent]失败，", e);
log.error("[解析CatalogChannelEvent]失败原文: \n{}", new String(evt.getRequest().getRawContent(), Charset.forName(device.getCharset())));
continue;
}
if (catalogChannelEvent == null) {
continue;
}
if (channel.getParentId() != null && channel.getParentId().equals(sipConfig.getId())) {
channel.setParentId(null);
}
channel.setDeviceId(device.getDeviceId());
logger.info("[收到目录订阅]：{}/{}", device.getDeviceId(), channel.getDeviceId());
switch (event) {
log.info("[收到目录订阅]：{}/{}-{}", device.getDeviceId(),
catalogChannelEvent.getChannel().getDeviceId(), catalogChannelEvent.getEvent());
switch (catalogChannelEvent.getEvent()) {
case CatalogEvent.ON:
// 上线
logger.info("[收到通道上线通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
updateChannelOnlineList.add(channel);
log.info("[收到通道上线通知] 来自设备: {}, 通道 {}", device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId());
updateChannelOnlineList.add(catalogChannelEvent.getChannel());
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), channel.getChannelId(), true);
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId(), true);
}
break;
case CatalogEvent.OFF:
// 离线
logger.info("[收到通道离线通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
log.info("[收到通道离线通知] 来自设备: {}, 通道 {}", device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId());
if (userSetting.getRefuseChannelStatusChannelFormNotify()) {
logger.info("[收到通道离线通知] 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
log.info("[收到通道离线通知] 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId());
} else {
updateChannelOfflineList.add(channel);
updateChannelOfflineList.add(catalogChannelEvent.getChannel());
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), channel.getChannelId(), false);
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId(), false);
}
}
break;
case CatalogEvent.VLOST:
// 视频丢失
logger.info("[收到通道视频丢失通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
log.info("[收到通道视频丢失通知] 来自设备: {}, 通道 {}", device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId());
if (userSetting.getRefuseChannelStatusChannelFormNotify()) {
logger.info("[收到通道视频丢失通知] 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
log.info("[收到通道视频丢失通知] 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId());
} else {
updateChannelOfflineList.add(channel);
updateChannelOfflineList.add(catalogChannelEvent.getChannel());
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), channel.getChannelId(), false);
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId(), false);
}
}
break;
case CatalogEvent.DEFECT:
// 故障
logger.info("[收到通道视频故障通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
log.info("[收到通道视频故障通知] 来自设备: {}, 通道 {}", device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId());
if (userSetting.getRefuseChannelStatusChannelFormNotify()) {
logger.info("[收到通道视频故障通知] 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
log.info("[收到通道视频故障通知] 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId());
} else {
updateChannelOfflineList.add(channel);
updateChannelOfflineList.add(catalogChannelEvent.getChannel());
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), channel.getChannelId(), false);
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId(), false);
}
}
break;
case CatalogEvent.ADD:
// 增加
logger.info("[收到增加通道通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
log.info("[收到增加通道通知] 来自设备: {}, 通道 {}", device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId());
// 判断此通道是否存在
DeviceChannel deviceChannel = deviceChannelService.getOne(deviceId, channel.getChannelId());
DeviceChannel deviceChannel = deviceChannelService.getOne(deviceId, catalogChannelEvent.getChannel().getDeviceId());
if (deviceChannel != null) {
logger.info("[增加通道] 已存在，不发送通知只更新，设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
log.info("[增加通道] 已存在，不发送通知只更新，设备: {}, 通道 {}", device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId());
DeviceChannel channel = catalogChannelEvent.getChannel();
channel.setId(deviceChannel.getId());
channel.setHasAudio(null);
updateChannelMap.put(channel.getChannelId(), channel);
channel.setHasAudio(deviceChannel.getHasAudio());
channel.setUpdateTime(DateUtil.getNow());
updateChannelMap.put(catalogChannelEvent.getChannel().getDeviceId(), channel);
} else {
addChannelMap.put(channel.getChannelId(), channel);
addChannelMap.put(catalogChannelEvent.getChannel().getDeviceId(), catalogChannelEvent.getChannel());
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendChannelAddOrDelete(device.getDeviceId(), channel.getChannelId(), true);
redisCatchStorage.sendChannelAddOrDelete(device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId(), true);
}
}
break;
case CatalogEvent.DEL:
// 删除
logger.info("[收到删除通道通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
deleteChannelList.add(channel);
log.info("[收到删除通道通知] 来自设备: {}, 通道 {}", device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId());
deleteChannelList.add(catalogChannelEvent.getChannel());
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendChannelAddOrDelete(device.getDeviceId(), channel.getChannelId(), false);
redisCatchStorage.sendChannelAddOrDelete(device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId(), false);
}
break;
case CatalogEvent.UPDATE:
// 更新
logger.info("[收到更新通道通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
log.info("[收到更新通道通知] 来自设备: {}, 通道 {}", device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId());
// 判断此通道是否存在
DeviceChannel deviceChannelForUpdate = deviceChannelService.getOne(deviceId, channel.getChannelId());
DeviceChannel deviceChannelForUpdate = deviceChannelService.getOne(deviceId, catalogChannelEvent.getChannel().getDeviceId());
if (deviceChannelForUpdate != null) {
DeviceChannel channel = catalogChannelEvent.getChannel();
channel.setId(deviceChannelForUpdate.getId());
channel.setHasAudio(deviceChannelForUpdate.getHasAudio());
channel.setUpdateTime(DateUtil.getNow());
channel.setHasAudio(null);
updateChannelMap.put(channel.getChannelId(), channel);
updateChannelMap.put(catalogChannelEvent.getChannel().getDeviceId(), channel);
} else {
addChannelMap.put(channel.getChannelId(), channel);
addChannelMap.put(catalogChannelEvent.getChannel().getDeviceId(), catalogChannelEvent.getChannel());
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendChannelAddOrDelete(device.getDeviceId(), channel.getChannelId(), true);
redisCatchStorage.sendChannelAddOrDelete(device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId(), true);
}
}
break;
default:
logger.warn("[ NotifyCatalog ] event not found ： {}", event);
log.warn("[ NotifyCatalog ] event not found ： {}", catalogChannelEvent.getEvent());
}
// 转发变化信息
eventPublisher.catalogEventPublish(null, channel, event);
eventPublisher.catalogEventPublish(null, catalogChannelEvent.getChannel(), catalogChannelEvent.getEvent());
}
}
} catch (DocumentException e) {
logger.error("未处理的异常 ", e);
log.error("未处理的异常 ", e);
}
}
taskQueue.clear();
@@ -255,33 +256,33 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
try {
executeSaveForAdd();
} catch (Exception e) {
logger.error("[存储收到的增加通道] 异常： ", e );
log.error("[存储收到的增加通道] 异常： ", e );
}
try {
executeSaveForOnline();
} catch (Exception e) {
logger.error("[存储收到的通道上线] 异常： ", e );
log.error("[存储收到的通道上线] 异常： ", e );
}
try {
executeSaveForOffline();
} catch (Exception e) {
logger.error("[存储收到的通道离线] 异常： ", e );
log.error("[存储收到的通道离线] 异常： ", e );
}
try {
executeSaveForUpdate();
} catch (Exception e) {
logger.error("[存储收到的更新通道] 异常： ", e );
log.error("[存储收到的更新通道] 异常： ", e );
}
try {
executeSaveForDelete();
} catch (Exception e) {
logger.error("[存储收到的删除通道] 异常： ", e );
log.error("[存储收到的删除通道] 异常： ", e );
}
}
private void executeSaveForUpdate(){
if (!updateChannelMap.values().isEmpty()) {
logger.info("[存储收到的更新通道], 数量： {}", updateChannelMap.size());
log.info("[存储收到的更新通道], 数量： {}", updateChannelMap.size());
ArrayList<DeviceChannel> deviceChannels = new ArrayList<>(updateChannelMap.values());
deviceChannelService.batchUpdateChannel(deviceChannels);
updateChannelMap.clear();
@@ -319,6 +320,6 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
@Scheduled(fixedRate = 10000) //每1秒执行一次
public void execute(){
logger.info("[待处理Notify-目录订阅消息数量]: {}", taskQueue.size());
log.info("[待处理Notify-目录订阅消息数量]: {}", taskQueue.size());
}
}

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

@@ -7,7 +7,6 @@ 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.SipUtils;
import com.genersoft.iot.vmp.gb28181.utils.XmlUtil;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import gov.nist.javax.sip.message.SIPRequest;
import org.dom4j.DocumentException;
@@ -114,8 +113,9 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp
if (channelDeviceElement == null) {
continue;
}
DeviceChannel channel = XmlUtil.channelContentHandler(itemDevice, device, null);
if (channel == null) {
DeviceChannel channel = DeviceChannel.decode(itemDevice);
// DeviceChannel channel = XmlUtil.channelContentHandler(itemDevice, device, null);
if (channel == null || channel.getDeviceId() == null) {
logger.info("[收到目录订阅]：但是解析失败 {}", new String(evt.getRequest().getRawContent()));
continue;
}