ogg_in.c

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

process_stream:
	/*live insertion (not supported yet, just a reminder)*/
	if (!st->ch) {
		read->resync_stream = st;
		return;
	}

	while (ogg_stream_packetout(&st->os, &oggpacket ) > 0 ) {
		if (oggpacket.granulepos != -1) {
			st->last_granule = oggpacket.granulepos;
		}
		/*stream reinit, don't resend headers*/
		if (st->parse_headers) st->parse_headers--;
		else if (st->map_time) {
			Double t;
			if (read->start_range && (oggpacket.granulepos==-1)) continue;
			t = OGG_GranuleToMediaTime(&st->info, st->last_granule);
			if (t>=read->start_range) {
				GF_NetworkCommand map;
				map.command_type = GF_NET_CHAN_MAP_TIME;
				map.map_time.on_channel = st->ch;
				map.map_time.reset_buffers = (read->start_range>0.2) ? 1 : 0;
				map.map_time.timestamp = st->ogg_ts = 0;
				map.map_time.media_time = t;
				gf_term_on_command(read->service, &map, GF_OK);
				st->map_time = 0;
				OGG_SendPackets(read, st, &oggpacket);
			}
		}
		else {
			OGG_SendPackets(read, st, &oggpacket);
		}
	}
}

static u32 OggDemux(void *par)
{
	GF_NetworkCommand com;
	Bool go;
	u32 i, count;
	OGGStream *st;
	OGGReader *read = (OGGReader *) par;

	read->bos_done = 0;
	memset(&com, 0, sizeof(GF_NetworkCommand));
	com.command_type = GF_NET_CHAN_BUFFER_QUERY;

	if (read->needs_connection) {
		read->needs_connection=0;
		gf_term_on_connect(read->service, NULL, GF_OK);
	}

    ogg_sync_init(&read->oy);


	while (!read->kill_demux) {
		OGG_Process(read);

		if (!read->bos_done) continue;
		
#if 0
		/*idle*/
		while (!read->kill_demux && !read->nb_playing) {
			/*send OD updates*/
			OGG_SendStreams(read);
			gf_sleep(20);
		}
#endif

		/*(re)starting, seek*/
		if (read->do_seek) {
			read->do_seek = 0;
			ogg_sync_clear(&read->oy);
			ogg_sync_init(&read->oy);
//			OGG_SendStreams(read);

			if (read->ogfile) {
				u32 seek_to = 0;
				read->resync_stream = NULL;
				if (read->dur) seek_to = (u32) (read->file_size * (read->start_range/read->dur) * 0.6f);
				if ((s32) seek_to > ftell(read->ogfile) ) {
					fseek(read->ogfile, seek_to, SEEK_SET);
				} else {
					fseek(read->ogfile, 0, SEEK_SET);
				}
			}
		}


		/*sleep untill the buffer occupancy is too low - note that this work because all streams in this
		demuxer are synchronized*/
		go = read->nb_playing;
		while (go && !read->kill_demux) {
			count = gf_list_count(read->streams);
			for (i=0; i<count; i++) {
				st = gf_list_get(read->streams, i);
				if (!st->ch) continue;
				com.base.on_channel = st->ch;
				gf_term_on_command(read->service, &com, GF_OK);
				if (com.buffer.occupancy < read->data_buffer_ms) {
					GF_LOG(GF_LOG_DEBUG, GF_LOG_CONTAINER, ("[OGG] channel %d needs fill (%d ms data, %d max buffer)\n", st->ESID, com.buffer.occupancy, read->data_buffer_ms));
					go = 0;
					break;
				}
			}
			if (!i || !read->nb_playing) break;
			gf_sleep(10);
		}
	}
    ogg_sync_clear(&read->oy);
	read->kill_demux=2;
	return 0;
}

