rtsp_datasource.cpp

彩信浏览器

// This file is part of Ambulant Player, www.ambulantplayer.org.
//
// Copyright (C) 2003-2007 Stichting CWI, 
// Kruislaan 413, 1098 SJ Amsterdam, The Netherlands.
//
// Ambulant Player 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.1 of the License, or
// (at your option) any later version.
//
// Ambulant Player 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 Ambulant Player; if not, write to the Free Software
// Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA

#include "ambulant/config/config.h"
#include "ambulant/net/rtsp_datasource.h"
#include "ambulant/net/demux_datasource.h"
#include "ambulant/lib/logger.h"
#include "GroupsockHelper.hh"

using namespace ambulant;
using namespace net;

//#define AM_DBG
#ifndef AM_DBG
#define AM_DBG if(0)
#endif

#define MIN_VIDEO_PACKET_SIZE 1024

// Forward declarations
static void after_reading_audio(void* data, unsigned sz, unsigned truncated, 
	struct timeval pts, unsigned duration);
static void  after_reading_video(void* data, unsigned sz, unsigned truncated, 
	struct timeval pts, unsigned duration);
static void on_source_close(void* data);

ambulant::net::rtsp_demux::rtsp_demux(rtsp_context_t* context, timestamp_t clip_begin, timestamp_t clip_end)
:	m_context(context),
	m_clip_begin(clip_begin),
	m_clip_end(clip_end),
	m_clip_begin_set(false)
//,	m_critical_section()
{
	m_context->audio_fmt.parameters = (void*) m_context->audio_codec_name;
	m_context->video_fmt.parameters = (void*) m_context->video_codec_name;
	m_context->vbuffer = (unsigned char*)malloc(20000);
	m_context->vbufferlen = 0;
	AM_DBG lib::logger::get_logger()->debug("ambulant::net::rtsp_demux::rtsp_demux(0x%x)", (void*) this);
	
	if ( m_clip_end < 0 || m_clip_end > m_context->time_left) 
		m_clip_end = m_context->time_left;	
}

ambulant::net::rtsp_demux::~rtsp_demux() {
	AM_DBG lib::logger::get_logger()->debug("ambulant::net::rtsp_demux::~rtsp_demux(0x%x)", (void*) this);
	delete m_context;
}

//#define	DUMMYTASK
#ifdef	DUMMYTASK
/*
[Live-devel] Shutdown of testRTSPonDemandServer
Ross Finlayson finlayson at live555.com
Fri Mar 31 11:01:58 PST 2006

    * Previous message: [Live-devel] Non Blocking doGetNextFrame
    * Messages sorted by: [ date ] [ thread ] [ subject ] [ author ]

>  I need to shutdown my hardware cleanly when the program terminates so I
>have added a watch variable to Eventloop. The variable gets set when a
>signal handler is invoked.
>That should stop the eventloop and then I shutdown my hardware.
>
>This works well when the RTSP server is streaming to clients. However,
>when there are no clients the eventloop does not seem respond to the
>watch variable.
>
>Is this to be expected?

This was unexpected, but not really a bug.  It happens because - when 
the server is sitting around waiting for a request - no 'events' are 
happening (no incoming packets, no delayed tasks), so the server sits 
forever in "select()", and so never gets to check the watch variable.

>  Is there a way to get around this?

Yes.  You can schedule a dummy task (that does nothing) to run 
periodically (e.g., every 100 ms).  This will ensure that the server 
leaves "select()" (to check the watch variable) at least every 100ms

Add the following just before "main()":

static void dummyTask(void* / * clientData * /) {
   // Call this again, after a brief delay:
   int uSecsToDelay = 100000; // 100 ms
   env->taskScheduler().scheduleDelayedTask(uSecsToDelay,
                                            (TaskFunc*)dummyTask, NULL);
}

And then, just before the call to "doEventLoop()", do

   dummyTask(NULL);


	Ross Finlayson
	Live Networks, Inc. (LIVE555.COM)
	<http://www.live555.com/>

-----------------
Also see: http://lists.live555.com/pipermail/live-devel/2006-April/004215.html
*/

static void dummyTask (UsageEnvironment* env /*clientData*/) {
	// Call this again, after a brief delay:
	int uSecsToDelay = 100000; // 100 ms
	env->taskScheduler().scheduleDelayedTask(uSecsToDelay,
						 (TaskFunc*)dummyTask, env);
}
#endif/*DUMMYTASK*/

void 
ambulant::net::rtsp_demux::add_datasink(demux_datasink *parent, int stream_index)
{
	m_critical_section.enter();
	assert(stream_index >= 0 && stream_index < MAX_STREAMS);
	assert(m_context && m_context->sinks && m_context->sinks[stream_index] == 0);
	m_context->sinks[stream_index] = parent;
	m_context->nsinks++;
	m_critical_section.leave();
}

void
ambulant::net::rtsp_demux::remove_datasink(int stream_index)
{
	m_critical_section.enter();
	assert(stream_index >= 0 && stream_index < MAX_STREAMS);
	assert(m_context && m_context->sinks && m_context->sinks[stream_index] != 0);
	if (m_context->sinks[stream_index])
		// signal EOF
		m_context->sinks[stream_index]->data_avail(0, 0, 0);
	m_context->sinks[stream_index] = 0;
	m_context->nsinks--;
	if (m_context->nsinks <= 0) _cancel();
	m_critical_section.leave();
}

