panlinlin
2021-05-08 e48fa711a3664bece9b3e58840a75fe7c05bc47c
src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java
@@ -2,24 +2,16 @@
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.DialogTerminatedEvent;
import javax.sip.IOExceptionEvent;
import javax.sip.ListeningPoint;
import javax.sip.PeerUnavailableException;
import javax.sip.RequestEvent;
import javax.sip.ResponseEvent;
import javax.sip.SipFactory;
import javax.sip.SipListener;
import javax.sip.SipProvider;
import javax.sip.SipStack;
import javax.sip.TimeoutEvent;
import javax.sip.TransactionTerminatedEvent;
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;
@@ -44,6 +36,9 @@
   @Autowired
   private SIPProcessorFactory processorFactory;
   @Autowired
   private SipSubscribe sipSubscribe;
   private SipStack sipStack;
   private SipFactory sipFactory;
@@ -54,14 +49,14 @@
   private ThreadPoolExecutor processThreadPool;
   @Bean("initSipServer")
   @DependsOn("allOffline")
   private void 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")
@@ -93,11 +88,17 @@
   @Bean("tcpSipProvider")
   @DependsOn("sipStack")
   private SipProvider startTcpListener() throws Exception {
      ListeningPoint tcpListeningPoint = sipStack.createListeningPoint(sipConfig.getSipIp(), sipConfig.getSipPort(), "TCP");
      SipProvider tcpSipProvider = sipStack.createSipProvider(tcpListeningPoint);
      tcpSipProvider.addSipListener(this);
      logger.info("Sip Server TCP 启动成功 port {" + sipConfig.getSipPort() + "}");
   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;
   }
   
@@ -107,7 +108,7 @@
      ListeningPoint udpListeningPoint = sipStack.createListeningPoint(sipConfig.getSipIp(), sipConfig.getSipPort(), "UDP");
      SipProvider udpSipProvider = sipStack.createSipProvider(udpListeningPoint);
      udpSipProvider.addSipListener(this);
      logger.info("Sip Server TCP 启动成功 port {" + sipConfig.getSipPort() + "}");
      logger.info("Sip Server UDP 启动成功 port {" + sipConfig.getSipPort() + "}");
      return udpSipProvider;
   }
@@ -117,17 +118,21 @@
    */
   @Override
   public void processRequest(RequestEvent evt) {
//      logger.debug(evt.getRequest().toString());
      // 由于jainsip是单线程程序,为提高性能并发处理
      processThreadPool.execute(() -> {
         processorFactory.createRequestProcessor(evt).process();
         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)) { // Success!
      if (((status >= 200) && (status < 300)) || status == 401) { // Success!
         ISIPResponseProcessor processor = processorFactory.createResponseProcessor(evt);
         try {
            processor.process(evt, this, sipConfig);
@@ -135,17 +140,32 @@
            // TODO Auto-generated catch block
            e.printStackTrace();
         }
      // } else if (status == Response.TRYING) {
         // trying不会回复
         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);
               }
            }
         }
      }
      // trying不会回复
      // if (status == Response.TRYING) {
      // }
   }
   /**