rtp_in.c

一个用于智能手机的多媒体库适合S60 WinCE的跨平台开发库

/*
 *			GPAC - Multimedia Framework C SDK
 *
 *			Copyright (c) Jean Le Feuvre 2000-2005 
 *					All rights reserved
 *
 *  This file is part of GPAC / RTP input module
 *
 *  GPAC is free software; you can redistribute it and/or modify
 *  it under the terms of the GNU Lesser General Public License as published by
 *  the Free Software Foundation; either version 2, or (at your option)
 *  any later version.
 *   
 *  GPAC 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 Lesser General Public License for more details.
 *   
 *  You should have received a copy of the GNU Lesser General Public
 *  License along with this library; see the file COPYING.  If not, write to
 *  the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA. 
 *		
 */

#include "rtp_in.h"

static void RT_LoadPrefs(GF_InputService *plug, RTPClient *rtp)
{
	const char *sOpt;

	sOpt = gf_modules_get_option((GF_BaseInterface *)plug, "Streaming", "DefaultPort");
	if (sOpt) {
		rtp->default_port = atoi(sOpt);
	} else {
		rtp->default_port = 554;
	}
	if ((rtp->default_port == 80) || (rtp->default_port == 8080))
		gf_modules_set_option((GF_BaseInterface *)plug, "Streaming", "RTPoverRTSP", "yes");
	
	sOpt = gf_modules_get_option((GF_BaseInterface *)plug, "Streaming", "RTPoverRTSP");
	if (sOpt && !stricmp(sOpt, "yes")) {
		rtp->transport_mode = 1;
	} else if (sOpt && !stricmp(sOpt, "OnlyCritical")) {
		rtp->transport_mode = 2;
	} else {
		rtp->transport_mode = 0;
	}

	/*
		get heneral network config for UDP
	*/
	/*if UDP not available don't try it*/
	sOpt = gf_modules_get_option((GF_BaseInterface *)plug, "Network", "UDPNotAvailable");
	if (!rtp->transport_mode && sOpt && !stricmp(sOpt, "yes")) rtp->transport_mode = 1;
	
	if (!rtp->transport_mode) {
		sOpt = gf_modules_get_option((GF_BaseInterface *)plug, "Network", "UDPTimeout");
		if (sOpt ) {
			rtp->udp_time_out = atoi(sOpt);
		} else {
			rtp->udp_time_out = 10000;
		}
	}
	
	sOpt = gf_modules_get_option((GF_BaseInterface *)plug, "Streaming", "RTSPTimeout");
	if (sOpt ) {
		rtp->time_out = atoi(sOpt);
	} else {
		rtp->time_out = 30000;
	}

	sOpt = gf_modules_get_option((GF_BaseInterface *)plug, "Streaming", "StreamControl");
	if (sOpt && !stricmp(sOpt, "independent")) {
		rtp->stream_control_type = RTSP_CONTROL_INDEPENDENT;
	} else if (sOpt && !stricmp(sOpt, "RTSP2")) {
		rtp->stream_control_type = RTSP_CONTROL_RTSP_V2;
	} else {
		rtp->stream_control_type = RTSP_CONTROL_AGGREGATE;
	}



	/*packet drop emulation*/
	sOpt = gf_modules_get_option((GF_BaseInterface *)plug, "Streaming", "FirstPacketDrop");
	if (sOpt) {
		rtp->first_packet_drop = atoi(sOpt);
	} else {
		rtp->first_packet_drop = 0;
	}
	sOpt = gf_modules_get_option((GF_BaseInterface *)plug, "Streaming", "PacketDropFrequency");
	if (sOpt) {
		rtp->frequency_drop = atoi(sOpt);
	} else {
		rtp->frequency_drop = 0;
	}

//	rtp->handle_announce = 0;
}

static void RP_cleanup(RTPClient *rtp)
{
	RTSPSession *sess;

	while (gf_list_count(rtp->channels)) {
		RTPStream *ch = (RTPStream *)gf_list_get(rtp->channels, 0);
		gf_list_rem(rtp->channels, 0);
		RP_DeleteStream(ch);
	}

	while ( (sess = (RTSPSession *)gf_list_last(rtp->sessions)) ) {
		gf_list_rem_last(rtp->sessions);
		RP_DelSession(sess);
	}

	if (rtp->session_desc) gf_odf_desc_del(rtp->session_desc);
	rtp->session_desc = NULL;

	if (rtp->sdp_temp) {
		free(rtp->sdp_temp->remote_url);
		free(rtp->sdp_temp);
	}
	rtp->sdp_temp = NULL;
}

