648540858
2022-02-24 a42dda2bd3cc1cf8c20cc61e7ad9211eadecbaf3
src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java
@@ -1,17 +1,9 @@
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 com.genersoft.iot.vmp.conf.SipConfig;
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;
@@ -19,14 +11,15 @@
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;
import javax.sip.*;
import java.util.Properties;
import java.util.TooManyListenersException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@Component
public class SipLayer implements SipListener {
public class SipLayer{
   private final static Logger logger = LoggerFactory.getLogger(SipLayer.class);
@@ -34,33 +27,14 @@
   private SipConfig sipConfig;
   @Autowired
   private SIPProcessorFactory processorFactory;
   private ISIPProcessorObserver sipProcessorObserver;
   @Autowired
   private SipSubscribe sipSubscribe;
   private SipStack sipStack;
   private SipStackImpl 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");
@@ -68,7 +42,7 @@
   }
   
   @Bean("sipStack")
   @DependsOn({"initSipServer", "sipFactory"})
   @DependsOn({"sipFactory"})
   private SipStack createSipStack() throws PeerUnavailableException {
      Properties properties = new Properties();
      properties.setProperty("javax.sip.STACK_NAME", "GB28181_SIP");
@@ -86,150 +60,51 @@
      return sipStack;
   }
   @Bean("tcpSipProvider")
   @Bean(name = "tcpSipProvider")
   @DependsOn("sipStack")
   private SipProvider startTcpListener() {
   private SipProviderImpl startTcpListener() {
      ListeningPoint tcpListeningPoint = null;
      SipProvider tcpSipProvider  = null;
      SipProviderImpl 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()));
         tcpListeningPoint = sipStack.createListeningPoint(sipConfig.getMonitorIp(), sipConfig.getPort(), "TCP");
         tcpSipProvider = (SipProviderImpl)sipStack.createSipProvider(tcpListeningPoint);
         tcpSipProvider.setDialogErrorsAutomaticallyHandled();
         tcpSipProvider.addSipListener(sipProcessorObserver);
         logger.info("Sip Server TCP 启动成功 port {" + sipConfig.getMonitorIp() + ":" + sipConfig.getPort() + "}");
      } catch (TransportNotSupportedException e) {
         e.printStackTrace();
      } catch (InvalidArgumentException e) {
         logger.error("无法使用 [ {}:{} ]作为SIP[ TCP ]服务,可排查: 1. sip.monitor-ip 是否为本机网卡IP; 2. sip.port 是否已被占用"
               , sipConfig.getMonitorIp(), sipConfig.getPort());
      } catch (TooManyListenersException e) {
         e.printStackTrace();
      } catch (ObjectInUseException e) {
         e.printStackTrace();
      }
      return tcpSipProvider;
   }
   
   @Bean("udpSipProvider")
   @Bean(name = "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 里面是GBK编码 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响应!status:" + 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);
               }
            }
         }
   private SipProviderImpl startUdpListener() {
      ListeningPoint udpListeningPoint = null;
      SipProviderImpl udpSipProvider = null;
      try {
         udpListeningPoint = sipStack.createListeningPoint(sipConfig.getMonitorIp(), sipConfig.getPort(), "UDP");
         udpSipProvider = (SipProviderImpl)sipStack.createSipProvider(udpListeningPoint);
         udpSipProvider.addSipListener(sipProcessorObserver);
      } catch (TransportNotSupportedException e) {
         e.printStackTrace();
      } catch (InvalidArgumentException e) {
         logger.error("无法使用 [ {}:{} ]作为SIP[ UDP ]服务,可排查: 1. sip.monitor-ip 是否为本机网卡IP; 2. sip.port 是否已被占用"
               , sipConfig.getMonitorIp(), sipConfig.getPort());
      } catch (TooManyListenersException e) {
         e.printStackTrace();
      } catch (ObjectInUseException e) {
         e.printStackTrace();
      }
   }
   /**
    * <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
      logger.info("Sip Server UDP 启动成功 port [" + sipConfig.getMonitorIp() + ":" + sipConfig.getPort() + "]");
      return udpSipProvider;
   }
}