From 588b1da35a1a51ddfca76fb8ca9c2c0c6cd70038 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期一, 13 五月 2024 11:12:20 +0800
Subject: [PATCH] Merge branch 'refs/heads/2.7.0' into 271-优化notify存储
---
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestForCatalogProcessor.java | 389 ++++++++++-----------
src/main/java/com/genersoft/iot/vmp/service/IDeviceChannelService.java | 5
src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java | 3
src/main/java/com/genersoft/iot/vmp/service/impl/DeviceChannelServiceImpl.java | 75 +++
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestForMobilePositionProcessor.java | 294 ++++++++-------
src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceMobilePositionMapper.java | 31 +
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/SIPRequestProcessorParent.java | 2
src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java | 5
src/main/java/com/genersoft/iot/vmp/gb28181/bean/DeviceChannel.java | 4
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java | 2
src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java | 2
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java | 6
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java | 205 ----------
src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java | 20
src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java | 4
src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java | 17
16 files changed, 492 insertions(+), 572 deletions(-)
diff --git a/src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java b/src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java
index a9f5c88..a9b17ae 100644
--- a/src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java
+++ b/src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java
@@ -66,7 +66,7 @@
private List<String> allowedOrigins = new ArrayList<>();
- private int maxNotifyCountQueue = 10000;
+ private int maxNotifyCountQueue = 100000;
private int registerAgainAfterTime = 60;
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/DeviceChannel.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/DeviceChannel.java
index 8290a45..6e230af 100644
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/DeviceChannel.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/DeviceChannel.java
@@ -535,11 +535,11 @@
this.subCount = subCount;
}
- public boolean isHasAudio() {
+ public Boolean getHasAudio() {
return hasAudio;
}
- public void setHasAudio(boolean hasAudio) {
+ public void setHasAudio(Boolean hasAudio) {
this.hasAudio = hasAudio;
}
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java b/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java
index 557563e..18ad2b0 100755
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java
@@ -49,6 +49,7 @@
ParentPlatform parentPlatform = null;
Map<String, List<ParentPlatform>> parentPlatformMap = new HashMap<>();
+ Map<String, DeviceChannel> channelMap = new HashMap<>();
if (!ObjectUtils.isEmpty(event.getPlatformId())) {
subscribe = subscribeHolder.getCatalogSubscribe(event.getPlatformId());
if (subscribe == null) {
@@ -67,6 +68,7 @@
for (DeviceChannel deviceChannel : event.getDeviceChannels()) {
List<ParentPlatform> parentPlatformsForGB = storager.queryPlatFormListForGBWithGBId(deviceChannel.getChannelId(), platforms);
parentPlatformMap.put(deviceChannel.getChannelId(), parentPlatformsForGB);
+ channelMap.put(deviceChannel.getChannelId(), deviceChannel);
}
}
}else if (event.getGbStreams() != null) {
@@ -174,7 +176,7 @@
}
logger.info("[Catalog浜嬩欢: {}]骞冲彴锛歿}锛屽奖鍝嶉�氶亾{}", event.getType(), platform.getServerGBId(), gbId);
List<DeviceChannel> deviceChannelList = new ArrayList<>();
- DeviceChannel deviceChannel = storager.queryChannelInParentPlatform(platform.getServerGBId(), gbId);
+ DeviceChannel deviceChannel = channelMap.get(gbId);
deviceChannelList.add(deviceChannel);
GbStream gbStream = storager.queryStreamInParentPlatform(platform.getServerGBId(), gbId);
if(gbStream != null){
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java
index 2082d39..cbf431a 100755
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java
@@ -520,9 +520,7 @@
if (parentPlatform == null) {
return;
}
- if (logger.isDebugEnabled()) {
- logger.debug("[鍙戦�� 绉诲姩浣嶇疆璁㈤槄] {}/{}->{},{}", parentPlatform.getServerGBId(), gpsMsgInfo.getId(), gpsMsgInfo.getLng(), gpsMsgInfo.getLat());
- }
+ logger.info("[鍙戦�� 绉诲姩浣嶇疆璁㈤槄] {}/{}->{},{}", parentPlatform.getServerGBId(), gpsMsgInfo.getId(), gpsMsgInfo.getLng(), gpsMsgInfo.getLat());
String characterSet = parentPlatform.getCharacterSet();
StringBuffer deviceStatusXml = new StringBuffer(600);
@@ -597,6 +595,7 @@
Integer finalIndex = index;
String catalogXmlContent = getCatalogXmlContentForCatalogAddOrUpdate(parentPlatform, channels,
deviceChannels.size(), type, subscribeInfo);
+ System.out.println(catalogXmlContent);
logger.info("[鍙戦�丯OTIFY閫氱煡]绫诲瀷锛� {}锛屽彂閫佹暟閲忥細 {}", type, channels.size());
sendNotify(parentPlatform, catalogXmlContent, subscribeInfo, eventResult -> {
logger.error("鍙戦�丯OTIFY閫氱煡娑堟伅澶辫触銆傞敊璇細{} {}", eventResult.statusCode, eventResult.msg);
@@ -626,7 +625,6 @@
private String getCatalogXmlContentForCatalogAddOrUpdate(ParentPlatform parentPlatform, List<DeviceChannel> channels, int sumNum, String type, SubscribeInfo subscribeInfo) {
StringBuffer catalogXml = new StringBuffer(600);
-
String characterSet = parentPlatform.getCharacterSet();
catalogXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet + "\"?>\r\n")
.append("<Notify>\r\n")
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/SIPRequestProcessorParent.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/SIPRequestProcessorParent.java
index 7cbfe70..f3f7431 100755
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/SIPRequestProcessorParent.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/SIPRequestProcessorParent.java
@@ -6,7 +6,6 @@
import com.google.common.primitives.Bytes;
import gov.nist.javax.sip.message.SIPRequest;
import gov.nist.javax.sip.message.SIPResponse;
-import org.apache.commons.lang3.ArrayUtils;
import org.dom4j.Document;
import org.dom4j.DocumentException;
import org.dom4j.Element;
@@ -172,6 +171,7 @@
return getRootElement(evt, "gb2312");
}
public Element getRootElement(RequestEvent evt, String charset) throws DocumentException {
+
if (charset == null) {
charset = "gb2312";
}
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestForCatalogProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestForCatalogProcessor.java
index cde70eb..6185cda 100755
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestForCatalogProcessor.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestForCatalogProcessor.java
@@ -1,11 +1,11 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;
-import com.genersoft.iot.vmp.conf.CivilCodeFileConf;
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;
@@ -19,7 +19,9 @@
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;
@@ -28,6 +30,7 @@
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CopyOnWriteArrayList;
/**
@@ -46,6 +49,7 @@
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;
@@ -63,222 +67,193 @@
private DynamicTask dynamicTask;
@Autowired
- private CivilCodeFileConf civilCodeFileConf;
-
- @Autowired
private SipConfig sipConfig;
- private final static String talkKey = "notify-request-for-catalog-task";
-
+ @Transactional
public void process(RequestEvent evt) {
- try {
- long start = System.currentTimeMillis();
- FromHeader fromHeader = (FromHeader) evt.getRequest().getHeader(FromHeader.NAME);
- String deviceId = SipUtils.getUserIdFromFromHeader(fromHeader);
+ if (taskQueue.size() >= userSetting.getMaxNotifyCountQueue()) {
+ logger.error("[notify-鐩綍璁㈤槄] 寰呭鐞嗘秷鎭槦鍒楀凡婊� {}锛岃繑鍥�486 BUSY_HERE锛屾秷鎭笉鍋氬鐞�", userSetting.getMaxNotifyCountQueue());
+ return;
+ }
+ taskQueue.offer(new HandlerCatchData(evt, null, null));
+ }
- Device device = redisCatchStorage.getDevice(deviceId);
- if (device == null || !device.isOnLine()) {
- logger.warn("[鏀跺埌鐩綍璁㈤槄]锛歿}, 浣嗘槸璁惧宸茬粡绂荤嚎", (device != null ? device.getDeviceId():"" ));
- return;
+ @Scheduled(fixedRate = 400) //姣�400姣鎵ц涓�娆�
+ public void executeTaskQueue(){
+ if (taskQueue.isEmpty()) {
+ return;
+ }
+ for (HandlerCatchData take : taskQueue) {
+ if (take == null) {
+ continue;
}
- Element rootElement = getRootElement(evt, device.getCharset());
- if (rootElement == null) {
- logger.warn("[ 鏀跺埌鐩綍璁㈤槄 ] content cannot be null, {}", evt.getRequest());
- return;
- }
- Element deviceListElement = rootElement.element("DeviceList");
- if (deviceListElement == null) {
- return;
- }
- Iterator<Element> deviceListIterator = deviceListElement.elementIterator();
- if (deviceListIterator != null) {
+ RequestEvent evt = take.getEvt();
+ try {
+ long start = System.currentTimeMillis();
+ FromHeader fromHeader = (FromHeader) evt.getRequest().getHeader(FromHeader.NAME);
+ String deviceId = SipUtils.getUserIdFromFromHeader(fromHeader);
- // 閬嶅巻DeviceList
- while (deviceListIterator.hasNext()) {
- Element itemDevice = deviceListIterator.next();
- Element channelDeviceElement = itemDevice.element("DeviceID");
- if (channelDeviceElement == null) {
- continue;
- }
- 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());
- logger.info("[鏀跺埌鐩綍璁㈤槄]锛歿}/{}", device.getDeviceId(), channel.getChannelId());
- switch (event) {
- case CatalogEvent.ON:
- // 涓婄嚎
- logger.info("[鏀跺埌閫氶亾涓婄嚎閫氱煡] 鏉ヨ嚜璁惧: {}, 閫氶亾 {}", device.getDeviceId(), channel.getChannelId());
- updateChannelOnlineList.add(channel);
- if (updateChannelOnlineList.size() > 300) {
- executeSaveForOnline();
- }
- if (userSetting.getDeviceStatusNotify()) {
- // 鍙戦�乺edis娑堟伅
- redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), channel.getChannelId(), true);
- }
+ Device device = redisCatchStorage.getDevice(deviceId);
+ if (device == null || !device.isOnLine()) {
+ logger.warn("[鏀跺埌鐩綍璁㈤槄]锛歿}, 浣嗘槸璁惧宸茬粡绂荤嚎", (device != null ? device.getDeviceId() : ""));
+ return;
+ }
+ Element rootElement = getRootElement(evt, device.getCharset());
+ if (rootElement == null) {
+ logger.warn("[ 鏀跺埌鐩綍璁㈤槄 ] content cannot be null, {}", evt.getRequest());
+ return;
+ }
+ Element deviceListElement = rootElement.element("DeviceList");
+ if (deviceListElement == null) {
+ return;
+ }
+ Iterator<Element> deviceListIterator = deviceListElement.elementIterator();
+ if (deviceListIterator != null) {
- break;
- case CatalogEvent.OFF :
- // 绂荤嚎
- logger.info("[鏀跺埌閫氶亾绂荤嚎閫氱煡] 鏉ヨ嚜璁惧: {}, 閫氶亾 {}", device.getDeviceId(), channel.getChannelId());
- if (userSetting.getRefuseChannelStatusChannelFormNotify()) {
- logger.info("[鏀跺埌閫氶亾绂荤嚎閫氱煡] 浣嗘槸骞冲彴宸查厤缃嫆缁濇娑堟伅锛屾潵鑷澶�: {}, 閫氶亾 {}", device.getDeviceId(), channel.getChannelId());
- }else {
- updateChannelOfflineList.add(channel);
- if (updateChannelOfflineList.size() > 300) {
- executeSaveForOffline();
- }
- if (userSetting.getDeviceStatusNotify()) {
- // 鍙戦�乺edis娑堟伅
- 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 (updateChannelOfflineList.size() > 300) {
- executeSaveForOffline();
- }
- if (userSetting.getDeviceStatusNotify()) {
- // 鍙戦�乺edis娑堟伅
- 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 (updateChannelOfflineList.size() > 300) {
- executeSaveForOffline();
- }
- if (userSetting.getDeviceStatusNotify()) {
- // 鍙戦�乺edis娑堟伅
- 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());
- updateChannelMap.put(channel.getChannelId(), channel);
- if (updateChannelMap.keySet().size() > 300) {
- executeSaveForUpdate();
- }
- }else {
- addChannelMap.put(channel.getChannelId(), channel);
- if (userSetting.getDeviceStatusNotify()) {
- // 鍙戦�乺edis娑堟伅
- redisCatchStorage.sendChannelAddOrDelete(device.getDeviceId(), channel.getChannelId(), true);
- }
-
- if (addChannelMap.keySet().size() > 300) {
- executeSaveForAdd();
- }
- }
-
- break;
- case CatalogEvent.DEL:
- // 鍒犻櫎
- logger.info("[鏀跺埌鍒犻櫎閫氶亾閫氱煡] 鏉ヨ嚜璁惧: {}, 閫氶亾 {}", device.getDeviceId(), channel.getChannelId());
- deleteChannelList.add(channel);
- if (userSetting.getDeviceStatusNotify()) {
- // 鍙戦�乺edis娑堟伅
- redisCatchStorage.sendChannelAddOrDelete(device.getDeviceId(), channel.getChannelId(), false);
- }
- if (deleteChannelList.size() > 300) {
- executeSaveForDelete();
- }
- 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());
- updateChannelMap.put(channel.getChannelId(), channel);
- if (updateChannelMap.keySet().size() > 300) {
- executeSaveForUpdate();
- }
- }else {
- addChannelMap.put(channel.getChannelId(), channel);
- if (addChannelMap.keySet().size() > 300) {
- executeSaveForAdd();
- }
- if (userSetting.getDeviceStatusNotify()) {
- // 鍙戦�乺edis娑堟伅
- redisCatchStorage.sendChannelAddOrDelete(device.getDeviceId(), channel.getChannelId(), true);
- }
- }
- break;
- default:
- logger.warn("[ NotifyCatalog ] event not found 锛� {}", event );
-
- }
- // 杞彂鍙樺寲淇℃伅
- eventPublisher.catalogEventPublish(null, channel, event);
-
- if (!updateChannelMap.keySet().isEmpty()
- || !addChannelMap.keySet().isEmpty()
- || !updateChannelOnlineList.isEmpty()
- || !updateChannelOfflineList.isEmpty()
- || !deleteChannelList.isEmpty()) {
-
- if (!dynamicTask.contains(talkKey)) {
- dynamicTask.startDelay(talkKey, this::executeSave, 1000);
+ // 閬嶅巻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());
+ logger.info("[鏀跺埌鐩綍璁㈤槄]锛歿}/{}", device.getDeviceId(), channel.getChannelId());
+ switch (event) {
+ case CatalogEvent.ON:
+ // 涓婄嚎
+ logger.info("[鏀跺埌閫氶亾涓婄嚎閫氱煡] 鏉ヨ嚜璁惧: {}, 閫氶亾 {}", device.getDeviceId(), channel.getChannelId());
+ updateChannelOnlineList.add(channel);
+ if (userSetting.getDeviceStatusNotify()) {
+ // 鍙戦�乺edis娑堟伅
+ 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 {
+ updateChannelOfflineList.add(channel);
+ if (userSetting.getDeviceStatusNotify()) {
+ // 鍙戦�乺edis娑堟伅
+ 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()) {
+ // 鍙戦�乺edis娑堟伅
+ 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()) {
+ // 鍙戦�乺edis娑堟伅
+ 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()) {
+ // 鍙戦�乺edis娑堟伅
+ redisCatchStorage.sendChannelAddOrDelete(device.getDeviceId(), channel.getChannelId(), true);
+ }
+ }
+
+ break;
+ case CatalogEvent.DEL:
+ // 鍒犻櫎
+ logger.info("[鏀跺埌鍒犻櫎閫氶亾閫氱煡] 鏉ヨ嚜璁惧: {}, 閫氶亾 {}", device.getDeviceId(), channel.getChannelId());
+ deleteChannelList.add(channel);
+ if (userSetting.getDeviceStatusNotify()) {
+ // 鍙戦�乺edis娑堟伅
+ 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()) {
+ // 鍙戦�乺edis娑堟伅
+ 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);
}
- } catch (DocumentException e) {
- logger.error("鏈鐞嗙殑寮傚父 ", e);
+ }
+ taskQueue.clear();
+ if (!updateChannelMap.keySet().isEmpty()
+ || !addChannelMap.keySet().isEmpty()
+ || !updateChannelOnlineList.isEmpty()
+ || !updateChannelOfflineList.isEmpty()
+ || !deleteChannelList.isEmpty()) {
+ executeSave();
}
}
- private void executeSave(){
+ public void executeSave(){
try {
executeSaveForAdd();
} catch (Exception e) {
logger.error("[瀛樺偍鏀跺埌鐨勫鍔犻�氶亾] 寮傚父锛� ", e );
- }
- try {
- executeSaveForUpdate();
- } catch (Exception e) {
- logger.error("[瀛樺偍鏀跺埌鐨勬洿鏂伴�氶亾] 寮傚父锛� ", e );
- }
- try {
- executeSaveForDelete();
- } catch (Exception e) {
- logger.error("[瀛樺偍鏀跺埌鐨勫垹闄ら�氶亾] 寮傚父锛� ", e );
}
try {
executeSaveForOnline();
@@ -290,7 +265,16 @@
} catch (Exception e) {
logger.error("[瀛樺偍鏀跺埌鐨勯�氶亾绂荤嚎] 寮傚父锛� ", e );
}
- dynamicTask.stop(talkKey);
+ try {
+ executeSaveForUpdate();
+ } catch (Exception e) {
+ logger.error("[瀛樺偍鏀跺埌鐨勬洿鏂伴�氶亾] 寮傚父锛� ", e );
+ }
+ try {
+ executeSaveForDelete();
+ } catch (Exception e) {
+ logger.error("[瀛樺偍鏀跺埌鐨勫垹闄ら�氶亾] 寮傚父锛� ", e );
+ }
}
private void executeSaveForUpdate(){
@@ -300,7 +284,6 @@
deviceChannelService.batchUpdateChannel(deviceChannels);
updateChannelMap.clear();
}
-
}
private void executeSaveForAdd(){
@@ -332,4 +315,8 @@
}
}
+ @Scheduled(fixedRate = 10000) //姣�1绉掓墽琛屼竴娆�
+ public void execute(){
+ logger.info("[寰呭鐞哊otify-鐩綍璁㈤槄娑堟伅鏁伴噺]: {}", taskQueue.size());
+ }
}
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestForMobilePositionProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestForMobilePositionProcessor.java
index 5157506..013d95e 100755
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestForMobilePositionProcessor.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestForMobilePositionProcessor.java
@@ -1,17 +1,15 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;
-import com.genersoft.iot.vmp.conf.CivilCodeFileConf;
-import com.genersoft.iot.vmp.conf.DynamicTask;
-import com.genersoft.iot.vmp.conf.SipConfig;
+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;
import com.genersoft.iot.vmp.gb28181.utils.NumericUtil;
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;
@@ -20,7 +18,9 @@
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;
import javax.sip.RequestEvent;
@@ -29,7 +29,7 @@
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ConcurrentLinkedQueue;
/**
* SIP鍛戒护绫诲瀷锛� NOTIFY璇锋眰涓殑绉诲姩浣嶇疆璇锋眰澶勭悊
@@ -40,10 +40,7 @@
private final static Logger logger = LoggerFactory.getLogger(NotifyRequestForMobilePositionProcessor.class);
- private final Map<String, DeviceChannel> updateChannelMap = new ConcurrentHashMap<>();
-
- private final List<MobilePosition> addMobilePositionList = new CopyOnWriteArrayList();
-
+ private ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>();
@Autowired
private UserSetting userSetting;
@@ -57,158 +54,169 @@
@Autowired
private IDeviceChannelService deviceChannelService;
- @Autowired
- private DynamicTask dynamicTask;
-
- @Autowired
- private CivilCodeFileConf civilCodeFileConf;
-
- @Autowired
- private SipConfig sipConfig;
-
- private final static String talkKey = "notify-request-for-mobile-position-task";
-
public void process(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;
+ 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 executeTaskQueue() {
+ if (taskQueue.isEmpty()) {
+ return;
+ }
+ Map<String, DeviceChannel> updateChannelMap = new ConcurrentHashMap<>();
+ List<MobilePosition> addMobilePositionList = new ArrayList<>();
+ for (HandlerCatchData take : taskQueue) {
+ if (take == null) {
+ continue;
}
-
- 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);
+ RequestEvent evt = take.getEvt();
+ try {
+ FromHeader fromHeader = (FromHeader) evt.getRequest().getHeader(FromHeader.NAME);
+ String deviceId = SipUtils.getUserIdFromFromHeader(fromHeader);
+ long startTime = System.currentTimeMillis();
+ // 鍥炲 200 OK
+ Element rootElement = getRootElement(evt);
+ if (rootElement == null) {
+ logger.error("澶勭悊MobilePosition绉诲姩浣嶇疆Notify鏃舵湭鑾峰彇鍒版秷鎭綋,{}", evt.getRequest());
+ return;
+ }
+ Device device = redisCatchStorage.getDevice(deviceId);
if (device == null) {
- // 鏍规嵁閫氶亾id鏌ヨ璁惧Id
- List<Device> deviceList = deviceChannelService.getDeviceByChannelId(channelId);
- if (deviceList.size() > 0) {
- device = deviceList.get(0);
+ logger.error("澶勭悊MobilePosition绉诲姩浣嶇疆Notify鏃舵湭鑾峰彇鍒癲evice,{}", deviceId);
+ return;
+ }
+ MobilePosition mobilePosition = new MobilePosition();
+ mobilePosition.setDeviceId(device.getDeviceId());
+ mobilePosition.setDeviceName(device.getName());
+ mobilePosition.setCreateTime(DateUtil.getNow());
+ List<Element> elements = rootElement.elements();
+ for (Element element : elements) {
+ switch (element.getName()){
+ case "DeviceID":
+ String channelId = element.getStringValue();
+ if (!deviceId.equals(channelId)) {
+ mobilePosition.setChannelId(channelId);
+ }
+ continue;
+ case "Time":
+ String timeVal = element.getStringValue();
+ if (ObjectUtils.isEmpty(timeVal)) {
+ mobilePosition.setTime(DateUtil.getNow());
+ } else {
+ mobilePosition.setTime(SipUtils.parseTime(timeVal));
+ }
+ continue;
+ case "Longitude":
+ mobilePosition.setLongitude(Double.parseDouble(element.getStringValue()));
+ continue;
+ case "Latitude":
+ mobilePosition.setLatitude(Double.parseDouble(element.getStringValue()));
+ continue;
+ case "Speed":
+ String speedVal = element.getStringValue();
+ if (NumericUtil.isDouble(speedVal)) {
+ mobilePosition.setSpeed(Double.parseDouble(speedVal));
+ } else {
+ mobilePosition.setSpeed(0.0);
+ }
+ continue;
+ case "Direction":
+ String directionVal = element.getStringValue();
+ if (NumericUtil.isDouble(directionVal)) {
+ mobilePosition.setDirection(Double.parseDouble(directionVal));
+ } else {
+ mobilePosition.setDirection(0.0);
+ }
+ continue;
+ case "Altitude":
+ String altitudeVal = element.getStringValue();
+ if (NumericUtil.isDouble(altitudeVal)) {
+ mobilePosition.setAltitude(Double.parseDouble(altitudeVal));
+ } else {
+ mobilePosition.setAltitude(0.0);
+ }
+ continue;
+
}
}
- }
- 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();
+
+// logger.info("[鏀跺埌绉诲姩浣嶇疆璁㈤槄閫氱煡]锛歿}/{}->{}.{}, 鏃堕棿锛� {}", mobilePosition.getDeviceId(), mobilePosition.getChannelId(),
+// mobilePosition.getLongitude(), mobilePosition.getLatitude(), System.currentTimeMillis() - startTime);
+ mobilePosition.setReportSource("Mobile Position");
+
+ // 鏇存柊device channel 鐨勭粡绾害
+ DeviceChannel deviceChannel = new DeviceChannel();
+ deviceChannel.setDeviceId(device.getDeviceId());
+ deviceChannel.setLongitude(mobilePosition.getLongitude());
+ deviceChannel.setLatitude(mobilePosition.getLatitude());
+ deviceChannel.setGpsTime(mobilePosition.getTime());
+ updateChannelMap.put(deviceId + mobilePosition.getChannelId(), deviceChannel);
+ addMobilePositionList.add(mobilePosition);
+
+
+ // 鍚戝叧鑱斾簡璇ラ�氶亾骞朵笖寮�鍚Щ鍔ㄤ綅缃闃呯殑涓婄骇骞冲彴鍙戦�佺Щ鍔ㄤ綅缃闃呮秷鎭�
+ try {
+ eventPublisher.mobilePositionEventPublish(mobilePosition);
+ }catch (Exception e) {
+ logger.error("[鍚戜笂绾ц浆鍙戠Щ鍔ㄤ綅缃け璐 ", e);
}
- }
- if (!ObjectUtils.isEmpty(device.getName())) {
- mobilePosition.setDeviceName(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());
- updateChannelMap.put(channelId, deviceChannel);
- addMobilePositionList.add(mobilePosition);
- if(updateChannelMap.size() > 300) {
- executeSaveChannel();
- }
- if (userSetting.isSavePositionHistory()) {
- if(addMobilePositionList.size() > 300) {
- executeSaveMobilePosition();
+ if (mobilePosition.getChannelId() == null || mobilePosition.getChannelId().equals(mobilePosition.getDeviceId()) ) {
+ List<DeviceChannel> channels = deviceChannelService.queryChaneListByDeviceId(mobilePosition.getDeviceId());
+ channels.forEach(channel -> {
+ // 鍙戦�乺edis娑堟伅銆� 閫氱煡浣嶇疆淇℃伅鐨勫彉鍖�
+ JSONObject jsonObject = new JSONObject();
+ jsonObject.put("time", DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(mobilePosition.getTime()));
+ jsonObject.put("serial", channel.getDeviceId());
+ jsonObject.put("code", channel.getChannelId());
+ jsonObject.put("longitude", mobilePosition.getLongitude());
+ jsonObject.put("latitude", mobilePosition.getLatitude());
+ jsonObject.put("altitude", mobilePosition.getAltitude());
+ jsonObject.put("direction", mobilePosition.getDirection());
+ jsonObject.put("speed", mobilePosition.getSpeed());
+ redisCatchStorage.sendMobilePositionMsg(jsonObject);
+ });
+ }else {
+ // 鍙戦�乺edis娑堟伅銆� 閫氱煡浣嶇疆淇℃伅鐨勫彉鍖�
+ JSONObject jsonObject = new JSONObject();
+ jsonObject.put("time", DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(mobilePosition.getTime()));
+ jsonObject.put("serial", mobilePosition.getDeviceId());
+ jsonObject.put("code", mobilePosition.getChannelId());
+ jsonObject.put("longitude", mobilePosition.getLongitude());
+ jsonObject.put("latitude", mobilePosition.getLatitude());
+ jsonObject.put("altitude", mobilePosition.getAltitude());
+ jsonObject.put("direction", mobilePosition.getDirection());
+ jsonObject.put("speed", mobilePosition.getSpeed());
+ redisCatchStorage.sendMobilePositionMsg(jsonObject);
}
+ } catch (DocumentException e) {
+ logger.error("鏈鐞嗙殑寮傚父 ", e);
}
-
-// 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);
-
- if (!dynamicTask.contains(talkKey)) {
- dynamicTask.startDelay(talkKey, this::executeSave, 1000);
- }
-
- } catch (DocumentException e) {
- logger.error("鏈鐞嗙殑寮傚父 ", e);
}
-
-
- }
-
- private void executeSave(){
- executeSaveChannel();
- executeSaveMobilePosition();
- dynamicTask.stop(talkKey);
- }
-
- private void executeSaveChannel(){
- try {
- logger.info("[绉诲姩浣嶇疆璁㈤槄]鏇存柊閫氶亾浣嶇疆锛� {}", updateChannelMap.size());
- ArrayList<DeviceChannel> deviceChannels = new ArrayList<>(updateChannelMap.values());
- deviceChannelService.batchUpdateChannelGPS(deviceChannels);
+ taskQueue.clear();
+ if(!updateChannelMap.isEmpty()) {
+ List<DeviceChannel> channels = new ArrayList<>(updateChannelMap.values());
+ logger.info("[绉诲姩浣嶇疆璁㈤槄]鏇存柊閫氶亾浣嶇疆锛� {}", channels.size());
+ deviceChannelService.batchUpdateChannel(channels);
updateChannelMap.clear();
- }catch (Exception e) {
-
}
- }
-
- private void executeSaveMobilePosition(){
- if (userSetting.isSavePositionHistory()) {
+ if (userSetting.isSavePositionHistory() && !addMobilePositionList.isEmpty()) {
try {
logger.info("[绉诲姩浣嶇疆璁㈤槄] 娣诲姞閫氶亾杞ㄨ抗鐐逛綅锛� {}", addMobilePositionList.size());
deviceChannelService.batchAddMobilePosition(addMobilePositionList);
- addMobilePositionList.clear();
}catch (Exception e) {
logger.info("[绉诲姩浣嶇疆璁㈤槄] b娣诲姞閫氶亾杞ㄨ抗鐐逛綅淇濆瓨澶辫触锛� {}", addMobilePositionList.size());
}
+ addMobilePositionList.clear();
}
}
-
-
-
+ @Scheduled(fixedRate = 10000)
+ public void execute(){
+ logger.info("[寰呭鐞哊otify-绉诲姩浣嶇疆璁㈤槄娑堟伅鏁伴噺]: {}", taskQueue.size());
+ }
}
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
index 84f44b5..8da07a1 100755
--- 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,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.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,10 +19,7 @@
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.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
-import org.springframework.util.ObjectUtils;
import javax.sip.InvalidArgumentException;
import javax.sip.RequestEvent;
@@ -35,8 +27,6 @@
import javax.sip.header.FromHeader;
import javax.sip.message.Response;
import java.text.ParseException;
-import java.util.List;
-import java.util.concurrent.ConcurrentLinkedQueue;
/**
* SIP鍛戒护绫诲瀷锛� NOTIFY璇锋眰,杩欐槸浣滀负涓婄骇鍙戦�佽闃呰姹傚悗锛岃澶囨墠浼氬搷搴旂殑
@@ -46,15 +36,6 @@
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;
@@ -76,13 +57,8 @@
@Autowired
private NotifyRequestForCatalogProcessor notifyRequestForCatalogProcessor;
- private ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>();
-
- @Qualifier("taskExecutor")
@Autowired
- private ThreadPoolTaskExecutor taskExecutor;
-
- private int maxQueueCount = 30000;
+ private NotifyRequestForMobilePositionProcessor notifyRequestForMobilePositionProcessor;
@Override
public void afterPropertiesSet() throws Exception {
@@ -93,159 +69,33 @@
@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();
- 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());
- continue;
- }
- String cmd = XmlUtil.getText(rootElement, "CmdType");
-
- if (CmdType.CATALOG.equals(cmd)) {
- logger.info("鎺ユ敹鍒癈atalog閫氱煡");
- notifyRequestForCatalogProcessor.process(take.getEvt());
- } else if (CmdType.ALARM.equals(cmd)) {
- logger.info("鎺ユ敹鍒癆larm閫氱煡");
- processNotifyAlarm(take.getEvt());
- } else if (CmdType.MOBILE_POSITION.equals(cmd)) {
- logger.info("鎺ユ敹鍒癕obilePosition閫氱煡");
- processNotifyMobilePosition(take.getEvt());
- } else {
- logger.info("鎺ユ敹鍒版秷鎭細" + cmd);
- }
- } catch (DocumentException e) {
- logger.error("澶勭悊NOTIFY娑堟伅鏃堕敊璇�", e);
- }
- }
- });
- }
- }
-
- /**
- * 澶勭悊MobilePosition绉诲姩浣嶇疆Notify
- *
- * @param evt
- */
- private void processNotifyMobilePosition(RequestEvent evt) {
- try {
- FromHeader fromHeader = (FromHeader) evt.getRequest().getHeader(FromHeader.NAME);
- String deviceId = SipUtils.getUserIdFromFromHeader(fromHeader);
-
- // 鍥炲 200 OK
+ responseAck((SIPRequest) evt.getRequest(), Response.OK, null, null);
Element rootElement = getRootElement(evt);
if (rootElement == null) {
- logger.error("澶勭悊MobilePosition绉诲姩浣嶇疆Notify鏃舵湭鑾峰彇鍒版秷鎭綋,{}", evt.getRequest());
+ logger.error("澶勭悊NOTIFY娑堟伅鏃舵湭鑾峰彇鍒版秷鎭綋,{}", evt.getRequest());
+ responseAck((SIPRequest) evt.getRequest(), Response.OK, null, null);
return;
}
+ String cmd = XmlUtil.getText(rootElement, "CmdType");
- 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")));
+ if (CmdType.CATALOG.equals(cmd)) {
+ notifyRequestForCatalogProcessor.process(evt);
+ } else if (CmdType.ALARM.equals(cmd)) {
+ processNotifyAlarm(evt);
+ } else if (CmdType.MOBILE_POSITION.equals(cmd)) {
+ notifyRequestForMobilePositionProcessor.process(evt);
} else {
- mobilePosition.setSpeed(0.0);
+ logger.info("鎺ユ敹鍒版秷鎭細" + cmd);
}
- 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
- *
- * @param evt
*/
private void processNotifyAlarm(RequestEvent evt) {
if (!sipConfig.isAlarm()) {
@@ -336,28 +186,5 @@
}
}
- 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;
- }
}
diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
index de98916..f3d7477 100755
--- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
+++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
@@ -289,7 +289,7 @@
String channelId = ssrcTransactionForAll.get(0).getChannelId();
DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
if (deviceChannel != null) {
- result.setEnable_audio(deviceChannel.isHasAudio());
+ result.setEnable_audio(deviceChannel.getHasAudio());
}
// 濡傛灉鏄綍鍍忎笅杞藉氨璁剧疆瑙嗛闂撮殧鍗佺
if (ssrcTransactionForAll.get(0).getType() == InviteSessionType.DOWNLOAD) {
diff --git a/src/main/java/com/genersoft/iot/vmp/service/IDeviceChannelService.java b/src/main/java/com/genersoft/iot/vmp/service/IDeviceChannelService.java
index 7dd04e8..a8cd581 100755
--- a/src/main/java/com/genersoft/iot/vmp/service/IDeviceChannelService.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/IDeviceChannelService.java
@@ -102,4 +102,9 @@
void batchAddMobilePosition(List<MobilePosition> addMobilePositionList);
+ void online(DeviceChannel channel);
+
+ void offline(DeviceChannel channel);
+
+ void delete(DeviceChannel channel);
}
diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceChannelServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceChannelServiceImpl.java
index 1836636..58e7b02 100755
--- a/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceChannelServiceImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceChannelServiceImpl.java
@@ -23,6 +23,7 @@
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.ObjectUtils;
import java.util.ArrayList;
@@ -248,8 +249,24 @@
}
@Override
+ public void online(DeviceChannel channel) {
+ channelMapper.online(channel.getDeviceId(), channel.getChannelId());
+ }
+
+ @Override
public int channelsOffline(List<DeviceChannel> channels) {
return channelMapper.batchOffline(channels);
+ }
+
+
+ @Override
+ public void offline(DeviceChannel channel) {
+ channelMapper.offline(channel.getDeviceId(), channel.getChannelId());
+ }
+
+ @Override
+ public void delete(DeviceChannel channel) {
+ channelMapper.del(channel.getDeviceId(), channel.getChannelId());
}
@Override
@@ -258,15 +275,23 @@
}
@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);
}
- channelMapper.batchUpdate(channels);
- for (DeviceChannel channel : channels) {
- if (channel.getParentId() != null) {
- channelMapper.updateChannelSubCount(channel.getDeviceId(), channel.getParentId());
+ 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);
}
}
}
@@ -355,15 +380,45 @@
}
@Override
+ @Transactional
public void batchUpdateChannelGPS(List<DeviceChannel> channelList) {
- channelMapper.batchUpdatePosition(channelList);
+ for (DeviceChannel deviceChannel : channelList) {
+ deviceChannel.setUpdateTime(DateUtil.getNow());
+ if (deviceChannel.getGpsTime() == null) {
+ deviceChannel.setGpsTime(DateUtil.getNow());
+ }
+ }
+ int count = 1000;
+ if (channelList.size() > count) {
+ for (int i = 0; i < channelList.size(); i+=count) {
+ int toIndex = i+count;
+ if ( i + count > channelList.size()) {
+ toIndex = channelList.size();
+ }
+ List<DeviceChannel> channels = channelList.subList(i, toIndex);
+ channelMapper.batchUpdatePosition(channels);
+ }
+ }else {
+ channelMapper.batchUpdatePosition(channelList);
+ }
}
@Override
+ @Transactional
public void batchAddMobilePosition(List<MobilePosition> mobilePositions) {
- deviceMobilePositionMapper.batchInsert(mobilePositions);
+// int count = 500;
+// if (mobilePositions.size() > count) {
+// for (int i = 0; i < mobilePositions.size(); i+=count) {
+// int toIndex = i+count;
+// if ( i + count > mobilePositions.size()) {
+// toIndex = mobilePositions.size();
+// }
+// List<MobilePosition> mobilePositionsSub = mobilePositions.subList(i, toIndex);
+// deviceMobilePositionMapper.batchadd(mobilePositionsSub);
+// }
+// }else {
+// deviceMobilePositionMapper.batchadd(mobilePositions);
+// }
+ deviceMobilePositionMapper.batchadd(mobilePositions);
}
-
-
-
}
diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
index bc05276..26867ee 100755
--- a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
@@ -149,7 +149,6 @@
logger.warn("[鐐规挱] 鏈壘鍒板彲鐢ㄧ殑zlm deviceId: {},channelId:{}", deviceId, channelId);
throw new ControllerException(ErrorCode.ERROR100.getCode(), "鏈壘鍒板彲鐢ㄧ殑zlm");
}
-
Device device = redisCatchStorage.getDevice(deviceId);
if (device.getStreamMode().equalsIgnoreCase("TCP-ACTIVE") && !mediaServerItem.isRtpEnable()) {
logger.warn("[鐐规挱] 鍗曠鍙ｆ敹娴佹椂涓嶆敮鎸乀CP涓诲姩鏂瑰紡鏀舵祦 deviceId: {},channelId:{}", deviceId, channelId);
@@ -163,6 +162,8 @@
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId);
if (inviteInfo != null ) {
if (inviteInfo.getStreamInfo() == null) {
+ // 閲婃斁鐢熸垚鐨剆src锛屼娇鐢ㄤ笂涓�娆＄敵璇风殑
+ ssrcFactory.releaseSsrc(mediaServerItem.getId(), ssrc);
// 鐐规挱鍙戣捣浜嗕絾鏄皻鏈垚鍔�, 浠呮敞鍐屽洖璋冪瓑寰呯粨鏋滃嵆鍙�
inviteStreamService.once(InviteSessionType.PLAY, deviceId, channelId, null, callback);
logger.info("[鐐规挱寮�濮媇 宸茬粡璇锋眰涓紝绛夊緟缁撴灉锛� deviceId: {}, channelId: {}", device.getDeviceId(), channelId);
diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
index f0230f7..3869f0b 100755
--- a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
@@ -170,15 +170,19 @@
callback.run(ErrorCode.ERROR100.getCode(), "淇濆瓨澶辫触", null);
return;
}
- HookSubscribeForStreamChange hookSubscribeForStreamChange = HookSubscribeFactory.on_stream_changed(param.getApp(), param.getStream(), true, "rtsp", mediaInfo.getId());
- hookSubscribe.addSubscribe(hookSubscribeForStreamChange, (mediaServerItem, response) -> {
- StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(
- mediaInfo, param.getApp(), param.getStream(), null, null);
- callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), streamInfo);
- });
+
if (param.isEnable()) {
String talkKey = UUID.randomUUID().toString();
String delayTalkKey = UUID.randomUUID().toString();
+ HookSubscribeForStreamChange hookSubscribeForStreamChange = HookSubscribeFactory.on_stream_changed(param.getApp(), param.getStream(), true, "rtsp", mediaInfo.getId());
+ hookSubscribe.addSubscribe(hookSubscribeForStreamChange, (mediaServerItem, response) -> {
+ dynamicTask.stop(delayTalkKey);
+ hookSubscribe.removeSubscribe(hookSubscribeForStreamChange);
+ StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(
+ mediaInfo, param.getApp(), param.getStream(), null, null);
+ callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), streamInfo);
+ });
+
dynamicTask.startDelay(delayTalkKey, ()->{
StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStreamWithCheck(param.getApp(), param.getStream(), mediaInfo.getId(), false);
if (streamInfo != null) {
@@ -324,6 +328,9 @@
zlmresTfulUtils.closeStreams(mediaServerItem, param.getApp(), param.getStream());
}
if ("ffmpeg".equalsIgnoreCase(param.getType())){
+ if (param.getTimeoutMs() == 0) {
+ param.setTimeoutMs(15);
+ }
result = zlmresTfulUtils.addFFmpegSource(mediaServerItem, param.getSrcUrl().trim(), param.getDstUrl(),
param.getTimeoutMs() + "", param.isEnableAudio(), param.isEnableMp4(),
param.getFfmpegCmdKey());
@@ -379,6 +386,7 @@
gbStreamMapper.del(app, stream);
videoManagerStorager.deleteStreamProxy(app, stream);
redisCatchStorage.removeStream(streamProxyItem.getMediaServerId(), "PULL", app, stream);
+ redisCatchStorage.removeStream(streamProxyItem.getMediaServerId(), "PUSH", app, stream);
JSONObject jsonObject = removeStreamProxyFromZlm(streamProxyItem);
if (jsonObject != null && jsonObject.getInteger("code") == 0) {
logger.info("[绉婚櫎浠ｇ悊]锛� 浠ｇ悊锛� {}/{}, 浠巣lm绉婚櫎鎴愬姛", app, stream);
diff --git a/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java b/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java
index 1555221..be78a52 100755
--- a/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java
+++ b/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java
@@ -401,6 +401,23 @@
" </script>"})
int updatePosition(DeviceChannel deviceChannel);
+ @Update({"<script>" +
+ "<foreach collection='deviceChannelList' item='item' separator=';'>" +
+ " UPDATE" +
+ " wvp_device_channel" +
+ " SET gps_time=#{item.gpsTime}" +
+ "<if test='item.longitude != null'>, longitude=#{item.longitude}</if>" +
+ "<if test='item.latitude != null'>, latitude=#{item.latitude}</if>" +
+ "<if test='item.longitudeGcj02 != null'>, longitude_gcj02=#{item.longitudeGcj02}</if>" +
+ "<if test='item.latitudeGcj02 != null'>, latitude_gcj02=#{item.latitudeGcj02}</if>" +
+ "<if test='item.longitudeWgs84 != null'>, longitude_wgs84=#{item.longitudeWgs84}</if>" +
+ "<if test='item.latitudeWgs84 != null'>, latitude_wgs84=#{item.latitudeWgs84}</if>" +
+ "WHERE device_id=#{item.deviceId} " +
+ " <if test='item.channelId != null' > AND channel_id=#{item.channelId}</if>" +
+ "</foreach>" +
+ "</script>"})
+ int batchUpdatePosition(List<DeviceChannel> deviceChannelList);
+
@Select("SELECT * FROM wvp_device_channel WHERE length(trim(stream_id)) > 0")
List<DeviceChannel> getAllChannelInPlay();
diff --git a/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceMobilePositionMapper.java b/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceMobilePositionMapper.java
index 19494f0..c28b16e 100755
--- a/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceMobilePositionMapper.java
+++ b/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceMobilePositionMapper.java
@@ -13,8 +13,7 @@
public interface DeviceMobilePositionMapper {
@Insert("INSERT INTO wvp_device_mobile_position (device_id,channel_id, device_name,time,longitude,latitude,altitude,speed,direction,report_source,longitude_gcj02,latitude_gcj02,longitude_wgs84,latitude_wgs84,create_time)"+
- "VALUES (#{deviceId}, #{channelId}, #{deviceName}, #{time}, #{longitude}, #{latitude}, #{altitude}, " +
- "#{speed}, #{direction}, #{reportSource}, #{longitudeGcj02}, #{latitudeGcj02}, #{longitudeWgs84}, #{latitudeWgs84}, #{createTime})")
+ "VALUES (#{deviceId}, #{channelId}, #{deviceName}, #{time}, #{longitude}, #{latitude}, #{altitude}, #{speed}, #{direction}, #{reportSource}, #{longitudeGcj02}, #{latitudeGcj02}, #{longitudeWgs84}, #{latitudeWgs84}, #{createTime})")
int insertNewPosition(MobilePosition mobilePosition);
@Select(value = {" <script>" +
@@ -34,19 +33,33 @@
@Delete("DELETE FROM wvp_device_mobile_position WHERE device_id = #{deviceId}")
int clearMobilePositionsByDeviceId(String deviceId);
+
@Insert("<script> " +
"insert into wvp_device_mobile_position " +
- "(device_id,channel_id, device_name,time,longitude," +
- "latitude,altitude,speed,direction," +
- "report_source,longitude_gcj02,latitude_gcj02," +
- "longitude_wgs84,latitude_wgs84,create_time)"+
+ "(device_id,channel_id, device_name,time,longitude,latitude,altitude,speed,direction,report_source," +
+ "longitude_gcj02,latitude_gcj02,longitude_wgs84,latitude_wgs84,create_time)"+
"values " +
"<foreach collection='mobilePositions' index='index' item='item' separator=','> " +
"(#{item.deviceId}, #{item.channelId}, #{item.deviceName}, #{item.time}, #{item.longitude}, " +
"#{item.latitude}, #{item.altitude}, #{item.speed},#{item.direction}," +
- "#{item.reportSource}, #{item.longitudeGcj02}, #{item.latitudeGcj02}, " +
- "#{item.longitudeWgs84}, #{item.latitudeWgs84}, #{item.createTime}) " +
+ "#{item.reportSource}, #{item.longitudeGcj02}, #{item.latitudeGcj02}, #{item.longitudeWgs84}, #{item.latitudeWgs84}, " +
+ "#{item.createTime}) " +
"</foreach> " +
"</script>")
- void batchInsert(List<MobilePosition> mobilePositions);
+ void batchadd2(List<MobilePosition> mobilePositions);
+
+ @Insert("<script> " +
+ "<foreach collection='mobilePositions' index='index' item='item' separator=','> " +
+ "insert into wvp_device_mobile_position " +
+ "(device_id,channel_id, device_name,time,longitude,latitude,altitude,speed,direction,report_source," +
+ "longitude_gcj02,latitude_gcj02,longitude_wgs84,latitude_wgs84,create_time)"+
+ "values " +
+ "(#{item.deviceId}, #{item.channelId}, #{item.deviceName}, #{item.time}, #{item.longitude}, " +
+ "#{item.latitude}, #{item.altitude}, #{item.speed},#{item.direction}," +
+ "#{item.reportSource}, #{item.longitudeGcj02}, #{item.latitudeGcj02}, #{item.longitudeWgs84}, #{item.latitudeWgs84}, " +
+ "#{item.createTime}); " +
+ "</foreach> " +
+ "</script>")
+ void batchadd(List<MobilePosition> mobilePositions);
+
}
diff --git a/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java b/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java
index 509240f..b1eef31 100755
--- a/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java
@@ -7,7 +7,6 @@
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 @@
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 @@
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)){
--
Gitblit v1.8.0