u32 RP_Thread(void *param)
{
	u32 i;
	GF_NetworkCommand com;
	RTSPSession *sess;
	RTPStream *ch;
	RTPClient *rtp = (RTPClient *)param;

	rtp->th_state = 1;
	com.command_type = GF_NET_CHAN_BUFFER_QUERY;
	while (rtp->th_state) {
		gf_mx_p(rtp->mx);

		/*fecth data on udp*/
		i=0;
		while ((ch = (RTPStream *)gf_list_enum(rtp->channels, &i))) {
			if ((ch->flags & RTP_EOS) || (ch->status!=RTP_Running) ) continue;
			/*for interleaved channels don't read too fast, query the buffer occupancy*/
			if (ch->flags & RTP_INTERLEAVED) {
				com.base.on_channel = ch->channel;
				gf_term_on_command(rtp->service, &com, GF_OK);
				/*if no buffering, use a default value (3 sec of data should do it)*/
				if (!com.buffer.max) com.buffer.max = 3000;
				if (com.buffer.occupancy <= com.buffer.max) ch->rtsp->flags |= RTSP_TCP_FLUSH;
			} else {
				RP_ReadStream(ch);
			}
		}
		
		/*and process commands / flush TCP*/
		i=0;
		while ((sess = (RTSPSession *)gf_list_enum(rtp->sessions, &i))) {
			RP_ProcessCommands(sess);
		}

		gf_mx_v(rtp->mx);

		gf_sleep(1);
	}

	if (rtp->dnload) gf_term_download_del(rtp->dnload);
	rtp->dnload = NULL;

	rtp->th_state = 2;
	return 0;
}


static Bool RP_CanHandleURL(GF_InputService *plug, const char *url)
{
	char *sExt = strrchr(url, '.');

	if (sExt && gf_term_check_extension(plug, "application/sdp", "sdp", "OnDemand Media/Multicast Session", sExt)) return 1;

	/*local */
	if (strstr(url, "data:application/sdp")) return 1;
	/*embedded data*/
	if (strstr(url, "data:application/mpeg4-od-au;base64") || 
		strstr(url, "data:application/mpeg4-bifs-au;base64") ||
		strstr(url, "data:application/mpeg4-es-au;base64")) return 1;

	/*we need rtsp/tcp , rtsp/udp or direct RTP sender (no control)*/
	if (!strnicmp(url, "rtsp://", 7) || !strnicmp(url, "rtspu://", 8) || !strnicmp(url, "rtp://", 6)) return 1;
	/*we don't check extensions*/
	return 0;
}

static GF_Err RP_ConnectService(GF_InputService *plug, GF_ClientService *serv, const char *url)
{
	RTSPSession *sess;
	RTPClient *priv = (RTPClient *)plug->priv;

	/*store user address*/
	priv->service = serv;

	if (priv->dnload) gf_term_download_del(priv->dnload);
	priv->dnload = NULL;

	GF_LOG(GF_LOG_DEBUG, GF_LOG_SERVICE, ("[RTP] Opening service %s\n", url));

	/*load preferences*/
	RT_LoadPrefs(plug, priv);

	/*start thread*/
	gf_th_run(priv->th, RP_Thread, priv);

	/*local or remote SDP*/
	if (strstr(url, "data:application/sdp") || (strnicmp(url, "rtsp", 4) && strstr(url, ".sdp")) ) {
		RP_FetchSDP(plug, (char *) url, NULL);
	}
	/*rtsp and rtsp over udp*/
	else if (!strnicmp(url, "rtsp://", 7) || !strnicmp(url, "rtspu://", 8)) {
		char *the_url = strdup(url);
		char *the_ext = strrchr(the_url, '#');
		if (the_ext) {
			if (!stricmp(the_ext, "#audio")) priv->media_type = GF_MEDIA_OBJECT_AUDIO;
			else if (!stricmp(the_ext, "#video")) priv->media_type = GF_MEDIA_OBJECT_VIDEO;
			the_ext[0] = 0;
		}
		sess = RP_NewSession(priv, (char *) the_url);
		free(the_url);
		if (!sess) {
			gf_term_on_connect(serv, NULL, GF_NOT_SUPPORTED);
		} else {
			RP_Describe(sess, 0, NULL);
		}
	}
	/*direct RTP (no control) or embedded data - this means the service is attached to a single channel (no IOD)
	reply right away*/
	else {
		gf_term_on_connect(serv, NULL, GF_OK);
		RP_SetupObjects(priv);
	}
	return GF_OK;
}

