From e6ee7fe7477b485676ce506b4e971c9a50dfa588 Mon Sep 17 00:00:00 2001 From: 648540858 <456panlinlin> Date: 星期四, 14 四月 2022 16:52:48 +0800 Subject: [PATCH] 优化级联移动位置订阅位置更新 --- src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeInfo.java | 24 ++++ src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/CatalogSubscribeTask.java | 1 src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java | 26 ++-- src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeHolder.java | 51 ++++++++ src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/SubscribeRequestProcessor.java | 32 ++-- src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java | 3 src/main/java/com/genersoft/iot/vmp/vmanager/server/ServerController.java | 42 ++++++ /dev/null | 50 -------- src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/MobilePositionSubscribeTask.java | 26 +++ src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java | 6 src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/MobilePositionSubscribeHandlerTask.java | 30 ++-- src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java | 55 ++++++++- src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java | 2 13 files changed, 227 insertions(+), 121 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java b/src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java index b842382..bd10317 100644 --- a/src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java +++ b/src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java @@ -1,6 +1,9 @@ package com.genersoft.iot.vmp.conf; import com.genersoft.iot.vmp.gb28181.task.ISubscribeTask; +import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd.CatalogResponseMessageHandler; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; @@ -18,6 +21,8 @@ @Component public class DynamicTask { + private Logger logger = LoggerFactory.getLogger(DynamicTask.class); + @Autowired private ThreadPoolTaskScheduler threadPoolTaskScheduler; @@ -26,7 +31,12 @@ @Bean public ThreadPoolTaskScheduler threadPoolTaskScheduler() { - return new ThreadPoolTaskScheduler(); + ThreadPoolTaskScheduler schedulerPool = new ThreadPoolTaskScheduler(); + schedulerPool.setPoolSize(300); + schedulerPool.setWaitForTasksToCompleteOnShutdown(true); + schedulerPool.setAwaitTerminationSeconds(10); + return schedulerPool; + } /** @@ -37,11 +47,24 @@ * @return */ public void startCron(String key, Runnable task, int cycleForCatalog) { - stop(key); + ScheduledFuture future = futureMap.get(key); + if (future != null) { + if (future.isCancelled()) { + logger.info("浠诲姟銆恵}銆戝凡瀛樺湪浣嗘槸鍏抽棴鐘舵�侊紒锛侊紒", key); + } else { + logger.info("浠诲姟銆恵}銆戝凡瀛樺湪涓斿凡鍚姩锛侊紒锛�", key); + return; + } + } // scheduleWithFixedDelay 蹇呴』绛夊緟涓婁竴涓换鍔$粨鏉熸墠寮�濮嬭鏃秔eriod锛� cycleForCatalog琛ㄧず鎵ц鐨勯棿闅� - ScheduledFuture future = threadPoolTaskScheduler.scheduleAtFixedRate(task, cycleForCatalog * 1000L); - futureMap.put(key, future); - runnableMap.put(key, task); + future = threadPoolTaskScheduler.scheduleAtFixedRate(task, cycleForCatalog * 1000L); + if (future != null){ + futureMap.put(key, future); + runnableMap.put(key, task); + logger.info("浠诲姟銆恵}銆戝惎鍔ㄦ垚鍔燂紒锛侊紒", key); + }else { + logger.info("浠诲姟銆恵}銆戝惎鍔ㄥけ璐ワ紒锛侊紒", key); + } } /** @@ -53,13 +76,31 @@ */ public void startDelay(String key, Runnable task, int delay) { stop(key); + System.out.println("瀹氭椂浠诲姟寮�濮嬩簡"); Date starTime = new Date(System.currentTimeMillis() + delay); + + ScheduledFuture future = futureMap.get(key); + if (future != null) { + if (future.isCancelled()) { + logger.info("浠诲姟銆恵}銆戝凡瀛樺湪浣嗘槸鍏抽棴鐘舵�侊紒锛侊紒", key); + } else { + logger.info("浠诲姟銆恵}銆戝凡瀛樺湪涓斿凡鍚姩锛侊紒锛�", key); + return; + } + } // scheduleWithFixedDelay 蹇呴』绛夊緟涓婁竴涓换鍔$粨鏉熸墠寮�濮嬭鏃秔eriod锛� cycleForCatalog琛ㄧず鎵ц鐨勯棿闅� - ScheduledFuture future = threadPoolTaskScheduler.schedule(task, starTime); - futureMap.put(key, future); + future = threadPoolTaskScheduler.schedule(task, starTime); + if (future != null){ + futureMap.put(key, future); + runnableMap.put(key, task); + logger.info("浠诲姟銆恵}銆戝惎鍔ㄦ垚鍔燂紒锛侊紒", key); + }else { + logger.info("浠诲姟銆恵}銆戝惎鍔ㄥけ璐ワ紒锛侊紒", key); + } } public void stop(String key) { + System.out.println("瀹氭椂浠诲姟缁撴潫浜�"); if (futureMap.get(key) != null && !futureMap.get(key).isCancelled()) { futureMap.get(key).cancel(true); Runnable runnable = runnableMap.get(key); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeHolder.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeHolder.java index 62a45d5..981fe1e 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeHolder.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeHolder.java @@ -1,5 +1,12 @@ package com.genersoft.iot.vmp.gb28181.bean; +import com.genersoft.iot.vmp.common.VideoManagerConstants; +import com.genersoft.iot.vmp.conf.DynamicTask; +import com.genersoft.iot.vmp.gb28181.task.impl.MobilePositionSubscribeHandlerTask; +import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform; +import com.genersoft.iot.vmp.storager.IRedisCatchStorage; +import com.genersoft.iot.vmp.storager.IVideoManagerStorage; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import java.util.ArrayList; @@ -9,12 +16,32 @@ @Component public class SubscribeHolder { + @Autowired + private DynamicTask dynamicTask; + + @Autowired + private IRedisCatchStorage redisCatchStorage; + + @Autowired + private ISIPCommanderForPlatform sipCommanderForPlatform; + + @Autowired + private IVideoManagerStorage storager; + + private final String taskOverduePrefix = "subscribe_overdue_"; + private static ConcurrentHashMap<String, SubscribeInfo> catalogMap = new ConcurrentHashMap<>(); private static ConcurrentHashMap<String, SubscribeInfo> mobilePositionMap = new ConcurrentHashMap<>(); public void putCatalogSubscribe(String platformId, SubscribeInfo subscribeInfo) { catalogMap.put(platformId, subscribeInfo); + // 娣诲姞璁㈤槄鍒版湡 + String taskOverdueKey = taskOverduePrefix + "catalog_" + platformId; + dynamicTask.stop(taskOverdueKey); + // 娣诲姞浠诲姟澶勭悊璁㈤槄杩囨湡 + dynamicTask.startDelay(taskOverdueKey, () -> removeCatalogSubscribe(subscribeInfo.getId()), + subscribeInfo.getExpires() * 1000); } public SubscribeInfo getCatalogSubscribe(String platformId) { @@ -23,10 +50,24 @@ public void removeCatalogSubscribe(String platformId) { catalogMap.remove(platformId); + String taskOverdueKey = taskOverduePrefix + "catalog_" + platformId; + // 娣诲姞浠诲姟澶勭悊璁㈤槄杩囨湡 + dynamicTask.stop(taskOverdueKey); } public void putMobilePositionSubscribe(String platformId, SubscribeInfo subscribeInfo) { mobilePositionMap.put(platformId, subscribeInfo); + String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + "MobilePosition_" + platformId; + // 娣诲姞浠诲姟澶勭悊GPS瀹氭椂鎺ㄩ�� + dynamicTask.startCron(key, new MobilePositionSubscribeHandlerTask(redisCatchStorage, sipCommanderForPlatform, storager, platformId, subscribeInfo.getSn(), key, this), subscribeInfo.getGpsInterval()); + String taskOverdueKey = taskOverduePrefix + "MobilePosition_" + platformId; + dynamicTask.stop(taskOverdueKey); + // 娣诲姞浠诲姟澶勭悊璁㈤槄杩囨湡 + dynamicTask.startDelay(taskOverdueKey, () -> { + System.out.println("璁㈤槄杩囨湡"); + removeMobilePositionSubscribe(subscribeInfo.getId()); + }, + subscribeInfo.getExpires() * 1000); } public SubscribeInfo getMobilePositionSubscribe(String platformId) { @@ -35,6 +76,12 @@ public void removeMobilePositionSubscribe(String platformId) { mobilePositionMap.remove(platformId); + String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + "MobilePosition_" + platformId; + // 缁撴潫浠诲姟澶勭悊GPS瀹氭椂鎺ㄩ�� + dynamicTask.stop(key); + String taskOverdueKey = taskOverduePrefix + "MobilePosition_" + platformId; + // 娣诲姞浠诲姟澶勭悊璁㈤槄杩囨湡 + dynamicTask.stop(taskOverdueKey); } public List<String> getAllCatalogSubscribePlatform() { @@ -48,7 +95,7 @@ } public void removeAllSubscribe(String platformId) { - mobilePositionMap.remove(platformId); - catalogMap.remove(platformId); + removeMobilePositionSubscribe(platformId); + removeCatalogSubscribe(platformId); } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeInfo.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeInfo.java index 434a639..feb6a72 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeInfo.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeInfo.java @@ -33,6 +33,14 @@ private ServerTransaction transaction; private Dialog dialog; + /** + * 浠ヤ笅涓哄彲閫夊瓧娈� + * @return + */ + private String sn; + private int gpsInterval; + + public String getId() { return id; } @@ -88,4 +96,20 @@ public void setDialog(Dialog dialog) { this.dialog = dialog; } + + public String getSn() { + return sn; + } + + public void setSn(String sn) { + this.sn = sn; + } + + public int getGpsInterval() { + return gpsInterval; + } + + public void setGpsInterval(int gpsInterval) { + this.gpsInterval = gpsInterval; + } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/SubscribeListenerForPlatform.java b/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/SubscribeListenerForPlatform.java deleted file mode 100644 index 898e51d..0000000 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/SubscribeListenerForPlatform.java +++ /dev/null @@ -1,50 +0,0 @@ -package com.genersoft.iot.vmp.gb28181.event.subscribe; - -import com.genersoft.iot.vmp.common.VideoManagerConstants; -import com.genersoft.iot.vmp.conf.DynamicTask; -import com.genersoft.iot.vmp.conf.RedisKeyExpirationEventMessageListener; -import com.genersoft.iot.vmp.conf.UserSetting; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.data.redis.connection.Message; -import org.springframework.data.redis.listener.RedisMessageListenerContainer; -import org.springframework.stereotype.Component; - -/** - * 骞冲彴璁㈤槄鍒版湡浜嬩欢 - */ -@Component -public class SubscribeListenerForPlatform extends RedisKeyExpirationEventMessageListener { - - private Logger logger = LoggerFactory.getLogger(SubscribeListenerForPlatform.class); - - @Autowired - private UserSetting userSetting; - - @Autowired - private DynamicTask dynamicTask; - - public SubscribeListenerForPlatform(RedisMessageListenerContainer listenerContainer, UserSetting userSetting) { - super(listenerContainer, userSetting); - } - - - /** - * 鐩戝惉澶辨晥鐨刱ey - * @param message - * @param pattern - */ - @Override - public void onMessage(Message message, byte[] pattern) { - // 鑾峰彇澶辨晥鐨刱ey - String expiredKey = message.toString(); - logger.debug(expiredKey); - // 璁㈤槄鍒版湡 - String PLATFORM_KEEPLIVEKEY_PREFIX = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetting.getServerId() + "_"; - if (expiredKey.startsWith(PLATFORM_KEEPLIVEKEY_PREFIX)) { - // 鍙栨秷瀹氭椂浠诲姟 - dynamicTask.stop(expiredKey); - } - } -} 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 f9546f0..f9ef10c 100644 --- 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 @@ -61,8 +61,6 @@ if (event.getPlatformId() != null) { parentPlatform = storager.queryParentPlatByServerGBId(event.getPlatformId()); if (parentPlatform != null && !parentPlatform.isStatus())return; - String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetting.getServerId() + "_Catalog_" + event.getPlatformId(); -// subscribe = redisCatchStorage.getSubscribe(key); subscribe = subscribeHolder.getCatalogSubscribe(event.getPlatformId()); if (subscribe == null) { diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/CatalogSubscribeTask.java b/src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/CatalogSubscribeTask.java index 433eb3b..bee5fba 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/CatalogSubscribeTask.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/CatalogSubscribeTask.java @@ -46,7 +46,6 @@ }); } - @Async @Override public void stop() { /** diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/MobilePositionSubscribeHandlerTask.java b/src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/MobilePositionSubscribeHandlerTask.java index 569a9b7..f20baf9 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/MobilePositionSubscribeHandlerTask.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/MobilePositionSubscribeHandlerTask.java @@ -24,52 +24,50 @@ private IVideoManagerStorage storager; private ISIPCommanderForPlatform sipCommanderForPlatform; private SubscribeHolder subscribeHolder; - private String platformId; + private ParentPlatform platform; private String sn; private String key; public MobilePositionSubscribeHandlerTask(IRedisCatchStorage redisCatchStorage, ISIPCommanderForPlatform sipCommanderForPlatform, IVideoManagerStorage storager, String platformId, String sn, String key, SubscribeHolder subscribeInfo) { + System.out.println("MobilePositionSubscribeHandlerTask 鍒濆鍖�"); this.redisCatchStorage = redisCatchStorage; this.storager = storager; - this.platformId = platformId; + this.platform = storager.queryParentPlatByServerGBId(platformId); this.sn = sn; this.key = key; this.sipCommanderForPlatform = sipCommanderForPlatform; this.subscribeHolder = subscribeInfo; } - @Async @Override public void run() { logger.info("鎵цMobilePositionSubscribeHandlerTask"); - SubscribeInfo subscribe = subscribeHolder.getMobilePositionSubscribe(platformId); + if (platform == null) return; + SubscribeInfo subscribe = subscribeHolder.getMobilePositionSubscribe(platform.getServerGBId()); if (subscribe != null) { - ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformId); - if (parentPlatform == null ) { - logger.info("鍙戦�佽闃呮椂鏈壘鍒板钩鍙颁俊鎭細{}", platformId); - return; - } - if (!parentPlatform.isStatus()) { - logger.info("鍙戦�佽闃呮椂鍙戠幇骞冲彴宸茬粡绂荤嚎锛歿}", platformId); - return; - } + +// if (!parentPlatform.isStatus()) { +// logger.info("鍙戦�佽闃呮椂鍙戠幇骞冲彴宸茬粡绂荤嚎锛歿}", platformId); +// return; +// } // TODO 鏆傛椂鍙鐞嗚棰戞祦鐨勫洖澶�,鍚庣画澧炲姞瀵瑰浗鏍囪澶囩殑鏀寔 - List<GbStream> gbStreams = storager.queryGbStreamListInPlatform(platformId); + List<GbStream> gbStreams = storager.queryGbStreamListInPlatform(platform.getServerGBId()); if (gbStreams.size() == 0) { - logger.info("鍙戦�佽闃呮椂鍙戠幇骞冲彴宸茬粡娌℃湁鍏宠仈鐨勭洿鎾祦锛歿}", platformId); + logger.info("鍙戦�佽闃呮椂鍙戠幇骞冲彴宸茬粡娌℃湁鍏宠仈鐨勭洿鎾祦锛歿}", platform.getServerGBId()); return; } for (GbStream gbStream : gbStreams) { String gbId = gbStream.getGbId(); GPSMsgInfo gpsMsgInfo = redisCatchStorage.getGpsMsgInfo(gbId); if (gpsMsgInfo != null) { // 鏃犳渶鏂颁綅缃笉鍙戦�� + logger.info("鏃犳渶鏂颁綅缃笉鍙戦��"); // 缁忕含搴﹂兘涓�0涓嶅彂閫� if (gpsMsgInfo.getLng() == 0 && gpsMsgInfo.getLat() == 0) { continue; } // 鍙戦�丟PS娑堟伅 - sipCommanderForPlatform.sendNotifyMobilePosition(parentPlatform, gpsMsgInfo, subscribe); + sipCommanderForPlatform.sendNotifyMobilePosition(platform, gpsMsgInfo, subscribe); } } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/MobilePositionSubscribeTask.java b/src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/MobilePositionSubscribeTask.java index 7203ee2..884f040 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/MobilePositionSubscribeTask.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/MobilePositionSubscribeTask.java @@ -11,6 +11,8 @@ import javax.sip.Dialog; import javax.sip.DialogState; import javax.sip.ResponseEvent; +import java.util.Timer; +import java.util.TimerTask; /** * 绉诲姩浣嶇疆璁㈤槄鐨勫畾鏃舵洿鏂� @@ -21,18 +23,23 @@ private ISIPCommander sipCommander; private Dialog dialog; + private Timer timer ; + public MobilePositionSubscribeTask(Device device, ISIPCommander sipCommander) { this.device = device; this.sipCommander = sipCommander; } - @Async @Override public void run() { + if (timer != null ) { + timer.cancel(); + timer = null; + } sipCommander.mobilePositionSubscribe(device, dialog, eventResult -> { - if (eventResult.dialog != null || eventResult.dialog.getState().equals(DialogState.CONFIRMED)) { - dialog = eventResult.dialog; - } +// if (eventResult.dialog != null || eventResult.dialog.getState().equals(DialogState.CONFIRMED)) { +// dialog = eventResult.dialog; +// } ResponseEvent event = (ResponseEvent) eventResult.event; if (event.getResponse().getRawContent() != null) { // 鎴愬姛 @@ -45,6 +52,13 @@ dialog = null; // 澶辫触 logger.warn("[绉诲姩浣嶇疆璁㈤槄]澶辫触锛屼俊浠ゅ彂閫佸け璐ワ細 {}-{} ", device.getDeviceId(), eventResult.msg); + timer = new Timer(); + timer.schedule(new TimerTask() { + @Override + public void run() { + MobilePositionSubscribeTask.this.run(); + } + }, 2000); }); } @@ -58,6 +72,10 @@ * COMPLETED-> Completed Dialog鐘舵��-宸插畬鎴� * TERMINATED-> Terminated Dialog鐘舵��-缁堟 */ + if (timer != null ) { + timer.cancel(); + timer = null; + } if (dialog != null && dialog.getState().equals(DialogState.CONFIRMED)) { logger.info("鍙栨秷绉诲姩璁㈤槄鏃禿ialog鐘舵�佷负{}", dialog.getState()); device.setSubscribeCycleForMobilePosition(0); 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 7768ed4..c5cdae0 100644 --- 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 @@ -385,7 +385,7 @@ if (parentPlatform == null) { return false; } - + logger.info("[鍙戦�� 绉诲姩浣嶇疆璁㈤槄] {}/{}->{},{}", parentPlatform.getServerGBId(), gpsMsgInfo.getId(), gpsMsgInfo.getLng(), gpsMsgInfo.getLat()); try { String characterSet = parentPlatform.getCharacterSet(); StringBuffer deviceStatusXml = new StringBuffer(600); @@ -405,7 +405,7 @@ CallIdHeader callIdHeader = parentPlatform.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId() : udpSipProvider.getNewCallId(); callIdHeader.setCallId(subscribeInfo.getCallId()); - logger.info("[鍙戦�� 绉诲姩浣嶇疆璁㈤槄] {}/{}->{},{}", parentPlatform.getServerGBId(), gpsMsgInfo.getId(), gpsMsgInfo.getLng(), gpsMsgInfo.getLat()); + sendNotify(parentPlatform, deviceStatusXml.toString(), subscribeInfo, eventResult -> { logger.error("鍙戦�丯OTIFY閫氱煡娑堟伅澶辫触銆傞敊璇細{} {}", eventResult.statusCode, eventResult.msg); }, null); @@ -459,7 +459,7 @@ // 璁剧疆缂栫爜锛� 闃叉涓枃涔辩爜 messageFactory.setDefaultContentEncodingCharset(characterSet); Dialog dialog = subscribeInfo.getDialog(); - if (dialog == null) return; + if (dialog == null || !dialog.getState().equals(DialogState.CONFIRMED)) return; SIPRequest notifyRequest = (SIPRequest)dialog.createRequest(Request.NOTIFY); ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("Application", "MANSCDP+xml"); notifyRequest.setContent(catalogXmlContent, contentTypeHeader); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/SubscribeRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/SubscribeRequestProcessor.java index 979849e..da1088a 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/SubscribeRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/SubscribeRequestProcessor.java @@ -149,7 +149,6 @@ subscribeInfo.setDialog(dialog); } String sn = XmlUtil.getText(rootElement, "SN"); - String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetting.getServerId() + "_MobilePosition_" + platformId; logger.info("[鍥炲 绉诲姩浣嶇疆璁㈤槄]: {}", platformId); StringBuilder resultXml = new StringBuilder(200); resultXml.append("<?xml version=\"1.0\" ?>\r\n") @@ -161,23 +160,25 @@ .append("</Response>\r\n"); if (subscribeInfo.getExpires() > 0) { - - if (subscribeHolder.getMobilePositionSubscribe(platformId) == null ) { - String interval = XmlUtil.getText(rootElement, "Interval"); // GPS涓婃姤鏃堕棿闂撮殧 - subscribeHolder.putMobilePositionSubscribe(platformId, subscribeInfo); - dynamicTask.startCron(key, new MobilePositionSubscribeHandlerTask(redisCatchStorage, sipCommanderForPlatform, storager, platformId, sn, key, subscribeHolder), Integer.parseInt(interval)); + String interval = XmlUtil.getText(rootElement, "Interval"); // GPS涓婃姤鏃堕棿闂撮殧 + if (interval == null) { + subscribeInfo.setGpsInterval(5); }else { - if (subscribeHolder.getMobilePositionSubscribe(platformId).getDialog() != null - && subscribeHolder.getMobilePositionSubscribe(platformId).getDialog().getState() != null - && !subscribeHolder.getMobilePositionSubscribe(platformId).getDialog().getState().equals(DialogState.CONFIRMED)) { - dynamicTask.stop(key); - String interval = XmlUtil.getText(rootElement, "Interval"); // GPS涓婃姤鏃堕棿闂撮殧 - subscribeHolder.putMobilePositionSubscribe(platformId, subscribeInfo); - dynamicTask.startCron(key, new MobilePositionSubscribeHandlerTask(redisCatchStorage, sipCommanderForPlatform, storager, platformId, sn, key, subscribeHolder), Integer.parseInt(interval)); - } + subscribeInfo.setGpsInterval(Integer.parseInt(interval)); } + + subscribeInfo.setSn(sn); + subscribeHolder.putMobilePositionSubscribe(platformId, subscribeInfo); +// if (subscribeHolder.getMobilePositionSubscribe(platformId) == null ) { +// subscribeHolder.putMobilePositionSubscribe(platformId, subscribeInfo); +// }else { +// if (subscribeHolder.getMobilePositionSubscribe(platformId).getDialog() != null +// && subscribeHolder.getMobilePositionSubscribe(platformId).getDialog().getState() != null +// && !subscribeHolder.getMobilePositionSubscribe(platformId).getDialog().getState().equals(DialogState.CONFIRMED)) { +// subscribeHolder.putMobilePositionSubscribe(platformId, subscribeInfo); +// } +// } }else if (subscribeInfo.getExpires() == 0) { - dynamicTask.stop(key); subscribeHolder.removeMobilePositionSubscribe(platformId); } @@ -211,7 +212,6 @@ subscribeInfo.setDialog(dialog); } String sn = XmlUtil.getText(rootElement, "SN"); - String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetting.getServerId() + "_Catalog_" + platformId; logger.info("[鍥炲 鐩綍璁㈤槄]: {}/{}", platformId, deviceID); StringBuilder resultXml = new StringBuilder(200); resultXml.append("<?xml version=\"1.0\" ?>\r\n") diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java index 1bafb59..dbc25fc 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java @@ -145,9 +145,6 @@ } // 鍥炲200 OK responseAck(evt, Response.OK); - if (offLineDetector.isOnline(device.getDeviceId())) { - publisher.onlineEventPublish(device, VideoManagerConstants.EVENT_ONLINE_MESSAGE); - } } } catch (DocumentException e) { e.printStackTrace(); diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java index 8cd2c77..d343286 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java @@ -14,6 +14,8 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; +import javax.sip.DialogState; + /** * 璁惧涓氬姟锛堢洰褰曡闃咃級 */ @@ -39,19 +41,17 @@ if (device == null || device.getSubscribeCycleForCatalog() < 0) { return false; } - if (dynamicTask.contains(device.getDeviceId() + "catalog")) { - // 瀛樺湪鍒欏仠姝㈢幇鏈夌殑锛屽紑鍚柊鐨� - dynamicTask.stop(device.getDeviceId() + "catalog"); + CatalogSubscribeTask task = (CatalogSubscribeTask)dynamicTask.get(device.getDeviceId() + "catalog"); + if (task != null && task.getDialogState() != null && task.getDialogState().equals(DialogState.CONFIRMED)) { // 宸插瓨鍦ㄤ笉闇�瑕佸啀娆℃坊鍔� + return true; } logger.info("[娣诲姞鐩綍璁㈤槄] 璁惧{}", device.getDeviceId()); // 娣诲姞鐩綍璁㈤槄 CatalogSubscribeTask catalogSubscribeTask = new CatalogSubscribeTask(device, sipCommander); - catalogSubscribeTask.run(); // 鎻愬墠寮�濮嬪埛鏂拌闃� - int subscribeCycleForCatalog = device.getSubscribeCycleForCatalog(); + int subscribeCycleForCatalog = Math.max(device.getSubscribeCycleForCatalog(),30); // 璁剧疆鏈�灏忓�间负30 - subscribeCycleForCatalog = Math.max(subscribeCycleForCatalog, 30); - dynamicTask.startCron(device.getDeviceId() + "catalog", catalogSubscribeTask, subscribeCycleForCatalog); + dynamicTask.startCron(device.getDeviceId() + "catalog", catalogSubscribeTask, subscribeCycleForCatalog -1); return true; } @@ -70,18 +70,16 @@ if (device == null || device.getSubscribeCycleForMobilePosition() < 0) { return false; } - if (dynamicTask.contains(device.getDeviceId() + "mobile_position")) { - // 瀛樺湪鍒欏仠姝㈢幇鏈夌殑锛屽紑鍚柊鐨� - dynamicTask.stop(device.getDeviceId() + "mobile_position"); - } logger.info("[娣诲姞绉诲姩浣嶇疆璁㈤槄] 璁惧{}", device.getDeviceId()); + MobilePositionSubscribeTask task = (MobilePositionSubscribeTask)dynamicTask.get(device.getDeviceId() + "mobile_position"); + if (task != null && task.getDialogState() != null && task.getDialogState().equals(DialogState.CONFIRMED)) { // 宸插瓨鍦ㄤ笉闇�瑕佸啀娆℃坊鍔� + return true; + } // 娣诲姞鐩綍璁㈤槄 MobilePositionSubscribeTask mobilePositionSubscribeTask = new MobilePositionSubscribeTask(device, sipCommander); - mobilePositionSubscribeTask.run(); // 鎻愬墠寮�濮嬪埛鏂拌闃� - int subscribeCycleForCatalog = device.getSubscribeCycleForCatalog(); // 璁剧疆鏈�灏忓�间负30 - subscribeCycleForCatalog = Math.max(subscribeCycleForCatalog, 30); + int subscribeCycleForCatalog = Math.max(device.getSubscribeCycleForMobilePosition(),30); dynamicTask.startCron(device.getDeviceId() + "mobile_position" , mobilePositionSubscribeTask, subscribeCycleForCatalog -1 ); return true; } diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/server/ServerController.java b/src/main/java/com/genersoft/iot/vmp/vmanager/server/ServerController.java index a0e7a73..faed2c8 100644 --- a/src/main/java/com/genersoft/iot/vmp/vmanager/server/ServerController.java +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/server/ServerController.java @@ -4,6 +4,7 @@ import com.alibaba.fastjson.JSONObject; import com.genersoft.iot.vmp.VManageBootstrap; import com.genersoft.iot.vmp.common.VersionPo; +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.conf.VersionInfo; @@ -27,6 +28,7 @@ import javax.sip.SipProvider; import java.util.Iterator; import java.util.List; +import java.util.Set; @SuppressWarnings("rawtypes") @Api(tags = "鏈嶅姟鎺у埗") @@ -42,13 +44,16 @@ private IMediaServerService mediaServerService; @Autowired - VersionInfo versionInfo; + private VersionInfo versionInfo; @Autowired - SipConfig sipConfig; + private SipConfig sipConfig; @Autowired - UserSetting userSetting; + private UserSetting userSetting; + + @Autowired + private DynamicTask dynamicTask; @Value("${server.port}") private int serverPort; @@ -248,4 +253,35 @@ result.setData(jsonObject); return result; } + +// @ApiOperation("褰撳墠杩涜涓殑鍔ㄦ�佷换鍔�") +// @GetMapping(value = "/dynamicTask") +// @ResponseBody +// public WVPResult<JSONObject> getDynamicTask(){ +// WVPResult<JSONObject> result = new WVPResult<>(); +// result.setCode(0); +// result.setMsg("success"); +// +// JSONObject jsonObject = new JSONObject(); +// +// Set<String> allKeys = dynamicTask.getAllKeys(); +// jsonObject.put("server.port", serverPort); +// if (StringUtils.isEmpty(type)) { +// jsonObject.put("sip", JSON.toJSON(sipConfig)); +// jsonObject.put("base", JSON.toJSON(userSetting)); +// }else { +// switch (type){ +// case "sip": +// jsonObject.put("sip", sipConfig); +// break; +// case "base": +// jsonObject.put("base", userSetting); +// break; +// default: +// break; +// } +// } +// result.setData(jsonObject); +// return result; +// } } -- Gitblit v1.8.0