From c084d6c98af1ef4d36a61adc719df5db76589428 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期六, 07 十月 2023 18:00:08 +0800
Subject: [PATCH] 优化国标级联心跳失败判断逻辑

---
 src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java |  376 ++++++++++++++++++++---------------------------------
 1 files changed, 141 insertions(+), 235 deletions(-)

diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java b/src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java
index 9c3e230..56663f1 100644
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java
@@ -1,235 +1,141 @@
-package com.genersoft.iot.vmp.gb28181;
-
-import java.text.ParseException;
-import java.util.Properties;
-import java.util.TooManyListenersException;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-
-import javax.sip.*;
-import javax.sip.header.CallIdHeader;
-import javax.sip.message.Response;
-
-import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.DependsOn;
-import org.springframework.stereotype.Component;
-
-import com.genersoft.iot.vmp.conf.SipConfig;
-import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorFactory;
-import com.genersoft.iot.vmp.gb28181.transmit.response.ISIPResponseProcessor;
-
-import gov.nist.javax.sip.SipStackImpl;
-
-@Component
-public class SipLayer implements SipListener {
-
-	private final static Logger logger = LoggerFactory.getLogger(SipLayer.class);
-
-	@Autowired
-	private SipConfig sipConfig;
-
-	@Autowired
-	private SIPProcessorFactory processorFactory;
-
-	@Autowired
-	private SipSubscribe sipSubscribe;
-
-	private SipStack sipStack;
-
-	private SipFactory sipFactory;
-
-	/**   
-	 * 娑堟伅澶勭悊鍣ㄧ嚎绋嬫睜
-	 */
-	private ThreadPoolExecutor processThreadPool;
-
-	@Bean("initSipServer")
-	private ThreadPoolExecutor initSipServer() {
-		
-		int processThreadNum = Runtime.getRuntime().availableProcessors() * 10;
-		LinkedBlockingQueue<Runnable> processQueue = new LinkedBlockingQueue<Runnable>(10000);
-		processThreadPool = new ThreadPoolExecutor(processThreadNum,processThreadNum,
-				0L,TimeUnit.MILLISECONDS,processQueue,
-				new ThreadPoolExecutor.CallerRunsPolicy());
-		return processThreadPool;
-	}
-	
-	@Bean("sipFactory")
-	@DependsOn("initSipServer")
-	private SipFactory createSipFactory() {
-		sipFactory = SipFactory.getInstance();
-		sipFactory.setPathName("gov.nist");
-		return sipFactory;
-	}
-	
-	@Bean("sipStack")
-	@DependsOn({"initSipServer", "sipFactory"})
-	private SipStack createSipStack() throws PeerUnavailableException {
-		Properties properties = new Properties();
-		properties.setProperty("javax.sip.STACK_NAME", "GB28181_SIP");
-		properties.setProperty("javax.sip.IP_ADDRESS", sipConfig.getMonitorIp());
-		properties.setProperty("gov.nist.javax.sip.LOG_MESSAGE_CONTENT", "false");
-		/**
-		 * sip_server_log.log 鍜� sip_debug_log.log public static final int TRACE_NONE =
-		 * 0; public static final int TRACE_MESSAGES = 16; public static final int
-		 * TRACE_EXCEPTION = 17; public static final int TRACE_DEBUG = 32;
-		 */
-		properties.setProperty("gov.nist.javax.sip.TRACE_LEVEL", "0");
-		properties.setProperty("gov.nist.javax.sip.SERVER_LOG", "sip_server_log");
-		properties.setProperty("gov.nist.javax.sip.DEBUG_LOG", "sip_debug_log");
-		sipStack = (SipStackImpl) sipFactory.createSipStack(properties);
-		return sipStack;
-	}
-
-	@Bean("tcpSipProvider")
-	@DependsOn("sipStack")
-	private SipProvider startTcpListener() {
-		ListeningPoint tcpListeningPoint = null;
-		SipProvider tcpSipProvider  = null;
-		try {
-			tcpListeningPoint = sipStack.createListeningPoint(sipConfig.getSipIp(), sipConfig.getSipPort(), "TCP");
-			tcpSipProvider = sipStack.createSipProvider(tcpListeningPoint);
-			tcpSipProvider.addSipListener(this);
-			logger.info("Sip Server TCP 鍚姩鎴愬姛 port {" + sipConfig.getSipPort() + "}");
-		} catch (TransportNotSupportedException | InvalidArgumentException | TooManyListenersException | ObjectInUseException e) {
-			logger.error(String.format("鍒涘缓SIP鏈嶅姟澶辫触: %s", e.getMessage()));
-		}
-		return tcpSipProvider;
-	}
-	
-	@Bean("udpSipProvider")
-	@DependsOn("sipStack")
-	private SipProvider startUdpListener() throws Exception {
-		ListeningPoint udpListeningPoint = sipStack.createListeningPoint(sipConfig.getSipIp(), sipConfig.getSipPort(), "UDP");
-		SipProvider udpSipProvider = sipStack.createSipProvider(udpListeningPoint);
-		udpSipProvider.addSipListener(this);
-		logger.info("Sip Server UDP 鍚姩鎴愬姛 port {" + sipConfig.getSipPort() + "}");
-		return udpSipProvider;
-	}
-
-	/**
-	 * SIP鏈嶅姟绔帴鏀舵秷鎭殑鏂规硶 Content 閲岄潰鏄疓BK缂栫爜 This method is called by the SIP stack when a
-	 * new request arrives.
-	 */
-	@Override
-	public void processRequest(RequestEvent evt) {
-//		logger.debug(evt.getRequest().toString());
-		// 鐢变簬jainsip鏄崟绾跨▼绋嬪簭锛屼负鎻愰珮鎬ц兘骞跺彂澶勭悊
-		processThreadPool.execute(() -> {
-			if (processorFactory != null) {
-				processorFactory.createRequestProcessor(evt).process();
-			}
-		});
-	}
-
-	@Override
-	public void processResponse(ResponseEvent evt) {
-		Response response = evt.getResponse();
-//		logger.debug(evt.getResponse().toString());
-		int status = response.getStatusCode();
-		if (((status >= 200) && (status < 300)) || status == 401) { // Success!
-			ISIPResponseProcessor processor = processorFactory.createResponseProcessor(evt);
-			try {
-				processor.process(evt, this, sipConfig);
-			} catch (ParseException e) {
-				// TODO Auto-generated catch block
-				e.printStackTrace();
-			}
-			if (evt.getResponse() != null && sipSubscribe.getOkSubscribesSize() > 0 ) {
-				CallIdHeader callIdHeader = (CallIdHeader)evt.getResponse().getHeader(CallIdHeader.NAME);
-				if (callIdHeader != null) {
-					SipSubscribe.Event subscribe = sipSubscribe.getOkSubscribe(callIdHeader.getCallId());
-					if (subscribe != null) {
-						subscribe.response(evt);
-					}
-				}
-			}
-		} else if ((status >= 100) && (status < 200)) {
-			// 澧炲姞鍏跺畠鏃犻渶鍥炲鐨勫搷搴旓紝濡�101銆�180绛�
-		} else {
-			logger.warn("鎺ユ敹鍒板け璐ョ殑response鍝嶅簲锛乻tatus锛�" + status + ",message:" + response.getReasonPhrase()/* .getContent().toString()*/);
-			if (evt.getResponse() != null && sipSubscribe.getErrorSubscribesSize() > 0 ) {
-				CallIdHeader callIdHeader = (CallIdHeader)evt.getResponse().getHeader(CallIdHeader.NAME);
-				if (callIdHeader != null) {
-					SipSubscribe.Event subscribe = sipSubscribe.getErrorSubscribe(callIdHeader.getCallId());
-					if (subscribe != null) {
-						subscribe.response(evt);
-					}
-				}
-			}
-		}
-
-
-
-	}
-
-	/**
-	 * <p>
-	 * Title: processTimeout
-	 * </p>
-	 * <p>
-	 * Description:
-	 * </p>
-	 * 
-	 * @param timeoutEvent
-	 */
-	@Override
-	public void processTimeout(TimeoutEvent timeoutEvent) {
-		// TODO Auto-generated method stub
-
-	}
-
-	/**
-	 * <p>
-	 * Title: processIOException
-	 * </p>
-	 * <p>
-	 * Description:
-	 * </p>
-	 * 
-	 * @param exceptionEvent
-	 */
-	@Override
-	public void processIOException(IOExceptionEvent exceptionEvent) {
-		// TODO Auto-generated method stub
-
-	}
-
-	/**
-	 * <p>
-	 * Title: processTransactionTerminated
-	 * </p>
-	 * <p>
-	 * Description:
-	 * </p>
-	 * 
-	 * @param transactionTerminatedEvent
-	 */
-	@Override
-	public void processTransactionTerminated(TransactionTerminatedEvent transactionTerminatedEvent) {
-		// TODO Auto-generated method stub
-
-	}
-
-	/**
-	 * <p>
-	 * Title: processDialogTerminated
-	 * </p>
-	 * <p>
-	 * Description:
-	 * </p>
-	 * 
-	 * @param dialogTerminatedEvent
-	 */
-	@Override
-	public void processDialogTerminated(DialogTerminatedEvent dialogTerminatedEvent) {
-		// TODO Auto-generated method stub
-
-	}
-
-}
+package com.genersoft.iot.vmp.gb28181;
+
+import com.genersoft.iot.vmp.conf.SipConfig;
+import com.genersoft.iot.vmp.conf.UserSetting;
+import com.genersoft.iot.vmp.gb28181.bean.GbStringMsgParserFactory;
+import com.genersoft.iot.vmp.gb28181.conf.DefaultProperties;
+import com.genersoft.iot.vmp.gb28181.transmit.ISIPProcessorObserver;
+import gov.nist.javax.sip.SipProviderImpl;
+import gov.nist.javax.sip.SipStackImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.core.annotation.Order;
+import org.springframework.stereotype.Component;
+import org.springframework.util.ObjectUtils;
+
+import javax.sip.*;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+
+@Component
+@Order(value=10)
+public class SipLayer implements CommandLineRunner {
+
+	private final static Logger logger = LoggerFactory.getLogger(SipLayer.class);
+
+	@Autowired
+	private SipConfig sipConfig;
+
+	@Autowired
+	private ISIPProcessorObserver sipProcessorObserver;
+
+	@Autowired
+	private UserSetting userSetting;
+
+	private final Map<String, SipProviderImpl> tcpSipProviderMap = new ConcurrentHashMap<>();
+	private final Map<String, SipProviderImpl> udpSipProviderMap = new ConcurrentHashMap<>();
+
+	@Override
+	public void run(String... args) {
+		List<String> monitorIps = new ArrayList<>();
+		// 浣跨敤閫楀彿鍒嗗壊澶氫釜ip
+		String separator = ",";
+		if (sipConfig.getIp().indexOf(separator) > 0) {
+			String[] split = sipConfig.getIp().split(separator);
+			monitorIps.addAll(Arrays.asList(split));
+		}else {
+			monitorIps.add(sipConfig.getIp());
+		}
+
+		SipFactory.getInstance().setPathName("gov.nist");
+		if (monitorIps.size() > 0) {
+			for (String monitorIp : monitorIps) {
+				addListeningPoint(monitorIp, sipConfig.getPort());
+			}
+			if (udpSipProviderMap.size() + tcpSipProviderMap.size() == 0) {
+				System.exit(1);
+			}
+		}
+	}
+
+	private void addListeningPoint(String monitorIp, int port){
+		SipStackImpl sipStack;
+		try {
+			sipStack = (SipStackImpl)SipFactory.getInstance().createSipStack(DefaultProperties.getProperties("GB28181_SIP", userSetting.getSipLog()));
+			sipStack.setMessageParserFactory(new GbStringMsgParserFactory());
+		} catch (PeerUnavailableException e) {
+			logger.error("[SIP SERVER] SIP鏈嶅姟鍚姩澶辫触锛� 鐩戝惉鍦板潃{}澶辫触,璇锋鏌p鏄惁姝g‘", monitorIp);
+			return;
+		}
+
+		try {
+			ListeningPoint tcpListeningPoint = sipStack.createListeningPoint(monitorIp, port, "TCP");
+			SipProviderImpl tcpSipProvider = (SipProviderImpl)sipStack.createSipProvider(tcpListeningPoint);
+
+			tcpSipProvider.setDialogErrorsAutomaticallyHandled();
+			tcpSipProvider.addSipListener(sipProcessorObserver);
+			tcpSipProviderMap.put(monitorIp, tcpSipProvider);
+			logger.info("[SIP SERVER] tcp://{}:{} 鍚姩鎴愬姛", monitorIp, port);
+		} catch (TransportNotSupportedException
+				 | TooManyListenersException
+				 | ObjectInUseException
+				 | InvalidArgumentException e) {
+			logger.error("[SIP SERVER] tcp://{}:{} SIP鏈嶅姟鍚姩澶辫触,璇锋鏌ョ鍙f槸鍚﹁鍗犵敤鎴栬�卛p鏄惁姝g‘"
+					, monitorIp, port);
+		}
+
+		try {
+			ListeningPoint udpListeningPoint = sipStack.createListeningPoint(monitorIp, port, "UDP");
+
+			SipProviderImpl udpSipProvider = (SipProviderImpl)sipStack.createSipProvider(udpListeningPoint);
+			udpSipProvider.addSipListener(sipProcessorObserver);
+
+			udpSipProviderMap.put(monitorIp, udpSipProvider);
+
+			logger.info("[SIP SERVER] udp://{}:{} 鍚姩鎴愬姛", monitorIp, port);
+		} catch (TransportNotSupportedException
+				 | TooManyListenersException
+				 | ObjectInUseException
+				 | InvalidArgumentException e) {
+			logger.error("[SIP SERVER] udp://{}:{} SIP鏈嶅姟鍚姩澶辫触,璇锋鏌ョ鍙f槸鍚﹁鍗犵敤鎴栬�卛p鏄惁姝g‘"
+					, monitorIp, port);
+		}
+	}
+
+	public SipProviderImpl getUdpSipProvider(String ip) {
+		if (ObjectUtils.isEmpty(ip)) {
+			return null;
+		}
+		return udpSipProviderMap.get(ip);
+	}
+
+	public SipProviderImpl getUdpSipProvider() {
+		if (udpSipProviderMap.size() != 1) {
+			return null;
+		}
+		return udpSipProviderMap.values().stream().findFirst().get();
+	}
+
+	public SipProviderImpl getTcpSipProvider() {
+		if (tcpSipProviderMap.size() != 1) {
+			return null;
+		}
+		return tcpSipProviderMap.values().stream().findFirst().get();
+	}
+
+	public SipProviderImpl getTcpSipProvider(String ip) {
+		if (ObjectUtils.isEmpty(ip)) {
+			return null;
+		}
+		return tcpSipProviderMap.get(ip);
+	}
+
+	public String getLocalIp(String deviceLocalIp) {
+		if (!ObjectUtils.isEmpty(deviceLocalIp)) {
+			return deviceLocalIp;
+		}
+		return getUdpSipProvider().getListeningPoint().getIPAddress();
+	}
+}

--
Gitblit v1.8.0