Thirdparty
/
wvp-GB28181-pro
mirror of https://github.com/648540858/wvp-GB28181-pro
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

优化notify消息处理

2.7.1
648540858 2024-10-11 14:09:18 +08:00
parent 2a2a3943f4
commit bbbadccb79
9 changed files with 375 additions and 281 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
src/main/java/com/genersoft/iot/vmp/gb28181/bean/NotifyCatalogChannel.java Normal file
View File

@ -0,0 +1,38 @@
package com.genersoft.iot.vmp.gb28181.bean;
public class NotifyCatalogChannel {
private Type type;
private DeviceChannel channel;
public enum Type {
ADD, DELETE, UPDATE, STATUS_CHANGED
}
public static NotifyCatalogChannel getInstance(Type type, DeviceChannel channel) {
NotifyCatalogChannel notifyCatalogChannel = new NotifyCatalogChannel();
notifyCatalogChannel.setType(type);
notifyCatalogChannel.setChannel(channel);
return notifyCatalogChannel;
}
public Type getType() {
return type;
}
public void setType(Type type) {
this.type = type;
}
public DeviceChannel getChannel() {
return channel;
}
public void setChannel(DeviceChannel channel) {
this.channel = channel;
}
}

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

@ -39,7 +39,7 @@ public class SIPRequestHeaderPlarformProvider {
@Autowired @Autowired
private SipConfig sipConfig; private SipConfig sipConfig;
@Autowired @Autowired
private SipLayer sipLayer; private SipLayer sipLayer;

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

@ -596,7 +596,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
String catalogXmlContent = getCatalogXmlContentForCatalogAddOrUpdate(parentPlatform, channels, String catalogXmlContent = getCatalogXmlContentForCatalogAddOrUpdate(parentPlatform, channels,
deviceChannels.size(), type, subscribeInfo); deviceChannels.size(), type, subscribeInfo);
System.out.println(catalogXmlContent); System.out.println(catalogXmlContent);
logger.info("[发送NOTIFY通知]类型： {}，发送数量： {}", type, channels.size()); logger.info("[发送NOTIFY通知]类型： {}，平台：{}， 发送数量： {}", type, parentPlatform.getServerGBId(), channels.size());
sendNotify(parentPlatform, catalogXmlContent, subscribeInfo, eventResult -> { sendNotify(parentPlatform, catalogXmlContent, subscribeInfo, eventResult -> {
logger.error("发送NOTIFY通知消息失败。错误：{} {}", eventResult.statusCode, eventResult.msg); logger.error("发送NOTIFY通知消息失败。错误：{} {}", eventResult.statusCode, eventResult.msg);
}, (eventResult -> { }, (eventResult -> {
@ -702,7 +702,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
}else { }else {
channels = deviceChannels.subList(index, deviceChannels.size()); channels = deviceChannels.subList(index, deviceChannels.size());
} }
logger.info("[发送NOTIFY通知]类型： {}，发送数量： {}", type, channels.size()); logger.info("[发送NOTIFY通知]类型： {}，平台：{}， 发送数量： {}", type, parentPlatform.getServerGBId(), channels.size());
Integer finalIndex = index; Integer finalIndex = index;
String catalogXmlContent = getCatalogXmlContentForCatalogOther(parentPlatform, channels, type); String catalogXmlContent = getCatalogXmlContentForCatalogOther(parentPlatform, channels, type);
sendNotify(parentPlatform, catalogXmlContent, subscribeInfo, eventResult -> { sendNotify(parentPlatform, catalogXmlContent, subscribeInfo, eventResult -> {

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

@ -5,6 +5,7 @@ import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.Device; import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel; import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.HandlerCatchData; import com.genersoft.iot.vmp.gb28181.bean.HandlerCatchData;
import com.genersoft.iot.vmp.gb28181.bean.NotifyCatalogChannel;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher; import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent; 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.transmit.event.request.SIPRequestProcessorParent;
@ -13,6 +14,7 @@ import com.genersoft.iot.vmp.gb28181.utils.XmlUtil;
import com.genersoft.iot.vmp.service.IDeviceChannelService; import com.genersoft.iot.vmp.service.IDeviceChannelService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.utils.DateUtil; import com.genersoft.iot.vmp.utils.DateUtil;
import gov.nist.javax.sip.message.SIPRequest;
import org.dom4j.DocumentException; import org.dom4j.DocumentException;
import org.dom4j.Element; import org.dom4j.Element;
import org.slf4j.Logger; import org.slf4j.Logger;
@ -23,12 +25,12 @@ import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.annotation.Transactional;
import javax.sip.RequestEvent; import javax.sip.RequestEvent;
import javax.sip.header.CSeqHeader;
import javax.sip.header.FromHeader; import javax.sip.header.FromHeader;
import javax.sip.message.Request;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Iterator; import java.util.Iterator;
import java.util.List; import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CopyOnWriteArrayList;
@ -38,281 +40,257 @@ import java.util.concurrent.CopyOnWriteArrayList;
@Component @Component
public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent { public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent {
private final static Logger logger = LoggerFactory.getLogger(NotifyRequestForCatalogProcessor.class); private final static Logger logger = LoggerFactory.getLogger(NotifyRequestForCatalogProcessor.class);
private final List<DeviceChannel> updateChannelForStatusChange = new CopyOnWriteArrayList<>(); private final ConcurrentLinkedQueue<NotifyCatalogChannel> channelList = new ConcurrentLinkedQueue<>();
// private final List<DeviceChannel> updateChannelOfflineList = new CopyOnWriteArrayList<>();
private final Map<String, DeviceChannel> updateChannelMap = new ConcurrentHashMap<>();
private final Map<String, DeviceChannel> addChannelMap = new ConcurrentHashMap<>(); private ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>();
private final List<DeviceChannel> deleteChannelList = new CopyOnWriteArrayList<>();
private ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>(); @Autowired
private UserSetting userSetting;
@Autowired @Autowired
private UserSetting userSetting; private EventPublisher eventPublisher;
@Autowired @Autowired
private EventPublisher eventPublisher; private IRedisCatchStorage redisCatchStorage;
@Autowired @Autowired
private IRedisCatchStorage redisCatchStorage; private IDeviceChannelService deviceChannelService;
@Autowired @Autowired
private IDeviceChannelService deviceChannelService; private SipConfig sipConfig;
@Autowired @Transactional
private SipConfig sipConfig; public void process(RequestEvent evt) {
if (taskQueue.size() >= userSetting.getMaxNotifyCountQueue()) {
@Transactional logger.error("[notify-目录订阅] 待处理消息队列已满 {}，返回486 BUSY_HERE，消息不做处理", userSetting.getMaxNotifyCountQueue());
public void process(RequestEvent evt) { return;
if (taskQueue.size() >= userSetting.getMaxNotifyCountQueue()) { }
logger.error("[notify-目录订阅] 待处理消息队列已满 {}，返回486 BUSY_HERE，消息不做处理", userSetting.getMaxNotifyCountQueue()); taskQueue.offer(new HandlerCatchData(evt, null, null));
return; }
}
taskQueue.offer(new HandlerCatchData(evt, null, null));
}
// @Scheduled(fixedRate = 2000) //每400毫秒执行一次 // @Scheduled(fixedRate = 2000) //每400毫秒执行一次
// public void showSize(){ // public void showSize(){
// logger.warn("[notify-目录订阅] 待处理消息数量： {}", taskQueue.size() ); // logger.warn("[notify-目录订阅] 待处理消息数量： {}", taskQueue.size() );
// } // }
@Scheduled(fixedRate = 400) //每400毫秒执行一次 @Scheduled(fixedRate = 400) //每400毫秒执行一次
public void executeTaskQueue(){ @Transactional
if (taskQueue.isEmpty()) { public void executeTaskQueue() {
return; if (taskQueue.isEmpty()) {
} return;
List<HandlerCatchData> handlerCatchDataList = new ArrayList<>(); }
while (!taskQueue.isEmpty()) { List<HandlerCatchData> handlerCatchDataList = new ArrayList<>();
handlerCatchDataList.add(taskQueue.poll()); int size = taskQueue.size();
} for (int i = 0; i < size; i++) {
if (handlerCatchDataList.isEmpty()) { HandlerCatchData poll = taskQueue.poll();
return; if (poll != null) {
} handlerCatchDataList.add(poll);
for (HandlerCatchData take : handlerCatchDataList) { }
if (take == null) { }
continue; // while (!taskQueue.isEmpty()) {
} // handlerCatchDataList.add(taskQueue.poll());
RequestEvent evt = take.getEvt(); // }
if (handlerCatchDataList.isEmpty()) {
return;
}
for (HandlerCatchData take : handlerCatchDataList) {
if (take == null) {
logger.warn("[收到目录订阅]：但是队列内任务为空");
continue;
}
RequestEvent evt = take.getEvt();
try {
FromHeader fromHeader = (FromHeader) evt.getRequest().getHeader(FromHeader.NAME);
String deviceId = SipUtils.getUserIdFromFromHeader(fromHeader);
Device device = redisCatchStorage.getDevice(deviceId);
if (device == null || !device.isOnLine()) {
logger.warn("[收到目录订阅]：{}, 但是设备已经离线", (device != null ? device.getDeviceId() : ""));
continue;
}
Element rootElement = getRootElement(evt, device.getCharset());
if (rootElement == null) {
logger.warn("[ 收到目录订阅 ] content cannot be null, {}", evt.getRequest());
continue;
}
Element deviceListElement = rootElement.element("DeviceList");
if (deviceListElement == null) {
logger.warn("[ 收到目录订阅 ] 解析xml获取DeviceList失败, {}", evt.getRequest());
continue;
}
Iterator<Element> deviceListIterator = deviceListElement.elementIterator();
if (deviceListIterator != null) {
// 遍历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()));
continue;
}
if (channel.getParentId() != null && channel.getParentId().equals(sipConfig.getId())) {
channel.setParentId(null);
}
channel.setDeviceId(device.getDeviceId());
if (logger.isDebugEnabled()) {
logger.debug("[收到目录订阅]：{}/{}", device.getDeviceId(), channel.getChannelId());
}
switch (event) {
case CatalogEvent.ON:
// 上线
logger.info("[收到通道上线通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
channel.setStatus(true);
channelList.add(NotifyCatalogChannel.getInstance(NotifyCatalogChannel.Type.STATUS_CHANGED, channel));
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), channel.getChannelId(), true);
}
break;
case CatalogEvent.OFF:
// 离线
logger.info("[收到通道离线通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
if (userSetting.getRefuseChannelStatusChannelFormNotify()) {
logger.info("[收到通道离线通知] 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
} else {
channel.setStatus(false);
channelList.add(NotifyCatalogChannel.getInstance(NotifyCatalogChannel.Type.STATUS_CHANGED, channel));
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), channel.getChannelId(), false);
}
}
break;
case CatalogEvent.VLOST:
// 视频丢失
logger.info("[收到通道视频丢失通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
if (userSetting.getRefuseChannelStatusChannelFormNotify()) {
logger.info("[收到通道视频丢失通知] 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
} else {
channel.setStatus(false);
channelList.add(NotifyCatalogChannel.getInstance(NotifyCatalogChannel.Type.STATUS_CHANGED, channel));
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), channel.getChannelId(), false);
}
}
break;
case CatalogEvent.DEFECT:
// 故障
logger.info("[收到通道视频故障通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
if (userSetting.getRefuseChannelStatusChannelFormNotify()) {
logger.info("[收到通道视频故障通知] 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
} else {
channel.setStatus(false);
channelList.add(NotifyCatalogChannel.getInstance(NotifyCatalogChannel.Type.STATUS_CHANGED, channel));
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), channel.getChannelId(), false);
}
}
break;
case CatalogEvent.ADD:
// 增加
logger.info("[收到增加通道通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
// 判断此通道是否存在
DeviceChannel deviceChannel = deviceChannelService.getOne(deviceId, channel.getChannelId());
if (deviceChannel != null) {
logger.info("[增加通道] 已存在，不发送通知只更新，设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
channel.setId(deviceChannel.getId());
channel.setHasAudio(deviceChannel.getHasAudio());
channelList.add(NotifyCatalogChannel.getInstance(NotifyCatalogChannel.Type.UPDATE, channel));
} else {
channelList.add(NotifyCatalogChannel.getInstance(NotifyCatalogChannel.Type.ADD, channel));
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendChannelAddOrDelete(device.getDeviceId(), channel.getChannelId(), true);
}
}
break;
case CatalogEvent.DEL:
// 删除
logger.info("[收到删除通道通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
channelList.add(NotifyCatalogChannel.getInstance(NotifyCatalogChannel.Type.DELETE, channel));
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendChannelAddOrDelete(device.getDeviceId(), channel.getChannelId(), false);
}
break;
case CatalogEvent.UPDATE:
// 更新
logger.info("[收到更新通道通知] 来自设备: {}, 通道 {}, 状态: {}", device.getDeviceId(), channel.getChannelId(), channel.isStatus());
// 判断此通道是否存在
DeviceChannel deviceChannelInDb = deviceChannelService.getOne(deviceId, channel.getChannelId());
if (deviceChannelInDb != null) {
channel.setId(deviceChannelInDb.getId());
channel.setUpdateTime(DateUtil.getNow());
channel.setHasAudio(deviceChannelInDb.getHasAudio());
channelList.add(NotifyCatalogChannel.getInstance(NotifyCatalogChannel.Type.UPDATE, channel));
} else {
channelList.add(NotifyCatalogChannel.getInstance(NotifyCatalogChannel.Type.ADD, channel));
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendChannelAddOrDelete(device.getDeviceId(), channel.getChannelId(), true);
}
}
break;
default:
logger.warn("[ NotifyCatalog ] event not found ： {}", event);
}
// 转发变化信息
eventPublisher.catalogEventPublish(null, channel, event);
}
}
} catch (DocumentException e) {
logger.error("未处理的异常 ", e);
}
}
if (!channelList.isEmpty()) {
executeSave();
}
}
@Transactional
public void executeSave() {
int size = channelList.size();
List<NotifyCatalogChannel> channelListForSave = new ArrayList<>();
for (int i = 0; i < size; i++) {
channelListForSave.add(channelList.poll());
}
for (NotifyCatalogChannel notifyCatalogChannel : channelListForSave) {
try { try {
long start = System.currentTimeMillis(); switch (notifyCatalogChannel.getType()) {
FromHeader fromHeader = (FromHeader) evt.getRequest().getHeader(FromHeader.NAME); case STATUS_CHANGED:
String deviceId = SipUtils.getUserIdFromFromHeader(fromHeader); deviceChannelService.updateChannelStatus(notifyCatalogChannel.getChannel());
break;
Device device = redisCatchStorage.getDevice(deviceId); case ADD:
if (device == null || !device.isOnLine()) { deviceChannelService.addChannel(notifyCatalogChannel.getChannel());
logger.warn("[收到目录订阅]：{}, 但是设备已经离线", (device != null ? device.getDeviceId() : "")); break;
return; case UPDATE:
deviceChannelService.updateChannelForNotify(notifyCatalogChannel.getChannel());
break;
case DELETE:
deviceChannelService.delete(notifyCatalogChannel.getChannel());
break;
} }
Element rootElement = getRootElement(evt, device.getCharset()); }catch (Exception e) {
if (rootElement == null) { logger.error("[存储收到的通道]类型：{}，编号：{}", notifyCatalogChannel.getType(),
logger.warn("[ 收到目录订阅 ] content cannot be null, {}", evt.getRequest()); notifyCatalogChannel.getChannel().getDeviceId(), e);
return;
}
Element deviceListElement = rootElement.element("DeviceList");
if (deviceListElement == null) {
return;
}
Iterator<Element> deviceListIterator = deviceListElement.elementIterator();
if (deviceListIterator != null) {
// 遍历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()));
continue;
}
if (channel.getParentId() != null && channel.getParentId().equals(sipConfig.getId())) {
channel.setParentId(null);
}
channel.setDeviceId(device.getDeviceId());
if (logger.isDebugEnabled()){
logger.debug("[收到目录订阅]：{}/{}", device.getDeviceId(), channel.getChannelId());
}
switch (event) {
case CatalogEvent.ON:
// 上线
logger.info("[收到通道上线通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
channel.setStatus(true);
updateChannelForStatusChange.add(channel);
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), channel.getChannelId(), true);
}
break;
case CatalogEvent.OFF:
// 离线
logger.info("[收到通道离线通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
if (userSetting.getRefuseChannelStatusChannelFormNotify()) {
logger.info("[收到通道离线通知] 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
} else {
channel.setStatus(false);
updateChannelForStatusChange.add(channel);
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), channel.getChannelId(), false);
}
}
break;
case CatalogEvent.VLOST:
// 视频丢失
logger.info("[收到通道视频丢失通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
if (userSetting.getRefuseChannelStatusChannelFormNotify()) {
logger.info("[收到通道视频丢失通知] 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
} else {
channel.setStatus(false);
updateChannelForStatusChange.add(channel);
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), channel.getChannelId(), false);
}
}
break;
case CatalogEvent.DEFECT:
// 故障
logger.info("[收到通道视频故障通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
if (userSetting.getRefuseChannelStatusChannelFormNotify()) {
logger.info("[收到通道视频故障通知] 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
} else {
channel.setStatus(false);
updateChannelForStatusChange.add(channel);
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), channel.getChannelId(), false);
}
}
break;
case CatalogEvent.ADD:
// 增加
logger.info("[收到增加通道通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
// 判断此通道是否存在
DeviceChannel deviceChannel = deviceChannelService.getOne(deviceId, channel.getChannelId());
if (deviceChannel != null) {
logger.info("[增加通道] 已存在，不发送通知只更新，设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
channel.setId(deviceChannel.getId());
channel.setHasAudio(null);
updateChannelMap.put(channel.getChannelId(), channel);
} else {
addChannelMap.put(channel.getChannelId(), channel);
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendChannelAddOrDelete(device.getDeviceId(), channel.getChannelId(), true);
}
}
break;
case CatalogEvent.DEL:
// 删除
logger.info("[收到删除通道通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
deleteChannelList.add(channel);
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendChannelAddOrDelete(device.getDeviceId(), channel.getChannelId(), false);
}
break;
case CatalogEvent.UPDATE:
// 更新
logger.info("[收到更新通道通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
// 判断此通道是否存在
DeviceChannel deviceChannelForUpdate = deviceChannelService.getOne(deviceId, channel.getChannelId());
if (deviceChannelForUpdate != null) {
channel.setId(deviceChannelForUpdate.getId());
channel.setUpdateTime(DateUtil.getNow());
channel.setHasAudio(null);
updateChannelMap.put(channel.getChannelId(), channel);
} else {
addChannelMap.put(channel.getChannelId(), channel);
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendChannelAddOrDelete(device.getDeviceId(), channel.getChannelId(), true);
}
}
break;
default:
logger.warn("[ NotifyCatalog ] event not found ： {}", event);
}
// 转发变化信息
eventPublisher.catalogEventPublish(null, channel, event);
}
}
} catch (DocumentException e) {
logger.error("未处理的异常 ", e);
} }
} }
taskQueue.clear(); }
if (!updateChannelMap.keySet().isEmpty()
|| !addChannelMap.keySet().isEmpty()
|| !updateChannelForStatusChange.isEmpty()
|| !deleteChannelList.isEmpty()) {
executeSave();
}
}
public void executeSave(){
try {
executeSaveForAdd();
} catch (Exception e) {
logger.error("[存储收到的增加通道] 异常： ", e );
}
try {
executeSaveForStatus();
} catch (Exception e) {
logger.error("[存储收到的通道状态变化] 异常： ", e );
}
try {
executeSaveForUpdate();
} catch (Exception e) {
logger.error("[存储收到的更新通道] 异常： ", e );
}
try {
executeSaveForDelete();
} catch (Exception e) {
logger.error("[存储收到的删除通道] 异常： ", e );
}
}
private void executeSaveForUpdate(){
if (!updateChannelMap.values().isEmpty()) {
logger.info("[存储收到的更新通道], 数量： {}", updateChannelMap.size());
ArrayList<DeviceChannel> deviceChannels = new ArrayList<>(updateChannelMap.values());
deviceChannelService.batchUpdateChannel(deviceChannels);
updateChannelMap.clear();
}
}
private void executeSaveForAdd(){
if (!addChannelMap.values().isEmpty()) {
ArrayList<DeviceChannel> deviceChannels = new ArrayList<>(addChannelMap.values());
addChannelMap.clear();
deviceChannelService.batchAddChannel(deviceChannels);
}
}
private void executeSaveForDelete(){
if (!deleteChannelList.isEmpty()) {
deviceChannelService.deleteChannels(deleteChannelList);
deleteChannelList.clear();
}
}
private void executeSaveForStatus(){
if (!updateChannelForStatusChange.isEmpty()) {
deviceChannelService.updateChannelsStaus(updateChannelForStatusChange);
updateChannelForStatusChange.clear();
}
}
} }

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

