stream.c

一个开源的sip源代码

	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_int16_t*)
			  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 >=
	(long)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;
}