stream.c

来自「基于sip协议的网络电话源码」· C语言 代码 · 共 1,670 行 · 第 1/3 页

    /* Create RTP and RTCP sessions: */

    status = pjmedia_rtp_session_init(&channel->rtp, pt, param->ssrc);
    if (status != PJ_SUCCESS)
	return status;

    /* Done. */
    *p_channel = channel;
    return PJ_SUCCESS;
}


/*
 * Create media stream.
 */
PJ_DEF(pj_status_t) pjmedia_stream_create( pjmedia_endpt *endpt,
					   pj_pool_t *pool,
					   const pjmedia_stream_info *info,
					   pjmedia_transport *tp,
					   void *user_data,
					   pjmedia_stream **p_stream)

{
    enum { M = 32 };
    pjmedia_stream *stream;
    pj_str_t name;
    unsigned jb_init, jb_max, jb_min_pre, jb_max_pre;
    pj_status_t status;

    PJ_ASSERT_RETURN(pool && info && p_stream, PJ_EINVAL);


    /* Allocate the media stream: */

    stream = pj_pool_zalloc(pool, sizeof(pjmedia_stream));
    PJ_ASSERT_RETURN(stream != NULL, PJ_ENOMEM);

    /* Init stream/port name */
    name.ptr = pj_pool_alloc(pool, M);
    name.slen = pj_ansi_snprintf(name.ptr, M, "strm%p", stream);


    /* Init some port-info. Some parts of the info will be set later
     * once we have more info about the codec.
     */
    pjmedia_port_info_init(&stream->port.info, &name,
			   PJMEDIA_PORT_SIGNATURE('S', 'T', 'R', 'M'),
			   info->fmt.clock_rate, info->fmt.channel_cnt,
			   16, 80);

    /* Init port. */

    pj_strdup(pool, &stream->port.info.encoding_name, &info->fmt.encoding_name);
    stream->port.info.clock_rate = info->fmt.clock_rate;
    stream->port.info.channel_count = info->fmt.channel_cnt;
    stream->port.port_data.pdata = stream;
    stream->port.put_frame = &put_frame;
    stream->port.get_frame = &get_frame;


    /* Init stream: */
    stream->endpt = endpt;
    stream->codec_mgr = pjmedia_endpt_get_codec_mgr(endpt);
    stream->dir = info->dir;
    stream->user_data = user_data;
    stream->rtcp_interval = (PJMEDIA_RTCP_INTERVAL + (pj_rand() % 8000)) * 
			    info->fmt.clock_rate / 1000;

    stream->tx_event_pt = info->tx_event_pt ? info->tx_event_pt : -1;
    stream->rx_event_pt = info->rx_event_pt ? info->rx_event_pt : -1;
    stream->last_dtmf = -1;


    /* Create mutex to protect jitter buffer: */

    status = pj_mutex_create_simple(pool, NULL, &stream->jb_mutex);
    if (status != PJ_SUCCESS)
	goto err_cleanup;


    /* Create and initialize codec: */

    status = pjmedia_codec_mgr_alloc_codec( stream->codec_mgr,
					    &info->fmt, &stream->codec);
    if (status != PJ_SUCCESS)
	goto err_cleanup;


    /* Get codec param: */
    if (info->param)
	stream->codec_param = *info->param;
    else {
	status = pjmedia_codec_mgr_get_default_param(stream->codec_mgr, 
						     &info->fmt, 
						     &stream->codec_param);
	if (status != PJ_SUCCESS)
	    goto err_cleanup;
    }

    /* Check for invalid frame per packet. */
    if (stream->codec_param.setting.frm_per_pkt < 1)
	stream->codec_param.setting.frm_per_pkt = 1;

    /* Set additional info. */
    stream->port.info.bits_per_sample = 16;
    stream->port.info.samples_per_frame = info->fmt.clock_rate * 
					  stream->codec_param.info.frm_ptime *
					  stream->codec_param.setting.frm_per_pkt /
					  1000;
    stream->port.info.bytes_per_frame = stream->codec_param.info.avg_bps/8 * 
					stream->codec_param.info.frm_ptime *
					stream->codec_param.setting.frm_per_pkt /
					1000;


    /* Open the codec: */

    status = stream->codec->op->open(stream->codec, &stream->codec_param);
    if (status != PJ_SUCCESS)
	goto err_cleanup;

    /* If encoder and decoder's ptime are asymmetric, then we need to
     * create buffer on the encoder side. This could happen for example
     * with iLBC 
     */
    if (stream->codec_param.info.enc_ptime!=0 &&
	stream->codec_param.info.enc_ptime!=stream->codec_param.info.frm_ptime)
    {
	unsigned ptime;

	stream->enc_samples_per_frame = stream->codec_param.info.enc_ptime *
					stream->port.info.clock_rate / 1000;

	/* Set buffer size as twice the largest ptime value between
	 * stream's ptime, encoder ptime, or decoder ptime.
	 */

	ptime = stream->port.info.samples_per_frame * 1000 /
		stream->port.info.clock_rate;

	if (stream->codec_param.info.enc_ptime > ptime)
	    ptime = stream->codec_param.info.enc_ptime;

	if (stream->codec_param.info.frm_ptime > ptime)
	    ptime = stream->codec_param.info.frm_ptime;

	ptime <<= 1;

	/* Allocate buffer */
	stream->enc_buf_size = stream->port.info.clock_rate * ptime / 1000;
	stream->enc_buf = pj_pool_alloc(pool, stream->enc_buf_size * 2);

    } else {
	stream->enc_samples_per_frame = stream->port.info.samples_per_frame;
    }

    /* Initially disable the VAD in the stream, to help traverse NAT better */
    stream->vad_enabled = stream->codec_param.setting.vad;
    if (stream->vad_enabled) {
	stream->codec_param.setting.vad = 0;
	stream->ts_vad_disabled = 0;
	stream->codec->op->modify(stream->codec, &stream->codec_param);
	PJ_LOG(4,(stream->port.info.name.ptr,"VAD temporarily disabled"));
    }

    /* Get the frame size: */

    stream->frame_size = ((stream->codec_param.info.avg_bps + 7) / 8) * 
			  stream->codec_param.info.frm_ptime / 1000;


    /* Init RTCP session: */

    pjmedia_rtcp_init(&stream->rtcp, stream->port.info.name.ptr,
		      info->fmt.clock_rate, 
		      stream->port.info.samples_per_frame, 
		      info->ssrc);


    /* Init jitter buffer parameters: */
    if (info->jb_max > 0)
	jb_max = info->jb_max;
    else
	jb_max = 360 / stream->codec_param.info.frm_ptime;

    if (info->jb_min_pre >= 0)
	jb_min_pre = info->jb_min_pre;
    else
	jb_min_pre = 60 / stream->codec_param.info.frm_ptime;

    if (info->jb_max_pre > 0)
	jb_max_pre = info->jb_max_pre;
    else
	jb_max_pre = 240 / stream->codec_param.info.frm_ptime;

    if (info->jb_init >= 0)
	jb_init = info->jb_init;
    else
	jb_init = (jb_min_pre + jb_max_pre) / 2;


    /* Create jitter buffer */
    status = pjmedia_jbuf_create(pool, &stream->port.info.name,
				 stream->frame_size, 
				 stream->codec_param.info.frm_ptime,
				 jb_max, &stream->jb);
    if (status != PJ_SUCCESS)
	goto err_cleanup;


    /* Set up jitter buffer */
    pjmedia_jbuf_set_adaptive( stream->jb, jb_init, jb_min_pre, jb_max_pre);

    /* Create decoder channel: */

    status = create_channel( pool, stream, PJMEDIA_DIR_DECODING, 
			     info->fmt.pt, info, &stream->dec);
    if (status != PJ_SUCCESS)
	goto err_cleanup;


    /* Create encoder channel: */

    status = create_channel( pool, stream, PJMEDIA_DIR_ENCODING, 
			     info->tx_pt, info, &stream->enc);
    if (status != PJ_SUCCESS)
	goto err_cleanup;


    /* Only attach transport when stream is ready. */
    status = pjmedia_transport_attach(tp, stream, &info->rem_addr, 
				      &info->rem_rtcp, sizeof(info->rem_addr), 
                                      &on_rx_rtp, &on_rx_rtcp);
    if (status != PJ_SUCCESS)
	goto err_cleanup;

    stream->transport = tp;


    /* Success! */
    *p_stream = stream;

    PJ_LOG(5,(THIS_FILE, "Stream %s created", stream->port.info.name.ptr));
    return PJ_SUCCESS;


err_cleanup:
    pjmedia_stream_destroy(stream);
    return status;
}


