通用订阅器添加 compile 方法定义

gb28181-service/src/main/java/cn/skcks/docking/gb28181/core/sip/message/subscribe/GenericSubscribe.java

@@ -19,6 +19,8 @@ public interface GenericSubscribe<T> {
     void addSubscribe(String key,Flow.Subscriber<T> subscribe);
     void delPublisher(String key);
 
+    void compile(String key);
+
     class Helper {
         public final static String SEPARATOR = ":";
         public static String getKey(String prefix,String... ids){

gb28181-service/src/main/java/cn/skcks/docking/gb28181/core/sip/message/subscribe/InviteSubscribe.java

@@ -30,6 +30,10 @@ public class InviteSubscribe implements GenericSubscribe<SIPResponse> {
         Helper.addSubscribe(publishers, key, subscribe);
     }
 
+    public void compile(String key){
+        delPublisher(key);
+    }
+
     @Override
     public void delPublisher(String key) {
         Helper.delPublisher(publishers, key);

gb28181-service/src/main/java/cn/skcks/docking/gb28181/core/sip/message/subscribe/RecordInfoSubscribe.java

@@ -30,6 +30,10 @@ public class RecordInfoSubscribe implements GenericSubscribe<RecordInfoResponseD
         Helper.addSubscribe(publishers, key, subscribe);
     }
 
+    public void compile(String key){
+        delPublisher(key);
+    }
+
     @Override
     public void delPublisher(String key) {
         Helper.delPublisher(publishers, key);

gb28181-service/src/main/java/cn/skcks/docking/gb28181/core/sip/message/subscribe/SipRequestSubscribe.java

@@ -33,6 +33,10 @@ public class SipRequestSubscribe implements GenericTimeoutSubscribe<SIPRequest>,
         Helper.addSubscribe(publishers, key, subscribe);
     }
 
+    public void compile(String key){
+        delPublisher(key);
+    }
+
     @Override
     public void delPublisher(String key) {
         ScheduledFuture<?> schedule = scheduledFutureManager.remove(key);

gb28181-service/src/main/java/cn/skcks/docking/gb28181/core/sip/message/subscribe/SipResponseSubscribe.java

@@ -33,6 +33,10 @@ public class SipResponseSubscribe implements GenericTimeoutSubscribe<SIPResponse
         Helper.addSubscribe(publishers, key, subscribe);
     }
 
+    public void compile(String key){
+        delPublisher(key);
+    }
+
     @Override
     public void delPublisher(String key) {
         ScheduledFuture<?> schedule = scheduledFutureManager.remove(key);

gb28181-service/src/main/java/cn/skcks/docking/gb28181/service/catalog/CatalogService.java

@@ -85,7 +85,7 @@ public class CatalogService {
                 data.addAll(catalogResponseDTO.getDeviceList().getDeviceList());
                 if(curNum >= sumNum){
                     log.info("获取完成 {}", key);
-                    subscribe.getSipRequestSubscribe().delPublisher(key);
+                    subscribe.getSipRequestSubscribe().compile(key);
                 } else {
                     subscription.request(1);
                 }
@@ -100,6 +100,7 @@ public class CatalogService {
             @Override
             public void onComplete() {
                 log.info("{} 返回结果 {}", key, result.complete(data));
+                subscribe.getSipRequestSubscribe().delPublisher(key);
             }
         });
         provider.sendRequest(request);