rtp_in.c

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

	/*data: url*/
	else 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")
		) {
		
		GF_SAFEALLOC(ch, RTPStream);
		ch->control = strdup(url);
		ch->owner = priv;
		ch->channel = channel;
		ch->status = RTP_Connected;
		/*register*/
		gf_list_add(priv->channels, ch);
		RP_ConfirmChannelConnect(ch, GF_OK);

		return GF_OK;
	}
	
	/*send a DESCRIBE (not a setup) on the channel. If the channel is already created then the
	describe is skipped and a SETUP is sent directly, otherwise the channel is first described then setup*/
	if (sess) RP_Describe(sess, es_url, channel);
	/*otherwise confirm channel connection*/
	else RP_ConfirmChannelConnect(ch, GF_OK);

	return GF_OK;
}


static GF_Err RP_DisconnectChannel(GF_InputService *plug, LPNETCHANNEL channel)
{
	RTPStream *ch;
	RTPClient *priv = (RTPClient *)plug->priv;

	GF_LOG(GF_LOG_DEBUG, GF_LOG_SERVICE, ("[RTP] Disconnecting channel @%08x\n", channel));

	ch = RP_FindChannel(priv, channel, 0, NULL, 0);
	if (!ch) return GF_STREAM_NOT_FOUND;
	gf_mx_p(priv->mx);
	/*disconnect stream BUT DO NOT DELETE IT since we don't store SDP*/
	RP_DisconnectStream(ch);
	gf_mx_v(priv->mx);
	gf_term_on_disconnect(priv->service, channel, GF_OK);
	return GF_OK;
}

