实现级联点播RTP推送

lawrencehj

2021-03-09 df34954c9eee1e870248368dd77a441509ac1dc2

实现级联点播RTP推送

 src/main/java/com/genersoft/iot/vmp/gb28181/bean/SendRtpItem.java

@@ -23,6 +23,11 @@
    private String platformId;

    /**
     * 对应设备id
     */
    private String deviceId;

   /**
     * 通道id
     */
    private String channelId;
@@ -92,6 +97,14 @@
        this.platformId = platformId;
    }

    public String getDeviceId() {
        return deviceId;
    }

    public void setDeviceId(String deviceId) {
        this.deviceId = deviceId;
    }

    public String getChannelId() {
        return channelId;
    }

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

@@ -4,25 +4,17 @@
import javax.sip.ResponseEvent;

import javax.sip.SipProvider;

import javax.sip.header.CSeqHeader;

import javax.sip.header.CallIdHeader;

import javax.sip.header.Header;

import javax.sip.message.Request;

import javax.sip.message.Response;



import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;

import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;

import com.genersoft.iot.vmp.storager.IRedisCatchStorage;

import com.alibaba.fastjson.JSON;

import com.genersoft.iot.vmp.gb28181.transmit.response.impl.*;

import com.genersoft.iot.vmp.gb28181.transmit.response.impl.*;

import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;

import com.genersoft.iot.vmp.vmanager.service.IPlayService;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

// import org.slf4j.Logger;

// import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.beans.factory.annotation.Qualifier;

import org.springframework.context.annotation.Lazy;

import org.springframework.stereotype.Component;



@@ -59,7 +51,7 @@
@Component

public class SIPProcessorFactory {

    

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

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

    

    @Autowired

