feat: 新增事件订阅

guzijian
1 parent d81f3f3a
Showing 6 changed files with 137 additions and 8 deletions
src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/AlarmSubscribeTask.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
src/main/java/com/genersoft/iot/vmp/service/IDeviceService.java
src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java
+package com.genersoft.iot.vmp.gb28181.task.impl;
+
+import com.genersoft.iot.vmp.common.CommonCallback;
+import com.genersoft.iot.vmp.conf.DynamicTask;
+import com.genersoft.iot.vmp.gb28181.bean.Device;
+import com.genersoft.iot.vmp.gb28181.task.ISubscribeTask;
+import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
+import gov.nist.javax.sip.message.SIPRequest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.sip.InvalidArgumentException;
+import javax.sip.SipException;
+import java.text.ParseException;
+
+/**
+ * @author 20412
+ */
+public class AlarmSubscribeTask implements ISubscribeTask {
+    private final Logger logger = LoggerFactory.getLogger(MobilePositionSubscribeTask.class);
+    private Device device;
+    private ISIPCommander sipCommander;
+
+    private SIPRequest request;
+    private DynamicTask dynamicTask;
+    private String taskKey = "alarm-subscribe-timeout";
+
+    public AlarmSubscribeTask(Device device, ISIPCommander sipCommander, DynamicTask dynamicTask) {
+        this.device = device;
+        this.sipCommander = sipCommander;
+        this.dynamicTask = dynamicTask;
+    }
+
+    @Override
+    public void run() {
+        if (dynamicTask.get(taskKey) != null) {
+            dynamicTask.stop(taskKey);
+        }
+        SIPRequest sipRequest = null;
+        try {
+            // 事件订阅 相当一部分设备不支持事件订阅 如海康。。。 这一部分通过布防处理 因为有部分设备不支持布防，如华为
+            sipRequest = sipCommander.alarmSubscribe(device,1000,"0","9","0",null,null);
+            logger.info("[事件订阅]成功： {}", device.getDeviceId());
+        } catch (InvalidArgumentException | SipException | ParseException e) {
+            logger.error("[命令发送失败] 事件订阅: {}", e.getMessage());
+        }
+        if (sipRequest != null) {
+            this.request = sipRequest;
+        }
+
+    }
+
+    @Override
+    public void stop(CommonCallback<Boolean> callback) {
+        /**
+         * dialog 的各个状态
+         * EARLY-> Early state状态-初始请求发送以后，收到了一个临时响应消息
+         * CONFIRMED-> Confirmed Dialog状态-已确认
+         * COMPLETED-> Completed Dialog状态-已完成
+         * TERMINATED-> Terminated Dialog状态-终止
+         */
+        if (dynamicTask.get(taskKey) != null) {
+            dynamicTask.stop(taskKey);
+        }
+        try {
+            // 当expires为0时取消订阅
+            sipCommander.alarmSubscribe(device,0,"0","9","0",null,null);
+            logger.info("[事件订阅]取消成功： {}", device.getDeviceId());
+        } catch (InvalidArgumentException | SipException | ParseException e) {
+            logger.error("[命令发送失败] 事件订阅取消失败: {}", e.getMessage());
+        }
+        device.setSubscribeCycleForAlarm(0);
+    }
+}
 \ No newline at end of file
@@ -336,7 +336,7 @@ public interface ISIPCommander {
 	 * @param endTime		报警发生终止时间（可选）
 	 * @return				true = 命令发送成功
 	 */
-	void alarmSubscribe(Device device, int expires, String startPriority, String endPriority, String alarmMethod, String startTime, String endTime) throws InvalidArgumentException, SipException, ParseException;
+	SIPRequest alarmSubscribe(Device device, int expires, String startPriority, String endPriority, String alarmMethod, String startTime, String endTime) throws InvalidArgumentException, SipException, ParseException;
 	/**
 	 * 订阅、取消订阅目录信息
@@ -1279,7 +1279,7 @@ public class SIPCommander implements ISIPCommander {
      * @return true = 命令发送成功
      */
     @Override
-    public void alarmSubscribe(Device device, int expires, String startPriority, String endPriority, String alarmMethod, String startTime, String endTime) throws InvalidArgumentException, SipException, ParseException {
+    public SIPRequest alarmSubscribe(Device device, int expires, String startPriority, String endPriority, String alarmMethod, String startTime, String endTime) throws InvalidArgumentException, SipException, ParseException {
         StringBuffer cmdXml = new StringBuffer(200);
         String charset = device.getCharset();
@@ -1307,9 +1307,9 @@ public class SIPCommander implements ISIPCommander {
-        Request request = headerProvider.createSubscribeRequest(device, cmdXml.toString(), null, expires, "presence",sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport()));
+        SIPRequest request = (SIPRequest)headerProvider.createSubscribeRequest(device, cmdXml.toString(), null, expires, "presence",sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport()));
         sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), request);
