stream.c

一个开源SIP协议栈

/* $Id: stream.c 1047 2007-03-07 11:22:14Z bennylp $ */
/* 
 * Copyright (C) 2003-2007 Benny Prijono <[e-mail]>
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA 
 */
#include <pjmedia/stream.h>
#include <pjmedia/errno.h>
#include <pjmedia/rtp.h>
#include <pjmedia/rtcp.h>
#include <pjmedia/jbuf.h>
#include <pj/array.h>
#include <pj/assert.h>
#include <pj/ctype.h>
#include <pj/compat/socket.h>
#include <pj/errno.h>
#include <pj/ioqueue.h>
#include <pj/log.h>
#include <pj/os.h>
#include <pj/pool.h>
#include <pj/rand.h>
#include <pj/sock_select.h>
#include <pj/string.h>	    /* memcpy() */


#define THIS_FILE			"stream.c"
#define ERRLEVEL			1
#define LOGERR_(expr)			stream_perror expr
#define TRC_(expr)			PJ_LOG(5,expr)

#define BYTES_PER_SAMPLE		2

/**
 * Media channel.
 */
struct pjmedia_channel
{
    pjmedia_stream	   *stream;	    /**< Parent stream.		    */
    pjmedia_dir		    dir;	    /**< Channel direction.	    */
    unsigned		    pt;		    /**< Payload type.		    */
    pj_bool_t		    paused;	    /**< Paused?.		    */
    unsigned		    in_pkt_size;    /**< Size of input buffer.	    */
    void		   *in_pkt;	    /**< Input buffer.		    */
    unsigned		    out_pkt_size;   /**< Size of output buffer.	    */
    void		   *out_pkt;	    /**< Output buffer.		    */
    pjmedia_rtp_session	    rtp;	    /**< RTP session.		    */
};


struct dtmf
{
    int		    event;
    pj_uint32_t	    start_ts;
};

/**
 * This structure describes media stream.
 * A media stream is bidirectional media transmission between two endpoints.
 * It consists of two channels, i.e. encoding and decoding channels.
 * A media stream corresponds to a single "m=" line in a SDP session
 * description.
 */
struct pjmedia_stream
{
    pjmedia_endpt	    *endpt;	    /**< Media endpoint.	    */
    pjmedia_codec_mgr	    *codec_mgr;	    /**< Codec manager instance.    */

    pjmedia_port	     port;	    /**< Port interface.	    */
    pjmedia_channel	    *enc;	    /**< Encoding channel.	    */
    pjmedia_channel	    *dec;	    /**< Decoding channel.	    */

    pjmedia_dir		     dir;	    /**< Stream direction.	    */
    void		    *user_data;	    /**< User data.		    */

    pjmedia_transport	    *transport;	    /**< Stream transport.	    */

    pjmedia_codec	    *codec;	    /**< Codec instance being used. */
    pjmedia_codec_param	     codec_param;   /**< Codec param.		    */
    pj_int16_t		    *enc_buf;	    /**< Encoding buffer, when enc's
						 ptime is different than dec.
						 Otherwise it's NULL.	    */

    unsigned		     enc_samples_per_frame;
    unsigned		     enc_buf_size;  /**< Encoding buffer size, in
						 samples.		    */
    unsigned		     enc_buf_pos;   /**< First position in buf.	    */
    unsigned		     enc_buf_count; /**< Number of samples in the
						 encoding buffer.	    */

    unsigned		     vad_enabled;   /**< VAD enabled in param.	    */
    unsigned		     frame_size;    /**< Size of encoded base frame.*/
    pj_bool_t		     is_streaming;  /**< Currently streaming?. This
						 is used to put RTP marker
						 bit.			    */
    pj_uint32_t		     ts_vad_disabled;/**< TS when VAD was disabled. */
    pj_uint32_t		     tx_duration;   /**< TX duration in timestamp.  */

    pj_mutex_t		    *jb_mutex;
    pjmedia_jbuf	    *jb;	    /**< Jitter buffer.		    */
    char		     jb_last_frm;   /**< Last frame type from jb    */

    pjmedia_rtcp_session     rtcp;	    /**< RTCP for incoming RTP.	    */

    pj_uint32_t		     rtcp_last_tx;  /**< RTCP tx time in timestamp  */
    pj_uint32_t		     rtcp_interval; /**< Interval, in timestamp.    */

    /* RFC 2833 DTMF transmission queue: */
    int			     tx_event_pt;   /**< Outgoing pt for dtmf.	    */
    int			     tx_dtmf_count; /**< # of digits in tx dtmf buf.*/
    struct dtmf		     tx_dtmf_buf[32];/**< Outgoing dtmf queue.	    */

    /* Incoming DTMF: */
    int			     rx_event_pt;   /**< Incoming pt for dtmf.	    */
    int			     last_dtmf;	    /**< Current digit, or -1.	    */
    pj_uint32_t		     last_dtmf_dur; /**< Start ts for cur digit.    */
    unsigned		     rx_dtmf_count; /**< # of digits in dtmf rx buf.*/
    char		     rx_dtmf_buf[32];/**< Incoming DTMF buffer.	    */

