From 1de344674afd6bb35b51b165bbad76dbe6299b7e Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期四, 28 三月 2024 18:08:33 +0800
Subject: [PATCH] 使用冒号分隔redis的key

---
 src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java |  224 ++++++++++++++++++++++++++++++++++++++++----------------
 1 files changed, 160 insertions(+), 64 deletions(-)

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
old mode 100644
new mode 100755
index 8a6257f..2c2674f
--- a/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java
@@ -1,15 +1,21 @@
 package com.genersoft.iot.vmp.service.impl;
 
+import com.baomidou.dynamic.datasource.annotation.DS;
+import com.genersoft.iot.vmp.common.CommonCallback;
 import com.genersoft.iot.vmp.common.VideoManagerConstants;
 import com.genersoft.iot.vmp.conf.DynamicTask;
 import com.genersoft.iot.vmp.conf.UserSetting;
 import com.genersoft.iot.vmp.gb28181.bean.*;
+import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;
 import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
 import com.genersoft.iot.vmp.gb28181.task.ISubscribeTask;
 import com.genersoft.iot.vmp.gb28181.task.impl.CatalogSubscribeTask;
 import com.genersoft.iot.vmp.gb28181.task.impl.MobilePositionSubscribeTask;
 import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
+import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
 import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd.CatalogResponseMessageHandler;
+import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
+import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
 import com.genersoft.iot.vmp.service.IDeviceChannelService;
 import com.genersoft.iot.vmp.service.IDeviceService;
 import com.genersoft.iot.vmp.service.IInviteStreamService;
@@ -34,19 +40,20 @@
 import javax.sip.SipException;
 import java.text.ParseException;
 import java.time.Instant;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
+import java.util.*;
 import java.util.concurrent.TimeUnit;
 
 /**
  * 璁惧涓氬姟锛堢洰褰曡闃咃級
  */
 @Service
