guzijian / wvp-GB28181-pro · Commits

Commit 22e1d92a9d8ae6aa257889f882722b8e48648abc

Authored by mk1990
Committed by GitHub
2 parents eef78c78 0dc1807f

Merge branch '648540858:wvp-28181-2.0' into wvp-28181-2.0

Inline Side-by-side
Showing 26 changed files with 543 additions and 246 deletions
bootstrap.sh 0 → 100755
  View file @22e1d92
  1 +#!/bin/bash
  2 +
  3 +######################################################
  4 +# Copyright 2019 Pham Ngoc Hoai
  5 +#
  6 +# Licensed under the Apache License, Version 2.0 (the "License");
  7 +# you may not use this file except in compliance with the License.
  8 +# You may obtain a copy of the License at
  9 +#
  10 +# http://www.apache.org/licenses/LICENSE-2.0
  11 +#
  12 +# Unless required by applicable law or agreed to in writing, software
  13 +# distributed under the License is distributed on an "AS IS" BASIS,
  14 +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15 +# See the License for the specific language governing permissions and
  16 +# limitations under the License.
  17 +#
  18 +# Repo: https://github.com/tyrion9/spring-boot-startup-script
  19 +#
  20 +######### PARAM ######################################
  21 +
  22 +JAVA_OPT=-Xmx1024m
  23 +JARFILE=`ls -1r *.jar 2>/dev/null | head -n 1`
  24 +PID_FILE=pid.file
  25 +RUNNING=N
  26 +PWD=`pwd`
  27 +
  28 +######### DO NOT MODIFY ########
  29 +
  30 +if [ -f $PID_FILE ]; then
  31 + PID=`cat $PID_FILE`
  32 + if [ ! -z "$PID" ] && kill -0 $PID 2>/dev/null; then
  33 + RUNNING=Y
  34 + fi
  35 +fi
  36 +
  37 +start()
  38 +{
  39 + if [ $RUNNING == "Y" ]; then
  40 + echo "Application already started"
  41 + else
  42 + if [ -z "$JARFILE" ]
  43 + then
  44 + echo "ERROR: jar file not found"
  45 + else
  46 + nohup java $JAVA_OPT -Djava.security.egd=file:/dev/./urandom -jar $PWD/$JARFILE > nohup.out 2>&1 &
  47 + echo $! > $PID_FILE
  48 + echo "Application $JARFILE starting..."
  49 + tail -f nohup.out
  50 + fi
  51 + fi
  52 +}
  53 +
  54 +stop()
  55 +{
  56 + if [ $RUNNING == "Y" ]; then
  57 + kill -9 $PID
  58 + rm -f $PID_FILE
  59 + echo "Application stopped"
  60 + else
  61 + echo "Application not running"
  62 + fi
  63 +}
  64 +
  65 +restart()
  66 +{
  67 + stop
  68 + start
  69 +}
  70 +
  71 +case "$1" in
  72 +
  73 + 'start')
  74 + start
  75 + ;;
  76 +
  77 + 'stop')
  78 + stop
  79 + ;;
  80 +
  81 + 'restart')
  82 + restart
  83 + ;;
  84 +
  85 + *)
  86 + echo "Usage: $0 { start | stop | restart }"
  87 + exit 1
  88 + ;;
  89 +esac
  90 +exit 0
  91 +
... ...
src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java
  View file @22e1d92
1 1 package com.genersoft.iot.vmp.conf;
2 2  
3 3 import com.genersoft.iot.vmp.gb28181.task.ISubscribeTask;
  4 +import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd.CatalogResponseMessageHandler;
  5 +import org.slf4j.Logger;
  6 +import org.slf4j.LoggerFactory;
