完善语音对讲级联

648540858

2023-03-21 82adc0cb23f3ee47322e78889cdaba57e9309000

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java

@@ -1,5 +1,6 @@
package com.genersoft.iot.vmp.gb28181.transmit;

import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor;
import com.genersoft.iot.vmp.gb28181.transmit.event.response.ISIPResponseProcessor;
@@ -12,10 +13,10 @@
import org.springframework.stereotype.Component;

import javax.sip.*;
import javax.sip.header.CSeqHeader;
import javax.sip.header.CallIdHeader;
import javax.sip.header.Header;
import javax.sip.header.*;
import javax.sip.message.Request;
import javax.sip.message.Response;
import java.net.InetAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@@ -29,16 +30,15 @@

    private final static Logger logger = LoggerFactory.getLogger(SIPProcessorObserver.class);

    private static Map<String, ISIPRequestProcessor> requestProcessorMap = new ConcurrentHashMap<>();
    private static Map<String,  ISIPRequestProcessor> requestProcessorMap = new ConcurrentHashMap<>();
    private static Map<String, ISIPResponseProcessor> responseProcessorMap = new ConcurrentHashMap<>();
    private static ITimeoutProcessor timeoutProcessor;

    @Autowired
    private SipSubscribe sipSubscribe;

//    @Autowired
//    @Qualifier(value = "taskExecutor")
//    private ThreadPoolTaskExecutor poolTaskExecutor;
    @Autowired
    private EventPublisher eventPublisher;

    /**
     * 添加 request订阅
@@ -63,7 +63,7 @@
     * @param processor 处理程序
     */
    public void addTimeoutProcessor(ITimeoutProcessor processor) {
        this.timeoutProcessor = processor;
        timeoutProcessor = processor;
    }

    /**
@@ -71,12 +71,13 @@
     * @param requestEvent RequestEvent事件
     */
    @Override
    @Async
    @Async("taskExecutor")
    public void processRequest(RequestEvent requestEvent) {
        String method = requestEvent.getRequest().getMethod();
        ISIPRequestProcessor sipRequestProcessor = requestProcessorMap.get(method);
        if (sipRequestProcessor == null) {
            logger.warn("不支持方法{}的request", method);
            // TODO 回复错误玛
            return;
        }
        requestProcessorMap.get(method).process(requestEvent);
@@ -88,34 +89,34 @@
     * @param responseEvent responseEvent事件
     */
    @Override
    @Async
    @Async("taskExecutor")
    public void processResponse(ResponseEvent responseEvent) {
        logger.debug(responseEvent.getResponse().toString());
        Response response = responseEvent.getResponse();
        logger.debug(responseEvent.getResponse().toString());
        int status = response.getStatusCode();
        if (((status >= 200) && (status < 300)) || status == 401) { // Success!

        // Success
        if (((status >= Response.OK) && (status < Response.MULTIPLE_CHOICES)) || status == Response.UNAUTHORIZED) {
            CSeqHeader cseqHeader = (CSeqHeader) responseEvent.getResponse().getHeader(CSeqHeader.NAME);
            String method = cseqHeader.getMethod();
            ISIPResponseProcessor sipRequestProcessor = responseProcessorMap.get(method);
            if (sipRequestProcessor != null) {
                sipRequestProcessor.process(responseEvent);
            }
            if (responseEvent.getResponse() != null && sipSubscribe.getOkSubscribesSize() > 0 ) {
            if (status != Response.UNAUTHORIZED && responseEvent.getResponse() != null && sipSubscribe.getOkSubscribesSize() > 0 ) {
                CallIdHeader callIdHeader = (CallIdHeader)responseEvent.getResponse().getHeader(CallIdHeader.NAME);
                if (callIdHeader != null) {
                    SipSubscribe.Event subscribe = sipSubscribe.getOkSubscribe(callIdHeader.getCallId());
                    if (subscribe != null) {
                        SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(responseEvent);
                        subscribe.response(eventResult);
                        sipSubscribe.removeOkSubscribe(callIdHeader.getCallId());
                        subscribe.response(eventResult);
                    }
                }
            }
        } else if ((status >= 100) && (status < 200)) {
        } else if ((status >= Response.TRYING) && (status < Response.OK)) {
            // 增加其它无需回复的响应，如101、180等
        } else {
            logger.warn("接收到失败的response响应！status：" + status + ",message:" + response.getReasonPhrase()/* .getContent().toString()*/);
            logger.warn("接收到失败的response响应！status：" + status + ",message:" + response.getReasonPhrase());
            if (responseEvent.getResponse() != null && sipSubscribe.getErrorSubscribesSize() > 0 ) {
                CallIdHeader callIdHeader = (CallIdHeader)responseEvent.getResponse().getHeader(CallIdHeader.NAME);
                if (callIdHeader != null) {
@@ -141,9 +142,28 @@
     */
    @Override
    public void processTimeout(TimeoutEvent timeoutEvent) {
        if(timeoutProcessor != null) {
            timeoutProcessor.process(timeoutEvent);
        logger.info("[消息发送超时]");
        ClientTransaction clientTransaction = timeoutEvent.getClientTransaction();

        if (clientTransaction != null) {
            logger.info("[发送错误订阅] clientTransaction != null");
            Request request = clientTransaction.getRequest();
            if (request != null) {
                logger.info("[发送错误订阅] request != null");
                CallIdHeader callIdHeader = (CallIdHeader) request.getHeader(CallIdHeader.NAME);
                if (callIdHeader != null) {
                    logger.info("[发送错误订阅]");
                    SipSubscribe.Event subscribe = sipSubscribe.getErrorSubscribe(callIdHeader.getCallId());
                    SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(timeoutEvent);
                    if (subscribe != null){
                        subscribe.response(eventResult);
                    }
                    sipSubscribe.removeOkSubscribe(callIdHeader.getCallId());
                    sipSubscribe.removeErrorSubscribe(callIdHeader.getCallId());
                }
            }
        }
        eventPublisher.requestTimeOut(timeoutEvent);
    }

    @Override
@@ -153,6 +173,12 @@

    @Override
    public void processTransactionTerminated(TransactionTerminatedEvent transactionTerminatedEvent) {
//        if (transactionTerminatedEvent.isServerTransaction()) {
//            ServerTransaction serverTransaction = transactionTerminatedEvent.getServerTransaction();
//            serverTransaction.get
//        }


//        Transaction transaction = null;
//        System.out.println("processTransactionTerminated");
//        if (transactionTerminatedEvent.isServerTransaction()) {