From c084d6c98af1ef4d36a61adc719df5db76589428 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期六, 07 十月 2023 18:00:08 +0800
Subject: [PATCH] 优化国标级联心跳失败判断逻辑

---
 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java |   53 ++++++++++++++++++++++++++++++++++++++++++++---------
 1 files changed, 44 insertions(+), 9 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 a0038ca..dbe49d5
--- 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.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.*;
@@ -76,11 +77,19 @@
 	@Autowired
 	private IDeviceChannelService deviceChannelService;
 
+	@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 {
@@ -91,17 +100,29 @@
 	@Override
 	public void process(RequestEvent evt) {
 		try {
-			responseAck((SIPRequest) evt.getRequest(), Response.OK, null, null);
+
+			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();
+		logger.info("[notify] 寰呭鐞嗘秷鎭暟閲忥細 {}", taskQueue.size());
 		taskQueue.offer(new HandlerCatchData(evt, null, null));
 		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());
@@ -112,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());
@@ -131,7 +153,7 @@
 
 	/**
 	 * 澶勭悊MobilePosition绉诲姩浣嶇疆Notify
-	 * 
+	 *
 	 * @param evt
 	 */
 	private void processNotifyMobilePosition(RequestEvent evt) {
@@ -174,7 +196,12 @@
 			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"))) {
@@ -219,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());
@@ -235,7 +262,7 @@
 
 	/***
 	 * 澶勭悊alarm璁惧鎶ヨNotify
-	 * 
+	 *
 	 * @param evt
 	 */
 	private void processNotifyAlarm(RequestEvent evt) {
@@ -292,6 +319,7 @@
 			logger.info("[鏀跺埌Notify-Alarm]锛歿}/{}", device.getDeviceId(), deviceAlarm.getChannelId());
 			if ("4".equals(deviceAlarm.getAlarmMethod())) {
 				MobilePosition mobilePosition = new MobilePosition();
+				mobilePosition.setChannelId(channelId);
 				mobilePosition.setCreateTime(DateUtil.getNow());
 				mobilePosition.setDeviceId(deviceAlarm.getDeviceId());
 				mobilePosition.setTime(deviceAlarm.getAlarmTime());
@@ -321,7 +349,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());
@@ -345,7 +373,7 @@
 
 	/***
 	 * 澶勭悊catalog璁惧鐩綍鍒楄〃Notify
-	 * 
+	 *
 	 * @param evt
 	 */
 	private void processNotifyCatalogList(RequestEvent evt) {
@@ -354,7 +382,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;
 			}
@@ -385,7 +413,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) {

--
Gitblit v1.8.0