@ -21,6 +21,7 @@ import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled; import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.ObjectUtils; import org.springframework.util.ObjectUtils;
import javax.sip.RequestEvent; import javax.sip.RequestEvent;
@ -71,7 +72,10 @@ public class NotifyRequestForMobilePositionProcessor extends SIPRequestProcessor
} }
List<HandlerCatchData> handlerCatchDataList = new ArrayList<>(); List<HandlerCatchData> handlerCatchDataList = new ArrayList<>();
while (!taskQueue.isEmpty()) { while (!taskQueue.isEmpty()) {
handlerCatchDataList.add(taskQueue.poll()); HandlerCatchData poll = taskQueue.poll();
if (poll != null) {
handlerCatchDataList.add(poll);
}
} }
if (handlerCatchDataList.isEmpty()) { if (handlerCatchDataList.isEmpty()) {
return; return;
@ -89,12 +93,12 @@ public class NotifyRequestForMobilePositionProcessor extends SIPRequestProcessor
Element rootElement = getRootElement(evt); Element rootElement = getRootElement(evt);
if (rootElement == null) { if (rootElement == null) {
logger.error("处理MobilePosition移动位置Notify时未获取到消息体,{}", evt.getRequest()); logger.error("处理MobilePosition移动位置Notify时未获取到消息体,{}", evt.getRequest());
return; continue;
} }
Device device = redisCatchStorage.getDevice(deviceId); Device device = redisCatchStorage.getDevice(deviceId);
if (device == null) { if (device == null) {
logger.error("处理MobilePosition移动位置Notify时未获取到device,{}", deviceId); logger.error("处理MobilePosition移动位置Notify时未获取到device,{}", deviceId);
return; continue;
} }
MobilePosition mobilePosition = new MobilePosition(); MobilePosition mobilePosition = new MobilePosition();
mobilePosition.setDeviceId(device.getDeviceId()); mobilePosition.setDeviceId(device.getDeviceId());
@ -151,7 +155,7 @@ public class NotifyRequestForMobilePositionProcessor extends SIPRequestProcessor
} }
} }
logger.debug("[收到移动位置订阅通知]：{}/{}->{}.{}, 时间： {}", mobilePosition.getDeviceId(), mobilePosition.getChannelId(), logger.debug("[收到移动位置订阅通知]：{}/{}->{}.{}, 时间： {}", mobilePosition.getDeviceId(), mobilePosition.getChannelId(),
mobilePosition.getLongitude(), mobilePosition.getLatitude(), System.currentTimeMillis() - startTime); mobilePosition.getLongitude(), mobilePosition.getLatitude(), System.currentTimeMillis() - startTime);
mobilePosition.setReportSource("Mobile Position"); mobilePosition.setReportSource("Mobile Position");
@ -194,7 +198,6 @@ public class NotifyRequestForMobilePositionProcessor extends SIPRequestProcessor
logger.error("未处理的异常 ", e); logger.error("未处理的异常 ", e);
} }
} }
taskQueue.clear();
} }
// @Scheduled(fixedRate = 10000) // @Scheduled(fixedRate = 10000)
// public void execute(){ // public void execute(){

8
src/main/java/com/genersoft/iot/vmp/service/IDeviceChannelService.java
View File

@ -25,7 +25,7 @@ public interface IDeviceChannelService {
* @param deviceId 设备id * @param deviceId 设备id
* @param channel 通道 * @param channel 通道
*/ */
void updateChannel(String deviceId, DeviceChannel channel); void updateChannelForNotify(String deviceId, DeviceChannel channel);
/** /**
* 批量添加设备通道 * 批量添加设备通道
@ -102,4 +102,10 @@ public interface IDeviceChannelService {
void offline(DeviceChannel channel); void offline(DeviceChannel channel);
void delete(DeviceChannel channel); void delete(DeviceChannel channel);
void updateChannelStatus(DeviceChannel channel);
void addChannel(DeviceChannel channel);
void updateChannelForNotify(DeviceChannel channel);
} }

17
src/main/java/com/genersoft/iot/vmp/service/impl/DeviceChannelServiceImpl.java
View File

@ -96,7 +96,7 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
} }
@Override @Override
public void updateChannel(String deviceId, DeviceChannel channel) { public void updateChannelForNotify(String deviceId, DeviceChannel channel) {
String channelId = channel.getChannelId(); String channelId = channel.getChannelId();
channel.setDeviceId(deviceId); channel.setDeviceId(deviceId);
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId); InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId);
@ -116,6 +116,11 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
channelMapper.updateChannelSubCount(deviceId,channel.getParentId()); channelMapper.updateChannelSubCount(deviceId,channel.getParentId());
} }
@Override
public void updateChannelForNotify(DeviceChannel channel) {
channelMapper.updateForNotify(channel);
}
@Override @Override
public int updateChannels(String deviceId, List<DeviceChannel> channels) { public int updateChannels(String deviceId, List<DeviceChannel> channels) {
List<DeviceChannel> addChannels = new ArrayList<>(); List<DeviceChannel> addChannels = new ArrayList<>();
@ -280,6 +285,16 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
return result; return result;
} }
@Override
public void updateChannelStatus(DeviceChannel channel) {
channelMapper.updateStatus(channel);
}
@Override
public void addChannel(DeviceChannel channel) {
channelMapper.add(channel);
}
@Override @Override
public void online(DeviceChannel channel) { public void online(DeviceChannel channel) {
channelMapper.online(channel.getDeviceId(), channel.getChannelId()); channelMapper.online(channel.getDeviceId(), channel.getChannelId());

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

@ -17,14 +17,22 @@ import java.util.List;
@Repository @Repository
public interface DeviceChannelMapper { public interface DeviceChannelMapper {
@Insert("INSERT INTO wvp_device_channel (channel_id, device_id, name, manufacture, model, owner, civil_code, block, " + @Insert("<script> " +
"address, parental, parent_id, safety_way, register_way, cert_num, certifiable, err_code, secrecy, " + "insert into wvp_device_channel " +
"ip_address, port, password, ptz_type, status, stream_id, longitude, latitude, longitude_gcj02, latitude_gcj02, " + "(channel_id, device_id, name, manufacture, model, owner, civil_code, block, sub_count, " +
"longitude_wgs84, latitude_wgs84, has_audio, create_time, update_time, business_group_id, gps_time, stream_identification) " + " address, parental, parent_id, safety_way, register_way, cert_num, certifiable, err_code, secrecy, " +
"VALUES (#{channelId}, #{deviceId}, #{name}, #{manufacture}, #{model}, #{owner}, #{civilCode}, #{block}," + " ip_address,port,password,ptz_type,status,stream_id,longitude,latitude,longitude_gcj02,latitude_gcj02,"+
"#{address}, #{parental}, #{parentId}, #{safetyWay}, #{registerWay}, #{certNum}, #{certifiable}, #{errCode}, #{secrecy}, " + " longitude_wgs84,latitude_wgs84,has_audio,create_time,update_time,business_group_id,gps_time,stream_identification)"+
"#{ipAddress}, #{port}, #{password}, #{ptzType}, #{status}, #{streamId}, #{longitude}, #{latitude}, #{longitudeGcj02}, " + "values " +
"#{latitudeGcj02}, #{longitudeWgs84}, #{latitudeWgs84}, #{hasAudio}, #{createTime}, #{updateTime}, #{businessGroupId}, #{gpsTime}, #{streamIdentification})") "(#{channelId}, #{deviceId}, #{name}, #{manufacture}, #{model}, " +
"#{owner}, #{civilCode}, #{block},#{subCount}," +
"#{address}, #{parental}, #{parentId}, #{safetyWay}, #{registerWay}, " +
"#{certNum}, #{certifiable}, #{errCode}, #{secrecy}, " +
"#{ipAddress}, #{port}, #{password}, #{ptzType}, #{status}, " +
"#{streamId}, #{longitude}, #{latitude},#{longitudeGcj02}, " +
"#{latitudeGcj02},#{longitudeWgs84}, #{latitudeWgs84}, #{hasAudio}, now(), now(), " +
"#{businessGroupId}, #{gpsTime}, #{streamIdentification}) " +
"</script>")
int add(DeviceChannel channel); int add(DeviceChannel channel);
@Update(value = {" <script>" + @Update(value = {" <script>" +
@ -65,6 +73,49 @@ public interface DeviceChannelMapper {
" </script>"}) " </script>"})
int update(DeviceChannel channel); int update(DeviceChannel channel);
@Update(value = {" <script>" +
" UPDATE" +
" wvp_device_channel" +
" SET update_time=#{updateTime}" +
"<if test='name != null'>, name=#{name}</if>" +
"<if test='manufacture != null'>, manufacture=#{manufacture}</if>" +
"<if test='model != null'>, model=#{model}</if>" +
"<if test='owner != null'>, owner=#{owner}</if>" +
"<if test='civilCode != null'>, civil_code=#{civilCode}</if>" +
"<if test='block != null'>, block=#{block}</if>" +
"<if test='subCount != null'>, sub_count=#{subCount}</if>" +
"<if test='address != null'>, address=#{address}</if>" +
"<if test='parental != null'>, parental=#{parental}</if>" +
"<if test='parentId != null'>, parent_id=#{parentId}</if>" +
"<if test='safetyWay != null'>, safety_way=#{safetyWay}</if>" +
"<if test='registerWay != null'>, register_way=#{registerWay}</if>" +
"<if test='certNum != null'>, cert_num=#{certNum}</if>" +
"<if test='certifiable != null'>, certifiable=#{certifiable}</if>" +
"<if test='errCode != null'>, err_code=#{errCode}</if>" +
"<if test='secrecy != null'>, secrecy=#{secrecy}</if>" +
"<if test='ipAddress != null'>, ip_address=#{ipAddress}</if>" +
"<if test='port != null'>, port=#{port}</if>" +
"<if test='password != null'>, password=#{password}</if>" +
"<if test='ptzType != null'>, ptz_type=#{ptzType}</if>" +
"<if test='status != null'>, status=#{status}</if>" +
"<if test='streamId != null'>, stream_id=#{streamId}</if>" +
"<if test='hasAudio != null'>, has_audio=#{hasAudio}</if>" +
"<if test='longitude != null'>, longitude=#{longitude}</if>" +
"<if test='latitude != null'>, latitude=#{latitude}</if>" +
"<if test='customLongitude != null'>, custom_longitude=#{customLongitude}</if>" +
"<if test='customLatitude != null'>, custom_latitude=#{customLatitude}</if>" +
"<if test='longitudeGcj02 != null'>, longitude_gcj02=#{longitudeGcj02}</if>" +
"<if test='latitudeGcj02 != null'>, latitude_gcj02=#{latitudeGcj02}</if>" +
"<if test='longitudeWgs84 != null'>, longitude_wgs84=#{longitudeWgs84}</if>" +
"<if test='latitudeWgs84 != null'>, latitude_wgs84=#{latitudeWgs84}</if>" +
"<if test='businessGroupId != null'>, business_group_id=#{businessGroupId}</if>" +
"<if test='gpsTime != null'>, gps_time=#{gpsTime}</if>" +
"<if test='streamIdentification != null'>, stream_identification=#{streamIdentification}</if>" +
"<if test='id > 0'>WHERE id=#{id}</if>" +
"<if test='id == 0'>WHERE device_id=#{deviceId} AND channel_id=#{channelId}</if>" +
" </script>"})
int updateForNotify(DeviceChannel channel);
@Select(value = {" <script>" + @Select(value = {" <script>" +
"SELECT " + "SELECT " +
"dc.id, " + "dc.id, " +
@ -574,4 +625,7 @@ public interface DeviceChannelMapper {
"</script>"}) "</script>"})
void batchUpdatePosition(List<DeviceChannel> channelList); void batchUpdatePosition(List<DeviceChannel> channelList);
@Update("UPDATE wvp_device_channel SET status=#{status} WHERE device_id=#{deviceId} AND channel_id=#{channelId}")
void updateStatus(DeviceChannel channel);
} }

2
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/device/DeviceQuery.java
View File

@ -261,7 +261,7 @@ public class DeviceQuery {
@Parameter(name = "channel", description = "通道信息", required = true) @Parameter(name = "channel", description = "通道信息", required = true)
@PostMapping("/channel/update/{deviceId}") @PostMapping("/channel/update/{deviceId}")
public void updateChannel(@PathVariable String deviceId,DeviceChannel channel){ public void updateChannel(@PathVariable String deviceId,DeviceChannel channel){
deviceChannelService.updateChannel(deviceId, channel); deviceChannelService.updateChannelForNotify(deviceId, channel);
} }
@Operation(summary = "修改通道的码流类型", security = @SecurityRequirement(name = JwtUtils.HEADER)) @Operation(summary = "修改通道的码流类型", security = @SecurityRequirement(name = JwtUtils.HEADER))