/*get streams & duration*/
Bool OGG_CheckFile(OGGReader *read)
{
	OGGInfo info, the_info;
	ogg_page oggpage;
	ogg_packet oggpacket;
	ogg_stream_state os, the_os;
	u64 max_gran;
	Bool has_stream = 0;
	fseek(read->ogfile, 0, SEEK_SET);

    ogg_sync_init(&read->oy);
	max_gran = 0;
	while (1) {
		if (!OGG_ReadPage(read, &oggpage)) break;

		if (ogg_page_bos(&oggpage)) {
			ogg_stream_init(&os, ogg_page_serialno(&oggpage));
			if (ogg_stream_pagein(&os, &oggpage) >= 0 ) {
				ogg_stream_packetpeek(&os, &oggpacket);
				if (ogg_stream_pagein(&os, &oggpage) >= 0 ) {
					ogg_stream_packetpeek(&os, &oggpacket);
					OGG_GetStreamInfo(&oggpacket, &info); 
				}
				if (!has_stream) {
					has_stream = 1;
					ogg_stream_init(&the_os, ogg_page_serialno(&oggpage));
					the_info = info;
				}
			}
			ogg_stream_clear(&os);
			continue;
		}
		if (has_stream && (ogg_stream_pagein(&the_os, &oggpage) >= 0) ) {
			while (ogg_stream_packetout(&the_os, &oggpacket ) > 0 ) {
				if ((oggpacket.granulepos>=0) && ((u64) oggpacket.granulepos>max_gran) ) {
					max_gran = oggpacket.granulepos;
				}
			}
		}
	}
    ogg_sync_clear(&read->oy);
	read->file_size = ftell(read->ogfile);
	fseek(read->ogfile, 0, SEEK_SET);
	read->dur = 0;
	if (has_stream) {
		ogg_stream_clear(&the_os);
		read->dur = (Double) (s64) OGG_GranuleToTime(&the_info, max_gran);
		if (the_info.sample_rate) read->dur /= the_info.sample_rate;
		else read->dur /= the_info.frame_rate_base;
	}
	return has_stream;
}

static Bool OGG_CanHandleURL(GF_InputService *plug, const char *url)
{
	char *sExt;
	sExt = strrchr(url, '.');
	if (!sExt) return 0;

	if (gf_term_check_extension(plug, "application/ogg", "ogg", "Xiph.org OGG Movie", sExt)) return 1;
	if (gf_term_check_extension(plug, "application/x-ogg", "ogg", "Xiph.org OGG Movie", sExt)) return 1;
	return 0;
}

static Bool ogg_is_local(const char *url)
{
	if (!strnicmp(url, "file://", 7)) return 1;
	if (strstr(url, "://")) return 0;
	return 1;
}

void OGG_NetIO(void *cbk, GF_NETIO_Parameter *param)
{
	OGGReader *read = (OGGReader *) cbk;

	gf_term_download_update_stats(read->dnload);
	
	/*done*/
	if ((param->msg_type==GF_NETIO_DATA_TRANSFERED) && read->ogfile) {
		read->is_remote = 0;
		/*reload file*/
		OGG_CheckFile(read);
		return;
	}
	if (param->error && read->needs_connection) {
		read->needs_connection = 0;
		read->kill_demux = 2;
		gf_term_on_connect(read->service, NULL, param->error);
	}
	/*we never receive data from here since the downloader is not threaded*/
}

void OGG_DownloadFile(GF_InputService *plug, char *url)
{
	OGGReader *read = (OGGReader*) plug->priv;

	read->dnload = gf_term_download_new(read->service, url, GF_NETIO_SESSION_NOT_THREADED, OGG_NetIO, read);
	if (!read->dnload) {
		read->kill_demux=2;
		read->needs_connection = 0;
		gf_term_on_connect(read->service, NULL, GF_NOT_SUPPORTED);
	}
	/*service confirm is done once fetched, but start the demuxer thread*/
	gf_th_run(read->demuxer, OggDemux, read);
}


static GF_Err OGG_ConnectService(GF_InputService *plug, GF_ClientService *serv, const char *url)
{
	char szURL[2048];
	char *ext;
	GF_Err reply;
	OGGReader *read = plug->priv;
	read->service = serv;

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

	read->service_type = 0;
	strcpy(szURL, url);
	ext = strrchr(szURL, '#');
	if (ext) {
		if (!strcmp(ext, "#video")) read->service_type = 1;
		else if (!strcmp(ext, "#audio")) read->service_type = 2;
		ext[0] = 0;
	}

	/*remote fetch*/
	read->is_remote = !ogg_is_local(szURL);
	if (read->is_remote) {
		read->needs_connection = 1;
		OGG_DownloadFile(plug, szURL);
		return GF_OK;
	} else {
		read->ogfile = fopen(szURL, "rb");
		if (!read->ogfile) {
			reply = GF_URL_ERROR;
		} else {
			reply = GF_OK;
			/*init ogg file in local mode*/
			if (!OGG_CheckFile(read)) {
				fclose(read->ogfile);
				reply = GF_NON_COMPLIANT_BITSTREAM;
			} else {
				read->needs_connection = 1;
				/*start the demuxer thread*/
				gf_th_run(read->demuxer, OggDemux, read);
				return GF_OK;
			}
		}
	}
	/*error*/
	read->kill_demux=2;
	gf_term_on_connect(serv, NULL, reply);
	return GF_OK;
}

