stream.c

一个开源的sip源代码

/**
 * check_tx_rtcp()
 *
 * This function is can be called by either put_frame() or get_frame(),
 * to transmit periodic RTCP SR/RR report.
 */
static void check_tx_rtcp(pjmedia_stream *stream, pj_uint32_t timestamp)
{
    /* Note that timestamp may represent local or remote timestamp, 
     * depending on whether this function is called from put_frame()
     * or get_frame().
     */


    if (stream->rtcp_last_tx == 0) {
	
	stream->rtcp_last_tx = timestamp;

    } else if (timestamp - stream->rtcp_last_tx >= stream->rtcp_interval) {
	
	pjmedia_rtcp_pkt *rtcp_pkt;
	int len;

	pjmedia_rtcp_build_rtcp(&stream->rtcp, &rtcp_pkt, &len);

	(*stream->transport->op->send_rtcp)(stream->transport, 
					    rtcp_pkt, len);

	stream->rtcp_last_tx = timestamp;
    }

}


/**
 * Rebuffer the frame when encoder and decoder has different ptime
 * (such as when different iLBC modes are used by local and remote)
 */
static void rebuffer(pjmedia_stream *stream,
		     pjmedia_frame *frame)
{
    /* How many samples are needed */
    unsigned count;

    /* Normalize frame */
    if (frame->type != PJMEDIA_FRAME_TYPE_AUDIO)
	frame->size = 0;

    /* Remove used frame from the buffer. */
    if (stream->enc_buf_pos) {
	if (stream->enc_buf_count) {
	    pj_memmove(stream->enc_buf,
		       stream->enc_buf + stream->enc_buf_pos,
		       (stream->enc_buf_count << 1));
	}
	stream->enc_buf_pos = 0;
    }

    /* Make sure we have space to store the new frame */
    pj_assert(stream->enc_buf_count + (frame->size >> 1) <
		stream->enc_buf_size);

    /* Append new frame to the buffer */
    if (frame->size) {
	pj_memcpy(stream->enc_buf + stream->enc_buf_count,
		  frame->buf, frame->size);
	stream->enc_buf_count += (frame->size >> 1);
    }

    /* How many samples are needed */
    count = stream->codec_param.info.enc_ptime * 
		stream->port.info.clock_rate / 1000;

    /* See if we have enough samples */
    if (stream->enc_buf_count >= count) {

	frame->type = PJMEDIA_FRAME_TYPE_AUDIO;
	frame->buf = stream->enc_buf;
	frame->size = (count << 1);

	stream->enc_buf_pos = count;
	stream->enc_buf_count -= count;

    } else {
	/* We don't have enough samples */
	frame->type = PJMEDIA_FRAME_TYPE_NONE;
    }
}


/**
 * put_frame_imp()
 */