    /* DTMF callback */
    void		    (*dtmf_cb)(pjmedia_stream*, void*, int);
    void		     *dtmf_cb_user_data;
};


/* RFC 2833 digit */
static const char digitmap[16] = { '0', '1', '2', '3', 
				   '4', '5', '6', '7', 
				   '8', '9', '*', '#',
				   'A', 'B', 'C', 'D'};

/* Zero PCM frame */
#define ZERO_PCM_MAX_SIZE   1920    /* 40ms worth of PCM @ 48KHz */
static pj_int16_t zero_frame[ZERO_PCM_MAX_SIZE];


/*
 * Print error.
 */
static void stream_perror(const char *sender, const char *title,
			  pj_status_t status)
{
    char errmsg[PJ_ERR_MSG_SIZE];

    pj_strerror(status, errmsg, sizeof(errmsg));
    PJ_LOG(4,(sender, "%s: %s [err:%d]", title, errmsg, status));
}


/*
 * play_callback()
 *
 * This callback is called by sound device's player thread when it
 * needs to feed the player with some frames.
 */
static pj_status_t get_frame( pjmedia_port *port, pjmedia_frame *frame)
{
    pjmedia_stream *stream = port->port_data.pdata;
    pjmedia_channel *channel = stream->dec;
    unsigned samples_count, samples_per_frame, samples_required;
    pj_int16_t *p_out_samp;
    pj_status_t status;


    /* Return no frame is channel is paused */
    if (channel->paused) {
	frame->type = PJMEDIA_FRAME_TYPE_NONE;
	return PJ_SUCCESS;
    }

    /* Repeat get frame from the jitter buffer and decode the frame
     * until we have enough frames according to codec's ptime.
     */

    /* Lock jitter buffer mutex first */
    pj_mutex_lock( stream->jb_mutex );

    samples_required = stream->port.info.samples_per_frame;
    samples_per_frame = stream->codec_param.info.frm_ptime *
			stream->codec_param.info.clock_rate *
			stream->codec_param.info.channel_cnt / 
			1000;
    p_out_samp = frame->buf;

    for (samples_count=0; samples_count < samples_required; 
	 samples_count += samples_per_frame) 
    {
	char frame_type;

	/* Get frame from jitter buffer. */
	pjmedia_jbuf_get_frame(stream->jb, channel->out_pkt, &frame_type);
	
	if (frame_type == PJMEDIA_JB_MISSING_FRAME) {
	    
	    /* Activate PLC */
	    if (stream->codec->op->recover && 
		stream->codec_param.setting.plc) 
	    {
		pjmedia_frame frame_out;

		frame_out.buf = p_out_samp + samples_count;
		frame_out.size = frame->size - samples_count*2;
		status = (*stream->codec->op->recover)(stream->codec,
						       frame_out.size,
						       &frame_out);

	    } else {
		status = -1;
	    }

	    if (status != PJ_SUCCESS) {
		/* Either PLC failed or PLC not supported/enabled */
		pjmedia_zero_samples(p_out_samp + samples_count,
				     samples_required - samples_count);
		PJ_LOG(5,(stream->port.info.name.ptr,  "Frame lost!"));

	    } else {
		PJ_LOG(5,(stream->port.info.name.ptr, 
			  "Lost frame recovered"));
	    }
	    
	} else if (frame_type == PJMEDIA_JB_ZERO_EMPTY_FRAME) {

	    /* Jitter buffer is empty. If this is the first "empty" state,
	     * activate PLC to smoothen the fade-out, otherwise zero
	     * the frame. 
	     */
	    if (frame_type != stream->jb_last_frm) {
		pjmedia_jb_state jb_state;

		/* Activate PLC to smoothen the missing frame */
		if (stream->codec->op->recover && 
		    stream->codec_param.setting.plc) 
		{
		    pjmedia_frame frame_out;

		    do {
			frame_out.buf = p_out_samp + samples_count;
			frame_out.size = frame->size - samples_count*2;
			status = (*stream->codec->op->recover)(stream->codec,
							       frame_out.size,
							       &frame_out);
			if (status != PJ_SUCCESS)
			    break;
			samples_count += samples_per_frame;

		    } while (samples_count < samples_required);

		} 

		/* Report the state of jitter buffer */
		pjmedia_jbuf_get_state(stream->jb, &jb_state);
		PJ_LOG(5,(stream->port.info.name.ptr, 
			  "Jitter buffer empty (prefetch=%d)", 
			  jb_state.prefetch));

	    }

	    if (samples_count < samples_required) {
		pjmedia_zero_samples(p_out_samp + samples_count,
				     samples_required - samples_count);
		samples_count = samples_required;
	    }

	    stream->jb_last_frm = frame_type;
	    break;

	} else if (frame_type != PJMEDIA_JB_NORMAL_FRAME) {

	    pjmedia_jb_state jb_state;

	    /* It can only be PJMEDIA_JB_ZERO_PREFETCH frame */
	    pj_assert(frame_type == PJMEDIA_JB_ZERO_PREFETCH_FRAME);

	    /* Get the state of jitter buffer */
	    pjmedia_jbuf_get_state(stream->jb, &jb_state);

	    /* Always activate PLC when it's available.. */
	    if (stream->codec->op->recover && 
		stream->codec_param.setting.plc) 
	    {
		pjmedia_frame frame_out;

		do {
		    frame_out.buf = p_out_samp + samples_count;
		    frame_out.size = frame->size - samples_count*2;
		    status = (*stream->codec->op->recover)(stream->codec,
							   frame_out.size,
							   &frame_out);
		    if (status != PJ_SUCCESS)
			break;
		    samples_count += samples_per_frame;

		} while (samples_count < samples_required);

		if (stream->jb_last_frm != frame_type) {
		    PJ_LOG(5,(stream->port.info.name.ptr, 
			      "Jitter buffer is bufferring with plc (prefetch=%d)",
			      jb_state.prefetch));
		}

	    } 

	    if (samples_count < samples_required) {
		pjmedia_zero_samples(p_out_samp + samples_count,
				     samples_required - samples_count);
		samples_count = samples_required;
		PJ_LOG(5,(stream->port.info.name.ptr, 
			  "Jitter buffer is bufferring (prefetch=%d)..", 
			  jb_state.prefetch));
	    }

	    stream->jb_last_frm = frame_type;
	    break;

	} else {
	    /* Got "NORMAL" frame from jitter buffer */
	    pjmedia_frame frame_in, frame_out;

	    /* Decode */
	    frame_in.buf = channel->out_pkt;
	    frame_in.size = stream->frame_size;
	    frame_in.type = PJMEDIA_FRAME_TYPE_AUDIO;  /* ignored */

	    frame_out.buf = p_out_samp + samples_count;
	    frame_out.size = frame->size - samples_count*BYTES_PER_SAMPLE;
	    status = stream->codec->op->decode( stream->codec, &frame_in,
						frame_out.size, &frame_out);
	    if (status != 0) {
		LOGERR_((port->info.name.ptr, "codec decode() error", 
			 status));

		pjmedia_zero_samples(p_out_samp + samples_count, 
				     samples_per_frame);
	    }
	}

	stream->jb_last_frm = frame_type;
    }


    /* Unlock jitter buffer mutex. */
    pj_mutex_unlock( stream->jb_mutex );

    /* Return PJMEDIA_FRAME_TYPE_NONE if we have no frames at all
     * (it can happen when jitter buffer returns PJMEDIA_JB_ZERO_EMPTY_FRAME).
     */
    if (samples_count == 0) {
	frame->type = PJMEDIA_FRAME_TYPE_NONE;
	frame->size = 0;
    } else {
	frame->type = PJMEDIA_FRAME_TYPE_AUDIO;
	frame->size = samples_count * BYTES_PER_SAMPLE;
	frame->timestamp.u64 = 0;
    }

    return PJ_SUCCESS;
}