static GF_Err OGG_CloseService(GF_InputService *plug)
{
	OGGReader *read = plug->priv;
	if (!read->kill_demux) {
		read->kill_demux = 1;
		while (read->kill_demux!=2) gf_sleep(2);
	}

	if (read->ogfile) fclose(read->ogfile);
	read->ogfile = NULL;
	if (read->dnload) gf_term_download_del(read->dnload);
	read->dnload = NULL;
	gf_term_on_disconnect(read->service, NULL, GF_OK);
	return GF_OK;
}

static GF_Descriptor *OGG_GetServiceDesc(GF_InputService *plug, u32 expect_type, const char *sub_url)
{
	u32 i;
	GF_ObjectDescriptor *od;
	OGGStream *st;
	OGGReader *read = plug->priv;
	/*since we don't handle multitrack in ogg yes, we don't need to check sub_url, only use expected type*/

	/*single object*/
	if ((expect_type==GF_MEDIA_OBJECT_AUDIO) || (expect_type==GF_MEDIA_OBJECT_VIDEO)) {
		if ((expect_type==GF_MEDIA_OBJECT_AUDIO) && !read->has_audio) return NULL;
		if ((expect_type==GF_MEDIA_OBJECT_VIDEO) && !read->has_video) return NULL;
		i=0;
		while ((st = gf_list_enum(read->streams, &i))) {
			if ((expect_type==GF_MEDIA_OBJECT_AUDIO) && (st->info.streamType!=GF_STREAM_AUDIO)) continue;
			if ((expect_type==GF_MEDIA_OBJECT_VIDEO) && (st->info.streamType!=GF_STREAM_VISUAL)) continue;
			
			od = OGG_GetOD(st);
			read->is_single_media = 1;
			return (GF_Descriptor *) od;
		}
		/*not supported yet - we need to know what's in the ogg stream for that*/
	}
	read->is_inline = 1;
	return NULL;
}

static GF_Err OGG_ConnectChannel(GF_InputService *plug, LPNETCHANNEL channel, const char *url, Bool upstream)
{
	u32 ES_ID, i;
	GF_Err e;
	OGGStream *st;
	OGGReader *read = plug->priv;

	e = GF_STREAM_NOT_FOUND;
	if (strstr(url, "ES_ID")) {
		sscanf(url, "ES_ID=%d", &ES_ID);
	}
	/*URL setup*/
//	else if (!read->es_ch && OGG_CanHandleURL(plug, url)) ES_ID = 3;

	i=0;
	while ((st = gf_list_enum(read->streams, &i))) {
		if (st->ESID==ES_ID) {
			st->ch = channel;
			e = GF_OK;
			break;
		}
	}
	gf_term_on_connect(read->service, channel, e);
	return e;
}

static GF_Err OGG_DisconnectChannel(GF_InputService *plug, LPNETCHANNEL channel)
{
	GF_Err e;
	OGGStream *st;
	u32 i=0;
	OGGReader *read = plug->priv;

	e = GF_STREAM_NOT_FOUND;
	while ((st = gf_list_enum(read->streams, &i))) {
		if (st->ch==channel) {
			st->ch = NULL;
			e = GF_OK;
			break;
		}
	}
	gf_term_on_disconnect(read->service, channel, e);
	return GF_OK;
}

