对级联点播信令进行处理

panlinlin

2021-03-02 2f165d595af3565bd17612ee0f5866ace8ab43f4

对级联点播信令进行处理

 src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java

@@ -30,6 +30,8 @@


    public static final String PLATFORM_REGISTER_INFO_PREFIX = "VMP_platform_register_info_";



    public static final String PLATFORM_SEND_RTP_INFO_PREFIX = "VMP_platform_send_rtp_info_";



    public static final String Pattern_Topic = "VMP_keeplive_platform_";



    public static final String EVENT_ONLINE_REGISTER = "1";


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

New file
@@ -0,0 +1,150 @@
package com.genersoft.iot.vmp.gb28181.bean;

public class SendRtpItem {

    /**
     * 推流ip
     */
    private String ip;

    /**
     * 推流端口
     */
    private int port;

    /**
     * 推流标识
     */
    private String ssrc;

    /**
     * 平台id
     */
    private String platformId;

    /**
     * 通道id
     */
    private String channelId;

    /**
     * 推流状态
     * 0 等待设备推流上来
     * 1 等待上级平台回复ack
     * 2 推流中
     */
    private int status = 0;

    /**
     * 设备推流的app
     */
    private String app = "rtp";

    /**
     * 设备推流的streamId
     */
    private String streamId;

    /**
     * 是否为tcp
     */
    private boolean tcp;

    /**
     * 是否为tcp主动模式
     */
    private boolean tcpActive;

    /**
     * 自己推流使用的端口
     */
    private int localPort;

    public String getIp() {
        return ip;
    }

    public void setIp(String ip) {
        this.ip = ip;
    }

    public int getPort() {
        return port;
    }

    public void setPort(int port) {
        this.port = port;
    }

    public String getSsrc() {
        return ssrc;
    }

    public void setSsrc(String ssrc) {
        this.ssrc = ssrc;
    }

    public String getPlatformId() {
        return platformId;
    }

    public void setPlatformId(String platformId) {
        this.platformId = platformId;
    }

    public String getChannelId() {
        return channelId;
    }

    public void setChannelId(String channelId) {
        this.channelId = channelId;
    }

    public int getStatus() {
        return status;
    }

    public void setStatus(int status) {
        this.status = status;
    }

    public String getApp() {
        return app;
    }

    public void setApp(String app) {
        this.app = app;
    }

    public String getStreamId() {
        return streamId;
    }

    public void setStreamId(String streamId) {
        this.streamId = streamId;
    }

    public boolean isTcp() {
        return tcp;
    }

    public void setTcp(boolean tcp) {
        this.tcp = tcp;
    }

    public int getLocalPort() {
        return localPort;
    }

    public void setLocalPort(int localPort) {
        this.localPort = localPort;
    }

    public boolean isTcpActive() {
        return tcpActive;
    }

    public void setTcpActive(boolean tcpActive) {
        this.tcpActive = tcpActive;
    }
}

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

@@ -136,6 +136,7 @@
            processor.setCmderFroPlatform(cmderFroPlatform);

            processor.setPlayService(playService);

            processor.setStorager(storager);

            processor.setRedisCatchStorage(redisCatchStorage);

            processor.setZlmrtpServerFactory(zlmrtpServerFactory);

            return processor;

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


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

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



import javax.sip.Dialog;

import javax.sip.InvalidArgumentException;

import javax.sip.RequestEvent;

import javax.sip.SipException;

import javax.sip.*;

import javax.sip.message.Request;



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

@@ -26,7 +23,11 @@
    public void process(RequestEvent evt) {

        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 查询并开始推流

        }

        try {

            Request ackRequest = null;

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


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

@@ -4,9 +4,12 @@
import javax.sip.InvalidArgumentException;

import javax.sip.RequestEvent;

import javax.sip.SipException;

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.message.Request;

import javax.sip.message.Response;

@@ -15,6 +18,7 @@
import com.genersoft.iot.vmp.conf.MediaServerConfig;

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

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

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

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

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

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

@@ -100,16 +104,18 @@
            platformId = uri.getUser();



            if (platformId == null || channelId == null) {

                response400Ack(evt); // 参数不全， 发400，请求错误

                logger.info("无法从FromHeader的Address中获取到平台id，返回404");

                responseAck(evt, Response.BAD_REQUEST); // 参数不全， 发400，请求错误

                return;

            }

            // 查询平台下是否有该通道

            DeviceChannel channel = storager.queryChannelInParentPlatform(platformId, channelId);

            if (channel == null) {

                response404Ack(evt); // 通道不存在，发404，资源不存在

                logger.info("通道不存在，返回404");

                responseAck(evt, Response.NOT_FOUND); // 通道不存在，发404，资源不存在

                return;

            }else {

                response100Ack(evt); // 通道存在，发100，trying

                responseAck(evt, Response.TRYING); // 通道存在，发100，trying

            }

            // 解析sdp消息, 使用jainsip 自带的sdp解析方式

            String contentString = new String(request.getRawContent());

@@ -152,107 +158,79 @@
                            }

                        }

                    }