static GF_Err RP_ServiceCommand(GF_InputService *plug, GF_NetworkCommand *com)
{
	RTPStream *ch;
	RTPClient *priv = (RTPClient *)plug->priv;

	/*ignore commands other than channels one*/
	if (!com->base.on_channel) {
		if (com->command_type==GF_NET_IS_CACHABLE) return GF_OK;
		return GF_NOT_SUPPORTED;
	}

	ch = RP_FindChannel(priv, com->base.on_channel, 0, NULL, 0);
	if (!ch) return GF_STREAM_NOT_FOUND;

	switch (com->command_type) {
	case GF_NET_CHAN_SET_PULL:
		if (ch->rtp_ch || ch->rtsp || !ch->control) return GF_NOT_SUPPORTED;
		/*embedded channels work in pull mode*/
		if (strstr(ch->control, "data:application/")) return GF_OK;
		return GF_NOT_SUPPORTED;
	case GF_NET_CHAN_INTERACTIVE:
		/*looks like pure RTP / multicast etc, not interactive*/
		if (!ch->control) return GF_NOT_SUPPORTED;
		/*emulated broadcast mode*/
		else if (ch->flags & RTP_FORCE_BROADCAST) return GF_NOT_SUPPORTED;
		/*regular rtsp mode*/
		else if (ch->flags & RTP_HAS_RANGE) return GF_OK;
		/*embedded data*/
		else if (strstr(ch->control, "application")) return GF_OK;
		return GF_NOT_SUPPORTED;
	case GF_NET_CHAN_BUFFER:
		if (!(ch->rtp_ch || ch->rtsp || !ch->control)) {
			com->buffer.max = com->buffer.min = 0;
		} else {
			const char *opt;
			/*amount of buffering in ms*/
			opt = gf_modules_get_option((GF_BaseInterface *)plug, "Network", "BufferLength");
			com->buffer.max = opt ? atoi(opt) : 1000;
			/*rebuffer low limit in ms - if the amount of buffering is less than this, rebuffering will never occur*/
			opt = gf_modules_get_option((GF_BaseInterface *)plug, "Network", "RebufferLength");
			if (opt) com->buffer.min = atoi(opt);
			else com->buffer.min = 500;
			if (com->buffer.min >= com->buffer.max ) com->buffer.min = 0;
		}
		return GF_OK;
	case GF_NET_CHAN_DURATION:
		com->duration.duration = (ch->flags & RTP_HAS_RANGE) ? (ch->range_end - ch->range_start) : 0;
		return GF_OK;
	/*RTP channel config is done upon connection, once the complete SL mapping is known
	however we must store some info not carried in SDP*/
	case GF_NET_CHAN_CONFIG:
		if (com->cfg.frame_duration) ch->depacketizer->sl_hdr.au_duration = com->cfg.frame_duration;
		return GF_OK;

	case GF_NET_CHAN_PLAY:
		GF_LOG(GF_LOG_DEBUG, GF_LOG_SERVICE, ("[RTP] Processing play on channel @%08x - %s\n", ch, ch->rtsp ? "RTSP control" : "No control (RTP)" ));
		/*is this RTSP or direct RTP?*/
		ch->flags &= ~RTP_EOS;
		if (ch->rtsp) {
			RP_UserCommand(ch->rtsp, ch, com);
		} else {
			ch->status = RTP_Running;
			if (ch->rtp_ch) {
				ch->check_rtp_time = 1;
				gf_mx_p(priv->mx);
				RP_InitStream(ch, 0);
				gf_mx_v(priv->mx);
				gf_rtp_set_info_rtp(ch->rtp_ch, 0, 0, 0);
			} else {
				/*direct channel, store current start*/
				ch->current_start = com->play.start_range;
				gf_rtp_depacketizer_reset(ch->depacketizer, 0);
			}
		}
		return GF_OK;
	case GF_NET_CHAN_STOP:
		/*is this RTSP or direct RTP?*/
		if (ch->rtsp) {
			RP_UserCommand(ch->rtsp, ch, com);
		} else {
			ch->status = RTP_Connected;
		}
		return GF_OK;
	case GF_NET_CHAN_SET_SPEED:
	case GF_NET_CHAN_PAUSE:
	case GF_NET_CHAN_RESUME:
		assert(ch->rtsp);
		RP_UserCommand(ch->rtsp, ch, com);
		return GF_OK;

	case GF_NET_CHAN_GET_DSI:
		if (ch->depacketizer->sl_map.configSize) {
			com->get_dsi.dsi_len = ch->depacketizer->sl_map.configSize;
			com->get_dsi.dsi = (char*)malloc(sizeof(char)*com->get_dsi.dsi_len);
			memcpy(com->get_dsi.dsi, ch->depacketizer->sl_map.config, sizeof(char)*com->get_dsi.dsi_len);
		} else {
			com->get_dsi.dsi = NULL;
			com->get_dsi.dsi_len = 0;
		}
		return GF_OK;

	
	case GF_NET_GET_STATS:
		memset(&com->net_stats, 0, sizeof(GF_NetComStats));
		if (ch->rtp_ch) {
			u32 time;
			Float bps;
			com->net_stats.pck_loss_percentage = gf_rtp_get_loss(ch->rtp_ch);
			if (ch->flags & RTP_INTERLEAVED) {
				com->net_stats.multiplex_port = gf_rtsp_get_session_port(ch->rtsp->session);
				com->net_stats.port = gf_rtp_get_low_interleave_id(ch->rtp_ch);
				com->net_stats.ctrl_port = gf_rtp_get_hight_interleave_id(ch->rtp_ch);
			} else {
				com->net_stats.multiplex_port = 0;
				gf_rtp_get_ports(ch->rtp_ch, &com->net_stats.port, &com->net_stats.ctrl_port);
			}
			if (ch->stat_stop_time) {
				time = ch->stat_stop_time - ch->stat_start_time;
			} else {
				time = gf_sys_clock() - ch->stat_start_time;
			}
			bps = 8.0f * ch->rtp_bytes; bps *= 1000; bps /= time; com->net_stats.bw_down = (u32) bps;
			bps = 8.0f * ch->rtcp_bytes; bps *= 1000; bps /= time; com->net_stats.ctrl_bw_down = (u32) bps;
			bps = 8.0f * gf_rtp_get_tcp_bytes_sent(ch->rtp_ch); bps *= 1000; bps /= time; com->net_stats.ctrl_bw_up = (u32) bps;
		}
		return GF_OK;
	}
	return GF_NOT_SUPPORTED;
}

static GF_Err RP_ChannelGetSLP(GF_InputService *plug, LPNETCHANNEL channel, char **out_data_ptr, u32 *out_data_size, GF_SLHeader *out_sl_hdr, Bool *sl_compressed, GF_Err *out_reception_status, Bool *is_new_data)
{
	char *data;
	RTPStream *ch;
	RTPClient *priv = (RTPClient *)plug->priv;

	ch = RP_FindChannel(priv, channel, 0, NULL, 0);
	if (!ch) return GF_STREAM_NOT_FOUND;
	if (ch->rtp_ch || ch->rtsp || !ch->control) return GF_SERVICE_ERROR;
	if (ch->status != RTP_Running) return GF_SERVICE_ERROR;
	data = strstr(ch->control, ";base64");
	if (!data) return GF_SERVICE_ERROR;

	if (ch->current_start>=0) {
		*sl_compressed = 0;
		memset(out_sl_hdr, 0, sizeof(GF_SLHeader));
		out_sl_hdr->accessUnitEndFlag = 1;
		out_sl_hdr->accessUnitStartFlag = 1;
		out_sl_hdr->compositionTimeStamp = (u64) (ch->current_start * 1000);
		out_sl_hdr->compositionTimeStampFlag = 1;
		out_sl_hdr->randomAccessPointFlag = 1;
		*out_reception_status = GF_OK;
		*is_new_data = (ch->flags & GF_RTP_NEW_AU) ? 1 : 0;

		/*decode data*/
		data = strstr(data, ",");
		data += 1;
		*out_data_size = gf_base64_decode(data, strlen(data), ch->buffer, RTP_BUFFER_SIZE);
		/*FIXME - currently only support for empty SL header*/
		*out_data_ptr = ch->buffer;
		ch->flags &= ~GF_RTP_NEW_AU;
	} else {
		*out_data_ptr = NULL;
		*out_data_size = 0;
		*out_reception_status = GF_EOS;
		ch->flags |= RTP_EOS;
	}
	return GF_OK;
}