static GF_Err OGG_ServiceCommand(GF_InputService *plug, GF_NetworkCommand *com)
{
	OGGStream *st;
	u32 i;
	OGGReader *read = plug->priv;

	if (!com->base.on_channel) {
		/*if live session we may cache*/
		if (read->is_live && (com->command_type==GF_NET_IS_CACHABLE)) return GF_OK;
		return GF_NOT_SUPPORTED;
	}
	switch (com->command_type) {
	case GF_NET_CHAN_SET_PULL:
		/*no way to demux streams independently, and we keep OD as dynamic ogfile to handle
		chained streams*/
		return GF_NOT_SUPPORTED;
	case GF_NET_CHAN_INTERACTIVE:
		//live: return GF_NOT_SUPPORTED;
		return GF_OK;
	case GF_NET_CHAN_BUFFER:
		com->buffer.min = com->buffer.max = 0;
		if (read->is_live) com->buffer.max = read->data_buffer_ms;
		return GF_OK;
	case GF_NET_CHAN_SET_PADDING: return GF_NOT_SUPPORTED;

	case GF_NET_CHAN_DURATION:
		com->duration.duration = read->dur;
		return GF_OK;
	case GF_NET_CHAN_PLAY:
		read->start_range = com->play.start_range;
		read->end_range = com->play.end_range;
		i=0;
		while ((st = gf_list_enum(read->streams, &i))) {
			if (st->ch == com->base.on_channel) {
				st->is_running = 1;
				st->map_time = read->dur ? 1 : 0;
				if (!read->nb_playing) read->do_seek = 1;
				read->nb_playing ++;
				break;
			}
		}
		/*recfg duration in case*/
		if (!read->is_remote && read->dur) { 
			GF_NetworkCommand rcfg;
			rcfg.base.on_channel = NULL;
			rcfg.base.command_type = GF_NET_CHAN_DURATION;
			rcfg.duration.duration = read->dur;
			gf_term_on_command(read->service, &rcfg, GF_OK);
		}
		return GF_OK;
	case GF_NET_CHAN_STOP:
		i=0;
		while ((st = gf_list_enum(read->streams, &i))) {
			if (st->ch == com->base.on_channel) {
				st->is_running = 0;
				read->nb_playing --;
				break;
			}
		}
		return GF_OK;
	default:
		return GF_OK;
	}
}

static Bool OGG_CanHandleURLInService(GF_InputService *plug, const char *url)
{
	char szURL[2048], *sep;
	OGGReader *read = (OGGReader *)plug->priv;
	const char *this_url = gf_term_get_service_url(read->service);
	if (!this_url || !url) return 0;

	strcpy(szURL, this_url);
	sep = strrchr(szURL, '#');
	if (sep) sep[0] = 0;

	if ((url[0] != '#') && strnicmp(szURL, url, sizeof(char)*strlen(szURL))) return 0;
	sep = strrchr(url, '#');
	if (!stricmp(sep, "#video") && (read->has_video)) return 1;
	if (!stricmp(sep, "#audio") && (read->has_audio)) return 1;
	return 0;
}

GF_InputService *OGG_LoadDemux()
{
	OGGReader *reader;
	GF_InputService *plug = malloc(sizeof(GF_InputService));
	memset(plug, 0, sizeof(GF_InputService));
	GF_REGISTER_MODULE_INTERFACE(plug, GF_NET_CLIENT_INTERFACE, "GPAC OGG Reader", "gpac distribution")

	plug->CanHandleURL = OGG_CanHandleURL;
	plug->ConnectService = OGG_ConnectService;
	plug->CloseService = OGG_CloseService;
	plug->GetServiceDescriptor = OGG_GetServiceDesc;
	plug->ConnectChannel = OGG_ConnectChannel;
	plug->DisconnectChannel = OGG_DisconnectChannel;
	plug->ServiceCommand = OGG_ServiceCommand;
	plug->CanHandleURLInService = OGG_CanHandleURLInService;

	reader = malloc(sizeof(OGGReader));
	memset(reader, 0, sizeof(OGGReader));
	reader->streams = gf_list_new();
	reader->demuxer = gf_th_new();
	reader->data_buffer_ms = 1000;

	plug->priv = reader;
	return plug;
}

void OGG_DeleteDemux(void *ifce)
{
	GF_InputService *plug = (GF_InputService *) ifce;
	OGGReader *read = plug->priv;
	gf_th_del(read->demuxer);

	/*just in case something went wrong*/
	while (gf_list_count(read->streams)) {
		OGGStream *st = gf_list_get(read->streams, 0);
		gf_list_rem(read->streams, 0);
		ogg_stream_clear(&st->os);
		if (st->dsi) free(st->dsi);
		free(st);
	}
	gf_list_del(read->streams);
	free(read);
	free(plug);
}