    private SipConfig sipConfig;

@@ -156,6 +148,8 @@
        } else if (Request.ACK.equals(method)) {

            AckRequestProcessor processor = new AckRequestProcessor();

            processor.setRequestEvent(evt);

            processor.setRedisCatchStorage(redisCatchStorage);

            processor.setZlmrtpServerFactory(zlmrtpServerFactory);

            return processor;

        } else if (Request.BYE.equals(method)) {

            ByeRequestProcessor processor = new ByeRequestProcessor();


 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/AckRequestProcessor.java

@@ -1,18 +1,33 @@
package com.genersoft.iot.vmp.gb28181.transmit.request.impl;



import java.util.HashMap;

import java.util.Map;



import javax.sip.*;

import javax.sip.message.Request;

//import javax.sip.message.Request;



import com.genersoft.iot.vmp.common.StreamInfo;

import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;

import com.genersoft.iot.vmp.gb28181.transmit.request.SIPRequestAbstractProcessor;

import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;

import com.genersoft.iot.vmp.storager.IRedisCatchStorage;



import gov.nist.javax.sip.header.CSeq;

import org.springframework.stereotype.Component;



/**    

 * @Description:ACK请求处理器  

 * @author: swwheihei

 * @date:   2020年5月3日 下午5:31:45     

 */

@Component

public class AckRequestProcessor extends SIPRequestAbstractProcessor {



    //@Autowired

    private IRedisCatchStorage redisCatchStorage;



    //@Autowired

    private ZLMRTPServerFactory zlmrtpServerFactory;



    

    /**   

     * 处理  ACK请求

@@ -21,25 +36,77 @@
     */

    @Override

    public void process(RequestEvent evt) {

        Request request = evt.getRequest();

        //Request request = evt.getRequest();

        Dialog dialog = evt.getDialog();

        DialogState state = dialog.getState();

        if (dialog == null) return;

        if (request.getMethod().equals(Request.INVITE) && dialog.getState()== DialogState.CONFIRMED) {

            // TODO 查询并开始推流

        }

        //DialogState state = dialog.getState();

        if (/*request.getMethod().equals(Request.INVITE) &&*/ dialog.getState()== DialogState.CONFIRMED) {

            String remoteUri = dialog.getRemoteParty().getURI().toString();

            String localUri = dialog.getLocalParty().getURI().toString();

            String platformGbId = remoteUri.substring(remoteUri.indexOf(":") + 1, remoteUri.indexOf("@"));

            String channelId = localUri.substring(remoteUri.indexOf(":") + 1, remoteUri.indexOf("@"));

            SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(platformGbId, channelId);

            String is_Udp = sendRtpItem.isTcp() ? "0" : "1";

            String deviceId = sendRtpItem.getDeviceId();

            StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);

            System.out.println(platformGbId);

            System.out.println(channelId);

            Map<String, Object> param = new HashMap<>();

            param.put("vhost","__defaultVhost__");

            param.put("app","rtp");

            param.put("stream",streamInfo.getStreamId());

            param.put("ssrc", sendRtpItem.getSsrc());

            param.put("dst_url",sendRtpItem.getIp());

            param.put("dst_port", sendRtpItem.getPort());

            param.put("is_udp", is_Udp);

            //param.put ("src_port", sendRtpItem.getLocalPort());

            // 设备推流查询，成功后才能转推

            boolean rtpPushed = false;

            long startTime = System.currentTimeMillis();

            while (!rtpPushed) {

        try {

            Request ackRequest = null;

            CSeq csReq = (CSeq) request.getHeader(CSeq.NAME);

            ackRequest = dialog.createAck(csReq.getSeqNumber());

            dialog.sendAck(ackRequest);

            System.out.println("send ack to callee:" + ackRequest.toString());

        } catch (SipException e) {

                    if (System.currentTimeMillis() - startTime < 30 * 1000) {

                        if (zlmrtpServerFactory.isRtpReady(streamInfo.getStreamId())) {

                            rtpPushed = true;

                            zlmrtpServerFactory.startSendRtpStream(param);

                        } else {

                            Thread.sleep(2000);

                            continue;

                        }

                    }

                } catch (InterruptedException e) {

            e.printStackTrace();

        } catch (InvalidArgumentException e) {

            e.printStackTrace();

                }

            }

        }

        // try {

        //     Request ackRequest = null;

        //     CSeq csReq = (CSeq) request.getHeader(CSeq.NAME);

        //     ackRequest = dialog.createAck(csReq.getSeqNumber());

        //     dialog.sendAck(ackRequest);

        //     System.out.println("send ack to callee:" + ackRequest.toString());

        // } catch (SipException e) {

        //     e.printStackTrace();

        // } catch (InvalidArgumentException e) {

        //     e.printStackTrace();

        // }

		
        }

        

    public IRedisCatchStorage getRedisCatchStorage() {

        return redisCatchStorage;

    }



    public void setRedisCatchStorage(IRedisCatchStorage redisCatchStorage) {

        this.redisCatchStorage = redisCatchStorage;

    }



    public ZLMRTPServerFactory getZlmrtpServerFactory() {

        return zlmrtpServerFactory;

    }



    public void setZlmrtpServerFactory(ZLMRTPServerFactory zlmrtpServerFactory) {

        this.zlmrtpServerFactory = zlmrtpServerFactory;

    }



}


 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/InviteRequestProcessor.java

@@ -7,14 +7,10 @@
import javax.sip.SipFactory;

import javax.sip.address.Address;

import javax.sip.address.SipURI;

import javax.sip.header.ContentTypeHeader;

import javax.sip.header.FromHeader;

import javax.sip.header.HeaderFactory;

import javax.sip.header.SubjectHeader;

import javax.sip.header.*;

import javax.sip.message.Request;

import javax.sip.message.Response;



import com.alibaba.fastjson.JSONObject;

import com.genersoft.iot.vmp.conf.MediaServerConfig;

import com.genersoft.iot.vmp.gb28181.bean.Device;

import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;

@@ -27,17 +23,12 @@
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;