static GF_Err RP_ChannelReleaseSLP(GF_InputService *plug, LPNETCHANNEL channel)
{
	RTPStream *ch;
	RTPClient *priv = (RTPClient *)plug->priv;

	ch = RP_FindChannel(priv, channel, 0, NULL, 0);
	if (!ch) return GF_STREAM_NOT_FOUND;
	if (ch->rtp_ch || ch->rtsp || !ch->control) return GF_SERVICE_ERROR;
	if (ch->status != RTP_Running) return GF_SERVICE_ERROR;

	/*this will trigger EOS at next fetch*/
	ch->current_start = -1.0;
	return GF_OK;
}

static Bool RP_CanHandleURLInService(GF_InputService *plug, const char *url)
{
	RTSPSession *sess;
	RTPClient *priv = (RTPClient *)plug->priv;

	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;

	if (!RP_CanHandleURL(plug, url)) return 0;
	/*if this URL is part of a running session then ok*/
	sess = RP_CheckSession(priv, (char *) url);
	if (sess) return 1;
	return 0;
}


GF_InputService *RTP_Load()
{
	RTPClient *priv;
	GF_InputService *plug;
	GF_SAFEALLOC(plug, GF_InputService);

	GF_REGISTER_MODULE_INTERFACE(plug, GF_NET_CLIENT_INTERFACE, "GPAC RTP/RTSP Client", "gpac distribution")

	plug->CanHandleURL = RP_CanHandleURL;
	plug->CanHandleURLInService = RP_CanHandleURLInService;
	plug->ConnectService = RP_ConnectService;
	plug->CloseService = RP_CloseService;
	plug->GetServiceDescriptor = RP_GetServiceDesc;
	plug->ConnectChannel = RP_ConnectChannel;
	plug->DisconnectChannel = RP_DisconnectChannel;
	plug->ServiceCommand = RP_ServiceCommand;

	/*PULL mode for embedded streams*/
	plug->ChannelGetSLP = RP_ChannelGetSLP;
	plug->ChannelReleaseSLP = RP_ChannelReleaseSLP;

	GF_SAFEALLOC(priv, RTPClient);
	priv->sessions = gf_list_new();
	priv->channels = gf_list_new();

	plug->priv = priv;
	
	priv->time_out = 30000;
	priv->mx = gf_mx_new();
	priv->th = gf_th_new();
	
	return plug;
}


void RTP_Delete(GF_BaseInterface *bi)
{
	RTPClient *rtp;
	u32 retry;
	GF_InputService *plug = (GF_InputService *) bi;
	rtp = (RTPClient *)plug->priv;

	/*shutdown thread*/
	if (rtp->th_state==1) rtp->th_state = 0;
	retry = 20;
	while ((rtp->th_state==1) && retry) {
		gf_sleep(10);
		retry--;
	}
	assert(retry);

	RP_cleanup(rtp);
	gf_th_del(rtp->th);
	gf_mx_del(rtp->mx);
	gf_list_del(rtp->sessions);
	gf_list_del(rtp->channels);
	free(rtp);
	free(bi);
}

GF_EXPORT
Bool QueryInterface(u32 InterfaceType) 
{
	if (InterfaceType == GF_NET_CLIENT_INTERFACE) return 1;
	return 0;
}

GF_EXPORT
GF_BaseInterface *LoadInterface(u32 InterfaceType) 
{
	if (InterfaceType == GF_NET_CLIENT_INTERFACE) return (GF_BaseInterface *)RTP_Load();
	return NULL;
}

GF_EXPORT
void ShutdownInterface(GF_BaseInterface *ifce)
{
	switch (ifce->InterfaceType) {
	case GF_NET_CLIENT_INTERFACE:
		RTP_Delete(ifce);
		break;
	}
}