+@DS("master")
 public class DeviceServiceImpl implements IDeviceService {
 
     private final static Logger logger = LoggerFactory.getLogger(DeviceServiceImpl.class);
 
+    @Autowired
+    private SIPCommander cmder;
     @Autowired
     private DynamicTask dynamicTask;
 
@@ -92,6 +99,12 @@
     @Autowired
     private IMediaServerService mediaServerService;
 
+    @Autowired
+    private AudioBroadcastManager audioBroadcastManager;
+
+    @Autowired
+    private ZLMRESTfulUtils zlmresTfulUtils;
+
     @Override
     public void online(Device device, SipTransactionInfo sipTransactionInfo) {
         logger.info("[璁惧涓婄嚎] deviceId锛歿}->{}:{}", device.getDeviceId(), device.getIp(), device.getPort());
@@ -104,6 +117,7 @@
             inviteStreamService.clearInviteInfo(device.getDeviceId());
         }
         device.setUpdateTime(now);
+        device.setKeepaliveTime(now);
         if (device.getKeepaliveIntervalTime() == 0) {
             // 榛樿蹇冭烦闂撮殧60
             device.setKeepaliveIntervalTime(60);
@@ -117,9 +131,10 @@
         }
 
         // 绗竴娆′笂绾� 鎴栧垯璁惧涔嬪墠鏄绾跨姸鎬�--杩涜閫氶亾鍚屾鍜岃澶囦俊鎭煡璇�
-        if (device.getCreateTime() == null) {
+        if (deviceInDb == null) {
             device.setOnLine(true);
             device.setCreateTime(now);
+            device.setUpdateTime(now);
             logger.info("[璁惧涓婄嚎,棣栨娉ㄥ唽]: {}锛屾煡璇㈣澶囦俊鎭互鍙婇�氶亾淇℃伅", device.getDeviceId());
             deviceMapper.add(device);
             redisCatchStorage.updateDevice(device);
@@ -145,6 +160,19 @@
                     sync(device);
                     // TODO 濡傛灉璁惧涓嬬殑閫氶亾绾ц仈鍒颁簡鍏朵粬骞冲彴锛岄偅涔堥渶瑕佸彂閫佷簨浠舵垨鑰卬otify缁欎笂绾у钩鍙�
                 }
+                // 涓婄嚎娣诲姞璁㈤槄
+                if (device.getSubscribeCycleForCatalog() > 0) {
+                    // 鏌ヨ鍦ㄧ嚎璁惧閭d簺寮�鍚簡璁㈤槄锛屼负璁惧寮�鍚畾鏃剁殑鐩綍璁㈤槄
+                    addCatalogSubscribe(device);
+                }
+                if (device.getSubscribeCycleForMobilePosition() > 0) {
+                    addMobilePositionSubscribe(device);
+                }
+                if (userSetting.getDeviceStatusNotify()) {
+                    // 鍙戦�乺edis娑堟伅
+                    redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), null, true);
+                }
+
             }else {
                 if (deviceChannelMapper.queryAllChannels(device.getDeviceId()).size() == 0) {
                     logger.info("[璁惧涓婄嚎]: {}锛岄�氶亾鏁颁负0,鏌ヨ閫氶亾淇℃伅", device.getDeviceId());
@@ -157,22 +185,21 @@
 
         }
 
-        // 涓婄嚎娣诲姞璁㈤槄
-        if (device.getSubscribeCycleForCatalog() > 0) {
-            // 鏌ヨ鍦ㄧ嚎璁惧閭d簺寮�鍚簡璁㈤槄锛屼负璁惧寮�鍚畾鏃剁殑鐩綍璁㈤槄
-            addCatalogSubscribe(device);
-        }
-        if (device.getSubscribeCycleForMobilePosition() > 0) {
-            addMobilePositionSubscribe(device);
-        }
         // 鍒锋柊杩囨湡浠诲姟
         String registerExpireTaskKey = VideoManagerConstants.REGISTER_EXPIRE_TASK_KEY_PREFIX + device.getDeviceId();
         // 濡傛灉绗竴娆℃敞鍐岄偅涔堝繀椤诲湪60 * 3鏃堕棿鍐呮敹鍒颁竴涓績璺筹紝鍚﹀垯璁惧绂荤嚎
         dynamicTask.startDelay(registerExpireTaskKey, ()-> offline(device.getDeviceId(), "棣栨娉ㄥ唽鍚庢湭鑳芥敹鍒板績璺�"), device.getKeepaliveIntervalTime() * 1000 * 3);
-        if (userSetting.getDeviceStatusNotify()) {
-            // 鍙戦�乺edis娑堟伅
-            redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), null, true);
-        }
+
+//
+//        try {
+//            cmder.alarmSubscribe(device, 600, "0", "4", "0", "2023-7-27T00:00:00", "2023-7-28T00:00:00");
+//        } catch (InvalidArgumentException e) {
+//            throw new RuntimeException(e);
+//        } catch (SipException e) {
+//            throw new RuntimeException(e);
+//        } catch (ParseException e) {
+//            throw new RuntimeException(e);
+//        }
 
     }
 
@@ -185,6 +212,13 @@
         }
         String registerExpireTaskKey = VideoManagerConstants.REGISTER_EXPIRE_TASK_KEY_PREFIX + deviceId;
         dynamicTask.stop(registerExpireTaskKey);
+        if (device.isOnLine()) {
+            if (userSetting.getDeviceStatusNotify()) {
+                // 鍙戦�乺edis娑堟伅
+                redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), null, false);
+            }
+        }
+
         device.setOnLine(false);
         redisCatchStorage.updateDevice(device);
         deviceMapper.update(device);
@@ -196,17 +230,31 @@
             for (SsrcTransaction ssrcTransaction : ssrcTransactions) {
                 mediaServerService.releaseSsrc(ssrcTransaction.getMediaServerId(), ssrcTransaction.getSsrc());
                 mediaServerService.closeRTPServer(ssrcTransaction.getMediaServerId(), ssrcTransaction.getStream());
-                streamSession.remove(deviceId, ssrcTransaction.getChannelId(), ssrcTransaction.getStream());
+                streamSession.removeByCallId(deviceId, ssrcTransaction.getChannelId(), ssrcTransaction.getCallId());
             }
         }
         // 绉婚櫎璁㈤槄