//                    Vector attributes = mediaDescription.getAttributes(false);

//                    for (Object attributeObj : attributes) {

//                        Attribute attribute = (Attribute)attributeObj;

//                        String name = attribute.getName();

//                        switch (name){

//                            case "recvonly":

//                                recvonly = true;

//                                break;

//                            case "rtpmap":

//                            case "connection":

//                                break;

//                            case "setup":

//                                mediaTransmissionTCP = true;

//                                if ("active".equals(attribute.getValue())) {  // tcp主动模式

//                                    tcpActive = true;

//                                }else if ("passive".equals(attribute.getValue())){ // tcp被动模式

//                                    tcpActive = false;

//                                }

//                                break;

//

//                        }

//                        if ("recvonly".equals(name)) {

//                            recvonly = true;

//                        }

//

//                        String value = attribute.getValue();

//                    }

                    break;

                }

            }

            if (port == -1) {

                logger.info("不支持的媒体格式，返回415");

                // 回复不支持的格式

                response415Ack(evt); // 不支持的格式，发415

                responseAck(evt, Response.UNSUPPORTED_MEDIA_TYPE); // 不支持的格式，发415

                return;

            }

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

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

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

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

//

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

//            if (device == null) {

//                logger.warn("点播平台{}的通道{}时未找到设备信息", platformId, channel);

//                response500Ack(evt);

//                return;

//            }

//

//            // 通知下级推流，

//            PlayResult playResult = playService.play(device.getDeviceId(), channelId, (responseJSON)->{

//                // 收到推流， 回复200OK

//                UUID uuid = UUID.randomUUID();

//                int rtpServer = zlmrtpServerFactory.createRTPServer(uuid.toString());

//                if (rtpServer == -1) {

//                    logger.error("为获取到可用端口");

//                    return;

//                }else {

//                    zlmrtpServerFactory.closeRTPServer(uuid.toString());

//                }

//                // TODO 添加对tcp的支持

//                MediaServerConfig mediaInfo = redisCatchStorage.getMediaInfo();

//                StringBuffer content = new StringBuffer(200);

//                content.append("v=0\r\n");

//                content.append("o="+"00000"+" 0 0 IN IP4 "+mediaInfo.getWanIp()+"\r\n");

//                content.append("s=Play\r\n");

//                content.append("c=IN IP4 "+mediaInfo.getWanIp()+"\r\n");

//                content.append("t=0 0\r\n");

//                content.append("m=video "+ rtpServer+" RTP/AVP 96\r\n");

//                content.append("a=sendonly\r\n");

//                content.append("a=rtpmap:96 PS/90000\r\n");

//                content.append("y="+ ssrc + "\r\n");

//                content.append("f=\r\n");

//

//                try {

//                    responseAck(evt, content.toString());

//                } catch (SipException e) {

//                    e.printStackTrace();

//                } catch (InvalidArgumentException e) {

//                    e.printStackTrace();

//                } catch (ParseException e) {

//                    e.printStackTrace();

//                }

//

//                // 写入redis， 超时时回复

////                redisCatchStorage.waiteAck()

//            },(event -> {

//                // 未知错误。直接转发设备点播的错误

//                Response response = null;

//                try {

//                    response = getMessageFactory().createResponse(event.getResponse().getStatusCode(), evt.getRequest());

//                    getServerTransaction(evt).sendResponse(response);

//

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

//                    e.printStackTrace();

//                }

//            }));