import com.genersoft.iot.vmp.vmanager.play.bean.PlayResult;

import com.genersoft.iot.vmp.vmanager.service.IPlayService;

import gov.nist.javax.sdp.fields.SDPFormat;

import gov.nist.javax.sip.address.AddressImpl;

import gov.nist.javax.sip.address.SipUri;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Autowired;



import java.io.IOException;

import java.text.ParseException;

import java.util.List;

import java.util.UUID;

import java.util.Vector;



/**    

@@ -45,6 +36,7 @@
 * @author: panll

 * @date:   2021年1月14日

 */

@SuppressWarnings("rawtypes")

public class InviteRequestProcessor extends SIPRequestAbstractProcessor {



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

@@ -122,8 +114,10 @@


            // jainSip不支持y=字段， 移除移除以解析。

            int ssrcIndex = contentString.indexOf("y=");

            String ssrc = contentString.substring(ssrcIndex + 2, contentString.length())

                    .replace("\r\n", "").replace("\n", "");

            String ssrc = contentString.substring(ssrcIndex + 2, ssrcIndex + 12);

            //ssrc规定长度为10字节，不取余下长度以避免后续还有“f=”字段

            // String ssrc = contentString.substring(ssrcIndex + 2, contentString.length())

            //         .replace("\r\n", "").replace("\n", "");



            String substring = contentString.substring(0, contentString.indexOf("y="));

            SessionDescription sdp = SdpFactory.getInstance().createSessionDescription(substring);

@@ -131,9 +125,9 @@
            //  获取支持的格式

            Vector mediaDescriptions = sdp.getMediaDescriptions(true);

            // 查看是否支持PS 负载96

            String ip = null;

            //String ip = null;

            int port = -1;

            boolean recvonly = false;

            //boolean recvonly = false;

            boolean mediaTransmissionTCP = false;

            Boolean tcpActive = null;

            for (int i = 0; i < mediaDescriptions.size(); i++) {

@@ -143,7 +137,7 @@
                Vector mediaFormats = media.getMediaFormats(false);

                if (mediaFormats.contains("96")) {

                    port = media.getMediaPort();

                    String mediaType = media.getMediaType();

                    //String mediaType = media.getMediaType();

                    String protocol = media.getProtocol();



                    // 区分TCP发流还是udp， 当前默认udp

@@ -169,7 +163,7 @@
            }

            String username = sdp.getOrigin().getUsername();

            String addressStr = sdp.getOrigin().getAddress();

            String sessionName = sdp.getSessionName().getValue();

            //String sessionName = sdp.getSessionName().getValue();

            logger.info("[上级点播]用户：{}， 地址：{}:{}， ssrc：{}", username, addressStr, port, ssrc);



            Device device = storager.queryVideoDeviceByPlatformIdAndChannelId(platformId, channelId);

@@ -178,7 +172,7 @@
                responseAck(evt, Response.SERVER_INTERNAL_ERROR);

                return;

            }

            SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(ip, port, platformId, ssrc, channelId,

            SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(addressStr, port, ssrc, platformId, device.getDeviceId(), channelId,

                    mediaTransmissionTCP);

            if (tcpActive != null) {

                sendRtpItem.setTcpActive(tcpActive);


 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/MessageRequestProcessor.java

@@ -370,9 +370,13 @@
    private void processMessageDeviceInfo(RequestEvent evt) {

        try {

            Element rootElement = getRootElement(evt);

            String requestName = rootElement.getName();

            Element deviceIdElement = rootElement.element("DeviceID");

            String deviceId = deviceIdElement.getTextTrim().toString();



            if (requestName.equals("Query")) {

                // 回复200 OK

                responseAck(evt);

            } else {

            Device device = storager.queryVideoDevice(deviceId);

            if (device == null) {

                return;

@@ -396,6 +400,7 @@
            if (offLineDetector.isOnline(deviceId)) {

                publisher.onlineEventPublish(deviceId, VideoManagerConstants.EVENT_ONLINE_KEEPLIVE);

            }

            }

        } catch (DocumentException | SipException | InvalidArgumentException | ParseException e) {

            e.printStackTrace();

        }


 src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java

@@ -119,4 +119,8 @@
    public JSONObject closeRtpServer(Map<String, Object> param) {
        return sendPost("closeRtpServer",param);
    }

    public JSONObject startSendRtp(Map<String, Object> param) {
        return sendPost("startSendRtp",param);
    }
}

 src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java

@@ -105,7 +105,7 @@
     * @param tcp 是否为tcp
     * @return SendRtpItem
     */
    public SendRtpItem createSendRtpItem(String ip, int port, String ssrc, String platformId, String channelId, boolean tcp){
    public SendRtpItem createSendRtpItem(String ip, int port, String ssrc, String platformId, String deviceId, String channelId, boolean tcp){
        String playSsrc = SsrcUtil.getPlaySsrc();
        int localPort = createRTPServer(SsrcUtil.getPlaySsrc());
        if (localPort != -1) {
@@ -119,9 +119,54 @@
        sendRtpItem.setPort(port);
        sendRtpItem.setSsrc(ssrc);
        sendRtpItem.setPlatformId(platformId);
        sendRtpItem.setDeviceId(deviceId);
        sendRtpItem.setChannelId(channelId);
        sendRtpItem.setTcp(tcp);
        sendRtpItem.setLocalPort(localPort);
        return sendRtpItem;
    }

    /**
     * 
     */
    public Boolean startSendRtpStream(Map<String, Object>param) {
        Boolean result = false;
        JSONObject jsonObject = zlmresTfulUtils.startSendRtp(param);
        System.out.println(jsonObject);
        if (jsonObject != null) {
            switch (jsonObject.getInteger("code")){
                case 0:
                    result= true;
                    logger.error("RTP推流请求成功，本地推流端口：" + jsonObject.getString("local_port"));
                    break;
                // case -300: // id已经存在
                //     result = false;
                //     break;
                // case -400: // 端口占用
                //     result= false;
                //     break;
                default:
                    logger.error("RTP推流失败: " + jsonObject.getString("msg"));
                    break;
            }
        }else {
            //  检查ZLM状态
            logger.error("RTP推流失败: 请检查ZLM服务");
        }
        return result;
    }

    /**
     * 
     */
    public Boolean isRtpReady(String streamId) {
        JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo("rtp", "rtmp", streamId);
        if (mediaInfo.getInteger("code") == 0 && mediaInfo.getBoolean("online")) {
            logger.info("设备RTP推流成功");
            return true;
        } else {
            logger.info("设备RTP推流未完成");
            return false;
        }
    }
}

 src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java

@@ -4,7 +4,6 @@
import com.genersoft.iot.vmp.conf.MediaServerConfig;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch;
import com.genersoft.iot.vmp.gb28181.bean.PlatformRegister;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;

import java.util.Map;
@@ -81,4 +80,13 @@
    void delPlatformRegisterInfo(String callId);

    void updateSendRTPSever(SendRtpItem sendRtpItem);

    /**
     * 查询RTP推送信息缓存
     * @param platformGbId
     * @param channelId
     * @return sendRtpItem
     */
    SendRtpItem querySendRTPServer(String platformGbId, String channelId);

}

 src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java

@@ -218,4 +218,11 @@
        String key = VideoManagerConstants.PLATFORM_SEND_RTP_INFO_PREFIX + sendRtpItem.getPlatformId() + "_" + sendRtpItem.getChannelId();
        redis.set(key, sendRtpItem);
    }

    @Override
    public SendRtpItem querySendRTPServer(String platformGbId, String channelId) {
        String key = VideoManagerConstants.PLATFORM_SEND_RTP_INFO_PREFIX + platformGbId + "_" + channelId;
        return (SendRtpItem)redis.get(key);
    }

}