static pj_status_t put_frame_imp( pjmedia_port *port, 
				  const pjmedia_frame *frame )
{
    pjmedia_stream *stream = (pjmedia_stream*) port->port_data.pdata;
    pjmedia_channel *channel = stream->enc;
    pj_status_t status = 0;
    pjmedia_frame frame_out;
    unsigned ts_len, samples_per_frame;
    void *rtphdr;
    int rtphdrlen;


    /* Don't do anything if stream is paused */
    if (channel->paused) {
	stream->enc_buf_pos = stream->enc_buf_count = 0;
	return PJ_SUCCESS;
    }

    /* Number of samples in the frame */
    if (frame->type == PJMEDIA_FRAME_TYPE_AUDIO)
	ts_len = (frame->size >> 1);
    else
	ts_len = 0;

    /* Increment transmit duration */
    stream->tx_duration += ts_len;

    /* Init frame_out buffer. */
    frame_out.buf = ((char*)channel->out_pkt) + sizeof(pjmedia_rtp_hdr);
    frame_out.size = 0;

    /* Calculate number of samples per frame */
    samples_per_frame = stream->enc_samples_per_frame;


    /* If we have DTMF digits in the queue, transmit the digits. 
     * Otherwise encode the PCM buffer.
     */
    if (stream->tx_dtmf_count) {

	create_dtmf_payload(stream, &frame_out);

	/* Encapsulate. */
	status = pjmedia_rtp_encode_rtp( &channel->rtp, 
					 stream->tx_event_pt, 0, 
					 frame_out.size, ts_len, 
					 (const void**)&rtphdr, 
					 &rtphdrlen);

    } else if (frame->type != PJMEDIA_FRAME_TYPE_NONE) {
	unsigned ts, codec_samples_per_frame;

	/* Repeatedly call encode if there are multiple frames to be
	 * sent.
	 */
	codec_samples_per_frame = stream->codec_param.info.enc_ptime *
				  stream->codec_param.info.clock_rate /
				  1000;
	if (codec_samples_per_frame == 0) {
	    codec_samples_per_frame = stream->codec_param.info.frm_ptime *
				      stream->codec_param.info.clock_rate /
				      1000;
	}

	for (ts=0; ts<ts_len; ts += codec_samples_per_frame) {
	    pjmedia_frame tmp_out_frame, tmp_in_frame;
	    unsigned bytes_per_sample, max_size;

	    /* Nb of bytes in PCM sample */
	    bytes_per_sample = stream->codec_param.info.pcm_bits_per_sample/8;

	    /* Split original PCM input frame into base frame size */
	    tmp_in_frame.timestamp.u64 = frame->timestamp.u64 + ts;
	    tmp_in_frame.buf = ((char*)frame->buf) + ts * bytes_per_sample;
	    tmp_in_frame.size = codec_samples_per_frame * bytes_per_sample;
	    tmp_in_frame.type = PJMEDIA_FRAME_TYPE_AUDIO;

	    /* Set output frame position */
	    tmp_out_frame.buf = ((char*)frame_out.buf) + frame_out.size;

	    max_size = channel->out_pkt_size - sizeof(pjmedia_rtp_hdr) -
		       frame_out.size;

	    /* Encode! */
	    status = stream->codec->op->encode( stream->codec, &tmp_in_frame, 
						max_size, &tmp_out_frame);
	    if (status != PJ_SUCCESS) {
		LOGERR_((stream->port.info.name.ptr, 
			"Codec encode() error", status));
		return status;
	    }

	    /* tmp_out_frame.size may be zero for silence frame. */
	    frame_out.size += tmp_out_frame.size;

	    /* Stop processing next PCM frame when encode() returns either 
	     * CNG frame or NULL frame.
	     */
	    if (tmp_out_frame.type!=PJMEDIA_FRAME_TYPE_AUDIO || 
		tmp_out_frame.size==0) 
	    {
		break;
	    }

	}

	/* Encapsulate. */
	status = pjmedia_rtp_encode_rtp( &channel->rtp, 
					 channel->pt, 0, 
					 frame_out.size, ts_len, 
					 (const void**)&rtphdr, 
					 &rtphdrlen);
    } else {

	/* Just update RTP session's timestamp. */
	status = pjmedia_rtp_encode_rtp( &channel->rtp, 
					 0, 0, 
					 0, ts_len, 
					 (const void**)&rtphdr, 
					 &rtphdrlen);

    }

    if (status != PJ_SUCCESS) {
	LOGERR_((stream->port.info.name.ptr, 
		"RTP encode_rtp() error", status));
	return status;
    }

    /* Check if now is the time to transmit RTCP SR/RR report. 
     * We only do this when stream direction is not "decoding only", because
     * when it is, check_tx_rtcp() will be handled by get_frame().
     */
    if (stream->dir != PJMEDIA_DIR_DECODING) {
	check_tx_rtcp(stream, pj_ntohl(channel->rtp.out_hdr.ts));
    }

    /* Do nothing if we have nothing to transmit */
    if (frame_out.size == 0) {
	if (stream->is_streaming) {
	    PJ_LOG(5,(stream->port.info.name.ptr,"Starting silence"));
	    stream->is_streaming = PJ_FALSE;
	}

	return PJ_SUCCESS;
    }


    /* Copy RTP header to the beginning of packet */
    pj_memcpy(channel->out_pkt, rtphdr, sizeof(pjmedia_rtp_hdr));


    /* Set RTP marker bit if currently not streaming */
    if (stream->is_streaming == PJ_FALSE) {
	pjmedia_rtp_hdr *rtp = (pjmedia_rtp_hdr*) channel->out_pkt;

	rtp->m = 1;
	PJ_LOG(5,(stream->port.info.name.ptr,"Start talksprut.."));
    }

    stream->is_streaming = PJ_TRUE;

    /* Send the RTP packet to the transport. */
    (*stream->transport->op->send_rtp)(stream->transport,
				       channel->out_pkt, 
				       frame_out.size + 
					    sizeof(pjmedia_rtp_hdr));


    /* Update stat */
    pjmedia_rtcp_tx_rtp(&stream->rtcp, frame_out.size);

    return PJ_SUCCESS;
}


/**
 * put_frame()
 *
 * This callback is called by upstream component when it has PCM frame
 * to transmit. This function encodes the PCM frame, pack it into
 * RTP packet, and transmit to peer.
 */