-        removeCatalogSubscribe(device);
-        removeMobilePositionSubscribe(device);
-        if (userSetting.getDeviceStatusNotify()) {
-            // 鍙戦�乺edis娑堟伅
-            redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), null, false);
-        }
+        removeCatalogSubscribe(device, null);
+        removeMobilePositionSubscribe(device, null);
 
+        List<AudioBroadcastCatch> audioBroadcastCatches = audioBroadcastManager.get(deviceId);
+        if (audioBroadcastCatches.size() > 0) {
+            for (AudioBroadcastCatch audioBroadcastCatch : audioBroadcastCatches) {
+
+                SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(deviceId, audioBroadcastCatch.getChannelId(), null, null);
+                if (sendRtpItem != null) {
+                    redisCatchStorage.deleteSendRTPServer(deviceId, sendRtpItem.getChannelId(), null, null);
+                    MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
+                    Map<String, Object> param = new HashMap<>();
+                    param.put("vhost", "__defaultVhost__");
+                    param.put("app", sendRtpItem.getApp());
+                    param.put("stream", sendRtpItem.getStream());
+                    zlmresTfulUtils.stopSendRtp(mediaInfo, param);
+                }
+
+                audioBroadcastManager.del(deviceId, audioBroadcastCatch.getChannelId());
+            }
+        }
     }
 
     @Override
@@ -221,11 +269,13 @@
         int subscribeCycleForCatalog = Math.max(device.getSubscribeCycleForCatalog(),30);
         // 璁剧疆鏈�灏忓�间负30
         dynamicTask.startCron(device.getDeviceId() + "catalog", catalogSubscribeTask, (subscribeCycleForCatalog -1) * 1000);
+
+        catalogSubscribeTask.run();
         return true;
     }
 
     @Override
-    public boolean removeCatalogSubscribe(Device device) {
+    public boolean removeCatalogSubscribe(Device device, CommonCallback<Boolean> callback) {
         if (device == null || device.getSubscribeCycleForCatalog() < 0) {
             return false;
         }
@@ -235,7 +285,7 @@
             Runnable runnable = dynamicTask.get(taskKey);
             if (runnable instanceof ISubscribeTask) {
                 ISubscribeTask subscribeTask = (ISubscribeTask) runnable;
-                subscribeTask.stop();
+                subscribeTask.stop(callback);
             }
         }
         dynamicTask.stop(taskKey);
@@ -253,12 +303,13 @@
         // 璁剧疆鏈�灏忓�间负30
         int subscribeCycleForCatalog = Math.max(device.getSubscribeCycleForMobilePosition(),30);
         // 鍒锋柊璁㈤槄
-        dynamicTask.startCron(device.getDeviceId() + "mobile_position" , mobilePositionSubscribeTask, (subscribeCycleForCatalog) * 1000);
+        dynamicTask.startCron(device.getDeviceId() + "mobile_position" , mobilePositionSubscribeTask, subscribeCycleForCatalog * 1000);
+        mobilePositionSubscribeTask.run();
         return true;
     }
 
     @Override