//            playResult.getResult();

            // 查找合适的端口推流，

            // 收到ack后调用推流接口



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

            if (device == null) {

                logger.warn("点播平台{}的通道{}时未找到设备信息", platformId, channel);

                responseAck(evt, Response.SERVER_INTERNAL_ERROR);

                return;

            }

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

                    mediaTransmissionTCP);

            if (tcpActive != null) {

                sendRtpItem.setTcpActive(tcpActive);

            }

            if (sendRtpItem == null) {

                logger.warn("服务器端口资源不足");

                responseAck(evt, Response.BUSY_HERE);

                return;

            }



            // 写入redis， 超时时回复

            redisCatchStorage.updateSendRTPSever(sendRtpItem);

            // 通知下级推流，

            PlayResult playResult = playService.play(device.getDeviceId(), channelId, (responseJSON)->{

                // 收到推流， 回复200OK, 等待ack

                sendRtpItem.setStatus(1);

                redisCatchStorage.updateSendRTPSever(sendRtpItem);

                // TODO 添加对tcp的支持

                MediaServerConfig mediaInfo = redisCatchStorage.getMediaInfo();

                StringBuffer content = new StringBuffer(200);

                content.append("v=0\r\n");

                content.append("o="+"00000"+" 0 0 IN IP4 "+mediaInfo.getWanIp()+"\r\n");

                content.append("s=Play\r\n");

                content.append("c=IN IP4 "+mediaInfo.getWanIp()+"\r\n");

                content.append("t=0 0\r\n");

                content.append("m=video "+ sendRtpItem.getLocalPort()+" RTP/AVP 96\r\n");

                content.append("a=sendonly\r\n");

                content.append("a=rtpmap:96 PS/90000\r\n");

                content.append("y="+ ssrc + "\r\n");

                content.append("f=\r\n");



                try {

                    responseAck(evt, content.toString());

                } catch (SipException e) {

                    e.printStackTrace();

                } catch (InvalidArgumentException e) {

                    e.printStackTrace();

                } catch (ParseException e) {

                    e.printStackTrace();

                }

            },(event -> {

                // 未知错误。直接转发设备点播的错误

                Response response = null;

                try {

                    response = getMessageFactory().createResponse(event.getResponse().getStatusCode(), evt.getRequest());

                    getServerTransaction(evt).sendResponse(response);



                } catch (ParseException | SipException | InvalidArgumentException e) {

                    e.printStackTrace();

                }

            }));

            playResult.getResult();



        } catch (SipException | InvalidArgumentException | ParseException e) {

            e.printStackTrace();

@@ -263,100 +241,46 @@
        } catch (SdpException e) {

            e.printStackTrace();

        }



    }



    /***

     * 回复100 trying

     * 回复状态码

     * 100 trying

     * 200 OK

     * 400

     * 404

     * @param evt

     * @throws SipException

     * @throws InvalidArgumentException

     * @throws ParseException

     */

    private void response100Ack(RequestEvent evt) throws SipException, InvalidArgumentException, ParseException {

    private void responseAck(RequestEvent evt, int statusCode) throws SipException, InvalidArgumentException, ParseException {

        Response response = getMessageFactory().createResponse(Response.TRYING, evt.getRequest());

        getServerTransaction(evt).sendResponse(response);

    }



    /***

     * 回复200 OK

    /**

     * 回复带sdp的200

     * @param evt

     * @param sdp

     * @throws SipException

     * @throws InvalidArgumentException

     * @throws ParseException

     */

    private void responseAck(RequestEvent evt, String sdp) throws SipException, InvalidArgumentException, ParseException {

        Response response = getMessageFactory().createResponse(Response.OK, evt.getRequest());

        ContentTypeHeader contentTypeHeader = getHeaderFactory().createContentTypeHeader("APPLICATION", "SDP");

        SipFactory sipFactory = SipFactory.getInstance();

        ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("APPLICATION", "SDP");

        response.setContent(sdp, contentTypeHeader);



        SipURI sipURI = (SipURI)evt.getRequest().getRequestURI();



        Address concatAddress = sipFactory.createAddressFactory().createAddress(

                sipFactory.createAddressFactory().createSipURI(sipURI.getUser(),  sipURI.getHost()+":"+sipURI.getPort()

                ));

        response.addHeader(sipFactory.createHeaderFactory().createContactHeader(concatAddress));

        getServerTransaction(evt).sendResponse(response);

    }



    /***

     * 回复400

     * @param evt

     * @throws SipException

     * @throws InvalidArgumentException

     * @throws ParseException

     */

    private void response400Ack(RequestEvent evt) throws SipException, InvalidArgumentException, ParseException {

        Response response = getMessageFactory().createResponse(Response.BAD_REQUEST, evt.getRequest());

        getServerTransaction(evt).sendResponse(response);

    }



    /***

     * 回复404

     * @param evt

     * @throws SipException

     * @throws InvalidArgumentException

     * @throws ParseException

     */

    private void response404Ack(RequestEvent evt) throws SipException, InvalidArgumentException, ParseException {

        Response response = getMessageFactory().createResponse(Response.NOT_FOUND, evt.getRequest());

        getServerTransaction(evt).sendResponse(response);

    }



    /***

     * 回复415 不支持的媒体类型

     * @param evt

     * @throws SipException

     * @throws InvalidArgumentException

     * @throws ParseException

     */

    private void response415Ack(RequestEvent evt) throws SipException, InvalidArgumentException, ParseException {

        Response response = getMessageFactory().createResponse(Response.UNSUPPORTED_MEDIA_TYPE, evt.getRequest());

        getServerTransaction(evt).sendResponse(response);

    }



    /***

     * 回复488

     * @param evt

     * @throws SipException

     * @throws InvalidArgumentException

     * @throws ParseException

     */

    private void response488Ack(RequestEvent evt) throws SipException, InvalidArgumentException, ParseException {

        Response response = getMessageFactory().createResponse(Response.NOT_ACCEPTABLE_HERE, evt.getRequest());

        getServerTransaction(evt).sendResponse(response);

    }



    /***

     * 回复500

     * @param evt

     * @throws SipException

     * @throws InvalidArgumentException

     * @throws ParseException

     */

    private void response500Ack(RequestEvent evt) throws SipException, InvalidArgumentException, ParseException {

        Response response = getMessageFactory().createResponse(Response.SERVER_INTERNAL_ERROR, evt.getRequest());

        getServerTransaction(evt).sendResponse(response);

    }



















@@ -394,4 +318,12 @@
    public void setPlayService(IPlayService playService) {

        this.playService = playService;

    }



    public IRedisCatchStorage getRedisCatchStorage() {

        return redisCatchStorage;

    }



    public void setRedisCatchStorage(IRedisCatchStorage redisCatchStorage) {

        this.redisCatchStorage = redisCatchStorage;

    }

}


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