-
+        return request;
     }
     @Override
@@ -222,16 +222,16 @@ public class ZLMHttpHookListener {
                     logger.info("推流鉴权失败： 缺少必要参数：sign=md5(user表的pushKey)");
                     return new HookResultForOnPublish(401, "Unauthorized");
                 }
+                // 鉴权配置
                 // 推流自定义播放鉴权码
-                String callId = paramMap.get("callId");
                 // 鉴权配置
-                boolean hasAuthority = userService.checkPushAuthority(callId, sign);
+                boolean hasAuthority = userService.checkPushAuthority(null, sign);
                 if (!hasAuthority) {
-                    logger.info("推流鉴权失败： sign 无权限: callId={}. sign={}", callId, sign);
+                    logger.info("推流鉴权失败： sign 无权限: sign={}", sign);
                     return new HookResultForOnPublish(401, "Unauthorized");
                 }
                 StreamAuthorityInfo streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(param);
-                streamAuthorityInfo.setCallId(callId);
+                streamAuthorityInfo.setCallId(sign);
                 streamAuthorityInfo.setSign(sign);
                 // 鉴权通过
                 redisCatchStorage.updateStreamAuthorityInfo(param.getApp(), param.getStream(), streamAuthorityInfo);
@@ -169,4 +169,7 @@ public interface IDeviceService {
      * 获取所有设备
      */
     List<Device> getAll();
+
+    boolean addAlarmDirectorySubscribe(Device device);
+    boolean removeAlarmDirectorySubscription(Device device);
 }
@@ -9,6 +9,7 @@ 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.AlarmSubscribeTask;
 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;
@@ -532,6 +533,23 @@ public class DeviceServiceImpl implements IDeviceService {
         if (!ObjectUtils.isEmpty(device.getStreamMode())) {
             deviceInStore.setStreamMode(device.getStreamMode());
         }
+        // 事件订阅相关的信息
+        if (deviceInStore.getSubscribeCycleForAlarm() != device.getSubscribeCycleForAlarm()) {
+            if (device.getSubscribeCycleForAlarm() > 0) {
+                // 若已开启订阅，但订阅周期不同，则先取消
+                if (deviceInStore.getSubscribeCycleForAlarm() != 0) {
+                    removeAlarmDirectorySubscription(deviceInStore);
+                }
+                // 开启订阅
+                deviceInStore.setSubscribeCycleForAlarm(device.getSubscribeCycleForAlarm());
+                addAlarmDirectorySubscribe(deviceInStore);
+            } else if (device.getSubscribeCycleForAlarm() == 0) {
+                // 取消订阅
+                deviceInStore.setSubscribeCycleForAlarm(device.getSubscribeCycleForAlarm());
+                removeAlarmDirectorySubscription(deviceInStore);
+            }
+        }
+
         //  目录订阅相关的信息
         if (deviceInStore.getSubscribeCycleForCatalog() != device.getSubscribeCycleForCatalog()) {
             if (device.getSubscribeCycleForCatalog() > 0) {
@@ -637,5 +655,39 @@ public class DeviceServiceImpl implements IDeviceService {
         return deviceMapper.getAll();
     }
+    @Override
+    public boolean addAlarmDirectorySubscribe(Device device) {
+        if (device == null || device.getSubscribeCycleForAlarm() < 0) {
+            return false;
+        }
+        logger.info("[添加事件订阅] 设备{}", device.getDeviceId());
+        // 添加目录订阅
+        AlarmSubscribeTask alarmSubscribeTask = new AlarmSubscribeTask(device, sipCommander, dynamicTask);
+        // 刷新订阅
+        int subscribeCycleForAlarm = Math.max(device.getSubscribeCycleForAlarm(), 30);
+        // 设置最小值为30
+        dynamicTask.startCron(device.getDeviceId() + "alarm", alarmSubscribeTask, (subscribeCycleForAlarm - 1) * 1000);
+        return true;
+    }
+
+
+    @Override
+    public boolean removeAlarmDirectorySubscription(Device device) {
+        if (device == null || device.getSubscribeCycleForAlarm() < 0) {
+            return false;
+        }
+        logger.info("[移除事件订阅]: {}", device.getDeviceId());
+        String taskKey = device.getDeviceId() + "alarm";
+        if (device.isOnLine()) {
+            Runnable runnable = dynamicTask.get(taskKey);
+            if (runnable instanceof ISubscribeTask) {
+                ISubscribeTask subscribeTask = (ISubscribeTask) runnable;
+                subscribeTask.stop(null);
+            }
+        }
+        dynamicTask.stop(taskKey);
+        return true;
+    }
+
 }