基于Janus房间服务器的POC对讲实现

瓜牛
2020-02-18 18:41:14
音视频
janus

一、理解Janus插件janus_videoroom的包转发实现

该插件模式具备SFU的包转发能力,每一个进入房间的与会者，可以通过Publish发布自己的音视频，也可以通过Subscribe订阅其他人的音视频，转发能力由libnice库提供，libnice可以修改为支持单端口转发RTP包；

二、理解POC对讲半双工模式下的包转发实现

POC，是PTT Over Cell 的简写，PTT是按下讲话的意思，要么说，要么听别人说，不会出现同时讲话&听的情况，所以类似半双工的通信方式；如果还沿用janus_videoroom的实现，如果有N个人参与的会议室，那么每个与会者都需要订阅其他N-1个人声音，会有N-1条下行信道，但每次只有一个信道可用，这对信道资源来说是明显的浪费，所以，我们希望改造成MCU模式，但又不需要做服务器端的混音操作，完成POC的业务对讲能力；

三、最后实现

1、给房间增加一个公共的publish对象，所有对房间的订阅都是订阅该publish对象，达到支持从SFU模式到支持MCU转发模式（适应POC对讲模式的MCU模式）

2、支持会场TBCP控制信息通过DataChannel通道传递

3、支持会场TBCP控制和会场通知；

部分逻辑：

给janus_videoroom结构体里面增加一个 janus_videoroom_publisher *room_publisher;
在创建房间成功后，既实例化这个对象room_publisher；
初始进入房间或者枚举房间的publisher的时候，只返回这个对象即可，不返回room结构体里面的participants列表中的用户id，这样每个subscriber都对应的是这个publisher对象；
然后在媒体包过来的时候，枚举room_publisher中的subscriber，逐个分发即可。

修改Janus服务器，支持datachannel能力：

Janus的datachannel支持的协议主要是：“DTLS/SCTP”、"UDP/DTLS/SCTP"，而RTP通道使用的是"UDP/TLS/RTP/SAVPF"，修改的思路包括SDP返回的修改，RTP数据包中提取datachannel包对应SSRC的包。

1、给janus_ice_stream结构体添加data的ssrc字段和初始化：

    guint32 data_ssrc_peer;
    guint32 data_ssrc, sequence_data;
 
    int janus_sdp_process(void *ice_handle, janus_sdp *remote_sdp, gboolean update) 方法中增加如下逻辑：
                    if(!strcasecmp(m->proto, "UDP/DTLS/SCTP")  || !strcasecmp(m->proto, "UDP/TLS/RTP/SAVPF")) {
                        stream->data_ssrc = janus_random_uint32();    /* FIXME Should we look for conflicts? */
                        stream->sequence_data = janus_random_uint32();    
                        janus_flags_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_NEW_DATACHAN_SDP);
                        janus_flags_clear(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_DATA_CHANNELS);//lyz@xdja.com add
                    } else {
                        janus_flags_clear(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_NEW_DATACHAN_SDP);
                    }

2、SDP返回数据：

m->fmts = g_list_append(m->fmts, g_strdup("109"));
            janus_sdp_attribute *aa = janus_sdp_attribute_create("rtpmap", "109 google-data/90000");
            m->attributes = g_list_append(m->attributes, aa);        
            m->ptypes = g_list_append(m->ptypes, GINT_TO_POINTER(109));

3、接收RTP数据函数janus_ice_cb_nice_recv中，增加对datachannel的数据处理的逻辑：

if (header->type == 109 || stream->data_ssrc_peer == packet_ssrc){ 
                        /* Pass the data to the responsible plugin */ 
                        janus_plugin *plugin = (janus_plugin *)handle->app;
                        
                        if(plugin && plugin->incoming_data &&
                                !g_atomic_int_get(&handle->app_handle->stopped) &&
                                !g_atomic_int_get(&handle->destroyed)){
                                
                            plugin->incoming_data(handle->app_handle, (char *)buf + RTP_HEADER_SIZE, buflen - RTP_HEADER_SIZE);
                        }
                        return;
                    }

4、发送RTP数据时，使用RTP通道发送datachannel的数据：

void janus_plugin_relay_data(janus_plugin_session *plugin_session, char *buf, int len) {
    if((plugin_session < (janus_plugin_session *)0x1000) || g_atomic_int_get(&plugin_session->stopped) || buf == NULL || len < 1)
        return;
    janus_ice_handle *handle = (janus_ice_handle *)plugin_session->gateway_handle;
    if(!handle || janus_flags_is_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_STOP)
            || janus_flags_is_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_ALERT)){
        return;
    }
#ifdef HAVE_SCTP
    if (janus_flags_is_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_NEW_DATACHAN_SDP)){
        janus_ice_relay_data_withrtp(handle, buf, len);//relay daa with rtp only.
    }else{
        janus_ice_relay_data(handle, buf, len);
    }
#else
    JANUS_LOG(LOG_WARN, "Asked to relay data, but Data Channels support has not been compiled...\n");
#endif
}
 
 
void janus_ice_relay_data_withrtp(janus_ice_handle *handle, char *buf, int len) {
    struct timeval now;
    if(!handle || handle->queued_packets == NULL || buf == NULL || len < 1){
        return;
    }
    /* Queue this packet */
    janus_ice_queued_packet *pkt = g_malloc(sizeof(janus_ice_queued_packet));
    pkt->data = g_malloc(RTP_HEADER_SIZE+len+SRTP_MAX_TAG_LEN+4);
    janus_rtp_header *header = (janus_rtp_header *)pkt->data;
    header->ssrc = (handle->stream->data_ssrc);
    header->type = 109;
    header->version = 2;
    header->markerbit = 1;
    header->extension = 0; 
    handle->stream->sequence_data++;
    header->seq_number = (handle->stream->sequence_data + 100);
    gettimeofday(&now,0);
    header->timestamp = htonl(now.tv_sec);
    
    memcpy((char *)pkt->data + RTP_HEADER_SIZE, buf, len);
    pkt->length = RTP_HEADER_SIZE + len;
    pkt->type = JANUS_ICE_PACKET_DATA_WITH_RTP;
    pkt->control = FALSE;
    pkt->encrypted = FALSE;
    pkt->retransmission = FALSE;
    pkt->added = janus_get_monotonic_time();
    janus_ice_queue_packet(handle, pkt);
}
发送函数static gboolean janus_ice_outgoing_traffic_handle的逻辑中，对ssrc的处理需要修改为：
     header->ssrc = htonl(video ? stream->video_ssrc : ((pkt->type == JANUS_ICE_PACKET_DATA_WITH_RTP)?stream->data_ssrc : stream->audio_ssrc));

5、体验Web和Android客户端: https://poc.it3q.com

呱牛笔记