-    public boolean removeMobilePositionSubscribe(Device device) {
+    public boolean removeMobilePositionSubscribe(Device device, CommonCallback<Boolean> callback) {
         if (device == null || device.getSubscribeCycleForCatalog() < 0) {
             return false;
         }
@@ -268,7 +319,7 @@
             Runnable runnable = dynamicTask.get(taskKey);
             if (runnable instanceof ISubscribeTask) {
                 ISubscribeTask subscribeTask = (ISubscribeTask) runnable;
-                subscribeTask.stop();
+                subscribeTask.stop(callback);
             }
         }
         dynamicTask.stop(taskKey);
@@ -380,8 +431,8 @@
         if (device == null) {
             return null;
         }
-        if (ObjectUtils.isEmpty(parentId) || parentId.equals(deviceId)) {
-            parentId = null;
+        if (ObjectUtils.isEmpty(parentId) ) {
+            parentId = deviceId;
         }
         List<DeviceChannel> rootNodes = deviceChannelMapper.getSubChannelsByDeviceId(deviceId, parentId, onlyCatalog);
         return transportChannelsToTree(rootNodes, "");
@@ -459,6 +510,7 @@
             logger.warn("鏇存柊璁惧鏃舵湭鎵惧埌璁惧淇℃伅");
             return;
         }
+
         if (!ObjectUtils.isEmpty(device.getName())) {
             deviceInStore.setName(device.getName());
         }
@@ -468,46 +520,87 @@
         if (!ObjectUtils.isEmpty(device.getMediaServerId())) {
             deviceInStore.setMediaServerId(device.getMediaServerId());
         }
-        deviceInStore.setSdpIp(device.getSdpIp());
-        deviceInStore.setCharset(device.getCharset());
-        deviceInStore.setTreeType(device.getTreeType());
-
+        if (!ObjectUtils.isEmpty(device.getCharset())) {
+            deviceInStore.setCharset(device.getCharset());
+        }
+        if (!ObjectUtils.isEmpty(device.getSdpIp())) {
+            deviceInStore.setSdpIp(device.getSdpIp());
+        }
+        if (!ObjectUtils.isEmpty(device.getPassword())) {
+            deviceInStore.setPassword(device.getPassword());
+        }
+        if (!ObjectUtils.isEmpty(device.getStreamMode())) {
+            deviceInStore.setStreamMode(device.getStreamMode());
+        }
         //  鐩綍璁㈤槄鐩稿叧鐨勪俊鎭�
-        if (device.getSubscribeCycleForCatalog() > 0) {
-            if (deviceInStore.getSubscribeCycleForCatalog() == 0 || deviceInStore.getSubscribeCycleForCatalog() != device.getSubscribeCycleForCatalog()) {
-                deviceInStore.setSubscribeCycleForCatalog(device.getSubscribeCycleForCatalog());
-                // 寮�鍚闃�
-                addCatalogSubscribe(deviceInStore);
-            }
-        }else if (device.getSubscribeCycleForCatalog() == 0) {
-            if (deviceInStore.getSubscribeCycleForCatalog() != 0) {
-                deviceInStore.setSubscribeCycleForCatalog(device.getSubscribeCycleForCatalog());
-                // 鍙栨秷璁㈤槄
-                removeCatalogSubscribe(deviceInStore);
-            }
-        }
+        if (deviceInStore.getSubscribeCycleForCatalog() != device.getSubscribeCycleForCatalog()) {
+            if (device.getSubscribeCycleForCatalog() > 0) {
+                // 鑻ュ凡寮�鍚闃咃紝浣嗚闃呭懆鏈熶笉鍚岋紝鍒欏厛鍙栨秷
+                if (deviceInStore.getSubscribeCycleForCatalog() != 0) {
+                    removeCatalogSubscribe(deviceInStore, result->{
+                        // 寮�鍚闃�
+                        deviceInStore.setSubscribeCycleForCatalog(device.getSubscribeCycleForCatalog());
+                        addCatalogSubscribe(deviceInStore);
+                        // 鍥犱负鏄紓姝ユ墽琛岋紝闇�瑕佸湪杩欓噷鏇存柊涓嬫暟鎹�
+                        deviceMapper.updateCustom(deviceInStore);
+                        redisCatchStorage.updateDevice(deviceInStore);
+                    });
+                }else {
+                    // 寮�鍚闃�
+                    deviceInStore.setSubscribeCycleForCatalog(device.getSubscribeCycleForCatalog());
+                    addCatalogSubscribe(deviceInStore);
+                }
 
-        // 绉诲姩浣嶇疆璁㈤槄鐩稿叧鐨勪俊鎭�
-        if (device.getSubscribeCycleForMobilePosition() > 0) {
-            if (deviceInStore.getSubscribeCycleForMobilePosition() == 0 || deviceInStore.getSubscribeCycleForMobilePosition() != device.getSubscribeCycleForMobilePosition()) {
-                deviceInStore.setMobilePositionSubmissionInterval(device.getMobilePositionSubmissionInterval());
-                deviceInStore.setSubscribeCycleForMobilePosition(device.getSubscribeCycleForMobilePosition());
-                // 寮�鍚闃�
-                addMobilePositionSubscribe(deviceInStore);
-            }
-        }else if (device.getSubscribeCycleForMobilePosition() == 0) {
-            if (deviceInStore.getSubscribeCycleForMobilePosition() != 0) {
+            }else if (device.getSubscribeCycleForCatalog() == 0) {
                 // 鍙栨秷璁㈤槄
-                removeMobilePositionSubscribe(deviceInStore);
+                deviceInStore.setSubscribeCycleForCatalog(0);
+                removeCatalogSubscribe(deviceInStore, null);
             }
         }
-        // 鍧愭爣绯诲彉鍖栵紝闇�瑕侀噸鏂拌绠桮CJ02鍧愭爣鍜學GS84鍧愭爣
-        if (!deviceInStore.getGeoCoordSys().equals(device.getGeoCoordSys())) {
-            updateDeviceChannelGeoCoordSys(device);
+        // 绉诲姩浣嶇疆璁㈤槄鐩稿叧鐨勪俊鎭�
+        if (deviceInStore.getSubscribeCycleForMobilePosition() != device.getSubscribeCycleForMobilePosition()) {
+            if (device.getSubscribeCycleForMobilePosition() > 0) {
+                // 鑻ュ凡寮�鍚闃咃紝浣嗚闃呭懆鏈熶笉鍚岋紝鍒欏厛鍙栨秷
+                if (deviceInStore.getSubscribeCycleForMobilePosition() != 0) {
+                    removeMobilePositionSubscribe(deviceInStore, result->{
+                        // 寮�鍚闃�
+                        deviceInStore.setSubscribeCycleForMobilePosition(device.getSubscribeCycleForMobilePosition());
+                        addMobilePositionSubscribe(deviceInStore);
+                        // 鍥犱负鏄紓姝ユ墽琛岋紝闇�瑕佸湪杩欓噷鏇存柊涓嬫暟鎹�
+                        deviceMapper.updateCustom(deviceInStore);
+                        redisCatchStorage.updateDevice(deviceInStore);
+                    });
+                }else {
+                    // 寮�鍚闃�
+                    deviceInStore.setSubscribeCycleForMobilePosition(device.getSubscribeCycleForMobilePosition());
+                    addMobilePositionSubscribe(deviceInStore);
+                }
+
+            }else if (device.getSubscribeCycleForMobilePosition() == 0) {
+                // 鍙栨秷璁㈤槄
+                deviceInStore.setSubscribeCycleForMobilePosition(0);
+                removeMobilePositionSubscribe(deviceInStore, null);
+            }
         }
-        // 鏇存柊redis
-        redisCatchStorage.updateDevice(device);
-        deviceMapper.updateCustom(device);
+        if (deviceInStore.getGeoCoordSys() != null) {
+            // 鍧愭爣绯诲彉鍖栵紝闇�瑕侀噸鏂拌绠桮CJ02鍧愭爣鍜學GS84鍧愭爣
+            if (!deviceInStore.getGeoCoordSys().equals(device.getGeoCoordSys())) {
+                deviceInStore.setGeoCoordSys(device.getGeoCoordSys());
+                updateDeviceChannelGeoCoordSys(deviceInStore);
+            }
+        }else {
+            deviceInStore.setGeoCoordSys("WGS84");
+        }
+        if (device.getCharset() == null) {
+            deviceInStore.setCharset("GB2312");
+        }
+        //SSRC鏍¢獙
+        deviceInStore.setSsrcCheck(device.isSsrcCheck());
+        //浣滀负娑堟伅閫氶亾
+        deviceInStore.setAsMessageChannel(device.isAsMessageChannel());
+
+        deviceMapper.updateCustom(deviceInStore);
+        redisCatchStorage.updateDevice(deviceInStore);
     }
 
     @Override
@@ -526,6 +619,9 @@
         }catch (Exception e) {
             dataSourceTransactionManager.rollback(transactionStatus);
         }
+        if (result) {
+            redisCatchStorage.removeDevice(deviceId);
+        }
         return result;
     }
 

--
Gitblit v1.8.0