优化级联注册，以及sip日志

648540858
1 parent 15169913
Showing 8 changed files with 77 additions and 146 deletions
src/main/java/com/genersoft/iot/vmp/gb28181/conf/DefaultProperties.java
src/main/java/com/genersoft/iot/vmp/gb28181/conf/SipLoggerPass.java
src/main/java/com/genersoft/iot/vmp/gb28181/utils/SipUtils.java
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java
src/main/java/com/genersoft/iot/vmp/service/impl/PlatformServiceImpl.java
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java
src/main/resources/all-application.yml
 package com.genersoft.iot.vmp.gb28181.conf;
  
+import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.notify.cmd.AlarmNotifyMessageHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.util.Properties;
  
 /**
@@ -12,6 +16,7 @@ public class DefaultProperties {
         Properties properties = new Properties();
         properties.setProperty("javax.sip.STACK_NAME", "GB28181_SIP");
         properties.setProperty("javax.sip.IP_ADDRESS", ip);
+        // 关闭自动会话
         properties.setProperty("javax.sip.AUTOMATIC_DIALOG_SUPPORT", "off");
         /**
          * 完整配置参考 gov.nist.javax.sip.SipStackImpl，需要下载源码
@@ -26,7 +31,7 @@ public class DefaultProperties {
         // 接收所有notify请求，即使没有订阅
         properties.setProperty("gov.nist.javax.sip.DELIVER_UNSOLICITED_NOTIFY", "true");
         properties.setProperty("gov.nist.javax.sip.AUTOMATIC_DIALOG_ERROR_HANDLING", "false");
-        properties.setProperty("gov.nist.javax.sip.CANCEL_CLIENT_TRANSACTION_CHECKED", "false");
+        properties.setProperty("gov.nist.javax.sip.CANCEL_CLIENT_TRANSACTION_CHECKED", "true");
         // 为_NULL _对话框传递_终止的_事件
         properties.setProperty("gov.nist.javax.sip.DELIVER_TERMINATED_EVENT_FOR_NULL_DIALOG", "true");
         // 会话清理策略
@@ -35,11 +40,33 @@ public class DefaultProperties {
         properties.setProperty("gov.nist.javax.sip.RELIABLE_CONNECTION_KEEP_ALIVE_TIMEOUT", "60");
         // 获取实际内容长度，不使用header中的长度信息
         properties.setProperty("gov.nist.javax.sip.COMPUTE_CONTENT_LENGTH_FROM_MESSAGE_BODY", "true");
+        // 线程可重入
+        properties.setProperty("gov.nist.javax.sip.REENTRANT_LISTENER", "true");
+        // 定义应用程序打算多久审计一次 SIP 堆栈，了解其内部线程的健康状况（该属性指定连续审计之间的时间（以毫秒为单位））
+        properties.setProperty("gov.nist.javax.sip.THREAD_AUDIT_INTERVAL_IN_MILLISECS", "30000");
  
         /**
          * sip_server_log.log 和 sip_debug_log.log ERROR, INFO, WARNING, OFF, DEBUG, TRACE
          */
-        properties.setProperty("gov.nist.javax.sip.TRACE_LEVEL", "ERROR");
+        Logger logger = LoggerFactory.getLogger(AlarmNotifyMessageHandler.class);
+        if (logger.isDebugEnabled()) {
+            System.out.println("DEBUG");
+            properties.setProperty("gov.nist.javax.sip.TRACE_LEVEL", "DEBUG");
+        }else if (logger.isInfoEnabled()) {
+            System.out.println("INFO1");
+            properties.setProperty("gov.nist.javax.sip.TRACE_LEVEL", "INFO");
+        }else if (logger.isWarnEnabled()) {
+            System.out.println("WARNING");
+            properties.setProperty("gov.nist.javax.sip.TRACE_LEVEL", "WARNING");
+        }else if (logger.isErrorEnabled()) {
+            System.out.println("ERROR");
+            properties.setProperty("gov.nist.javax.sip.TRACE_LEVEL", "ERROR");
+        }else {
+            System.out.println("INFO2");
+            properties.setProperty("gov.nist.javax.sip.TRACE_LEVEL", "INFO");
+        }
+        logger.info("[SIP日志]级别为: {}", properties.getProperty("gov.nist.javax.sip.TRACE_LEVEL"));
+
  
         return properties;
     }