/*
 * Destroy stream.
 */
PJ_DEF(pj_status_t) pjmedia_stream_destroy( pjmedia_stream *stream )
{

    PJ_ASSERT_RETURN(stream != NULL, PJ_EINVAL);


    /* This function may be called when stream is partly initialized. */
    if (stream->jb_mutex)
	pj_mutex_lock(stream->jb_mutex);


    /* Detach from transport */
    if (stream->transport) {
	(*stream->transport->op->detach)(stream->transport, stream);
	stream->transport = NULL;
    }

    /* Free codec. */

    if (stream->codec) {
	stream->codec->op->close(stream->codec);
	pjmedia_codec_mgr_dealloc_codec(stream->codec_mgr, stream->codec);
	stream->codec = NULL;
    }

    /* Free mutex */
    
    if (stream->jb_mutex) {
	pj_mutex_destroy(stream->jb_mutex);
	stream->jb_mutex = NULL;
    }

    return PJ_SUCCESS;
}



/*
 * Get the port interface.
 */
PJ_DEF(pj_status_t) pjmedia_stream_get_port( pjmedia_stream *stream,
					     pjmedia_port **p_port )
{
    *p_port = &stream->port;
    return PJ_SUCCESS;
}


/*
 * Get the transport object
 */
PJ_DEF(pjmedia_transport*) pjmedia_stream_get_transport(pjmedia_stream *st)
{
    return st->transport;
}