@@ -1,6 +1,8 @@
package com.genersoft.iot.vmp.media.zlm;

import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.gb28181.session.SsrcUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -92,4 +94,34 @@
            return currentPort++;
        }
    }

    /**
     * 创建一个推流
     * @param ip 推流ip
     * @param port 推流端口
     * @param ssrc 推流唯一标识
     * @param platformId 平台id
     * @param channelId 通道id
     * @param tcp 是否为tcp
     * @return SendRtpItem
     */
    public SendRtpItem createSendRtpItem(String ip, int port, String ssrc, String platformId, String channelId, boolean tcp){
        String playSsrc = SsrcUtil.getPlaySsrc();
        int localPort = createRTPServer(SsrcUtil.getPlaySsrc());
        if (localPort != -1) {
            closeRTPServer(playSsrc);
        }else {
            logger.error("没有可用的端口");
            return null;
        }
        SendRtpItem sendRtpItem = new SendRtpItem();
        sendRtpItem.setIp(ip);
        sendRtpItem.setPort(port);
        sendRtpItem.setSsrc(ssrc);
        sendRtpItem.setPlatformId(platformId);
        sendRtpItem.setChannelId(channelId);
        sendRtpItem.setTcp(tcp);
        sendRtpItem.setLocalPort(localPort);
        return sendRtpItem;
    }
}

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

@@ -5,6 +5,7 @@
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;

@@ -78,4 +79,6 @@
    String queryPlatformRegisterInfo(String callId);

    void delPlatformRegisterInfo(String callId);

    void updateSendRTPSever(SendRtpItem sendRtpItem);
}

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

@@ -3,10 +3,7 @@
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.MediaServerConfig;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
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.*;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.dao.DeviceChannelMapper;
import com.genersoft.iot.vmp.utils.redis.RedisUtil;
@@ -215,4 +212,10 @@
    public void delPlatformRegisterInfo(String callId) {
        redis.del(VideoManagerConstants.PLATFORM_REGISTER_INFO_PREFIX + callId);
    }

    @Override
    public void updateSendRTPSever(SendRtpItem sendRtpItem) {
        String key = VideoManagerConstants.PLATFORM_SEND_RTP_INFO_PREFIX + sendRtpItem.getPlatformId() + "_" + sendRtpItem.getChannelId();
        redis.set(key, sendRtpItem);
    }
}