-package com.genersoft.iot.vmp.gb28181.conf;
-
-import gov.nist.core.StackLogger;
-
-import java.util.Properties;
-
-/**
- * sip日志格式化
- * 暂不使用
- */
-public class SipLoggerPass implements StackLogger {
-
-    @Override
-    public void logStackTrace() {
-
-    }
-
-    @Override
-    public void logStackTrace(int traceLevel) {
-
-    }
-
-    @Override
-    public int getLineCount() {
-        return 0;
-    }
-
-    @Override
-    public void logException(Throwable ex) {
-
-    }
-
-    @Override
-    public void logDebug(String message) {
-
-    }
-
-    @Override
-    public void logDebug(String message, Exception ex) {
-
-    }
-
-    @Override
-    public void logTrace(String message) {
-
-    }
-
-    @Override
-    public void logFatalError(String message) {
-
-    }
-
-    @Override
-    public void logError(String message) {
-
-    }
-
-    @Override
-    public boolean isLoggingEnabled() {
-        return false;
-    }
-
-    @Override
-    public boolean isLoggingEnabled(int logLevel) {
-        return false;
-    }
-
-    @Override
-    public void logError(String message, Exception ex) {
-
-    }
-
-    @Override
-    public void logWarning(String string) {
-
-    }
-
-    @Override
-    public void logInfo(String string) {
-
-    }
-
-    @Override
-    public void disableLogging() {
-
-    }
-
-    @Override
-    public void enableLogging() {
-
-    }
-
-    @Override
-    public void setBuildTimeStamp(String buildTimeStamp) {
-
-    }
-
-    @Override
-    public void setStackProperties(Properties stackProperties) {
-
-    }
-
-    @Override
-    public String getLoggerName() {
-        return null;
-    }
-}
@@ -13,7 +13,6 @@ import javax.sip.SipFactory;
 import javax.sip.header.FromHeader;
 import javax.sip.header.Header;
 import javax.sip.header.UserAgentHeader;
-import javax.sip.header.ViaHeader;
 import javax.sip.message.Request;
 import java.text.ParseException;
 import java.util.ArrayList;
@@ -138,13 +137,12 @@ public class SipUtils {
         }else {
             // 判断RPort是否改变，改变则说明路由nat信息变化，修改设备信息
             // 获取到通信地址等信息
-            ViaHeader viaHeader = (ViaHeader) request.getHeader(ViaHeader.NAME);
-            remoteAddress = viaHeader.getReceived();
-            remotePort = viaHeader.getRPort();
+            remoteAddress = request.getTopmostViaHeader().getReceived();
+            remotePort = request.getTopmostViaHeader().getRPort();
             // 解析本地地址替代
             if (ObjectUtils.isEmpty(remoteAddress) || remotePort == -1) {
-                remoteAddress = viaHeader.getHost();
-                remotePort = viaHeader.getPort();
+                remoteAddress = request.getTopmostViaHeader().getHost();
+                remotePort = request.getTopmostViaHeader().getPort();
             }
         }
  