static pj_status_t put_frame( pjmedia_port *port, 
			      const pjmedia_frame *frame )
{
    pjmedia_stream *stream = (pjmedia_stream*) port->port_data.pdata;
    pjmedia_frame tmp_zero_frame;
    unsigned samples_per_frame;

    samples_per_frame = stream->enc_samples_per_frame;

    /* http://www.pjsip.org/trac/ticket/56:
     *  when input is PJMEDIA_FRAME_TYPE_NONE, feed zero PCM frame
     *  instead so that encoder can decide whether or not to transmit
     *  silence frame.
     */
    if (frame->type == PJMEDIA_FRAME_TYPE_NONE &&
	samples_per_frame <= ZERO_PCM_MAX_SIZE) 
    {
	pj_memcpy(&tmp_zero_frame, frame, sizeof(pjmedia_frame));
	frame = &tmp_zero_frame;

	tmp_zero_frame.buf = zero_frame;
	tmp_zero_frame.size = samples_per_frame * 2;
	tmp_zero_frame.type = PJMEDIA_FRAME_TYPE_AUDIO;
    }

#if 0
    // This is no longer needed because each TYPE_NONE frame will
    // be converted into zero frame above

    /* If VAD is temporarily disabled during creation, feed zero PCM frame
     * to the codec.
     */
    if (stream->vad_enabled != stream->codec_param.setting.vad &&
	stream->vad_enabled != 0 &&
	frame->type == PJMEDIA_FRAME_TYPE_NONE &&
	samples_per_frame <= ZERO_PCM_MAX_SIZE)
    {
	pj_memcpy(&tmp_in_frame, frame, sizeof(pjmedia_frame));
	frame = &tmp_in_frame;

	tmp_in_frame.buf = zero_frame;
	tmp_in_frame.size = samples_per_frame * 2;
	tmp_in_frame.type = PJMEDIA_FRAME_TYPE_AUDIO;
    }
#endif

    /* If VAD is temporarily disabled during creation, enable it
     * after transmitting for VAD_SUSPEND_SEC seconds.
     */
    if (stream->vad_enabled != stream->codec_param.setting.vad &&
	(stream->tx_duration - stream->ts_vad_disabled) > 
	stream->port.info.clock_rate * PJMEDIA_STREAM_VAD_SUSPEND_MSEC / 1000)
    {
	stream->codec_param.setting.vad = stream->vad_enabled;
	stream->codec->op->modify(stream->codec, &stream->codec_param);
	PJ_LOG(4,(stream->port.info.name.ptr,"VAD re-enabled"));
    }


    /* If encoder has different ptime than decoder, then the frame must
     * be passed through the encoding buffer via rebuffer() function.
     */
    if (stream->enc_buf != NULL) {
	pjmedia_frame tmp_rebuffer_frame;
	pj_status_t status = PJ_SUCCESS;

	/* Copy original frame to temporary frame since we need 
	 * to modify it.
	 */
	pj_memcpy(&tmp_rebuffer_frame, frame, sizeof(pjmedia_frame));

	/* Loop while we have full frame in enc_buffer */
	for (;;) {
	    pj_status_t st;

	    /* Run rebuffer() */
	    rebuffer(stream, &tmp_rebuffer_frame);

	    /* Process this frame */
	    st = put_frame_imp(port, &tmp_rebuffer_frame);
	    if (st != PJ_SUCCESS)
		status = st;

	    /* If we still have full frame in the buffer, re-run
	     * rebuffer() with NULL frame.
	     */
	    if (stream->enc_buf_count >= stream->enc_samples_per_frame) {

		tmp_rebuffer_frame.type = PJMEDIA_FRAME_TYPE_NONE;

	    } else {

		/* Otherwise break */
		break;
	    }
	}

	return status;

    } else {
	return put_frame_imp(port, frame);
    }
}


#if 0
static void dump_bin(const char *buf, unsigned len)
{
    unsigned i;

    PJ_LOG(3,(THIS_FILE, "begin dump"));
    for (i=0; i<len; ++i) {
	int j;
	char bits[9];
	unsigned val = buf[i] & 0xFF;

	bits[8] = '\0';
	for (j=0; j<8; ++j) {
	    if (val & (1 << (7-j)))
		bits[j] = '1';
	    else
		bits[j] = '0';
	}

	PJ_LOG(3,(THIS_FILE, "%2d %s [%d]", i, bits, val));
    }
    PJ_LOG(3,(THIS_FILE, "end dump"));
}
#endif

/*
 * Handle incoming DTMF digits.
 */
static void handle_incoming_dtmf( pjmedia_stream *stream, 
				  const void *payload, unsigned payloadlen)
{
    pjmedia_rtp_dtmf_event *event = (pjmedia_rtp_dtmf_event*) payload;

    /* Check compiler packing. */
    pj_assert(sizeof(pjmedia_rtp_dtmf_event)==4);