国标GB28181视频流对接和媒体服务器实现

ゞ浴缸里的玫瑰 2024-03-27 19:00 15阅读 0赞

视频流对接采用pjproject相关库实现sip服务器,采用pjmedia实现视频流接收。 媒体服务器完全自研，媒体服务器实现了rtsp协议，jrtplib库实现RTP的发送，ACE库实现垮平台底层支撑。

sipsvc以及mediasvc以及管理系统采用MQTT实现进程通信。

sipsvc核心代码分享：

SIP初始化：

status = pjsip_endpt_create(&_cp->factory, pj_gethostname()->ptr, &_sip_endpt);
            if (status != PJ_SUCCESS) {
                ERROR_LOG(THIS_FILE, "pjsip_endpt_create failed", status);
                return status;
            }
            /* Add UDP transport. */
            {
                pj_sockaddr_in addr;
                pjsip_host_port addrname;
                pjsip_transport *tp;
    
                pj_bzero(&addr, sizeof(addr));
                addr.sin_family = pj_AF_INET();
                addr.sin_port = pj_htons((pj_uint16_t)_sip_port);
    
                if (_localAddr.slen) {
    
                    addrname.host = _localAddr;
                    addrname.port = _sip_port;
    
                    status = pj_sockaddr_in_init(&addr, &_localAddr, (pj_uint16_t)_sip_port);
                    if (status != PJ_SUCCESS) {
                        ERROR_LOG(THIS_FILE, "Unable to resolve IP interface", status);
                        return status;
                    }
                }
    
                status = pjsip_udp_transport_start(_sip_endpt, &addr,(_localAddr.slen ? &addrname : NULL),1, &tp);
                if (status != PJ_SUCCESS) {
                    ERROR_LOG(THIS_FILE, "Unable to start UDP transport", status);
                    return status;
                }
    
                if (_localAddr.slen == 0) {
                    pj_strdup(_pool, &_localAddr, &tp->local_name.host);
                }
                
                char localAddr[32];
                char val[64];
                memset(localAddr, 0, sizeof(localAddr));
                memcpy(localAddr, _localAddr.ptr, _localAddr.slen);
                pj_ansi_sprintf(val, "<sip:%s@%s:%d>", _sipId.c_str(), localAddr, _sip_port);
                if (_localUri.slen == 0) {
                    _localUri = pj_strdup3(_pool, val);
                    pj_strdup(_pool, &_localContact, &_localUri);
                }
    
                SIP_LOG(SIP_INFO, (THIS_FILE, "SIP UDP listening on %.*s:%d",(int)tp->local_name.host.slen, tp->local_name.host.ptr,tp->local_name.port));
            }
            pjsip_tsx_layer_init_module(_sip_endpt);
            pjsip_ua_init_module(_sip_endpt, NULL);
            pj_str_t realm = pj_str((char*)"localhost");
            pjsip_auth_srv_init(_pool, &_auth_srv, &realm, pjsip_auth_lookup, 0);
            /* Init the callback for INVITE session: */
            pjsip_inv_callback inv_cb;
            pj_bzero(&inv_cb, sizeof(inv_cb));
            inv_cb.on_state_changed = &on_state_changed;
            inv_cb.on_new_session = &on_new_session;
            inv_cb.on_media_update = &on_media_update;
            inv_cb.on_rx_offer = &on_rx_offer;
            /* Initialize invite session module:  */
            status = pjsip_inv_usage_init(_sip_endpt, &inv_cb);
            if (status != PJ_SUCCESS) {
                ERROR_LOG(THIS_FILE, "pjsip_inv_usage_init failed", status);
                return status;
            }
            status = pjsip_100rel_init_module(_sip_endpt);
            if (status != PJ_SUCCESS) {
                ERROR_LOG(THIS_FILE, "pjsip_inv_usage_init failed", status);
                return status;
            }
            //register invite request module 
            status = pjsip_endpt_register_module(_sip_endpt, &app_module);
            if (status != PJ_SUCCESS) {
                ERROR_LOG(THIS_FILE, "register module failed", status);
                return status;
            }
            status = createWorkThread("sipecho", &worker_thread_proc, this, &_work_thread);
            if (status != PJ_SUCCESS) {
                ERROR_LOG(THIS_FILE, "create sip thread failed", status);
                return status;
            }

对于PS混合流编码 pjproject库默认不识别，需要自己注册ps编码函数：

PJ_DEF(pj_status_t) pjmedia_codec_ps_vid_init(pjmedia_vid_codec_mgr *mgr, pj_pool_factory *pf)
    {
        const pj_str_t ps_name = { (char*)"PS", 2 };
        pj_status_t status;
    
        if (ps_factory.pool != NULL) {
            /* Already initialized. */
            return PJ_SUCCESS;
        }
    
        if (!mgr) mgr = pjmedia_vid_codec_mgr_instance();
        PJ_ASSERT_RETURN(mgr, PJ_EINVAL);
    
        /* Create OpenH264 codec factory. */
        ps_factory.base.op = &ps_factory_op;
        ps_factory.base.factory_data = NULL;
        ps_factory.mgr = mgr;
        ps_factory.pf = pf;
        ps_factory.pool = pj_pool_create(pf, "psfactory", 256, 256, NULL);
        if (!ps_factory.pool)
            return PJ_ENOMEM;
    
    
        /* Register codec factory to codec manager. */
        status = pjmedia_vid_codec_mgr_register_factory(mgr, &ps_factory.base);
        if (status != PJ_SUCCESS)
            goto on_error;
    
        SIP_LOG(SIP_INFO, (THIS_FILE, "PS codec initialized"));
    
        /* Done. */
        return PJ_SUCCESS;
    
    on_error:
        pj_pool_release(ps_factory.pool);
        ps_factory.pool = NULL;
        return status;
    }

媒体服务器不仅要对混合流解码也要打包支持RTP over udp 或者 over TCP

分享实现RTSP的代码：

void RtspStream::open(ACE_HANDLE handle, ACE_Message_Block &message_block)
    {
        ACE_DEBUG((LM_DEBUG, "%s: open() called\n", THIS_FILE));
        handle_ = handle;
        _parser.set_parse_callback(&callback, this);   //add RTSP parser callback
        MQ::instance()->setCallbackObject(MEDIA_FILE_RESP, this);  //add MQ callback
        if (_writer.open(*this) == -1) {
            ACE_ERROR((LM_ERROR, ACE_TEXT("%p\n"), ACE_TEXT("ACE_Asynch_Write_Stream open")));
            return;
        }
        if (_reader.open(*this) == -1) {
            ACE_ERROR((LM_ERROR, ACE_TEXT("%p\n"), ACE_TEXT("ACE_Asynch_Read_Stream open")));
            return;
        }
        //if length is not 0 then fake a result and make handle_read_stream get it.
        if (message_block.length() != 0) {
            ACE_Message_Block &duplicate = *message_block.duplicate();
            ACE_Asynch_Read_Stream_Result_Impl *fake_result =
                ACE_Proactor::instance()->create_asynch_read_stream_result(this->proxy(),
                    this->handle_,
                    duplicate,
                    BUFSIZ,
                    0,
                    ACE_INVALID_HANDLE,
                    0,
                    0);
            size_t bytes_transferred = message_block.length();
            duplicate.wr_ptr(duplicate.wr_ptr() - bytes_transferred);
            // This will call the callback.
            fake_result->complete(message_block.length(), 1, 0);
            delete fake_result;
        }
        else {
            ready_read_stream();
        }
    }