4 7 import org.springframework.beans.factory.annotation.Autowired;
5 8 import org.springframework.context.annotation.Bean;
6 9 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
... ... @@ -18,6 +21,8 @@ import java.util.concurrent.ScheduledFuture;
18 21 @Component
19 22 public class DynamicTask {
20 23  
  24 + private Logger logger = LoggerFactory.getLogger(DynamicTask.class);
  25 +
21 26 @Autowired
22 27 private ThreadPoolTaskScheduler threadPoolTaskScheduler;
23 28  
... ... @@ -26,7 +31,12 @@ public class DynamicTask {
26 31  
27 32 @Bean
28 33 public ThreadPoolTaskScheduler threadPoolTaskScheduler() {
29   - return new ThreadPoolTaskScheduler();
  34 + ThreadPoolTaskScheduler schedulerPool = new ThreadPoolTaskScheduler();
  35 + schedulerPool.setPoolSize(300);
  36 + schedulerPool.setWaitForTasksToCompleteOnShutdown(true);
  37 + schedulerPool.setAwaitTerminationSeconds(10);
  38 + return schedulerPool;
  39 +
30 40 }
31 41  
32 42 /**
... ... @@ -37,11 +47,24 @@ public class DynamicTask {
37 47 * @return
38 48 */
39 49 public void startCron(String key, Runnable task, int cycleForCatalog) {
40   - stop(key);
  50 + ScheduledFuture future = futureMap.get(key);
  51 + if (future != null) {
  52 + if (future.isCancelled()) {
  53 + logger.info("任务【{}】已存在但是关闭状态！！！", key);
  54 + } else {
  55 + logger.info("任务【{}】已存在且已启动！！！", key);
  56 + return;
  57 + }
  58 + }
41 59 // scheduleWithFixedDelay 必须等待上一个任务结束才开始计时period， cycleForCatalog表示执行的间隔
42   - ScheduledFuture future = threadPoolTaskScheduler.scheduleWithFixedDelay(task, cycleForCatalog * 1000L);
43   - futureMap.put(key, future);
44   - runnableMap.put(key, task);
  60 + future = threadPoolTaskScheduler.scheduleAtFixedRate(task, cycleForCatalog * 1000L);
  61 + if (future != null){
  62 + futureMap.put(key, future);
  63 + runnableMap.put(key, task);
  64 + logger.info("任务【{}】启动成功！！！", key);
  65 + }else {
  66 + logger.info("任务【{}】启动失败！！！", key);
  67 + }
45 68 }
46 69  
47 70 /**
... ... @@ -54,9 +77,25 @@ public class DynamicTask {
54 77 public void startDelay(String key, Runnable task, int delay) {
55 78 stop(key);
56 79 Date starTime = new Date(System.currentTimeMillis() + delay);
  80 +
  81 + ScheduledFuture future = futureMap.get(key);
  82 + if (future != null) {
  83 + if (future.isCancelled()) {
  84 + logger.info("任务【{}】已存在但是关闭状态！！！", key);
  85 + } else {
  86 + logger.info("任务【{}】已存在且已启动！！！", key);
  87 + return;
  88 + }
  89 + }
57 90 // scheduleWithFixedDelay 必须等待上一个任务结束才开始计时period， cycleForCatalog表示执行的间隔
58   - ScheduledFuture future = threadPoolTaskScheduler.schedule(task, starTime);
59   - futureMap.put(key, future);
  91 + future = threadPoolTaskScheduler.schedule(task, starTime);
  92 + if (future != null){
  93 + futureMap.put(key, future);
  94 + runnableMap.put(key, task);
  95 + logger.info("任务【{}】启动成功！！！", key);
  96 + }else {
  97 + logger.info("任务【{}】启动失败！！！", key);
  98 + }
60 99 }
61 100  
62 101 public void stop(String key) {
... ... @@ -78,4 +117,7 @@ public class DynamicTask {
78 117 return futureMap.keySet();
79 118 }
80 119  
  120 + public Runnable get(String key) {
  121 + return runnableMap.get(key);
  122 + }
81 123 }
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java
  View file @22e1d92
... ... @@ -48,6 +48,7 @@ public class SipLayer{
48 48 properties.setProperty("javax.sip.STACK_NAME", "GB28181_SIP");
49 49 properties.setProperty("javax.sip.IP_ADDRESS", sipConfig.getMonitorIp());
50 50 properties.setProperty("gov.nist.javax.sip.LOG_MESSAGE_CONTENT", "true");
  51 + properties.setProperty("gov.nist.javax.sip.DELIVER_UNSOLICITED_NOTIFY", "true"); // 接收所有notify请求，即使没有订阅
51 52 /**
52 53 * sip_server_log.log 和 sip_debug_log.log public static final int TRACE_NONE =
53 54 * 0; public static final int TRACE_MESSAGES = 16; public static final int
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/bean/CatalogData.java
  View file @22e1d92
... ... @@ -4,6 +4,7 @@ import java.util.Date;
4 4 import java.util.List;
5 5  
6 6 public class CatalogData {
  7 + private int sn; // 命令序列号
7 8 private int total;
8 9 private List<DeviceChannel> channelList;
9 10 private Date lastTime;
... ... @@ -15,6 +16,15 @@ public class CatalogData {
15 16 }
16 17 private CatalogDataStatus status;
17 18  
  19 +
  20 + public int getSn() {
  21 + return sn;
  22 + }
  23 +
  24 + public void setSn(int sn) {
  25 + this.sn = sn;
  26 + }
  27 +
18 28 public int getTotal() {
19 29 return total;
20 30 }
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeHolder.java
  View file @22e1d92
1 1 package com.genersoft.iot.vmp.gb28181.bean;
2 2  
  3 +import com.genersoft.iot.vmp.common.VideoManagerConstants;
  4 +import com.genersoft.iot.vmp.conf.DynamicTask;
  5 +import com.genersoft.iot.vmp.gb28181.task.impl.MobilePositionSubscribeHandlerTask;
  6 +import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
  7 +import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
  8 +import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
  9 +import org.springframework.beans.factory.annotation.Autowired;
3 10 import org.springframework.stereotype.Component;
4 11  
5 12 import java.util.ArrayList;
... ... @@ -9,12 +16,32 @@ import java.util.concurrent.ConcurrentHashMap;
9 16 @Component
10 17 public class SubscribeHolder {
11 18  
  19 + @Autowired
  20 + private DynamicTask dynamicTask;
  21 +
  22 + @Autowired
  23 + private IRedisCatchStorage redisCatchStorage;
  24 +
  25 + @Autowired
  26 + private ISIPCommanderForPlatform sipCommanderForPlatform;
  27 +
  28 + @Autowired
  29 + private IVideoManagerStorage storager;
  30 +
  31 + private final String taskOverduePrefix = "subscribe_overdue_";
  32 +
12 33 private static ConcurrentHashMap<String, SubscribeInfo> catalogMap = new ConcurrentHashMap<>();
13 34 private static ConcurrentHashMap<String, SubscribeInfo> mobilePositionMap = new ConcurrentHashMap<>();
14 35  
15 36  
16 37 public void putCatalogSubscribe(String platformId, SubscribeInfo subscribeInfo) {
17 38 catalogMap.put(platformId, subscribeInfo);
  39 + // 添加订阅到期
  40 + String taskOverdueKey = taskOverduePrefix + "catalog_" + platformId;
  41 + dynamicTask.stop(taskOverdueKey);
  42 + // 添加任务处理订阅过期
  43 + dynamicTask.startDelay(taskOverdueKey, () -> removeCatalogSubscribe(subscribeInfo.getId()),
  44 + subscribeInfo.getExpires() * 1000);
18 45 }
19 46  
20 47 public SubscribeInfo getCatalogSubscribe(String platformId) {
... ... @@ -23,10 +50,24 @@ public class SubscribeHolder {
23 50  
24 51 public void removeCatalogSubscribe(String platformId) {
25 52 catalogMap.remove(platformId);
  53 + String taskOverdueKey = taskOverduePrefix + "catalog_" + platformId;
  54 + // 添加任务处理订阅过期
  55 + dynamicTask.stop(taskOverdueKey);
26 56 }
27 57  
28 58 public void putMobilePositionSubscribe(String platformId, SubscribeInfo subscribeInfo) {
29 59 mobilePositionMap.put(platformId, subscribeInfo);
  60 + String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + "MobilePosition_" + platformId;
  61 + // 添加任务处理GPS定时推送
  62 + dynamicTask.startCron(key, new MobilePositionSubscribeHandlerTask(redisCatchStorage, sipCommanderForPlatform, storager, platformId, subscribeInfo.getSn(), key, this), subscribeInfo.getGpsInterval());
  63 + String taskOverdueKey = taskOverduePrefix + "MobilePosition_" + platformId;
  64 + dynamicTask.stop(taskOverdueKey);
  65 + // 添加任务处理订阅过期
  66 + dynamicTask.startDelay(taskOverdueKey, () -> {
  67 + System.out.println("订阅过期");
  68 + removeMobilePositionSubscribe(subscribeInfo.getId());
  69 + },
  70 + subscribeInfo.getExpires() * 1000);
30 71 }
31 72  
32 73 public SubscribeInfo getMobilePositionSubscribe(String platformId) {
... ... @@ -35,6 +76,12 @@ public class SubscribeHolder {
35 76  
36 77 public void removeMobilePositionSubscribe(String platformId) {
37 78 mobilePositionMap.remove(platformId);
  79 + String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + "MobilePosition_" + platformId;
  80 + // 结束任务处理GPS定时推送
  81 + dynamicTask.stop(key);
  82 + String taskOverdueKey = taskOverduePrefix + "MobilePosition_" + platformId;
  83 + // 添加任务处理订阅过期
  84 + dynamicTask.stop(taskOverdueKey);
38 85 }
39 86  
40 87 public List<String> getAllCatalogSubscribePlatform() {
... ... @@ -48,7 +95,7 @@ public class SubscribeHolder {
48 95 }
49 96  
50 97 public void removeAllSubscribe(String platformId) {
51   - mobilePositionMap.remove(platformId);
52   - catalogMap.remove(platformId);
  98 + removeMobilePositionSubscribe(platformId);
  99 + removeCatalogSubscribe(platformId);
53 100 }
54 101 }
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeInfo.java
  View file @22e1d92
... ... @@ -33,6 +33,14 @@ public class SubscribeInfo {
33 33 private ServerTransaction transaction;
34 34 private Dialog dialog;
35 35  
  36 + /**
  37 + * 以下为可选字段
  38 + * @return
  39 + */
  40 + private String sn;
  41 + private int gpsInterval;
  42 +
  43 +
36 44 public String getId() {
37 45 return id;
38 46 }
... ... @@ -88,4 +96,20 @@ public class SubscribeInfo {
88 96 public void setDialog(Dialog dialog) {
89 97 this.dialog = dialog;
90 98 }
  99 +
  100 + public String getSn() {
  101 + return sn;
  102 + }
  103 +
  104 + public void setSn(String sn) {
  105 + this.sn = sn;
  106 + }
  107 +
  108 + public int getGpsInterval() {
  109 + return gpsInterval;
  110 + }
  111 +
  112 + public void setGpsInterval(int gpsInterval) {
  113 + this.gpsInterval = gpsInterval;
  114 + }
91 115 }
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/event/online/OnlineEventListener.java
  View file @22e1d92
... ... @@ -54,6 +54,7 @@ public class OnlineEventListener implements ApplicationListener&lt;OnlineEvent&gt; {
54 54 @Autowired
55 55 private SIPCommander cmder;
56 56  
  57 +
57 58 private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
58 59  
59 60 @Override
... ... @@ -76,7 +77,7 @@ public class OnlineEventListener implements ApplicationListener&lt;OnlineEvent&gt; {
76 77 if (deviceInStore == null) { //第一次上线
77 78 logger.info("[{}] 首次注册，查询设备信息以及通道信息", device.getDeviceId());
78 79 cmder.deviceInfoQuery(device);
79   - cmder.catalogQuery(device, null);
  80 + deviceService.sync(device);
80 81 }
81 82 break;
82 83 // 设备主动发送心跳触发的在线事件
... ... @@ -99,7 +100,10 @@ public class OnlineEventListener implements ApplicationListener&lt;OnlineEvent&gt; {
99 100 storager.updateDevice(device);
100 101 // 上线添加订阅
101 102 if (device.getSubscribeCycleForCatalog() > 0) {
  103 + // 查询在线设备那些开启了订阅，为设备开启定时的目录订阅
102 104 deviceService.addCatalogSubscribe(device);
  105 + }
  106 + if (device.getSubscribeCycleForMobilePosition() > 0) {
103 107 deviceService.addMobilePositionSubscribe(device);
104 108 }
105 109 }
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/SubscribeListenerForPlatform.java deleted 100644 → 0
  View file @eef78c7
1   -package com.genersoft.iot.vmp.gb28181.event.subscribe;
2   -
3   -import com.genersoft.iot.vmp.common.VideoManagerConstants;
4   -import com.genersoft.iot.vmp.conf.DynamicTask;
5   -import com.genersoft.iot.vmp.conf.RedisKeyExpirationEventMessageListener;
6   -import com.genersoft.iot.vmp.conf.UserSetting;
7   -import org.slf4j.Logger;
8   -import org.slf4j.LoggerFactory;
9   -import org.springframework.beans.factory.annotation.Autowired;
10   -import org.springframework.data.redis.connection.Message;
11   -import org.springframework.data.redis.listener.RedisMessageListenerContainer;
12   -import org.springframework.stereotype.Component;
13   -
14   -/**
15   - * 平台订阅到期事件
16   - */
17   -@Component
18   -public class SubscribeListenerForPlatform extends RedisKeyExpirationEventMessageListener {
19   -
20   - private Logger logger = LoggerFactory.getLogger(SubscribeListenerForPlatform.class);
21   -
22   - @Autowired
23   - private UserSetting userSetting;
24   -
25   - @Autowired
26   - private DynamicTask dynamicTask;
27   -
28   - public SubscribeListenerForPlatform(RedisMessageListenerContainer listenerContainer, UserSetting userSetting) {
29   - super(listenerContainer, userSetting);
30   - }
31   -
32   -
33   - /**
34   - * 监听失效的key
35   - * @param message
36   - * @param pattern
37   - */
38   - @Override
39   - public void onMessage(Message message, byte[] pattern) {
40   - // 获取失效的key
41   - String expiredKey = message.toString();
42   - logger.debug(expiredKey);
43   - // 订阅到期
44   - String PLATFORM_KEEPLIVEKEY_PREFIX = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetting.getServerId() + "_";
45   - if (expiredKey.startsWith(PLATFORM_KEEPLIVEKEY_PREFIX)) {
46   - // 取消定时任务
47   - dynamicTask.stop(expiredKey);
48   - }
49   - }
50   -}
src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java
  View file @22e1d92
... ... @@ -61,8 +61,6 @@ public class CatalogEventLister implements ApplicationListener&lt;CatalogEvent&gt; {
61 61 if (event.getPlatformId() != null) {
62 62 parentPlatform = storager.queryParentPlatByServerGBId(event.getPlatformId());
63 63 if (parentPlatform != null && !parentPlatform.isStatus())return;
64   - String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetting.getServerId() + "_Catalog_" + event.getPlatformId();
65   -// subscribe = redisCatchStorage.getSubscribe(key);
66 64 subscribe = subscribeHolder.getCatalogSubscribe(event.getPlatformId());
67 65  
68 66 if (subscribe == null) {
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/session/CatalogDataCatch.java
  View file @22e1d92
... ... @@ -26,28 +26,35 @@ public class CatalogDataCatch {
26 26 @Autowired
27 27 private IVideoManagerStorage storager;
28 28  
29   - public void addReady(String key) {
30   - CatalogData catalogData = data.get(key);
  29 + public void addReady(Device device, int sn ) {
  30 + CatalogData catalogData = data.get(device.getDeviceId());
31 31 if (catalogData == null || catalogData.getStatus().equals(CatalogData.CatalogDataStatus.end)) {
32 32 catalogData = new CatalogData();
33 33 catalogData.setChannelList(new ArrayList<>());
  34 + catalogData.setDevice(device);
  35 + catalogData.setSn(sn);
34 36 catalogData.setStatus(CatalogData.CatalogDataStatus.ready);
35 37 catalogData.setLastTime(new Date(System.currentTimeMillis()));
36   - data.put(key, catalogData);
  38 + data.put(device.getDeviceId(), catalogData);
37 39 }
38 40 }
39 41  
40   - public void put(String key, int total, Device device, List<DeviceChannel> deviceChannelList) {
41   - CatalogData catalogData = data.get(key);
  42 + public void put(String deviceId, int sn, int total, Device device, List<DeviceChannel> deviceChannelList) {
  43 + CatalogData catalogData = data.get(deviceId);
42 44 if (catalogData == null) {
43 45 catalogData = new CatalogData();
  46 + catalogData.setSn(sn);
44 47 catalogData.setTotal(total);
45 48 catalogData.setDevice(device);
46 49 catalogData.setChannelList(new ArrayList<>());
47 50 catalogData.setStatus(CatalogData.CatalogDataStatus.runIng);
48 51 catalogData.setLastTime(new Date(System.currentTimeMillis()));
49   - data.put(key, catalogData);
  52 + data.put(deviceId, catalogData);
50 53 }else {
  54 + // 同一个设备的通道同步请求只考虑一个，其他的直接忽略
  55 + if (catalogData.getSn() != sn) {
  56 + return;
  57 + }
51 58 catalogData.setTotal(total);
52 59 catalogData.setDevice(device);
53 60 catalogData.setStatus(CatalogData.CatalogDataStatus.runIng);
... ... @@ -56,20 +63,20 @@ public class CatalogDataCatch {
56 63 }
57 64 }
58 65  
59   - public List<DeviceChannel> get(String key) {
60   - CatalogData catalogData = data.get(key);
  66 + public List<DeviceChannel> get(String deviceId) {
  67 + CatalogData catalogData = data.get(deviceId);
61 68 if (catalogData == null) return null;
62 69 return catalogData.getChannelList();
63 70 }
64 71  
65   - public int getTotal(String key) {
66   - CatalogData catalogData = data.get(key);
  72 + public int getTotal(String deviceId) {
  73 + CatalogData catalogData = data.get(deviceId);
67 74 if (catalogData == null) return 0;
68 75 return catalogData.getTotal();
69 76 }
70 77  
71   - public SyncStatus getSyncStatus(String key) {
72   - CatalogData catalogData = data.get(key);
  78 + public SyncStatus getSyncStatus(String deviceId) {
  79 + CatalogData catalogData = data.get(deviceId);
73 80 if (catalogData == null) return null;
74 81 SyncStatus syncStatus = new SyncStatus();
75 82 syncStatus.setCurrent(catalogData.getChannelList().size());
... ... @@ -78,10 +85,6 @@ public class CatalogDataCatch {
78 85 return syncStatus;
79 86 }
80 87  
81   - public void del(String key) {
82   - data.remove(key);
83   - }
84   -
85 88 @Scheduled(fixedRate = 5 * 1000) //每5秒执行一次, 发现数据5秒未更新则移除数据并认为数据接收超时
86 89 private void timerTask(){
87 90 Set<String> keys = data.keySet();
... ... @@ -92,23 +95,30 @@ public class CatalogDataCatch {
92 95 Calendar calendarBefore30S = Calendar.getInstance();
93 96 calendarBefore30S.setTime(new Date());
94 97 calendarBefore30S.set(Calendar.SECOND, calendarBefore30S.get(Calendar.SECOND) - 30);
95   - for (String key : keys) {
96   - CatalogData catalogData = data.get(key);
97   - if (catalogData.getLastTime().before(calendarBefore5S.getTime())) { // 超过五秒收不到消息任务超时， 只更新这一部分数据
98   - storager.resetChannels(catalogData.getDevice().getDeviceId(), catalogData.getChannelList());
99   - String errorMsg = "更新成功，共" + catalogData.getTotal() + "条，已更新" + catalogData.getChannelList().size() + "条";
  98 + for (String deviceId : keys) {
  99 + CatalogData catalogData = data.get(deviceId);
  100 + if ( catalogData.getLastTime().before(calendarBefore5S.getTime())) { // 超过五秒收不到消息任务超时， 只更新这一部分数据
  101 + if (catalogData.getStatus().equals(CatalogData.CatalogDataStatus.runIng)) {
  102 + storager.resetChannels(catalogData.getDevice().getDeviceId(), catalogData.getChannelList());
  103 + if (catalogData.getTotal() != catalogData.getChannelList().size()) {
  104 + String errorMsg = "更新成功，共" + catalogData.getTotal() + "条，已更新" + catalogData.getChannelList().size() + "条";
  105 + catalogData.setErrorMsg(errorMsg);
  106 + }
  107 + }else if (catalogData.getStatus().equals(CatalogData.CatalogDataStatus.ready)) {
  108 + String errorMsg = "同步失败，等待回复超时";
  109 + catalogData.setErrorMsg(errorMsg);
  110 + }
100 111 catalogData.setStatus(CatalogData.CatalogDataStatus.end);
101   - catalogData.setErrorMsg(errorMsg);
102 112 }
103   - if (catalogData.getLastTime().before(calendarBefore30S.getTime())) { // 超过三十秒，如果标记为end则删除
104   - data.remove(key);
  113 + if (catalogData.getStatus().equals(CatalogData.CatalogDataStatus.end) && catalogData.getLastTime().before(calendarBefore30S.getTime())) { // 超过三十秒，如果标记为end则删除
  114 + data.remove(deviceId);
105 115 }
106 116 }
107 117 }
108 118  
109 119  
110   - public void setChannelSyncEnd(String key, String errorMsg) {
111   - CatalogData catalogData = data.get(key);
  120 + public void setChannelSyncEnd(String deviceId, String errorMsg) {
  121 + CatalogData catalogData = data.get(deviceId);
112 122 if (catalogData == null)return;
113 123 catalogData.setStatus(CatalogData.CatalogDataStatus.end);
114 124 catalogData.setErrorMsg(errorMsg);
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/task/ISubscribeTask.java
  View file @22e1d92
1 1 package com.genersoft.iot.vmp.gb28181.task;
2 2  
  3 +import javax.sip.DialogState;
  4 +
3 5 public interface ISubscribeTask extends Runnable{
4 6 void stop();
  7 +
  8 + DialogState getDialogState();
5 9 }
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/CatalogSubscribeTask.java
  View file @22e1d92
... ... @@ -5,6 +5,7 @@ import com.genersoft.iot.vmp.gb28181.task.ISubscribeTask;
5 5 import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
6 6 import org.slf4j.Logger;
7 7 import org.slf4j.LoggerFactory;
  8 +import org.springframework.scheduling.annotation.Async;
8 9  
9 10 import javax.sip.Dialog;
10 11 import javax.sip.DialogState;
... ... @@ -72,4 +73,10 @@ public class CatalogSubscribeTask implements ISubscribeTask {
72 73 });
73 74 }
74 75 }
  76 +
  77 + @Override
  78 + public DialogState getDialogState() {
  79 + if (dialog == null) return null;
  80 + return dialog.getState();
  81 + }
75 82 }
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/MobilePositionSubscribeHandlerTask.java
  View file @22e1d92
1 1 package com.genersoft.iot.vmp.gb28181.task.impl;
2 2  
3   -import com.genersoft.iot.vmp.gb28181.bean.GbStream;
4   -import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
5   -import com.genersoft.iot.vmp.gb28181.bean.SubscribeHolder;
6   -import com.genersoft.iot.vmp.gb28181.bean.SubscribeInfo;
  3 +import com.genersoft.iot.vmp.gb28181.bean.*;
7 4 import com.genersoft.iot.vmp.gb28181.task.ISubscribeTask;
8 5 import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
9 6 import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
10 7 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
11 8 import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
  9 +import org.slf4j.Logger;
  10 +import org.slf4j.LoggerFactory;
  11 +import org.springframework.scheduling.annotation.Async;
12 12  
13   -import java.text.SimpleDateFormat;
  13 +import javax.sip.DialogState;
14 14 import java.util.List;
15 15  
16 16 /**
... ... @@ -18,20 +18,21 @@ import java.util.List;
18 18 */
19 19 public class MobilePositionSubscribeHandlerTask implements ISubscribeTask {
20 20  
  21 + private Logger logger = LoggerFactory.getLogger(MobilePositionSubscribeHandlerTask.class);
  22 +
21 23 private IRedisCatchStorage redisCatchStorage;
22 24 private IVideoManagerStorage storager;
23 25 private ISIPCommanderForPlatform sipCommanderForPlatform;
24 26 private SubscribeHolder subscribeHolder;
25   - private String platformId;
  27 + private ParentPlatform platform;
26 28 private String sn;
27 29 private String key;
28 30  
29   - private final SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
30   -
31 31 public MobilePositionSubscribeHandlerTask(IRedisCatchStorage redisCatchStorage, ISIPCommanderForPlatform sipCommanderForPlatform, IVideoManagerStorage storager, String platformId, String sn, String key, SubscribeHolder subscribeInfo) {
  32 + System.out.println("MobilePositionSubscribeHandlerTask 初始化");
32 33 this.redisCatchStorage = redisCatchStorage;
33 34 this.storager = storager;
34   - this.platformId = platformId;
  35 + this.platform = storager.queryParentPlatByServerGBId(platformId);
35 36 this.sn = sn;
36 37 this.key = key;
37 38 this.sipCommanderForPlatform = sipCommanderForPlatform;
... ... @@ -41,30 +42,31 @@ public class MobilePositionSubscribeHandlerTask implements ISubscribeTask {
41 42 @Override
42 43 public void run() {
43 44  
44   - SubscribeInfo subscribe = subscribeHolder.getMobilePositionSubscribe(platformId);
45   -
  45 + if (platform == null) return;
  46 + SubscribeInfo subscribe = subscribeHolder.getMobilePositionSubscribe(platform.getServerGBId());
46 47 if (subscribe != null) {
47   - ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformId);
48   - if (parentPlatform == null || parentPlatform.isStatus()) {
49   - // TODO 暂时只处理视频流的回复,后续增加对国标设备的支持
50   - List<GbStream> gbStreams = storager.queryGbStreamListInPlatform(platformId);
51   - if (gbStreams.size() > 0) {
52   - for (GbStream gbStream : gbStreams) {
53   - String gbId = gbStream.getGbId();
54   - GPSMsgInfo gpsMsgInfo = redisCatchStorage.getGpsMsgInfo(gbId);
55   - if (gpsMsgInfo != null) {
56   - // 发送GPS消息
57   - sipCommanderForPlatform.sendNotifyMobilePosition(parentPlatform, gpsMsgInfo, subscribe);
58   - }else {
59   - // 没有在redis找到新的消息就使用数据库的消息
60   - gpsMsgInfo = new GPSMsgInfo();
61   - gpsMsgInfo.setId(gbId);
62   - gpsMsgInfo.setLat(gbStream.getLongitude());
63   - gpsMsgInfo.setLng(gbStream.getLongitude());
64   - // 发送GPS消息
65   - sipCommanderForPlatform.sendNotifyMobilePosition(parentPlatform, gpsMsgInfo, subscribe);
66   - }
  48 +
  49 +// if (!parentPlatform.isStatus()) {
  50 +// logger.info("发送订阅时发现平台已经离线：{}", platformId);
  51 +// return;
  52 +// }
  53 + // TODO 暂时只处理视频流的回复,后续增加对国标设备的支持
  54 + List<GbStream> gbStreams = storager.queryGbStreamListInPlatform(platform.getServerGBId());
  55 + if (gbStreams.size() == 0) {
  56 + logger.info("发送订阅时发现平台已经没有关联的直播流：{}", platform.getServerGBId());
  57 + return;
  58 + }
  59 + for (GbStream gbStream : gbStreams) {
  60 + String gbId = gbStream.getGbId();
  61 + GPSMsgInfo gpsMsgInfo = redisCatchStorage.getGpsMsgInfo(gbId);
  62 + if (gpsMsgInfo != null) { // 无最新位置不发送
  63 + logger.info("无最新位置不发送");
  64 + // 经纬度都为0不发送
  65 + if (gpsMsgInfo.getLng() == 0 && gpsMsgInfo.getLat() == 0) {
  66 + continue;
67 67 }
  68 + // 发送GPS消息
  69 + sipCommanderForPlatform.sendNotifyMobilePosition(platform, gpsMsgInfo, subscribe);
68 70 }
69 71 }
70 72 }
... ... @@ -74,4 +76,9 @@ public class MobilePositionSubscribeHandlerTask implements ISubscribeTask {
74 76 public void stop() {
75 77  
76 78 }
  79 +
  80 + @Override
  81 + public DialogState getDialogState() {
  82 + return null;
  83 + }
77 84 }
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/MobilePositionSubscribeTask.java
  View file @22e1d92
... ... @@ -6,10 +6,13 @@ import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
6 6 import org.dom4j.Element;
7 7 import org.slf4j.Logger;
8 8 import org.slf4j.LoggerFactory;
  9 +import org.springframework.scheduling.annotation.Async;
9 10  
10 11 import javax.sip.Dialog;
11 12 import javax.sip.DialogState;
12 13 import javax.sip.ResponseEvent;
  14 +import java.util.Timer;
  15 +import java.util.TimerTask;
13 16  
14 17 /**
15 18 * 移动位置订阅的定时更新
... ... @@ -20,6 +23,8 @@ public class MobilePositionSubscribeTask implements ISubscribeTask {
20 23 private ISIPCommander sipCommander;
21 24 private Dialog dialog;
22 25  
  26 + private Timer timer ;
  27 +
23 28 public MobilePositionSubscribeTask(Device device, ISIPCommander sipCommander) {
24 29 this.device = device;
25 30 this.sipCommander = sipCommander;
... ... @@ -27,10 +32,14 @@ public class MobilePositionSubscribeTask implements ISubscribeTask {
27 32  
28 33 @Override
29 34 public void run() {
  35 + if (timer != null ) {
  36 + timer.cancel();
  37 + timer = null;
  38 + }
30 39 sipCommander.mobilePositionSubscribe(device, dialog, eventResult -> {
31   - if (eventResult.dialog != null || eventResult.dialog.getState().equals(DialogState.CONFIRMED)) {
32   - dialog = eventResult.dialog;
33   - }
  40 +// if (eventResult.dialog != null || eventResult.dialog.getState().equals(DialogState.CONFIRMED)) {
  41 +// dialog = eventResult.dialog;
  42 +// }
34 43 ResponseEvent event = (ResponseEvent) eventResult.event;
35 44 if (event.getResponse().getRawContent() != null) {
36 45 // 成功
... ... @@ -43,6 +52,13 @@ public class MobilePositionSubscribeTask implements ISubscribeTask {
43 52 dialog = null;
44 53 // 失败
45 54 logger.warn("[移动位置订阅]失败，信令发送失败： {}-{} ", device.getDeviceId(), eventResult.msg);
  55 + timer = new Timer();
  56 + timer.schedule(new TimerTask() {
  57 + @Override
  58 + public void run() {
  59 + MobilePositionSubscribeTask.this.run();
  60 + }
  61 + }, 2000);
46 62 });
47 63  
48 64 }
... ... @@ -56,6 +72,10 @@ public class MobilePositionSubscribeTask implements ISubscribeTask {
56 72 * COMPLETED-> Completed Dialog状态-已完成
57 73 * TERMINATED-> Terminated Dialog状态-终止
58 74 */
  75 + if (timer != null ) {
  76 + timer.cancel();
  77 + timer = null;
  78 + }
59 79 if (dialog != null && dialog.getState().equals(DialogState.CONFIRMED)) {
60 80 logger.info("取消移动订阅时dialog状态为{}", dialog.getState());
61 81 device.setSubscribeCycleForMobilePosition(0);
... ... @@ -74,4 +94,9 @@ public class MobilePositionSubscribeTask implements ISubscribeTask {
74 94 });
75 95 }
76 96 }
  97 + @Override
  98 + public DialogState getDialogState() {
  99 + if (dialog == null) return null;
  100 + return dialog.getState();
  101 + }
77 102 }
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java
  View file @22e1d92
... ... @@ -250,7 +250,7 @@ public interface ISIPCommander {
250 250 *
251 251 * @param device 视频设备
252 252 */
253   - boolean catalogQuery(Device device, SipSubscribe.Event errorEvent);
  253 + boolean catalogQuery(Device device, int sn, SipSubscribe.Event errorEvent);
254 254  
255 255 /**
256 256 * 查询录像信息
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
  View file @22e1d92
... ... @@ -1208,14 +1208,14 @@ public class SIPCommander implements ISIPCommander {
1208 1208 * @param device 视频设备
1209 1209 */
1210 1210 @Override
1211   - public boolean catalogQuery(Device device, SipSubscribe.Event errorEvent) {
  1211 + public boolean catalogQuery(Device device, int sn, SipSubscribe.Event errorEvent) {
1212 1212 try {
1213 1213 StringBuffer catalogXml = new StringBuffer(200);
1214 1214 String charset = device.getCharset();
1215 1215 catalogXml.append("<?xml version=\"1.0\" encoding=\"" + charset + "\"?>\r\n");
1216 1216 catalogXml.append("<Query>\r\n");
1217 1217 catalogXml.append("<CmdType>Catalog</CmdType>\r\n");
1218   - catalogXml.append("<SN>" + (int)((Math.random()*9+1)*100000) + "</SN>\r\n");
  1218 + catalogXml.append("<SN>" + sn + "</SN>\r\n");
1219 1219 catalogXml.append("<DeviceID>" + device.getDeviceId() + "</DeviceID>\r\n");
1220 1220 catalogXml.append("</Query>\r\n");
1221 1221  
... ... @@ -1566,17 +1566,28 @@ public class SIPCommander implements ISIPCommander {
1566 1566 cmdXml.append("<DeviceID>" + device.getDeviceId() + "</DeviceID>\r\n");
1567 1567 cmdXml.append("</Query>\r\n");
1568 1568  
1569   - String tm = Long.toString(System.currentTimeMillis());
1570 1569  
1571   - CallIdHeader callIdHeader = device.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
1572   - : udpSipProvider.getNewCallId();
  1570 + Request request;
  1571 + if (dialog != null) {
  1572 + logger.info("发送目录订阅消息时 dialog的状态为： {}", dialog.getState());
  1573 + request = dialog.createRequest(Request.SUBSCRIBE);
  1574 + ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("Application", "MANSCDP+xml");
  1575 + request.setContent(cmdXml.toString(), contentTypeHeader);
  1576 + ExpiresHeader expireHeader = sipFactory.createHeaderFactory().createExpiresHeader(device.getSubscribeCycleForMobilePosition());
  1577 + request.addHeader(expireHeader);
  1578 + }else {
  1579 + String tm = Long.toString(System.currentTimeMillis());
1573 1580  
1574   - // 有效时间默认为60秒以上
1575   - Request request = headerProvider.createSubscribeRequest(device, cmdXml.toString(), "z9hG4bK-viaPos-" + tm,
1576   - "fromTagPos" + tm, null, device.getSubscribeCycleForCatalog(), "Catalog" ,
1577   - callIdHeader);
1578   - transmitRequest(device, request, errorEvent, okEvent);
  1581 + CallIdHeader callIdHeader = device.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
  1582 + : udpSipProvider.getNewCallId();
  1583 +
  1584 + // 有效时间默认为60秒以上
  1585 + request = headerProvider.createSubscribeRequest(device, cmdXml.toString(), "z9hG4bK-viaPos-" + tm,
  1586 + "fromTagPos" + tm, null, device.getSubscribeCycleForCatalog(), "Catalog" ,
  1587 + callIdHeader);
1579 1588  
  1589 + }
  1590 + transmitRequest(device, request, errorEvent, okEvent);
1580 1591 return true;
1581 1592  
1582 1593 } catch ( NumberFormatException | ParseException | InvalidArgumentException | SipException e) {
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java
  View file @22e1d92
... ... @@ -385,7 +385,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
385 385 if (parentPlatform == null) {
386 386 return false;
387 387 }
388   -
  388 + logger.info("[发送 移动位置订阅] {}/{}->{},{}", parentPlatform.getServerGBId(), gpsMsgInfo.getId(), gpsMsgInfo.getLng(), gpsMsgInfo.getLat());
389 389 try {
390 390 String characterSet = parentPlatform.getCharacterSet();
391 391 StringBuffer deviceStatusXml = new StringBuffer(600);
... ... @@ -405,7 +405,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
405 405 CallIdHeader callIdHeader = parentPlatform.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
406 406 : udpSipProvider.getNewCallId();
407 407 callIdHeader.setCallId(subscribeInfo.getCallId());
408   - logger.info("[发送Notify-MobilePosition] {}/{}->{},{}", parentPlatform.getServerGBId(), gpsMsgInfo.getId(), gpsMsgInfo.getLng(), gpsMsgInfo.getLat());
  408 +
409 409 sendNotify(parentPlatform, deviceStatusXml.toString(), subscribeInfo, eventResult -> {
410 410 logger.error("发送NOTIFY通知消息失败。错误：{} {}", eventResult.statusCode, eventResult.msg);
411 411 }, null);
... ... @@ -459,7 +459,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
459 459 // 设置编码， 防止中文乱码
460 460 messageFactory.setDefaultContentEncodingCharset(characterSet);
461 461 Dialog dialog = subscribeInfo.getDialog();
462   - if (dialog == null) return;
  462 + if (dialog == null || !dialog.getState().equals(DialogState.CONFIRMED)) return;
463 463 SIPRequest notifyRequest = (SIPRequest)dialog.createRequest(Request.NOTIFY);
464 464 ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("Application", "MANSCDP+xml");
465 465 notifyRequest.setContent(catalogXmlContent, contentTypeHeader);
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java
  View file @22e1d92
... ... @@ -146,7 +146,7 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
146 146 } else {
147 147 mobilePosition.setAltitude(0.0);
148 148 }
149   - logger.info("[收到Notify-MobilePosition]：{}/{}->{}.{}", mobilePosition.getDeviceId(), mobilePosition.getChannelId(),
  149 + logger.info("[收到 移动位置订阅]：{}/{}->{}.{}", mobilePosition.getDeviceId(), mobilePosition.getChannelId(),
150 150 mobilePosition.getLongitude(), mobilePosition.getLatitude());
151 151 mobilePosition.setReportSource("Mobile Position");
152 152 BaiduPoint bp = GpsUtil.Wgs84ToBd09(String.valueOf(mobilePosition.getLongitude()), String.valueOf(mobilePosition.getLatitude()));
... ... @@ -281,7 +281,7 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
281 281 Element eventElement = itemDevice.element("Event");
282 282 DeviceChannel channel = XmlUtil.channelContentHander(itemDevice);
283 283 channel.setDeviceId(device.getDeviceId());
284   - logger.info("[收到Notify-Catalog]：{}/{}", device.getDeviceId(), channel.getChannelId());
  284 + logger.info("[收到 目录订阅]：{}/{}", device.getDeviceId(), channel.getChannelId());
285 285 switch (eventElement.getText().toUpperCase()) {
286 286 case CatalogEvent.ON: // 上线
287 287 logger.info("收到来自设备【{}】的通道【{}】上线通知", device.getDeviceId(), channel.getChannelId());
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/SubscribeRequestProcessor.java
  View file @22e1d92
... ... @@ -149,8 +149,7 @@ public class SubscribeRequestProcessor extends SIPRequestProcessorParent impleme
149 149 subscribeInfo.setDialog(dialog);
150 150 }
151 151 String sn = XmlUtil.getText(rootElement, "SN");
152   - String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetting.getServerId() + "_MobilePosition_" + platformId;
153   - logger.info("[notify-MobilePosition]: {}", platformId);
  152 + logger.info("[回复 移动位置订阅]: {}", platformId);
154 153 StringBuilder resultXml = new StringBuilder(200);
155 154 resultXml.append("<?xml version=\"1.0\" ?>\r\n")
156 155 .append("<Response>\r\n")
... ... @@ -161,14 +160,25 @@ public class SubscribeRequestProcessor extends SIPRequestProcessorParent impleme
161 160 .append("</Response>\r\n");
162 161  
163 162 if (subscribeInfo.getExpires() > 0) {
164   - if (subscribeHolder.getMobilePositionSubscribe(platformId) != null) {
165   - dynamicTask.stop(key);
166   - }
167 163 String interval = XmlUtil.getText(rootElement, "Interval"); // GPS上报时间间隔
168   - dynamicTask.startCron(key, new MobilePositionSubscribeHandlerTask(redisCatchStorage, sipCommanderForPlatform, storager, platformId, sn, key, subscribeHolder), Integer.parseInt(interval) -1 );
  164 + if (interval == null) {
  165 + subscribeInfo.setGpsInterval(5);
  166 + }else {
  167 + subscribeInfo.setGpsInterval(Integer.parseInt(interval));
  168 + }
  169 +
  170 + subscribeInfo.setSn(sn);
169 171 subscribeHolder.putMobilePositionSubscribe(platformId, subscribeInfo);
  172 +// if (subscribeHolder.getMobilePositionSubscribe(platformId) == null ) {
  173 +// subscribeHolder.putMobilePositionSubscribe(platformId, subscribeInfo);
  174 +// }else {
  175 +// if (subscribeHolder.getMobilePositionSubscribe(platformId).getDialog() != null
  176 +// && subscribeHolder.getMobilePositionSubscribe(platformId).getDialog().getState() != null
  177 +// && !subscribeHolder.getMobilePositionSubscribe(platformId).getDialog().getState().equals(DialogState.CONFIRMED)) {
  178 +// subscribeHolder.putMobilePositionSubscribe(platformId, subscribeInfo);
  179 +// }
  180 +// }
170 181 }else if (subscribeInfo.getExpires() == 0) {
171   - dynamicTask.stop(key);
172 182 subscribeHolder.removeMobilePositionSubscribe(platformId);
173 183 }
174 184  
... ... @@ -202,8 +212,7 @@ public class SubscribeRequestProcessor extends SIPRequestProcessorParent impleme
202 212 subscribeInfo.setDialog(dialog);
203 213 }
204 214 String sn = XmlUtil.getText(rootElement, "SN");
205   - String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetting.getServerId() + "_Catalog_" + platformId;
206   - logger.info("[notify-Catalog]: {}", platformId);
  215 + logger.info("[回复 目录订阅]: {}/{}", platformId, deviceID);
207 216 StringBuilder resultXml = new StringBuilder(200);
208 217 resultXml.append("<?xml version=\"1.0\" ?>\r\n")
209 218 .append("<Response>\r\n")
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java
  View file @22e1d92
... ... @@ -86,23 +86,17 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp
86 86 rootElement = getRootElement(evt, device.getCharset());
87 87 Element deviceListElement = rootElement.element("DeviceList");
88 88 Element sumNumElement = rootElement.element("SumNum");
89   - if (sumNumElement == null || deviceListElement == null) {
  89 + Element snElement = rootElement.element("SN");
  90 + if (snElement == null || sumNumElement == null || deviceListElement == null) {
90 91 responseAck(evt, Response.BAD_REQUEST, "xml error");
91 92 return;
92 93 }
93 94 int sumNum = Integer.parseInt(sumNumElement.getText());
  95 +
94 96 if (sumNum == 0) {
95 97 // 数据已经完整接收
96 98 storager.cleanChannelsForDevice(device.getDeviceId());
97   - RequestMessage msg = new RequestMessage();
98   - msg.setKey(key);
99   - WVPResult<Object> result = new WVPResult<>();
100   - result.setCode(0);
101   - result.setData(device);
102   - msg.setData(result);
103   - result.setMsg("更新成功，共0条");
104   - deferredResultHolder.invokeAllResult(msg);
105   - catalogDataCatch.del(key);
  99 + catalogDataCatch.setChannelSyncEnd(device.getDeviceId(), null);
106 100 }else {
107 101 Iterator<Element> deviceListIterator = deviceListElement.elementIterator();
108 102 if (deviceListIterator != null) {
... ... @@ -123,31 +117,22 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp
123 117  
124 118 channelList.add(deviceChannel);
125 119 }
  120 + int sn = Integer.parseInt(snElement.getText());
126 121 logger.info("收到来自设备【{}】的通道: {}个，{}/{}", device.getDeviceId(), channelList.size(), catalogDataCatch.get(key) == null ? 0 :catalogDataCatch.get(key).size(), sumNum);
127   - catalogDataCatch.put(key, sumNum, device, channelList);
128   - if (catalogDataCatch.get(key).size() == sumNum) {
  122 + catalogDataCatch.put(device.getDeviceId(), sn, sumNum, device, channelList);
  123 + if (catalogDataCatch.get(device.getDeviceId()).size() == sumNum) {
129 124 // 数据已经完整接收
130   - boolean resetChannelsResult = storager.resetChannels(device.getDeviceId(), catalogDataCatch.get(key));
131   - RequestMessage msg = new RequestMessage();
132   - msg.setKey(key);
133   - WVPResult<Object> result = new WVPResult<>();
134   - result.setCode(0);
135   - result.setData(device);
136   - if (resetChannelsResult || sumNum ==0) {
137   - result.setMsg("更新成功，共" + sumNum + "条，已更新" + catalogDataCatch.get(key).size() + "条");
  125 + boolean resetChannelsResult = storager.resetChannels(device.getDeviceId(), catalogDataCatch.get(device.getDeviceId()));
  126 + if (!resetChannelsResult) {
  127 + String errorMsg = "接收成功，写入失败，共" + sumNum + "条，已接收" + catalogDataCatch.get(device.getDeviceId()).size() + "条";
  128 + catalogDataCatch.setChannelSyncEnd(device.getDeviceId(), errorMsg);
138 129 }else {
139   - result.setMsg("接收成功，写入失败，共" + sumNum + "条，已接收" + catalogDataCatch.get(key).size() + "条");
  130 + catalogDataCatch.setChannelSyncEnd(device.getDeviceId(), null);
140 131 }
141   - msg.setData(result);
142   - deferredResultHolder.invokeAllResult(msg);
143   - catalogDataCatch.del(key);
144 132 }
145 133 }
146 134 // 回复200 OK
147 135 responseAck(evt, Response.OK);
148   - if (offLineDetector.isOnline(device.getDeviceId())) {
149   - publisher.onlineEventPublish(device, VideoManagerConstants.EVENT_ONLINE_MESSAGE);
150   - }
151 136 }
152 137 } catch (DocumentException e) {
153 138 e.printStackTrace();
... ... @@ -231,21 +216,18 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp
231 216 }
232 217  
233 218 public SyncStatus getChannelSyncProgress(String deviceId) {
234   - String key = DeferredResultHolder.CALLBACK_CMD_CATALOG + deviceId;
235   - if (catalogDataCatch.get(key) == null) {
  219 + if (catalogDataCatch.get(deviceId) == null) {
236 220 return null;
237 221 }else {
238   - return catalogDataCatch.getSyncStatus(key);
  222 + return catalogDataCatch.getSyncStatus(deviceId);
239 223 }
240 224 }
241 225  
242   - public void setChannelSyncReady(String deviceId) {
243   - String key = DeferredResultHolder.CALLBACK_CMD_CATALOG + deviceId;
244   - catalogDataCatch.addReady(key);
  226 + public void setChannelSyncReady(Device device, int sn) {
  227 + catalogDataCatch.addReady(device, sn);
245 228 }
246 229  
247 230 public void setChannelSyncEnd(String deviceId, String errorMsg) {
248   - String key = DeferredResultHolder.CALLBACK_CMD_CATALOG + deviceId;
249   - catalogDataCatch.setChannelSyncEnd(key, errorMsg);
  231 + catalogDataCatch.setChannelSyncEnd(deviceId, errorMsg);
250 232 }
251 233 }
... ...
src/main/java/com/genersoft/iot/vmp/service/IDeviceService.java
  View file @22e1d92
... ... @@ -44,15 +44,8 @@ public interface IDeviceService {
44 44 SyncStatus getChannelSyncStatus(String deviceId);
45 45  
46 46 /**
47   - * 设置通道同步状态
48   - * @param deviceId 设备ID
49   - */
50   - void setChannelSyncReady(String deviceId);
51   -
52   - /**
53   - * 设置同步结束
54   - * @param deviceId 设备ID
55   - * @param errorMsg 错误信息
  47 + * 通道同步
  48 + * @param device
56 49 */
57   - void setChannelSyncEnd(String deviceId, String errorMsg);
  50 + void sync(Device device);
58 51 }
... ...
src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java
  View file @22e1d92
... ... @@ -14,6 +14,8 @@ import org.slf4j.LoggerFactory;
14 14 import org.springframework.beans.factory.annotation.Autowired;
15 15 import org.springframework.stereotype.Service;
16 16  
  17 +import javax.sip.DialogState;
  18 +
17 19 /**
18 20 * 设备业务（目录订阅）
19 21 */
... ... @@ -39,19 +41,17 @@ public class DeviceServiceImpl implements IDeviceService {
39 41 if (device == null || device.getSubscribeCycleForCatalog() < 0) {
40 42 return false;
41 43 }
42   - if (dynamicTask.contains(device.getDeviceId() + "catalog")) {
43   - // 存在则停止现有的，开启新的
44   - dynamicTask.stop(device.getDeviceId() + "catalog");
  44 + CatalogSubscribeTask task = (CatalogSubscribeTask)dynamicTask.get(device.getDeviceId() + "catalog");
  45 + if (task != null && task.getDialogState() != null && task.getDialogState().equals(DialogState.CONFIRMED)) { // 已存在不需要再次添加
  46 + return true;
45 47 }
46 48 logger.info("[添加目录订阅] 设备{}", device.getDeviceId());
47 49 // 添加目录订阅
48 50 CatalogSubscribeTask catalogSubscribeTask = new CatalogSubscribeTask(device, sipCommander);
49   - catalogSubscribeTask.run();
50 51 // 提前开始刷新订阅
51   - int subscribeCycleForCatalog = device.getSubscribeCycleForCatalog();
  52 + int subscribeCycleForCatalog = Math.max(device.getSubscribeCycleForCatalog(),30);
52 53 // 设置最小值为30
53   - subscribeCycleForCatalog = Math.max(subscribeCycleForCatalog, 30);
54   - dynamicTask.startCron(device.getDeviceId() + "catalog", catalogSubscribeTask, subscribeCycleForCatalog);
  54 + dynamicTask.startCron(device.getDeviceId() + "catalog", catalogSubscribeTask, subscribeCycleForCatalog -1);
55 55 return true;
56 56 }
57 57  
... ... @@ -70,18 +70,16 @@ public class DeviceServiceImpl implements IDeviceService {
70 70 if (device == null || device.getSubscribeCycleForMobilePosition() < 0) {
71 71 return false;
72 72 }
73   - if (dynamicTask.contains(device.getDeviceId() + "mobile_position")) {
74   - // 存在则停止现有的，开启新的
75   - dynamicTask.stop(device.getDeviceId() + "mobile_position");
76   - }
77 73 logger.info("[添加移动位置订阅] 设备{}", device.getDeviceId());
  74 + MobilePositionSubscribeTask task = (MobilePositionSubscribeTask)dynamicTask.get(device.getDeviceId() + "mobile_position");
  75 + if (task != null && task.getDialogState() != null && task.getDialogState().equals(DialogState.CONFIRMED)) { // 已存在不需要再次添加
  76 + return true;
  77 + }
78 78 // 添加目录订阅
79 79 MobilePositionSubscribeTask mobilePositionSubscribeTask = new MobilePositionSubscribeTask(device, sipCommander);
80   - mobilePositionSubscribeTask.run();
81 80 // 提前开始刷新订阅
82   - int subscribeCycleForCatalog = device.getSubscribeCycleForCatalog();
83 81 // 设置最小值为30
84   - subscribeCycleForCatalog = Math.max(subscribeCycleForCatalog, 30);
  82 + int subscribeCycleForCatalog = Math.max(device.getSubscribeCycleForMobilePosition(),30);
85 83 dynamicTask.startCron(device.getDeviceId() + "mobile_position" , mobilePositionSubscribeTask, subscribeCycleForCatalog -1 );
86 84 return true;
87 85 }
... ... @@ -102,12 +100,16 @@ public class DeviceServiceImpl implements IDeviceService {
102 100 }
103 101  
104 102 @Override
105   - public void setChannelSyncReady(String deviceId) {
106   - catalogResponseMessageHandler.setChannelSyncReady(deviceId);
107   - }
108   -
109   - @Override
110   - public void setChannelSyncEnd(String deviceId, String errorMsg) {
111   - catalogResponseMessageHandler.setChannelSyncEnd(deviceId, errorMsg);
  103 + public void sync(Device device) {
  104 + if (catalogResponseMessageHandler.getChannelSyncProgress(device.getDeviceId()) != null) {
  105 + logger.info("开启同步时发现同步已经存在");
  106 + return;
  107 + }
  108 + int sn = (int)((Math.random()*9+1)*100000);
  109 + catalogResponseMessageHandler.setChannelSyncReady(device, sn);
  110 + sipCommander.catalogQuery(device, sn, event -> {
  111 + String errorMsg = String.format("同步通道失败，错误码： %s, %s", event.statusCode, event.msg);
  112 + catalogResponseMessageHandler.setChannelSyncEnd(device.getDeviceId(), errorMsg);
  113 + });
112 114 }
113 115 }
... ...
src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java
  View file @22e1d92
... ... @@ -238,12 +238,15 @@ public class VideoManagerStorageImpl implements IVideoManagerStorage {
238 238  
239 239 @Override
240 240 public boolean resetChannels(String deviceId, List<DeviceChannel> deviceChannelList) {
  241 + if (deviceChannelList == null) {
  242 + return false;
  243 + }
241 244 TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
242 245 // 数据去重
243 246 List<DeviceChannel> channels = new ArrayList<>();
244 247 StringBuilder stringBuilder = new StringBuilder();
245 248 Map<String, Integer> subContMap = new HashMap<>();
246   - if (deviceChannelList.size() > 1) {
  249 + if (deviceChannelList != null && deviceChannelList.size() > 1) {
247 250 // 数据去重
248 251 Set<String> gbIdSet = new HashSet<>();
249 252 for (DeviceChannel deviceChannel : deviceChannelList) {
... ... @@ -300,6 +303,7 @@ public class VideoManagerStorageImpl implements IVideoManagerStorage {
300 303 dataSourceTransactionManager.commit(transactionStatus); //手动提交
301 304 return true;
302 305 }catch (Exception e) {
  306 + e.printStackTrace();
303 307 dataSourceTransactionManager.rollback(transactionStatus);
304 308 return false;
305 309 }
... ... @@ -415,10 +419,9 @@ public class VideoManagerStorageImpl implements IVideoManagerStorage {
415 419 TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
416 420 boolean result = false;
417 421 try {
418   - if (platformChannelMapper.delChannelForDeviceId(deviceId) <0 // 删除与国标平台的关联
419   - || deviceChannelMapper.cleanChannelsByDeviceId(deviceId) < 0 // 删除他的通道
420   - || deviceMapper.del(deviceId) < 0 // 移除设备信息
421   - ) {
  422 + platformChannelMapper.delChannelForDeviceId(deviceId);
  423 + deviceChannelMapper.cleanChannelsByDeviceId(deviceId);
  424 + if ( deviceMapper.del(deviceId) < 0 ) {
422 425 //事务回滚
423 426 dataSourceTransactionManager.rollback(transactionStatus);
424 427 }
... ...
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/device/DeviceQuery.java
  View file @22e1d92
... ... @@ -4,8 +4,13 @@ import com.alibaba.fastjson.JSONObject;
4 4 import com.genersoft.iot.vmp.conf.DynamicTask;
5 5 import com.genersoft.iot.vmp.gb28181.bean.Device;
6 6 import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
  7 +import com.genersoft.iot.vmp.gb28181.bean.SubscribeHolder;
7 8 import com.genersoft.iot.vmp.gb28181.bean.SyncStatus;
8 9 import com.genersoft.iot.vmp.gb28181.event.DeviceOffLineDetector;
  10 +import com.genersoft.iot.vmp.gb28181.task.ISubscribeTask;
  11 +import com.genersoft.iot.vmp.gb28181.task.impl.CatalogSubscribeTask;
  12 +import com.genersoft.iot.vmp.gb28181.task.impl.MobilePositionSubscribeHandlerTask;
  13 +import com.genersoft.iot.vmp.gb28181.task.impl.MobilePositionSubscribeTask;
9 14 import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
10 15 import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
11 16 import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
... ... @@ -29,9 +34,8 @@ import org.springframework.util.StringUtils;
29 34 import org.springframework.web.bind.annotation.*;
30 35 import org.springframework.web.context.request.async.DeferredResult;
31 36  
32   -import java.util.List;
33   -import java.util.Set;
34   -import java.util.UUID;
  37 +import javax.sip.DialogState;
  38 +import java.util.*;
35 39  
36 40 @Api(tags = "国标设备查询", value = "国标设备查询")
37 41 @SuppressWarnings("rawtypes")
... ... @@ -63,6 +67,9 @@ public class DeviceQuery {
63 67 @Autowired
64 68 private DynamicTask dynamicTask;
65 69  
  70 + @Autowired
  71 + private SubscribeHolder subscribeHolder;
  72 +
66 73 /**
67 74 * 使用ID查询国标设备
68 75 * @param deviceId 国标ID
... ... @@ -165,12 +172,8 @@ public class DeviceQuery {
165 172 wvpResult.setData(syncStatus);
166 173 return wvpResult;
167 174 }
168   - SyncStatus syncStatusReady = new SyncStatus();
169   - deviceService.setChannelSyncReady(deviceId);
170   - cmder.catalogQuery(device, event -> {
171   - String errorMsg = String.format("同步通道失败，错误码： %s, %s", event.statusCode, event.msg);
172   - deviceService.setChannelSyncEnd(deviceId, errorMsg);
173   - });
  175 + deviceService.sync(device);
  176 +
174 177 WVPResult<SyncStatus> wvpResult = new WVPResult<>();
175 178 wvpResult.setCode(0);
176 179 wvpResult.setMsg("开始同步");
... ... @@ -469,4 +472,29 @@ public class DeviceQuery {
469 472 }
470 473 return wvpResult;
471 474 }
  475 +
  476 + @GetMapping("/{deviceId}/subscribe_info")
  477 + @ApiOperation(value = "获取设备的订阅状态", notes = "获取设备的订阅状态")
  478 + public WVPResult<Map<String, String>> getSubscribeInfo(@PathVariable String deviceId) {
  479 + Set<String> allKeys = dynamicTask.getAllKeys();
  480 + Map<String, String> dialogStateMap = new HashMap<>();
  481 + for (String key : allKeys) {
  482 + if (key.startsWith(deviceId)) {
  483 + ISubscribeTask subscribeTask = (ISubscribeTask)dynamicTask.get(key);
  484 + DialogState dialogState = subscribeTask.getDialogState();
  485 + if (dialogState == null) {
  486 + continue;
  487 + }
  488 + if (subscribeTask instanceof CatalogSubscribeTask) {
  489 + dialogStateMap.put("catalog", dialogState.toString());
  490 + }else if (subscribeTask instanceof MobilePositionSubscribeTask) {
  491 + dialogStateMap.put("mobilePosition", dialogState.toString());
  492 + }
  493 + }
  494 + }
  495 + WVPResult<Map<String, String>> wvpResult = new WVPResult<>();
  496 + wvpResult.setCode(0);
  497 + wvpResult.setData(dialogStateMap);
  498 + return wvpResult;
  499 + }
472 500 }
... ...
src/main/java/com/genersoft/iot/vmp/vmanager/server/ServerController.java
  View file @22e1d92
... ... @@ -4,6 +4,7 @@ import com.alibaba.fastjson.JSON;
4 4 import com.alibaba.fastjson.JSONObject;
5 5 import com.genersoft.iot.vmp.VManageBootstrap;
6 6 import com.genersoft.iot.vmp.common.VersionPo;
  7 +import com.genersoft.iot.vmp.conf.DynamicTask;
7 8 import com.genersoft.iot.vmp.conf.SipConfig;
8 9 import com.genersoft.iot.vmp.conf.UserSetting;
9 10 import com.genersoft.iot.vmp.conf.VersionInfo;
... ... @@ -27,6 +28,7 @@ import javax.sip.ObjectInUseException;
27 28 import javax.sip.SipProvider;
28 29 import java.util.Iterator;
29 30 import java.util.List;
  31 +import java.util.Set;
30 32  
31 33 @SuppressWarnings("rawtypes")
32 34 @Api(tags = "服务控制")
... ... @@ -42,13 +44,16 @@ public class ServerController {
42 44 private IMediaServerService mediaServerService;
43 45  
44 46 @Autowired
45   - VersionInfo versionInfo;
  47 + private VersionInfo versionInfo;
46 48  
47 49 @Autowired
48   - SipConfig sipConfig;
  50 + private SipConfig sipConfig;
49 51  
50 52 @Autowired
51   - UserSetting userSetting;
  53 + private UserSetting userSetting;
  54 +
  55 + @Autowired
  56 + private DynamicTask dynamicTask;
52 57  
53 58 @Value("${server.port}")
54 59 private int serverPort;
... ... @@ -248,4 +253,35 @@ public class ServerController {
248 253 result.setData(jsonObject);
249 254 return result;
250 255 }
  256 +
  257 +// @ApiOperation("当前进行中的动态任务")
  258 +// @GetMapping(value = "/dynamicTask")
  259 +// @ResponseBody
  260 +// public WVPResult<JSONObject> getDynamicTask(){
  261 +// WVPResult<JSONObject> result = new WVPResult<>();
  262 +// result.setCode(0);
  263 +// result.setMsg("success");
  264 +//
  265 +// JSONObject jsonObject = new JSONObject();
  266 +//
  267 +// Set<String> allKeys = dynamicTask.getAllKeys();
  268 +// jsonObject.put("server.port", serverPort);
  269 +// if (StringUtils.isEmpty(type)) {
  270 +// jsonObject.put("sip", JSON.toJSON(sipConfig));
  271 +// jsonObject.put("base", JSON.toJSON(userSetting));
  272 +// }else {
  273 +// switch (type){
  274 +// case "sip":
  275 +// jsonObject.put("sip", sipConfig);
  276 +// break;
  277 +// case "base":
  278 +// jsonObject.put("base", userSetting);
  279 +// break;
  280 +// default:
  281 +// break;
  282 +// }
  283 +// }
  284 +// result.setData(jsonObject);
  285 +// return result;
  286 +// }
251 287 }
... ...
web_src/src/components/dialog/SyncChannelProgress.vue
  View file @22e1d92
... ... @@ -61,23 +61,36 @@ export default {
61 61 if (!this.syncFlag) {
62 62 this.syncFlag = true;
63 63 }
64   - if (res.data.data == null) {
65   - this.syncStatus = "success"
66   - this.percentage = 100;
67   - this.msg = '同步成功';
68   - }else if (res.data.data.total == 0){
69   - this.msg = `等待同步中`;
70   - this.timmer = setTimeout(this.getProgress, 300)
71   - }else if (res.data.data.errorMsg !== null ){
72   - this.msg = res.data.data.errorMsg;
73   - this.syncStatus = "exception"
74   - }else {
75   - this.total = res.data.data.total;
76   - this.current = res.data.data.current;
77   - this.percentage = Math.floor(Number(res.data.data.current)/Number(res.data.data.total)* 10000)/100;
78   - this.msg = `同步中...[${res.data.data.current}/${res.data.data.total}]`;
79   - this.timmer = setTimeout(this.getProgress, 300)
  64 +
  65 + if (res.data.data != null) {
  66 + if (res.data.data.total == 0) {
  67 + if (res.data.data.errorMsg !== null ){
  68 + this.msg = res.data.data.errorMsg;
  69 + this.syncStatus = "exception"
  70 + }else {
  71 + this.msg = `等待同步中`;
  72 + this.timmer = setTimeout(this.getProgress, 300)
  73 + }
  74 + }else {
  75 + if (res.data.data.total == res.data.data.current) {
  76 + this.syncStatus = "success"
  77 + this.percentage = 100;
  78 + this.msg = '同步成功';
  79 + }else {
  80 + if (res.data.data.errorMsg !== null ){
  81 + this.msg = res.data.data.errorMsg;
  82 + this.syncStatus = "exception"
  83 + }else {
  84 + this.total = res.data.data.total;
  85 + this.current = res.data.data.current;
  86 + this.percentage = Math.floor(Number(res.data.data.current)/Number(res.data.data.total)* 10000)/100;
  87 + this.msg = `同步中...[${res.data.data.current}/${res.data.data.total}]`;
  88 + this.timmer = setTimeout(this.getProgress, 300)
  89 + }
  90 + }
  91 + }
80 92 }
  93 +
81 94 }else {
82 95 if (this.syncFlag) {
83 96 this.syncStatus = "success"
... ...