优化目录订阅以及国标级联目录订阅回复

648540858

2022-03-10 c1d7f867c2ffcb1364334a5e013eb8f208819ef5

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java

@@ -5,8 +5,16 @@
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.SIPRequestHeaderPlarformProvider;
import com.genersoft.iot.vmp.gb28181.utils.DateUtil;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.utils.SerializeUtils;
import gov.nist.javax.sip.SipProviderImpl;
import gov.nist.javax.sip.SipStackImpl;
import gov.nist.javax.sip.message.SIPRequest;
import gov.nist.javax.sip.stack.SIPDialog;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -18,10 +26,12 @@
import org.springframework.util.StringUtils;

import javax.sip.*;
import javax.sip.header.CallIdHeader;
import javax.sip.header.WWWAuthenticateHeader;
import javax.sip.address.SipURI;
import javax.sip.header.*;
import javax.sip.message.Request;
import java.lang.reflect.Field;
import java.text.ParseException;
import java.util.HashSet;
import java.util.List;
import java.util.UUID;

@@ -38,41 +48,50 @@
    private IRedisCatchStorage redisCatchStorage;

    @Autowired
    private IMediaServerService mediaServerService;

    @Autowired
    private SipSubscribe sipSubscribe;

    @Autowired
    private ZLMRTPServerFactory zlmrtpServerFactory;

    @Lazy
    @Autowired
    @Qualifier(value="tcpSipProvider")
    private SipProvider tcpSipProvider;
    private SipProviderImpl tcpSipProvider;

    @Lazy
    @Autowired
    @Qualifier(value="udpSipProvider")
    private SipProvider udpSipProvider;
    private SipProviderImpl udpSipProvider;

    @Autowired
    private SipFactory sipFactory;

    @Override
    public boolean register(ParentPlatform parentPlatform, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent) {
        return register(parentPlatform, null, null, errorEvent, okEvent);
        return register(parentPlatform, null, null, errorEvent, okEvent, false);
    }

    @Override
    public boolean unregister(ParentPlatform parentPlatform, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent) {
        parentPlatform.setExpires("0");
        ParentPlatformCatch parentPlatformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId());
        if (parentPlatformCatch != null) {
            parentPlatformCatch.setParentPlatform(parentPlatform);
            redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
        }

        return register(parentPlatform, null, null, errorEvent, okEvent);
        parentPlatform.setExpires("0");
        return register(parentPlatform, null, null, errorEvent, okEvent, false);
    }

    @Override
    public boolean register(ParentPlatform parentPlatform, @Nullable String callId, @Nullable WWWAuthenticateHeader www, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent) {
    public boolean register(ParentPlatform parentPlatform, @Nullable String callId, @Nullable WWWAuthenticateHeader www,
                            SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent, boolean registerAgain) {
        try {
            Request request = null;
            Request request;
            String tm = Long.toString(System.currentTimeMillis());
            if (www == null ) {
            if (!registerAgain ) {
                //      //callid
                CallIdHeader callIdHeader = null;
                if(parentPlatform.getTransport().equals("TCP")) {
@@ -346,16 +365,18 @@
                    : udpSipProvider.getNewCallId();
            callIdHeader.setCallId(subscribeInfo.getCallId());

            String tm = Long.toString(System.currentTimeMillis());
//
            sendNotify(parentPlatform, deviceStatusXml.toString(), subscribeInfo, eventResult -> {
                logger.error("发送NOTIFY通知消息失败。错误：{} {}", eventResult.statusCode, eventResult.msg);
            }, null);

            Request request = headerProviderPlarformProvider.createNotifyRequest(parentPlatform,
                    deviceStatusXml.toString(),callIdHeader,
                    "z9hG4bK-" + UUID.randomUUID().toString().replace("-", ""),  subscribeInfo);
            transmitRequest(parentPlatform, request);

        } catch (SipException | ParseException | InvalidArgumentException e) {
        } catch (SipException | ParseException  e) {
            e.printStackTrace();
            return false;
        } catch (NoSuchFieldException e) {
            e.printStackTrace();
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        }
        return true;
    }
@@ -368,36 +389,88 @@
        if (index == null) {
            index = 0;
        }

        if (index >= deviceChannels.size()) {
            return true;
        }
        try {
            if (index > deviceChannels.size() - 1) {
                return true;
            }
            Request request = getCatalogNotifyRequestForCatalogAddOrUpdate(parentPlatform, deviceChannels.get(index), deviceChannels.size(), type, subscribeInfo);
            index += 1;
            Integer finalIndex = index;
            transmitRequest(parentPlatform, request, null, (eventResult -> {
                sendNotifyForCatalogAddOrUpdate(type, parentPlatform, deviceChannels, subscribeInfo, finalIndex);
            String catalogXmlContent = getCatalogXmlContentForCatalogAddOrUpdate(parentPlatform, deviceChannels.get(index ), deviceChannels.size(), type, subscribeInfo);
            sendNotify(parentPlatform, catalogXmlContent, subscribeInfo, eventResult -> {
                logger.error("发送NOTIFY通知消息失败。错误：{} {}", eventResult.statusCode, eventResult.msg);
            }, (eventResult -> {
                sendNotifyForCatalogAddOrUpdate(type, parentPlatform, deviceChannels, subscribeInfo, finalIndex + 1);
            }));
        } catch (SipException | ParseException | InvalidArgumentException e) {
        } catch (SipException | ParseException e) {
            e.printStackTrace();
            return false;
        } catch (NoSuchFieldException e) {
            e.printStackTrace();
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        }
        return true;
    }

    private Request getCatalogNotifyRequestForCatalogAddOrUpdate(ParentPlatform parentPlatform, DeviceChannel channel, int size, String type,
                                            SubscribeInfo subscribeInfo) throws ParseException, InvalidArgumentException,
            PeerUnavailableException {
        String catalogXmlContent = getCatalogXmlContentForCatalogAddOrUpdate(parentPlatform, channel, size, type, subscribeInfo);
    private void sendNotify(ParentPlatform parentPlatform, String catalogXmlContent,
                                   SubscribeInfo subscribeInfo, SipSubscribe.Event errorEvent,  SipSubscribe.Event okEvent )
            throws NoSuchFieldException, IllegalAccessException, SipException, ParseException {
        Dialog dialog  = subscribeInfo.getDialog();
        Request notifyRequest = dialog.createRequest(Request.NOTIFY);

        CallIdHeader callIdHeader = parentPlatform.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
                : udpSipProvider.getNewCallId();
        callIdHeader.setCallId(subscribeInfo.getCallId());
        Request request = headerProviderPlarformProvider.createNotifyRequest(parentPlatform, catalogXmlContent,
                callIdHeader, "z9hG4bK-" + UUID.randomUUID().toString().replace("-", ""), subscribeInfo);
        return request;
        ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("Application", "MANSCDP+xml");

        notifyRequest.setContent(catalogXmlContent, contentTypeHeader);

        SubscriptionStateHeader subscriptionState = sipFactory.createHeaderFactory()
                .createSubscriptionStateHeader(SubscriptionStateHeader.ACTIVE);
        notifyRequest.addHeader(subscriptionState);

        EventHeader event = sipFactory.createHeaderFactory().createEventHeader(subscribeInfo.getEventType());
        if (subscribeInfo.getEventId() != null) {
            event.setEventId(subscribeInfo.getEventId());
        }
        notifyRequest.addHeader(event);

        SipURI sipURI = (SipURI) notifyRequest.getRequestURI();
        SIPRequest request = (SIPRequest) subscribeInfo.getTransaction().getRequest();
        sipURI.setHost(request.getRemoteAddress().getHostName());
        sipURI.setPort(request.getRemotePort());
        ClientTransaction transaction = null;
        if ("TCP".equals(parentPlatform.getTransport())) {
            transaction = tcpSipProvider.getNewClientTransaction(notifyRequest);
        } else if ("UDP".equals(parentPlatform.getTransport())) {
            transaction = udpSipProvider.getNewClientTransaction(notifyRequest);
        }
        // 添加错误订阅
        if (errorEvent != null) {
            sipSubscribe.addErrorSubscribe(subscribeInfo.getCallId(), errorEvent);
        }
        // 添加订阅
        if (okEvent != null) {
            sipSubscribe.addOkSubscribe(subscribeInfo.getCallId(), okEvent);
        }
        if (transaction == null) {
            logger.error("平台{}的Transport错误：{}",parentPlatform.getServerGBId(), parentPlatform.getTransport());
            return;
        }
        dialog.sendRequest(transaction);

    }

//    private Request getCatalogNotifyRequestForCatalogAddOrUpdate(ParentPlatform parentPlatform, DeviceChannel channel, int size, String type,
//                                            SubscribeInfo subscribeInfo) throws ParseException, InvalidArgumentException,
//            PeerUnavailableException, NoSuchFieldException, IllegalAccessException {
//        String catalogXmlContent = getCatalogXmlContentForCatalogAddOrUpdate(parentPlatform, channel, size, type, subscribeInfo);
//
//        CallIdHeader callIdHeader = parentPlatform.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
//                : udpSipProvider.getNewCallId();
//        callIdHeader.setCallId(subscribeInfo.getCallId());
//        String tm = Long.toString(System.currentTimeMillis());
//
//        Request request = headerProviderPlarformProvider.createNotifyRequest(parentPlatform, catalogXmlContent,
//                callIdHeader, "z9hG4bK-" + UUID.randomUUID().toString().replace("-", ""),"FromRegister" + tm, subscribeInfo);
//        return request;
//    }

    private  String getCatalogXmlContentForCatalogAddOrUpdate(ParentPlatform parentPlatform, DeviceChannel channel, int sumNum, String type, SubscribeInfo subscribeInfo) {
        StringBuffer catalogXml = new StringBuffer(600);
@@ -447,34 +520,31 @@
        if (index == null) {
            index = 0;
        }

        if (index > deviceChannels.size() - 1) {
        if (index >= deviceChannels.size()) {
            return true;
        }
        try {
            String catalogXml = getCatalogXmlContentForCatalogOther(deviceChannels.get(index), type, parentPlatform);
            CallIdHeader callIdHeader = parentPlatform.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
                        : udpSipProvider.getNewCallId();
                Request request = headerProviderPlarformProvider.createNotifyRequest(parentPlatform, catalogXml,
                        callIdHeader,
                        "z9hG4bK-" + UUID.randomUUID().toString().replace("-", ""), subscribeInfo);
                index += 1;
            Integer finalIndex = index;
            transmitRequest(parentPlatform, request, null, eventResult -> {
                sendNotifyForCatalogOther(type, parentPlatform, deviceChannels, subscribeInfo, finalIndex);
            });
            String catalogXmlContent = getCatalogXmlContentForCatalogOther(parentPlatform, deviceChannels.get(index), type);
            sendNotify(parentPlatform, catalogXmlContent, subscribeInfo, eventResult -> {
                logger.error("发送NOTIFY通知消息失败。错误：{} {}", eventResult.statusCode, eventResult.msg);
            }, (eventResult -> {
                sendNotifyForCatalogOther(type, parentPlatform, deviceChannels, subscribeInfo, finalIndex + 1);
            }));
        } catch (SipException e) {
            e.printStackTrace();
        } catch (InvalidArgumentException e) {
            e.printStackTrace();
        } catch (ParseException e) {
            e.printStackTrace();
        } catch (NoSuchFieldException e) {
            e.printStackTrace();
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        }

        return true;
    }

    private String getCatalogXmlContentForCatalogOther(DeviceChannel channel, String type, ParentPlatform parentPlatform) {
    private String getCatalogXmlContentForCatalogOther(ParentPlatform parentPlatform, DeviceChannel channel, String type) {
        if (parentPlatform.getServerGBId().equals(channel.getParentId())) {
            channel.setParentId(parentPlatform.getDeviceGBId());
        }
@@ -542,4 +612,63 @@
        }
        return true;
    }

    @Override
    public void streamByeCmd(ParentPlatform platform, String callId) {
        if (platform == null) {
            return;
        }
        SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(platform.getServerGBId(), null, null, callId);
        if (sendRtpItem != null) {
            String mediaServerId = sendRtpItem.getMediaServerId();
            MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
            if (mediaServerItem != null) {
                mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpItem.getSsrc());
                zlmrtpServerFactory.closeRTPServer(mediaServerItem, sendRtpItem.getStreamId());
            }
            byte[] dialogByteArray = sendRtpItem.getDialog();
            if (dialogByteArray != null) {
                SIPDialog dialog = (SIPDialog) SerializeUtils.deSerialize(dialogByteArray);
                SipStack sipStack = udpSipProvider.getSipStack();
                SIPDialog sipDialog = ((SipStackImpl) sipStack).putDialog(dialog);
                if (dialog != sipDialog) {
                    dialog = sipDialog;
                } else {
                    try {
                        dialog.setSipProvider(udpSipProvider);
                        Field sipStackField = SIPDialog.class.getDeclaredField("sipStack");
                        sipStackField.setAccessible(true);
                        sipStackField.set(dialog, sipStack);
                        Field eventListenersField = SIPDialog.class.getDeclaredField("eventListeners");
                        eventListenersField.setAccessible(true);
                        eventListenersField.set(dialog, new HashSet<>());

                        byte[] transactionByteArray = sendRtpItem.getTransaction();
                        ClientTransaction clientTransaction = (ClientTransaction) SerializeUtils.deSerialize(transactionByteArray);
                        Request byeRequest = dialog.createRequest(Request.BYE);

                        SipURI byeURI = (SipURI) byeRequest.getRequestURI();
                        SIPRequest request = (SIPRequest) clientTransaction.getRequest();
                        byeURI.setHost(request.getRemoteAddress().getHostName());
                        byeURI.setPort(request.getRemotePort());
                        if ("TCP".equals(platform.getTransport())) {
                            clientTransaction = tcpSipProvider.getNewClientTransaction(byeRequest);
                        } else if ("UDP".equals(platform.getTransport())) {
                            clientTransaction = udpSipProvider.getNewClientTransaction(byeRequest);
                        }
                        dialog.sendRequest(clientTransaction);
                    } catch (SipException e) {
                        e.printStackTrace();
                    } catch (ParseException e) {
                        e.printStackTrace();
                    } catch (NoSuchFieldException e) {
                        e.printStackTrace();
                    } catch (IllegalAccessException e) {
                        e.printStackTrace();
                    }

                }
            }
        }
    }
}