/*
 * Transmit DTMF
 */
static void create_dtmf_payload(pjmedia_stream *stream, 
			  struct pjmedia_frame *frame_out)
{
    pjmedia_rtp_dtmf_event *event;
    struct dtmf *digit = &stream->tx_dtmf_buf[0];
    unsigned duration;
    pj_uint32_t cur_ts;

    pj_assert(sizeof(pjmedia_rtp_dtmf_event) == 4);

    event = frame_out->buf;
    cur_ts = pj_ntohl(stream->enc->rtp.out_hdr.ts);
    duration = cur_ts - digit->start_ts;

    event->event = (pj_uint8_t)digit->event;
    event->e_vol = 10;
    event->duration = pj_htons((pj_uint16_t)duration);

    if (duration >= PJMEDIA_DTMF_DURATION) {
	event->e_vol |= 0x80;

	/* Prepare next digit. */
	pj_mutex_lock(stream->jb_mutex);
	pj_array_erase(stream->tx_dtmf_buf, sizeof(stream->tx_dtmf_buf[0]),
		       stream->tx_dtmf_count, 0);
	--stream->tx_dtmf_count;

	stream->tx_dtmf_buf[0].start_ts = cur_ts;
	pj_mutex_unlock(stream->jb_mutex);

	if (stream->tx_dtmf_count) {
	    PJ_LOG(5,(stream->port.info.name.ptr,
		      "Sending DTMF digit id %c", 
		      digitmap[stream->tx_dtmf_buf[0].event]));
	}

    } else if (duration == 0) {
	PJ_LOG(5,(stream->port.info.name.ptr, "Sending DTMF digit id %c", 
		  digitmap[digit->event]));
    }


    frame_out->size = 4;
}