static void RP_FlushCommands(RTPClient *rtp)
{
	u32 i, nb_com;
	RTSPSession *sess;
	/*process teardown on all sessions*/
	while (1) {
		nb_com = 0;
		i=0;
		while ((sess = (RTSPSession *)gf_list_enum(rtp->sessions, &i))) {
			nb_com += gf_list_count(sess->rtsp_commands);
		}
		if (!nb_com) break;
		gf_sleep(10);
	}
}

static GF_Err RP_CloseService(GF_InputService *plug)
{
	u32 i;
	RTSPSession *sess;
	RTPClient *rtp = (RTPClient *)plug->priv;
	GF_LOG(GF_LOG_DEBUG, GF_LOG_SERVICE, ("[RTP] Closing service\n"));

	RP_FlushCommands(rtp);
	/*send teardown on all sessions*/
	i=0;
	while ((sess = (RTSPSession *)gf_list_enum(rtp->sessions, &i))) {
		RP_Teardown(sess, NULL);
	}
	RP_FlushCommands(rtp);

	/*shutdown thread*/
	if (rtp->th_state==1) rtp->th_state = 0;

	/*confirm close*/
	gf_term_on_disconnect(rtp->service, NULL, GF_OK);
	return GF_OK;
}

static GF_Descriptor *RP_GetServiceDesc(GF_InputService *plug, u32 expect_type, const char *sub_url)
{
	GF_Descriptor *desc;
	RTPClient *priv = (RTPClient *)plug->priv;

	GF_LOG(GF_LOG_DEBUG, GF_LOG_SERVICE, ("[RTP] Fetching service descriptor\n"));
	if ((expect_type!=GF_MEDIA_OBJECT_UNDEF) && (expect_type!=GF_MEDIA_OBJECT_SCENE)) {
		/*ignore the SDP IOD and regenerate one*/
		if (priv->session_desc) gf_odf_desc_del(priv->session_desc);
		priv->session_desc = NULL;
		priv->media_type = expect_type;
		return RP_EmulateIOD(priv, sub_url);
	}

	desc = priv->session_desc;
	priv->session_desc = NULL;
	return desc;
}

static GF_Err RP_ConnectChannel(GF_InputService *plug, LPNETCHANNEL channel, const char *url, Bool upstream)
{
	u32 ESID;
	RTPStream *ch;
	RTSPSession *sess;
	char *es_url;
	RTPClient *priv = (RTPClient *)plug->priv;
	if (upstream) return GF_NOT_SUPPORTED;


	GF_LOG(GF_LOG_DEBUG, GF_LOG_SERVICE, ("[RTP] Connecting channel @%08x - %s\n", channel, url));

	ch = RP_FindChannel(priv, channel, 0, (char *) url, 0);
	if (ch && (ch->status != RTP_Disconnected) ) return GF_SERVICE_ERROR;

	es_url = NULL;
	sess = NULL;
	if (strstr(url, "ES_ID=")) {
		sscanf(url, "ES_ID=%d", &ESID);
		/*first case: simple URL (same namespace)*/
		ch = RP_FindChannel(priv, NULL, ESID, NULL, 0);
		/*this should not happen, the sdp must describe all streams in the service*/
		if (!ch) return GF_STREAM_NOT_FOUND;
		
		/*assign app channel*/
		ch->channel = channel;
		sess = ch->rtsp;
	}
	/*rtsp url - create a session if needed*/
	else if (!strnicmp(url, "rtsp://", 7) || !strnicmp(url, "rtspu://", 8)) {
		sess = RP_CheckSession(priv, (char *) url);
		if (!sess) sess = RP_NewSession(priv, (char *) url);
		es_url = (char *) url;
	}