/*
 * Start stream.
 */
PJ_DEF(pj_status_t) pjmedia_stream_start(pjmedia_stream *stream)
{

    PJ_ASSERT_RETURN(stream && stream->enc && stream->dec, PJ_EINVALIDOP);

    if (stream->enc && (stream->dir & PJMEDIA_DIR_ENCODING)) {
	stream->enc->paused = 0;
	//pjmedia_snd_stream_start(stream->enc->snd_stream);
	PJ_LOG(4,(stream->port.info.name.ptr, "Encoder stream started"));
    } else {
	PJ_LOG(4,(stream->port.info.name.ptr, "Encoder stream paused"));
    }

    if (stream->dec && (stream->dir & PJMEDIA_DIR_DECODING)) {
	stream->dec->paused = 0;
	//pjmedia_snd_stream_start(stream->dec->snd_stream);
	PJ_LOG(4,(stream->port.info.name.ptr, "Decoder stream started"));
    } else {
	PJ_LOG(4,(stream->port.info.name.ptr, "Decoder stream paused"));
    }

    return PJ_SUCCESS;
}


/*
 * Get stream statistics.
 */
PJ_DEF(pj_status_t) pjmedia_stream_get_stat( const pjmedia_stream *stream,
					     pjmedia_rtcp_stat *stat)
{
    PJ_ASSERT_RETURN(stream && stat, PJ_EINVAL);

    pj_memcpy(stat, &stream->rtcp.stat, sizeof(pjmedia_rtcp_stat));
    return PJ_SUCCESS;
}


/*
 * Pause stream.
 */
PJ_DEF(pj_status_t) pjmedia_stream_pause( pjmedia_stream *stream,
					  pjmedia_dir dir)
{
    PJ_ASSERT_RETURN(stream, PJ_EINVAL);

    if ((dir & PJMEDIA_DIR_ENCODING) && stream->enc) {
	stream->enc->paused = 1;
	PJ_LOG(4,(stream->port.info.name.ptr, "Encoder stream paused"));
    }

    if ((dir & PJMEDIA_DIR_DECODING) && stream->dec) {
	stream->dec->paused = 1;

	/* Also reset jitter buffer */
	pj_mutex_lock( stream->jb_mutex );
	pjmedia_jbuf_reset(stream->jb);
	pj_mutex_unlock( stream->jb_mutex );

	PJ_LOG(4,(stream->port.info.name.ptr, "Decoder stream paused"));
    }

    return PJ_SUCCESS;
}


/*
 * Resume stream
 */
PJ_DEF(pj_status_t) pjmedia_stream_resume( pjmedia_stream *stream,
					   pjmedia_dir dir)
{
    PJ_ASSERT_RETURN(stream, PJ_EINVAL);

    if ((dir & PJMEDIA_DIR_ENCODING) && stream->enc) {
	stream->enc->paused = 0;
	PJ_LOG(4,(stream->port.info.name.ptr, "Encoder stream resumed"));
    }

    if ((dir & PJMEDIA_DIR_DECODING) && stream->dec) {
	stream->dec->paused = 0;
	PJ_LOG(4,(stream->port.info.name.ptr, "Decoder stream resumed"));
    }

    return PJ_SUCCESS;
}