@@ -424,7 +424,7 @@ public class ZLMHttpHookListener {
 		logger.info("[ZLM HOOK]流无人观看：{]->{}->{}/{}" + param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());
 		JSONObject ret = new JSONObject();
 		ret.put("code", 0);
-		// 录像下载
+		// 国标类型的流
 		if ("rtp".equals(param.getApp())){
 			ret.put("close", userSetting.getStreamOnDemand());
 			// 国标流， 点播/录像回放/录像下载
@@ -596,7 +596,7 @@ public class ZLMHttpHookListener {
 	@PostMapping(value = "/on_send_rtp_stopped", produces = "application/json;charset=UTF-8")
 	public JSONObject onSendRtpStopped(HttpServletRequest request, @RequestBody OnSendRtpStoppedHookParam param){
  
-		logger.info("[ZLM HOOK] 发送rtp被动关闭：{}->{}/{}", param.getMediaServerId(), param.getApp(), param.getStream());
+		logger.info("[ZLM HOOK] rtp发送关闭：{}->{}/{}", param.getMediaServerId(), param.getApp(), param.getStream());
  
 		JSONObject ret = new JSONObject();
 		ret.put("code", 0);
@@ -154,7 +154,8 @@ public class DeviceServiceImpl implements IDeviceService {
         }
         // 刷新过期任务
         String registerExpireTaskKey = registerExpireTaskKeyPrefix + device.getDeviceId();
-        dynamicTask.startDelay(registerExpireTaskKey, ()-> offline(device.getDeviceId()), device.getExpires() * 1000);
+        // 增加一个10秒给设备重发消息的机会
+        dynamicTask.startDelay(registerExpireTaskKey, ()-> offline(device.getDeviceId()), (device.getExpires() + 10) * 1000);
     }
  
     @Override
@@ -135,14 +135,7 @@ public class PlatformServiceImpl implements IPlatformService {
             dynamicTask.startCron(registerTaskKey,
                 // 注册失败（注册成功时由程序直接调用了online方法）
                 ()-> {
-                    try {
-                        logger.info("[国标级联] 平台：{}注册即将到期，重新注册", parentPlatform.getServerGBId());
-                        commanderForPlatform.register(parentPlatform, eventResult -> {
-                            offline(parentPlatform, false);
-                        },null);
-                    } catch (InvalidArgumentException | ParseException | SipException e) {
-                        logger.error("[命令发送失败] 国标级联定时注册: {}", e.getMessage());
-                    }
+                    registerTask(parentPlatform);
                 },
                 (parentPlatform.getExpires() - 10) *1000);
         }
@@ -194,6 +187,28 @@ public class PlatformServiceImpl implements IPlatformService {
         }
     }
  
+    private void registerTask(ParentPlatform parentPlatform){
+        try {
+            // 设置超时重发， 后续从底层支持消息重发
+            String key = KEEPALIVE_KEY_PREFIX + parentPlatform.getServerGBId() + "_timeout";
+            if (dynamicTask.isAlive(key)) {
+                return;
+            }
+            dynamicTask.startDelay(key, ()->{
+                registerTask(parentPlatform);
+            }, 1000);
+            logger.info("[国标级联] 平台：{}注册即将到期，重新注册", parentPlatform.getServerGBId());
+            commanderForPlatform.register(parentPlatform, eventResult -> {
+                dynamicTask.stop(key);
+                offline(parentPlatform, false);
+            },eventResult -> {
+                dynamicTask.stop(key);
+            });
+        } catch (InvalidArgumentException | ParseException | SipException e) {
+            logger.error("[命令发送失败] 国标级联定时注册: {}", e.getMessage());
+        }
+    }
+
     @Override
     public void offline(ParentPlatform parentPlatform, boolean stopRegister) {
         logger.info("[平台离线]：{}", parentPlatform.getServerGBId());
@@ -106,24 +106,21 @@ public class PlayController {
 			msg.setData(wvpResult);
 			resultHolder.invokeResult(msg);
 		});
-
-
-			// TODO 在点播未成功的情况下在此调用接口点播会导致返回的流地址ip错误
-			deferredResultEx.setFilter(result1 -> {
-				WVPResult<StreamInfo> wvpResult1 = (WVPResult<StreamInfo>)result1;
-				WVPResult<StreamContent> resultStream = null;
-				if (wvpResult1.getCode() == ErrorCode.SUCCESS.getCode()) {
-					StreamInfo data = wvpResult1.getData().clone();
-					if (userSetting.getUseSourceIpAsStreamIp()) {
-						data.channgeStreamIp(request.getLocalName());
-					}
-					resultStream = new WVPResult<>();
-					resultStream.setCode(wvpResult1.getCode());
-					resultStream.setMsg(wvpResult1.getMsg());
-					resultStream.setData(new StreamContent(wvpResult1.getData()));
+		// TODO 在点播未成功的情况下在此调用接口点播会导致返回的流地址ip错误
+		deferredResultEx.setFilter(result1 -> {
+			WVPResult<StreamInfo> wvpResult1 = (WVPResult<StreamInfo>)result1;
+			WVPResult<StreamContent> resultStream = new WVPResult<>();
+			resultStream.setCode(wvpResult1.getCode());
+			resultStream.setMsg(wvpResult1.getMsg());
+			if (wvpResult1.getCode() == ErrorCode.SUCCESS.getCode()) {
+				StreamInfo data = wvpResult1.getData().clone();
+				if (userSetting.getUseSourceIpAsStreamIp()) {
+					data.channgeStreamIp(request.getLocalName());
 				}
-				return resultStream;
-			});
+				resultStream.setData(new StreamContent(wvpResult1.getData()));
+			}
+			return resultStream;
+		});
  
  
 		// 录像查询以channelId作为deviceId查询
@@ -186,7 +186,7 @@ user-settings:
     use-pushing-as-status: true
     # 使用来源请求ip作为streamIp,当且仅当你只有zlm节点它与wvp在一起的情况下开启
     use-source-ip-as-stream-ip: true
-    # 按需拉流, true：有人观看拉流，无人观看释放， false：拉起后不自动释放
+    # 国标点播 按需拉流, true：有人观看拉流，无人观看释放， false：拉起后不自动释放
     stream-on-demand: true
     # 推流鉴权， 默认开启
     push-authority: true
@@ -195,8 +195,8 @@ user-settings:
     gb-send-stream-strict: false
     # 设备上线时是否自动同步通道
     sync-channel-on-device-online: false
-    # 设备上线时是否自动同步通道
-    sip-use-source-ip-as-remote-address: true
+    # 是否使用设备来源Ip作为回复IP， 不设置则为 false
+    sip-use-source-ip-as-remote-address: false
  
 # 关闭在线文档（生产环境建议关闭）
 springdoc: