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

Merge branch 'refs/heads/2.7.0' 

# Conflicts:
#	src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
pull/1450/head
648540858 2024-04-24 20:51:28 +08:00
parent 1a30992985 2113e8cf27
commit b89af9a693
6 changed files with 320 additions and 331 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -535,11 +535,11 @@ public class DeviceChannel {
this.subCount = subCount;
}
public boolean isHasAudio() {
public Boolean getHasAudio() {
return hasAudio;
}
public void setHasAudio(boolean hasAudio) {
public void setHasAudio(Boolean hasAudio) {
this.hasAudio = hasAudio;
}

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

@ -1,9 +1,11 @@
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.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.HandlerCatchData;
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;
@ -17,15 +19,18 @@ 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;
import org.springframework.transaction.annotation.Transactional;
import javax.sip.RequestEvent;
import javax.sip.header.FromHeader;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CopyOnWriteArrayList;
/**
@ -37,11 +42,14 @@ 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<>();
private final Map<String, DeviceChannel> addChannelMap = new ConcurrentHashMap<>();
private final List<DeviceChannel> deleteChannelList = new CopyOnWriteArrayList<>();
private ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>();
@Autowired
private UserSetting userSetting;
@ -55,15 +63,31 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
@Autowired
private IDeviceChannelService deviceChannelService;
@Autowired
private DynamicTask dynamicTask;
@Autowired
private SipConfig sipConfig;
@Transactional
public void process(List<RequestEvent> evtList) {
if (evtList.isEmpty()) {
public void process(RequestEvent evt) {
if (taskQueue.size() >= userSetting.getMaxNotifyCountQueue()) {
logger.error("[notify-目录订阅] 待处理消息队列已满 {}，返回486 BUSY_HERE，消息不做处理", userSetting.getMaxNotifyCountQueue());
return;
}
for (RequestEvent evt : evtList) {
taskQueue.offer(new HandlerCatchData(evt, null, null));
}
@Scheduled(fixedRate = 400) //每400毫秒执行一次
public void executeTaskQueue(){
if (taskQueue.isEmpty()) {
return;
}
for (HandlerCatchData take : taskQueue) {
if (take == null) {
continue;
}
RequestEvent evt = take.getEvt();
try {
long start = System.currentTimeMillis();
FromHeader fromHeader = (FromHeader) evt.getRequest().getHeader(FromHeader.NAME);
@ -71,7 +95,7 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
Device device = redisCatchStorage.getDevice(deviceId);
if (device == null || !device.isOnLine()) {
logger.warn("[收到目录订阅]：{}, 但是设备已经离线", (device != null ? device.getDeviceId():"" ));
logger.warn("[收到目录订阅]：{}, 但是设备已经离线", (device != null ? device.getDeviceId() : ""));
return;
}
Element rootElement = getRootElement(evt, device.getCharset());
@ -92,9 +116,9 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
Element eventElement = itemDevice.element("Event");
String event;
if (eventElement == null) {
logger.warn("[收到目录订阅]：{}, 但是Event为空, 设为默认值 ADD", (device != null ? device.getDeviceId():"" ));
logger.warn("[收到目录订阅]：{}, 但是Event为空, 设为默认值 ADD", (device != null ? device.getDeviceId() : ""));
event = CatalogEvent.ADD;
}else {
} else {
event = eventElement.getText().toUpperCase();
}
DeviceChannel channel = XmlUtil.channelContentHandler(itemDevice, device, event);
@ -106,60 +130,193 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
channel.setParentId(null);
}
channel.setDeviceId(device.getDeviceId());
logger.info("[收到目录订阅]：{}, {}/{}",event, device.getDeviceId(), channel.getChannelId());
logger.info("[收到目录订阅]：{}/{}", device.getDeviceId(), channel.getChannelId());
switch (event) {
case CatalogEvent.ON:
// 上线
deviceChannelService.online(channel);
logger.info("[收到通道上线通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
updateChannelOnlineList.add(channel);
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), channel.getChannelId(), true);
}
break;
case CatalogEvent.OFF :
case CatalogEvent.VLOST:
case CatalogEvent.DEFECT:
case CatalogEvent.OFF:
// 离线
logger.info("[收到通道离线通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
if (userSetting.getRefuseChannelStatusChannelFormNotify()) {
logger.info("[目录订阅] 离线 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
}else {
deviceChannelService.offline(channel);
logger.info("[收到通道离线通知] 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
} else {
updateChannelOfflineList.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 {
updateChannelOfflineList.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 {
updateChannelOfflineList.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:
// 删除
deviceChannelService.delete(channel);
logger.info("[收到删除通道通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
deleteChannelList.add(channel);
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendChannelAddOrDelete(device.getDeviceId(), channel.getChannelId(), false);
}
break;
case CatalogEvent.ADD:
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());
deviceChannelService.updateChannel(deviceId,channel);
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 );
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()
|| !updateChannelOnlineList.isEmpty()
|| !updateChannelOfflineList.isEmpty()
|| !deleteChannelList.isEmpty()) {
executeSave();
}
}
public void executeSave(){
try {
executeSaveForAdd();
} catch (Exception e) {
logger.error("[存储收到的增加通道] 异常： ", e );
}
try {
executeSaveForOnline();
} catch (Exception e) {
logger.error("[存储收到的通道上线] 异常： ", e );
}
try {
executeSaveForOffline();
} 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 executeSaveForOnline(){
if (!updateChannelOnlineList.isEmpty()) {
deviceChannelService.channelsOnline(updateChannelOnlineList);
updateChannelOnlineList.clear();
}
}
private void executeSaveForOffline(){
if (!updateChannelOfflineList.isEmpty()) {
deviceChannelService.channelsOffline(updateChannelOfflineList);
updateChannelOfflineList.clear();
}
}
@Scheduled(fixedRate = 10000) //每1秒执行一次
public void execute(){
logger.info("[待处理Notify-目录订阅消息数量]: {}", taskQueue.size());
}
}

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

@ -4,6 +4,7 @@ import com.alibaba.fastjson2.JSONObject;
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.bean.HandlerCatchData;
import com.genersoft.iot.vmp.gb28181.bean.MobilePosition;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
@ -17,6 +18,7 @@ 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;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.ObjectUtils;
@ -27,6 +29,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
/**
* SIP命令类型： NOTIFY请求中的移动位置请求处理
@ -37,6 +40,7 @@ public class NotifyRequestForMobilePositionProcessor extends SIPRequestProcessor
private final static Logger logger = LoggerFactory.getLogger(NotifyRequestForMobilePositionProcessor.class);
private ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>();
@Autowired
private UserSetting userSetting;
@ -50,14 +54,28 @@ public class NotifyRequestForMobilePositionProcessor extends SIPRequestProcessor
@Autowired
private IDeviceChannelService deviceChannelService;
public void process(RequestEvent evt) {
if (taskQueue.size() >= userSetting.getMaxNotifyCountQueue()) {
logger.error("[notify-移动位置] 待处理消息队列已满 {}，返回486 BUSY_HERE，消息不做处理", userSetting.getMaxNotifyCountQueue());
return;
}
taskQueue.offer(new HandlerCatchData(evt, null, null));
}
@Scheduled(fixedRate = 200) //每200毫秒执行一次
@Transactional
public void process(List<RequestEvent> eventList) {
if (eventList.isEmpty()) {
public void executeTaskQueue() {
if (taskQueue.isEmpty()) {
return;
}
Map<String, DeviceChannel> updateChannelMap = new ConcurrentHashMap<>();
List<MobilePosition> addMobilePositionList = new ArrayList<>();
for (RequestEvent evt : eventList) {
for (HandlerCatchData take : taskQueue) {
if (take == null) {
continue;
}
RequestEvent evt = take.getEvt();
try {
FromHeader fromHeader = (FromHeader) evt.getRequest().getHeader(FromHeader.NAME);
String deviceId = SipUtils.getUserIdFromFromHeader(fromHeader);
@ -180,10 +198,11 @@ public class NotifyRequestForMobilePositionProcessor extends SIPRequestProcessor
logger.error("未处理的异常 ", e);
}
}
taskQueue.clear();
if(!updateChannelMap.isEmpty()) {
List<DeviceChannel> channels = new ArrayList<>(updateChannelMap.values());
logger.info("[移动位置订阅]更新通道位置： {}", channels.size());
deviceChannelService.batchUpdateChannelGPS(channels);
deviceChannelService.batchUpdateChannel(channels);
updateChannelMap.clear();
}
if (userSetting.isSavePositionHistory() && !addMobilePositionList.isEmpty()) {
@ -196,4 +215,8 @@ public class NotifyRequestForMobilePositionProcessor extends SIPRequestProcessor
addMobilePositionList.clear();
}
}
@Scheduled(fixedRate = 10000)
public void execute(){
logger.info("[待处理Notify-移动位置订阅消息数量]: {}", taskQueue.size());
}
}

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

@ -1,12 +1,9 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
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.gb28181.utils.NumericUtil;
@ -14,9 +11,7 @@ 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.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.utils.redis.RedisUtil;
import gov.nist.javax.sip.message.SIPRequest;
import org.dom4j.DocumentException;
import org.dom4j.Element;
@ -24,12 +19,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;
import javax.sip.InvalidArgumentException;
import javax.sip.RequestEvent;
@ -37,9 +27,6 @@ import javax.sip.SipException;
import javax.sip.header.FromHeader;
import javax.sip.message.Response;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
/**
* SIP命令类型： NOTIFY请求,这是作为上级发送订阅请求后，设备才会响应的
@ -50,15 +37,6 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
private final static Logger logger = LoggerFactory.getLogger(NotifyRequestProcessor.class);
@Autowired
private UserSetting userSetting;
@Autowired
private IVideoManagerStorage storager;
@Autowired
private EventPublisher eventPublisher;
@Autowired
private SipConfig sipConfig;
@ -82,14 +60,6 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
@Autowired
private NotifyRequestForMobilePositionProcessor notifyRequestForMobilePositionProcessor;
private ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>();
@Qualifier("taskExecutor")
@Autowired
private ThreadPoolTaskExecutor taskExecutor;
private int maxQueueCount = 30000;
@Override
public void afterPropertiesSet() throws Exception {
// 添加消息处理的订阅
@ -99,176 +69,38 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
@Override
public void process(RequestEvent evt) {
try {
if (taskQueue.size() >= userSetting.getMaxNotifyCountQueue()) {
responseAck((SIPRequest) evt.getRequest(), Response.BUSY_HERE, null, null);
logger.error("[notify] 待处理消息队列已满 {}，返回486 BUSY_HERE，消息不做处理", userSetting.getMaxNotifyCountQueue());
return;
} else {
responseAck((SIPRequest) evt.getRequest(), Response.OK, null, null);
}
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("未处理的异常 ", e);
}
boolean runed = !taskQueue.isEmpty();
taskQueue.offer(new HandlerCatchData(evt, null, null));
}
@Scheduled(fixedRate = 200) //每200毫秒执行一次
public void executeTaskQueue(){
if (taskQueue.isEmpty()) {
return;
}
try {
List<RequestEvent> catalogEventList = new ArrayList<>();
List<RequestEvent> alarmEventList = new ArrayList<>();
List<RequestEvent> mobilePositionEventList = new ArrayList<>();
for (HandlerCatchData take : taskQueue) {
if (take == null) {
continue;
}
Element rootElement = getRootElement(take.getEvt());
Element rootElement = getRootElement(evt);
if (rootElement == null) {
logger.error("处理NOTIFY消息时未获取到消息体,{}", take.getEvt().getRequest());
continue;
logger.error("处理NOTIFY消息时未获取到消息体,{}", evt.getRequest());
responseAck((SIPRequest) evt.getRequest(), Response.OK, null, null);
return;
}
String cmd = XmlUtil.getText(rootElement, "CmdType");
if (CmdType.CATALOG.equals(cmd)) {
catalogEventList.add(take.getEvt());
notifyRequestForCatalogProcessor.process(evt);
} else if (CmdType.ALARM.equals(cmd)) {
alarmEventList.add(take.getEvt());
processNotifyAlarm(evt);
} else if (CmdType.MOBILE_POSITION.equals(cmd)) {
mobilePositionEventList.add(take.getEvt());
notifyRequestForMobilePositionProcessor.process(evt);
} else {
logger.info("接收到消息：" + cmd);
}
}
taskQueue.clear();
if (!alarmEventList.isEmpty()) {
processNotifyAlarm(alarmEventList);
}
if (!catalogEventList.isEmpty()) {
notifyRequestForCatalogProcessor.process(catalogEventList);
}
if (!mobilePositionEventList.isEmpty()) {
notifyRequestForMobilePositionProcessor.process(mobilePositionEventList);
}
} catch (DocumentException e) {
logger.error("处理NOTIFY消息时错误", e);
}
}
/**
* 处理MobilePosition移动位置Notify
*
* @param evt
*/
@Async("taskExecutor")
public void processNotifyMobilePosition(RequestEvent evt) {
try {
FromHeader fromHeader = (FromHeader) evt.getRequest().getHeader(FromHeader.NAME);
String deviceId = SipUtils.getUserIdFromFromHeader(fromHeader);
// 回复 200 OK
Element rootElement = getRootElement(evt);
if (rootElement == null) {
logger.error("处理MobilePosition移动位置Notify时未获取到消息体,{}", evt.getRequest());
return;
}
MobilePosition mobilePosition = new MobilePosition();
mobilePosition.setCreateTime(DateUtil.getNow());
Element deviceIdElement = rootElement.element("DeviceID");
String channelId = deviceIdElement.getTextTrim().toString();
Device device = redisCatchStorage.getDevice(deviceId);
if (device == null) {
device = redisCatchStorage.getDevice(channelId);
if (device == null) {
// 根据通道id查询设备Id
List<Device> deviceList = deviceChannelService.getDeviceByChannelId(channelId);
if (deviceList.size() > 0) {
device = deviceList.get(0);
}
}
}
if (device == null) {
logger.warn("[mobilePosition移动位置Notify] 未找到通道{}所属的设备", channelId);
return;
}
// 兼容设备部分设备上报是通道编号与设备编号一致的情况
if(deviceId.equals(channelId)) {
List<DeviceChannel> deviceChannels = deviceChannelService.queryChaneListByDeviceId(deviceId);
if (deviceChannels.size() == 1) {
channelId = deviceChannels.get(0).getChannelId();
}
}
if (!ObjectUtils.isEmpty(device.getName())) {
mobilePosition.setDeviceName(device.getName());
}
mobilePosition.setDeviceId(device.getDeviceId());
mobilePosition.setChannelId(channelId);
String time = XmlUtil.getText(rootElement, "Time");
if (ObjectUtils.isEmpty(time)){
mobilePosition.setTime(DateUtil.getNow());
}else {
mobilePosition.setTime(SipUtils.parseTime(time));
}
mobilePosition.setLongitude(Double.parseDouble(XmlUtil.getText(rootElement, "Longitude")));
mobilePosition.setLatitude(Double.parseDouble(XmlUtil.getText(rootElement, "Latitude")));
if (NumericUtil.isDouble(XmlUtil.getText(rootElement, "Speed"))) {
mobilePosition.setSpeed(Double.parseDouble(XmlUtil.getText(rootElement, "Speed")));
} else {
mobilePosition.setSpeed(0.0);
}
if (NumericUtil.isDouble(XmlUtil.getText(rootElement, "Direction"))) {
mobilePosition.setDirection(Double.parseDouble(XmlUtil.getText(rootElement, "Direction")));
} else {
mobilePosition.setDirection(0.0);
}
if (NumericUtil.isDouble(XmlUtil.getText(rootElement, "Altitude"))) {
mobilePosition.setAltitude(Double.parseDouble(XmlUtil.getText(rootElement, "Altitude")));
} else {
mobilePosition.setAltitude(0.0);
}
// logger.info("[收到移动位置订阅通知]：{}/{}->{}.{}", mobilePosition.getDeviceId(), mobilePosition.getChannelId(),
// mobilePosition.getLongitude(), mobilePosition.getLatitude());
mobilePosition.setReportSource("Mobile Position");
// 更新device channel 的经纬度
DeviceChannel deviceChannel = new DeviceChannel();
deviceChannel.setDeviceId(device.getDeviceId());
deviceChannel.setChannelId(channelId);
deviceChannel.setLongitude(mobilePosition.getLongitude());
deviceChannel.setLatitude(mobilePosition.getLatitude());
deviceChannel.setGpsTime(mobilePosition.getTime());
// deviceChannel = deviceChannelService.updateGps(deviceChannel, device);
//
// mobilePosition.setLongitudeWgs84(deviceChannel.getLongitudeWgs84());
// mobilePosition.setLatitudeWgs84(deviceChannel.getLatitudeWgs84());
// mobilePosition.setLongitudeGcj02(deviceChannel.getLongitudeGcj02());
// mobilePosition.setLatitudeGcj02(deviceChannel.getLatitudeGcj02());
deviceChannelService.updateChannelGPS(device, deviceChannel, mobilePosition);
} catch (DocumentException e) {
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("未处理的异常 ", e);
}
} catch (DocumentException e) {
throw new RuntimeException(e);
}
}
/***
* 处理alarm设备报警Notify
*/
private void processNotifyAlarm(List<RequestEvent> evtList) {
private void processNotifyAlarm(RequestEvent evt) {
if (!sipConfig.isAlarm()) {
return;
}
if (!evtList.isEmpty()) {
for (RequestEvent evt : evtList) {
try {
FromHeader fromHeader = (FromHeader) evt.getRequest().getHeader(FromHeader.NAME);
String deviceId = SipUtils.getUserIdFromFromHeader(fromHeader);
@ -353,36 +185,6 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
logger.error("未处理的异常 ", e);
}
}
}
}
public void setCmder(SIPCommander cmder) {
}
public void setStorager(IVideoManagerStorage storager) {
this.storager = storager;
}
public void setPublisher(EventPublisher publisher) {
this.publisher = publisher;
}
public void setRedis(RedisUtil redis) {
}
public void setDeferredResultHolder(DeferredResultHolder deferredResultHolder) {
}
public IRedisCatchStorage getRedisCatchStorage() {
return redisCatchStorage;
}
public void setRedisCatchStorage(IRedisCatchStorage redisCatchStorage) {
this.redisCatchStorage = redisCatchStorage;
}
@Scheduled(fixedRate = 10000) //每1秒执行一次
public void execute(){
logger.info("[待处理Notify消息数量]: {}", taskQueue.size());
}
}

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

@ -275,15 +275,23 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
}
@Override
public void batchUpdateChannel(List<DeviceChannel> channels) {
public synchronized void batchUpdateChannel(List<DeviceChannel> channels) {
String now = DateUtil.getNow();
for (DeviceChannel channel : channels) {
channel.setUpdateTime(now);
}
int limitCount = 1000;
if (!channels.isEmpty()) {
if (channels.size() > limitCount) {
for (int i = 0; i < channels.size(); i += limitCount) {
int toIndex = i + limitCount;
if (i + limitCount > channels.size()) {
toIndex = channels.size();
}
channelMapper.batchUpdate(channels.subList(i, toIndex));
}
}else {
channelMapper.batchUpdate(channels);
for (DeviceChannel channel : channels) {
if (channel.getParentId() != null) {
channelMapper.updateChannelSubCount(channel.getDeviceId(), channel.getParentId());
}
}
}

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

@ -7,7 +7,6 @@ import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
import com.genersoft.iot.vmp.service.IGbStreamService;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
@ -139,7 +138,7 @@ public class VideoManagerStorageImpl implements IVideoManagerStorage {
gbIdSet.add(deviceChannel.getChannelId());
if (allChannelMap.containsKey(deviceChannel.getChannelId())) {
deviceChannel.setStreamId(allChannelMap.get(deviceChannel.getChannelId()).getStreamId());
deviceChannel.setHasAudio(allChannelMap.get(deviceChannel.getChannelId()).isHasAudio());
deviceChannel.setHasAudio(allChannelMap.get(deviceChannel.getChannelId()).getHasAudio());
if (allChannelMap.get(deviceChannel.getChannelId()).isStatus() !=deviceChannel.isStatus()){
List<String> strings = platformChannelMapper.queryParentPlatformByChannelId(deviceChannel.getChannelId());
if (!CollectionUtils.isEmpty(strings)){
@ -275,7 +274,7 @@ public class VideoManagerStorageImpl implements IVideoManagerStorage {
deviceChannel.setUpdateTime(DateUtil.getNow());
if (allChannelMap.containsKey(deviceChannel.getChannelId())) {
deviceChannel.setStreamId(allChannelMap.get(deviceChannel.getChannelId()).getStreamId());
deviceChannel.setHasAudio(allChannelMap.get(deviceChannel.getChannelId()).isHasAudio());
deviceChannel.setHasAudio(allChannelMap.get(deviceChannel.getChannelId()).getHasAudio());
if (allChannelMap.get(deviceChannel.getChannelId()).isStatus() !=deviceChannel.isStatus()){
List<String> strings = platformChannelMapper.queryParentPlatformByChannelId(deviceChannel.getChannelId());
if (!CollectionUtils.isEmpty(strings)){