/*
 * Dial DTMF
 */
PJ_DEF(pj_status_t) pjmedia_stream_dial_dtmf( pjmedia_stream *stream,
					      const pj_str_t *digit_char)
{
    pj_status_t status = PJ_SUCCESS;

    /* By convention we use jitter buffer mutex to access DTMF
     * queue.
     */
    PJ_ASSERT_RETURN(stream && digit_char, PJ_EINVAL);

    /* Check that remote can receive DTMF events. */
    if (stream->tx_event_pt < 0) {
	return PJMEDIA_RTP_EREMNORFC2833;
    }
    
    pj_mutex_lock(stream->jb_mutex);
    
    if (stream->tx_dtmf_count+digit_char->slen >=
	PJ_ARRAY_SIZE(stream->tx_dtmf_buf))
    {
	status = PJ_ETOOMANY;
    } else {
	int i;

	/* convert ASCII digits into payload type first, to make sure
	 * that all digits are valid. 
	 */
	for (i=0; i<digit_char->slen; ++i) {
	    unsigned pt;
	    int dig = pj_tolower(digit_char->ptr[i]);

	    if (dig >= '0' && dig <= '9')
	    {
		pt = dig - '0';
	    } 
	    else if (dig >= 'a' && dig <= 'd')
	    {
		pt = dig - 'a' + 12;
	    }
	    else if (dig == '*')
	    {
		pt = 10;
	    }
	    else if (dig == '#')
	    {
		pt = 11;
	    }
	    else
	    {
		status = PJMEDIA_RTP_EINDTMF;
		break;
	    }

	    stream->tx_dtmf_buf[stream->tx_dtmf_count+i].event = pt;
	}

	if (status != PJ_SUCCESS)
	    goto on_return;

	/* Init start_ts and end_ts only for the first digit.
	 * Subsequent digits are initialized on the fly.
	 */
	if (stream->tx_dtmf_count ==0) {
	    pj_uint32_t start_ts;

	    start_ts = pj_ntohl(stream->enc->rtp.out_hdr.ts);
	    stream->tx_dtmf_buf[0].start_ts = start_ts;
	}

	/* Increment digit count only if all digits are valid. */
	stream->tx_dtmf_count += digit_char->slen;

    }

on_return:
    pj_mutex_unlock(stream->jb_mutex);

    return status;
}


/*
 * See if we have DTMF digits in the rx buffer.
 */
PJ_DEF(pj_bool_t) pjmedia_stream_check_dtmf(pjmedia_stream *stream)
{
    return stream->rx_dtmf_count != 0;
}


/*
 * Retrieve incoming DTMF digits from the stream's DTMF buffer.
 */
PJ_DEF(pj_status_t) pjmedia_stream_get_dtmf( pjmedia_stream *stream,
					     char *digits,
					     unsigned *size)
{
    PJ_ASSERT_RETURN(stream && digits && size, PJ_EINVAL);

    pj_assert(sizeof(stream->rx_dtmf_buf[0]) == 0);

    /* By convention, we use jitter buffer's mutex to access DTMF
     * digits resources.
     */
    pj_mutex_lock(stream->jb_mutex);

    if (stream->rx_dtmf_count < *size)
	*size = stream->rx_dtmf_count;

    if (*size) {
	pj_memcpy(digits, stream->rx_dtmf_buf, *size);
	stream->rx_dtmf_count -= *size;
	if (stream->rx_dtmf_count) {
	    pj_memmove(stream->rx_dtmf_buf,
		       &stream->rx_dtmf_buf[*size],
		       stream->rx_dtmf_count);
	}
    }

    pj_mutex_unlock(stream->jb_mutex);

    return PJ_SUCCESS;
}


/*
 * Set callback to be called upon receiving DTMF digits.
 */
PJ_DEF(pj_status_t)
pjmedia_stream_set_dtmf_callback(pjmedia_stream *stream,
				 void (*cb)(pjmedia_stream*, 
					    void *user_data, 
					    int digit), 
				 void *user_data)
{
    PJ_ASSERT_RETURN(stream, PJ_EINVAL);

    /* By convention, we use jitter buffer's mutex to access DTMF
     * digits resources.
     */
    pj_mutex_lock(stream->jb_mutex);

    stream->dtmf_cb = cb;
    stream->dtmf_cb_user_data = user_data;

    pj_mutex_unlock(stream->jb_mutex);

    return PJ_SUCCESS;
}