From 985082d33930868c3cc1723f28fd9aaae9013a8f Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期三, 16 八月 2023 09:11:22 +0800
Subject: [PATCH] 添加缺少的内容
---
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java | 108 ++++++++++++++++++++++++++++++++++++++++--------------
1 files changed, 80 insertions(+), 28 deletions(-)
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java
old mode 100644
new mode 100755
index 930ddb5..64a6478
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java
@@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;
-import com.alibaba.fastjson.JSONObject;
+import com.alibaba.fastjson2.JSONObject;
+import com.genersoft.iot.vmp.conf.CivilCodeFileConf;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.*;
@@ -11,7 +12,6 @@
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.Coordtransform;
import com.genersoft.iot.vmp.gb28181.utils.NumericUtil;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.gb28181.utils.XmlUtil;
@@ -20,6 +20,7 @@
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;
import org.slf4j.Logger;
@@ -30,16 +31,15 @@
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;
-import org.springframework.util.StringUtils;
import javax.sip.InvalidArgumentException;
import javax.sip.RequestEvent;
-import javax.sip.ServerTransaction;
import javax.sip.SipException;
import javax.sip.header.FromHeader;
import javax.sip.message.Response;
import java.text.ParseException;
import java.util.Iterator;
+import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
/**
@@ -77,13 +77,19 @@
@Autowired
private IDeviceChannelService deviceChannelService;
- private boolean taskQueueHandlerRun = false;
+ @Autowired
+ private NotifyRequestForCatalogProcessor notifyRequestForCatalogProcessor;
+
+ @Autowired
+ private CivilCodeFileConf civilCodeFileConf;
private ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>();
@Qualifier("taskExecutor")
@Autowired
private ThreadPoolTaskExecutor taskExecutor;
+
+ private int maxQueueCount = 30000;
@Override
public void afterPropertiesSet() throws Exception {
@@ -93,19 +99,30 @@
@Override
public void process(RequestEvent evt) {
- ServerTransaction serverTransaction = getServerTransaction(evt);
try {
- responseAck(serverTransaction, Response.OK);
+
+ 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) {
- e.printStackTrace();
+ logger.error("鏈鐞嗙殑寮傚父 ", e);
}
+ boolean runed = !taskQueue.isEmpty();
+ logger.info("[notify] 寰呭鐞嗘秷鎭暟閲忥細 {}", taskQueue.size());
taskQueue.offer(new HandlerCatchData(evt, null, null));
- if (!taskQueueHandlerRun) {
- taskQueueHandlerRun = true;
+ if (!runed) {
taskExecutor.execute(()-> {
while (!taskQueue.isEmpty()) {
try {
HandlerCatchData take = taskQueue.poll();
+ if (take == null) {
+ continue;
+ }
Element rootElement = getRootElement(take.getEvt());
if (rootElement == null) {
logger.error("澶勭悊NOTIFY娑堟伅鏃舵湭鑾峰彇鍒版秷鎭綋,{}", take.getEvt().getRequest());
@@ -116,6 +133,7 @@
if (CmdType.CATALOG.equals(cmd)) {
logger.info("鎺ユ敹鍒癈atalog閫氱煡");
processNotifyCatalogList(take.getEvt());
+ notifyRequestForCatalogProcessor.process(take.getEvt());
} else if (CmdType.ALARM.equals(cmd)) {
logger.info("鎺ユ敹鍒癆larm閫氱煡");
processNotifyAlarm(take.getEvt());
@@ -129,14 +147,13 @@
logger.error("澶勭悊NOTIFY娑堟伅鏃堕敊璇�", e);
}
}
- taskQueueHandlerRun = false;
});
}
}
/**
* 澶勭悊MobilePosition绉诲姩浣嶇疆Notify
- *
+ *
* @param evt
*/
private void processNotifyMobilePosition(RequestEvent evt) {
@@ -153,18 +170,38 @@
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) {
- if (!ObjectUtils.isEmpty(device.getName())) {
- mobilePosition.setDeviceName(device.getName());
+
+ 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);
+ }
}
}
- mobilePosition.setDeviceId(XmlUtil.getText(rootElement, "DeviceID"));
+ if (device == null) {
+ logger.warn("[mobilePosition绉诲姩浣嶇疆Notify] 鏈壘鍒伴�氶亾{}鎵�灞炵殑璁惧", channelId);
+ return;
+ }
+ if (!ObjectUtils.isEmpty(device.getName())) {
+ mobilePosition.setDeviceName(device.getName());
+ }
+
+ mobilePosition.setDeviceId(device.getDeviceId());
mobilePosition.setChannelId(channelId);
String time = XmlUtil.getText(rootElement, "Time");
- mobilePosition.setTime(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"))) {
@@ -209,7 +246,7 @@
// 鍙戦�乺edis娑堟伅銆� 閫氱煡浣嶇疆淇℃伅鐨勫彉鍖�
JSONObject jsonObject = new JSONObject();
- jsonObject.put("time", time);
+ jsonObject.put("time", DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(mobilePosition.getTime()));
jsonObject.put("serial", deviceId);
jsonObject.put("code", channelId);
jsonObject.put("longitude", mobilePosition.getLongitude());
@@ -219,13 +256,13 @@
jsonObject.put("speed", mobilePosition.getSpeed());
redisCatchStorage.sendMobilePositionMsg(jsonObject);
} catch (DocumentException e) {
- e.printStackTrace();
+ logger.error("鏈鐞嗙殑寮傚父 ", e);
}
}
/***
* 澶勭悊alarm璁惧鎶ヨNotify
- *
+ *
* @param evt
*/
private void processNotifyAlarm(RequestEvent evt) {
@@ -311,7 +348,7 @@
storager.updateChannelPosition(deviceChannel);
// 鍙戦�乺edis娑堟伅銆� 閫氱煡浣嶇疆淇℃伅鐨勫彉鍖�
JSONObject jsonObject = new JSONObject();
- jsonObject.put("time", mobilePosition.getTime());
+ jsonObject.put("time", DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(mobilePosition.getTime()));
jsonObject.put("serial", deviceChannel.getDeviceId());
jsonObject.put("code", deviceChannel.getChannelId());
jsonObject.put("longitude", mobilePosition.getLongitude());
@@ -329,13 +366,13 @@
publisher.deviceAlarmEventPublish(deviceAlarm);
}
} catch (DocumentException e) {
- e.printStackTrace();
+ logger.error("鏈鐞嗙殑寮傚父 ", e);
}
}
/***
* 澶勭悊catalog璁惧鐩綍鍒楄〃Notify
- *
+ *
* @param evt
*/
private void processNotifyCatalogList(RequestEvent evt) {
@@ -344,7 +381,7 @@
String deviceId = SipUtils.getUserIdFromFromHeader(fromHeader);
Device device = redisCatchStorage.getDevice(deviceId);
- if (device == null || device.getOnline() == 0) {
+ if (device == null || !device.isOnLine()) {
logger.warn("[鏀跺埌鐩綍璁㈤槄]锛歿}, 浣嗘槸璁惧宸茬粡绂荤嚎", (device != null ? device.getDeviceId():"" ));
return;
}
@@ -375,7 +412,14 @@
}else {
event = eventElement.getText().toUpperCase();
}
- DeviceChannel channel = XmlUtil.channelContentHander(itemDevice, device, event);
+ DeviceChannel channel = XmlUtil.channelContentHandler(itemDevice, device, event, civilCodeFileConf);
+ 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());
logger.info("[鏀跺埌鐩綍璁㈤槄]锛歿}/{}", device.getDeviceId(), channel.getChannelId());
switch (event) {
@@ -387,12 +431,20 @@
case CatalogEvent.OFF :
// 绂荤嚎
logger.info("[鏀跺埌閫氶亾绂荤嚎閫氱煡] 鏉ヨ嚜璁惧: {}, 閫氶亾 {}", device.getDeviceId(), channel.getChannelId());
- storager.deviceChannelOffline(deviceId, channel.getChannelId());
+ if (userSetting.getRefuseChannelStatusChannelFormNotify()) {
+ storager.deviceChannelOffline(deviceId, channel.getChannelId());
+ }else {
+ logger.info("[鏀跺埌閫氶亾绂荤嚎閫氱煡] 浣嗘槸骞冲彴宸查厤缃嫆缁濇娑堟伅锛屾潵鑷澶�: {}, 閫氶亾 {}", device.getDeviceId(), channel.getChannelId());
+ }
break;
case CatalogEvent.VLOST:
// 瑙嗛涓㈠け
logger.info("[鏀跺埌閫氶亾瑙嗛涓㈠け閫氱煡] 鏉ヨ嚜璁惧: {}, 閫氶亾 {}", device.getDeviceId(), channel.getChannelId());
- storager.deviceChannelOffline(deviceId, channel.getChannelId());
+ if (userSetting.getRefuseChannelStatusChannelFormNotify()) {
+ storager.deviceChannelOffline(deviceId, channel.getChannelId());
+ }else {
+ logger.info("[鏀跺埌閫氶亾瑙嗛涓㈠け閫氱煡] 浣嗘槸骞冲彴宸查厤缃嫆缁濇娑堟伅锛屾潵鑷澶�: {}, 閫氶亾 {}", device.getDeviceId(), channel.getChannelId());
+ }
break;
case CatalogEvent.DEFECT:
// 鏁呴殰
@@ -422,7 +474,7 @@
}
}
} catch (DocumentException e) {
- e.printStackTrace();
+ logger.error("鏈鐞嗙殑寮傚父 ", e);
}
}
--
Gitblit v1.8.0