rtsp_context_t::~rtsp_context_t()
{
	AM_DBG  lib::logger::get_logger()->debug("ambulant::net::rtsp_context_t::~rtsp_context_t()");
	//Have to tear down session here, so that the server is not left hanging till timeout.
	rtsp_client->teardownMediaSession(*media_session);
	//deleting stuff, in reverse order as created in rtsp_demux::supported()
	if (vbuffer)
		free(vbuffer);
	for (int i=0; i < MAX_STREAMS; i++) {
		if (sinks[i] != NULL)
			delete sinks[i];
	}
	if (media_session) {
		Medium::close(media_session);
	}
	if (sdp)
		free(sdp);
	if (rtsp_client) {
		Medium::close(rtsp_client);
	}
	if (env)
		env->reclaim();
	if (scheduler)
		delete scheduler;
}

rtsp_context_t*
ambulant::net::rtsp_demux::supported(const net::url& url) 
{
	if (url.get_protocol() != "rtsp") return NULL;
	
	rtsp_context_t* context = new rtsp_context_t;
	context->first_sync_time.tv_sec = 0;
	context->first_sync_time.tv_usec = 0;
	context->scheduler = NULL;
	context->env = NULL;
	context->rtsp_client = NULL;
	context->media_session = NULL;
	context->sdp = NULL;
	context->audio_stream = -1;
	context->video_stream = -1;
	context->nstream = 0;
	context->blocking_flag = 0;
	context->audio_packet = NULL;
	context->video_packet = NULL;
	context->last_pts = 0;
	context->audio_codec_name = NULL;
	context->video_codec_name = NULL;
	context->audio_fmt = audio_format("live");
	context->video_fmt = video_format("live");
	context->eof = false;
	context->need_video = true;
	context->need_audio = true;
	context->nsinks = 0;
	context->vbuffer = NULL;
	context->vbufferlen = 0;

	memset(context->sinks, 0, sizeof context->sinks);
	
	// setup the basics.
	context->scheduler = BasicTaskScheduler::createNew();
	if (!context->scheduler) {
		lib::logger::get_logger()->error("ambulant::net::rtsp_demux(net::url& url) failed to create scheduler");
		delete context;
		return NULL;
	}
	
	context->env = BasicUsageEnvironment::createNew(*context->scheduler);
	if (!context->env) {
		lib::logger::get_logger()->error("ambulant::net::rtsp_demux(net::url& url) failed to create UsageEnvironment");
		delete context;
		return NULL;
	}
	// setup a rtp session
	int verbose = 0;
	context->rtsp_client = RTSPClient::createNew(*context->env, verbose, "AmbulantPlayer");
	if (!context->rtsp_client) {
		lib::logger::get_logger()->error("ambulant::net::rtsp_demux(net::url& url) failed to create  a RTSP Client");
		delete context;
		return NULL;
	}
	
	std::string str_url = url.get_url();
	const char* ch_url = str_url.c_str();
	assert(ch_url);
	context->sdp = context->rtsp_client->describeURL(ch_url);
	if (!context->sdp) {
		lib::logger::get_logger()->error("%s: describeURL failed (no server available, not rtsp, or url not found?)", ch_url);
		//lib::logger::get_logger()->error("RTSP Connection Failed");
		delete context;	
		return NULL;
	}
	
	AM_DBG lib::logger::get_logger()->debug("rtsp_demux: describe(\"%s\") -> \"%s\"", ch_url, context->sdp);
	context->media_session = MediaSession::createNew(*context->env, context->sdp);
	if (!context->media_session) {
		lib::logger::get_logger()->error("%s: failed to create a MediaSession, sdp=%s", ch_url, context->sdp);
		//lib::logger::get_logger()->error("RTSP Connection Failed");
		delete context;		
		return NULL;
	}	
	context->duration = context->media_session->playEndTime();
	context->time_left = (timestamp_t) (context->duration*1000000 - 40000); // skip last frame
//	context->time_left = (timestamp_t) (context->duration*1000000); // do not skip last frame
	AM_DBG lib::logger::get_logger()->debug("rtps_demux::supported: time_left = %ld", context->time_left);
	// next set up the rtp subsessions.
	
	unsigned int desired_buf_size;
	MediaSubsession* subsession;
	MediaSubsessionIterator iter(*context->media_session);
	// Only audio/video session need to apply for a job !
	while ((subsession = iter.next()) != NULL) {
		if (!subsession->initiate()) {
			lib::logger::get_logger()->error("ambulant::net::rtsp_demux(net::url& url) failed to initiate subsession for medium \"%s\"", subsession->mediumName());
			//lib::logger::get_logger()->error("RTSP Connection Failed");
			delete context;
			return NULL;
		}
		if (strcmp(subsession->mediumName(), "audio") == 0) {
			desired_buf_size = 100000;
			if (context->audio_stream < 0) {
				context->audio_stream = context->nstream;
				context->audio_codec_name = subsession->codecName();
				AM_DBG lib::logger::get_logger()->debug("ambulant::net::rtsp_demux(net::url& url), audio codecname :%s ",context->audio_codec_name);
				context->audio_fmt.channels = 0; //Let the decoder (ffmpeg) find out the channels subsession->numChannels() returns channels -1 ???
				context->audio_fmt.bits = 16;
				//context->audio_fmt.samplerate = subsession->rtpSource()->timestampFrequency();
				context->audio_fmt.samplerate = 0;
			}
		} else if (strcmp(subsession->mediumName(), "video") == 0) {
			desired_buf_size = 200000;
			if (context->video_stream < 0) {
				context->video_stream = context->nstream;
				context->video_codec_name = subsession->codecName();
				AM_DBG lib::logger::get_logger()->debug("ambulant::net::rtsp_demux(net::url& url), video codecname :%s ",context->video_codec_name);
				//KB getting video_fmt from RTSP doesn't work 
				context->video_fmt.frameduration = (timestamp_t) (1000000.0/subsession->videoFPS());
				context->video_fmt.width = subsession->videoWidth();