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