Mercurial > mplayer.hg

changeset 9565:e74916774667

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Improved RTP packet buffering, by relying on the underlying OS's UDP socket buffering. Improve A/V sync by dropping packets when one stream gets too far behind the other. Now tries to figure out the video frame rate automatically (if "-fps" is not used). Added support for MPEG-4 Elementary Stream video and MPEG-4 Generic audio RTP streams.
author rsf
date Tue, 11 Mar 2003 19:08:31 +0000
parents 898e3692ca0d
children 015b404023f5
files libmpdemux/demux_rtp.cpp libmpdemux/demux_rtp_codec.cpp libmpdemux/demux_rtp_internal.h
diffstat 3 files changed, 348 insertions(+), 233 deletions(-) [+]
line wrap: on
line diff
--- a/libmpdemux/demux_rtp.cpp	Tue Mar 11 19:03:31 2003 +0000
+++ b/libmpdemux/demux_rtp.cpp	Tue Mar 11 19:08:31 2003 +0000
@@ -9,6 +9,7 @@
 
 #include "BasicUsageEnvironment.hh"
 #include "liveMedia.hh"
+#include "GroupsockHelper.hh"
 #include <unistd.h>
 
 extern "C" stream_t* stream_open_sdp(int fd, off_t fileSize,
@@ -43,41 +44,38 @@
   return 0;
 }
 
-// A data structure representing a buffer being read:
-class ReadBufferQueue; // forward
-class ReadBuffer {
-public:
-  ReadBuffer(ReadBufferQueue* ourQueue, demux_packet_t* dp);
-  virtual ~ReadBuffer();
-  Boolean enqueue();
-
-  demux_packet_t* dp() const { return fDP; }
-  ReadBufferQueue* ourQueue() { return fOurQueue; }
-
-  ReadBuffer* next;
-private:
-  demux_packet_t* fDP;
-  ReadBufferQueue* fOurQueue;
-};
-
+// A data structure representing input data for each stream:
 class ReadBufferQueue {
 public:
   ReadBufferQueue(MediaSubsession* subsession, demuxer_t* demuxer,
 		  char const* tag);
   virtual ~ReadBufferQueue();
 
-  ReadBuffer* dequeue();
-
   FramedSource* readSource() const { return fReadSource; }
   RTPSource* rtpSource() const { return fRTPSource; }
   demuxer_t* ourDemuxer() const { return fOurDemuxer; }
   char const* tag() const { return fTag; }
 
-  ReadBuffer* head;
-  ReadBuffer* tail;
   char blockingFlag; // used to implement synchronous reads
-  unsigned counter; // used for debugging
+
+  // For A/V synchronization:
+  Boolean prevPacketWasSynchronized;
+  float prevPacketPTS;
+  ReadBufferQueue** otherQueue;
+
+  // The 'queue' actually consists of just a single "demux_packet_t"
+  // (because the underlying OS does the actual queueing/buffering):
+  demux_packet_t* dp;
+
+  // However, we sometimes inspect buffers before delivering them.
+  // For this, we maintain a queue of pending buffers:
+  void savePendingBuffer(demux_packet_t* dp);
+  demux_packet_t* getPendingBuffer();
+
 private:
+  demux_packet_t* pendingDPHead;
+  demux_packet_t* pendingDPTail;
+
   FramedSource* fReadSource;
   RTPSource* fRTPSource;
   demuxer_t* fOurDemuxer;
@@ -99,10 +97,6 @@
 int rtspStreamOverTCP = 0; 
 
 extern "C" void demux_open_rtp(demuxer_t* demuxer) {
-  if (rtspStreamOverTCP && LIVEMEDIA_LIBRARY_VERSION_INT < 1033689600) {
-    fprintf(stderr, "TCP streaming of RTP/RTCP requires \"LIVE.COM Streaming Media\" library version 2002.10.04 or later - ignoring the \"-rtsp-stream-over-tcp\" flag\n");
-    rtspStreamOverTCP = 0;
-  }
   do {
     TaskScheduler* scheduler = BasicTaskScheduler::createNew();
     if (scheduler == NULL) break;
@@ -110,7 +104,6 @@
     if (env == NULL) break;
 
     RTSPClient* rtspClient = NULL;
-    unsigned flags = 0;
 
     if (demuxer == NULL || demuxer->stream == NULL) break;  // shouldn't happen
     demuxer->stream->eof = 0; // just in case 
@@ -120,7 +113,7 @@
     char* sdpDescription = (char*)(demuxer->stream->priv);
     if (sdpDescription == NULL) {
       // We weren't given a SDP description directly, so assume that
-      // we were give a RTSP URL
+      // we were given a RTSP URL:
       char const* url = demuxer->stream->streaming_ctrl->url->url;
 
       extern int verbose;
@@ -151,19 +144,20 @@
     rtpState->rtspClient = rtspClient;
     rtpState->mediaSession = mediaSession;
     rtpState->audioBufferQueue = rtpState->videoBufferQueue = NULL;
+    rtpState->flags = 0;
     rtpState->firstSyncTime.tv_sec = rtpState->firstSyncTime.tv_usec = 0;
     demuxer->priv = rtpState;
 
     // Create RTP receivers (sources) for each subsession:
     MediaSubsessionIterator iter(*mediaSession);
     MediaSubsession* subsession;
-    unsigned streamType = 0; // 0 => video; 1 => audio
+    unsigned desiredReceiveBufferSize;
     while ((subsession = iter.next()) != NULL) {
       // Ignore any subsession that's not audio or video:
       if (strcmp(subsession->mediumName(), "audio") == 0) {
-	streamType = 1;
+	desiredReceiveBufferSize = 100000;
       } else if (strcmp(subsession->mediumName(), "video") == 0) {
-	streamType = 0;
+	desiredReceiveBufferSize = 2000000;
       } else {
 	continue;
       }
@@ -173,27 +167,52 @@
       } else {
 	fprintf(stderr, "Initiated \"%s/%s\" RTP subsession\n", subsession->mediumName(), subsession->codecName());
 
-	if (rtspClient != NULL) {
-	  // Issue RTSP "SETUP" and "PLAY" commands on the chosen subsession:
-	  if (!rtspClient->setupMediaSubsession(*subsession, False,
-						rtspStreamOverTCP)) break;
-	  if (!rtspClient->playMediaSubsession(*subsession)) break;
+	// Set the OS's socket receive buffer sufficiently large to avoid
+	// incoming packets getting dropped between successive reads from this
+	// subsession's demuxer.  Depending on the bitrate(s) that you expect,
+	// you may wish to tweak the "desiredReceiveBufferSize" values above.
+	int rtpSocketNum = subsession->rtpSource()->RTPgs()->socketNum();
+	int receiveBufferSize
+	  = increaseReceiveBufferTo(*env, rtpSocketNum,
+				    desiredReceiveBufferSize);
+	if (verbose > 0) {
+	  fprintf(stderr, "Increased %s socket receive buffer to %d bytes \n",
+		  subsession->mediumName(), receiveBufferSize);
 	}
 
-	// Now that the subsession is ready to be read, do additional
-	// MPlayer codec-specific initialization on it:
-	if (streamType == 0) { // video
-	  rtpState->videoBufferQueue
-	    = new ReadBufferQueue(subsession, demuxer, "video");
-	  rtpCodecInitialize_video(demuxer, subsession, flags);
-	} else { // audio
-	  rtpState->audioBufferQueue
-	    = new ReadBufferQueue(subsession, demuxer, "audio");
-	  rtpCodecInitialize_audio(demuxer, subsession, flags);
+	if (rtspClient != NULL) {
+	  // Issue a RTSP "SETUP" command on the chosen subsession:
+	  if (!rtspClient->setupMediaSubsession(*subsession, False,
+						rtspStreamOverTCP)) break;
 	}
       }
     }
-    rtpState->flags = flags;
+
+    if (rtspClient != NULL) {
+      // Issue a RTSP aggregate "PLAY" command on the whole session:
+      if (!rtspClient->playMediaSession(*mediaSession)) break;
+    }
+
+    // Now that the session is ready to be read, do additional
+    // MPlayer codec-specific initialization on each subsession:
+    iter.reset();
+    while ((subsession = iter.next()) != NULL) {
+      if (subsession->readSource() == NULL) continue; // not reading this
+
+      unsigned flags = 0;
+      if (strcmp(subsession->mediumName(), "audio") == 0) {
+	rtpState->audioBufferQueue
+	  = new ReadBufferQueue(subsession, demuxer, "audio");
+	rtpState->audioBufferQueue->otherQueue = &(rtpState->videoBufferQueue);
+	rtpCodecInitialize_audio(demuxer, subsession, flags);
+      } else if (strcmp(subsession->mediumName(), "video") == 0) {
+	rtpState->videoBufferQueue
+	  = new ReadBufferQueue(subsession, demuxer, "video");
+	rtpState->videoBufferQueue->otherQueue = &(rtpState->audioBufferQueue);
+	rtpCodecInitialize_video(demuxer, subsession, flags);
+      }
+      rtpState->flags |= flags;
+    }
   } while (0);
 }
 
@@ -201,11 +220,12 @@
   // Get the RTP state that was stored in the demuxer's 'priv' field:
   RTPState* rtpState = (RTPState*)(demuxer->priv);
 
-  return (rtpState->flags&RTPSTATE_IS_MPEG) != 0;
+  return (rtpState->flags&RTPSTATE_IS_MPEG12_VIDEO) != 0;
 }
 
-static ReadBuffer* getBuffer(ReadBufferQueue* bufferQueue,
-			     demuxer_t* demuxer); // forward
+static demux_packet_t* getBuffer(demuxer_t* demuxer, demux_stream_t* ds,
+				 Boolean mustGetNewData,
+				 float& ptsBehind); // forward
 
 extern "C" int demux_rtp_fill_buffer(demuxer_t* demuxer, demux_stream_t* ds) {
   // Get a filled-in "demux_packet" from the RTP source, and deliver it.
@@ -213,7 +233,51 @@
   // to block in the (hopefully infrequent) case where no packet is
   // immediately available.
 
-  // Begin by finding the buffer queue that we want to read from:
+  while (1) {
+    float ptsBehind;
+    demux_packet_t* dp = getBuffer(demuxer, ds, False, ptsBehind); // blocking
+    if (dp == NULL) return 0;
+
+    if (demuxer->stream->eof) return 0; // source stream has closed down
+  
+    // Before using this packet, check to make sure that its presentation
+    // time is not far behind the other stream (if any).  If it is,
+    // then we discard this packet, and get another instead.  (The rest of
+    // MPlayer doesn't always do a good job of synchronizing when the
+    // audio and video streams get this far apart.)
+    // (We don't do this when streaming over TCP, because then the audio and
+    // video streams are interleaved.)
+    const float ptsBehindThreshold = 1.0; // seconds
+    if (ptsBehind < ptsBehindThreshold || rtspStreamOverTCP) { // packet's OK
+      ds_add_packet(ds, dp);
+      break;
+    }
+    
+    free_demux_packet(dp); // give back this packet, and get another one
+  }
+
+  return 1;
+}
+
+Boolean awaitRTPPacket(demuxer_t* demuxer, demux_stream_t* ds,
+		       unsigned char*& packetData, unsigned& packetDataLen,
+		       float& pts) {
+  // Similar to "demux_rtp_fill_buffer()", except that the "demux_packet"
+  // is not delivered to the "demux_stream".
+  float ptsBehind;
+  demux_packet_t* dp = getBuffer(demuxer, ds, True, ptsBehind); // blocking
+  if (dp == NULL) return False;
+
+  packetData = dp->buffer;
+  packetDataLen = dp->len;
+  pts = dp->pts;
+
+  return True;
+}
+
+Boolean insertRTPData(demuxer_t* demuxer, demux_stream_t* ds,
+		      unsigned char* data, unsigned dataLen) {
+  // Begin by finding the buffer queue that we want to add data to.
   // (Get this from the RTP state, which we stored in
   //  the demuxer's 'priv' field)
   RTPState* rtpState = (RTPState*)(demuxer->priv);
@@ -223,54 +287,23 @@
   } else if (ds == demuxer->audio) {
     bufferQueue = rtpState->audioBufferQueue;
   } else {
-    fprintf(stderr, "demux_rtp_fill_buffer: internal error: unknown stream\n");
-    return 0;
-  }
-
-  if (bufferQueue == NULL || bufferQueue->readSource() == NULL) {
-    fprintf(stderr, "demux_rtp_fill_buffer failed: no appropriate RTP subsession has been set up\n");
-    return 0;
-  }
-  
-  ReadBuffer* readBuffer = getBuffer(bufferQueue, demuxer); // blocking
-  if (readBuffer != NULL) ds_add_packet(ds, readBuffer->dp());
-
-  if (demuxer->stream->eof) return 0; // source stream has closed down
-
-  return 1;
-}
-
-Boolean awaitRTPPacket(demuxer_t* demuxer, unsigned streamType,
-		       unsigned char*& packetData, unsigned& packetDataLen) {
-  // Begin by finding the buffer queue that we want to read from:
-  // (Get this from the RTP state, which we stored in
-  //  the demuxer's 'priv' field)
-  RTPState* rtpState = (RTPState*)(demuxer->priv);
-  ReadBufferQueue* bufferQueue = NULL;
-  if (streamType == 0) {
-    bufferQueue = rtpState->videoBufferQueue;
-  } else if (streamType == 1) {
-    bufferQueue = rtpState->audioBufferQueue;
-  } else {
-    fprintf(stderr, "awaitRTPPacket: internal error: unknown streamType %d\n",
-	    streamType);
+    fprintf(stderr, "(demux_rtp)insertRTPData: internal error: unknown stream\n");
     return False;
   }
 
-  if (bufferQueue == NULL || bufferQueue->readSource() == NULL) {
-    fprintf(stderr, "awaitRTPPacket failed: no appropriate RTP subsession has been set up\n");
-    return False;
-  }
-  
-  ReadBuffer* readBuffer = getBuffer(bufferQueue, demuxer); // blocking
-  if (readBuffer == NULL) return False;
+  if (data == NULL || dataLen == 0) return False;
+
+  demux_packet_t* dp = new_demux_packet(dataLen);
+  if (dp == NULL) return False;
 
-  demux_packet_t* dp = readBuffer->dp();
-  packetData = dp->buffer;
-  packetDataLen = dp->len;
+  // Copy our data into the buffer, and save it:
+  memmove(dp->buffer, data, dataLen);
+  dp->len = dataLen;
+  dp->pts = 0;
+  bufferQueue->savePendingBuffer(dp);
+}
 
-  return True;
-}
+static void teardownRTSPSession(RTPState* rtpState); // forward
 
 extern "C" void demux_close_rtp(demuxer_t* demuxer) {
   // Reclaim all RTP-related state:
@@ -278,6 +311,9 @@
   // Get the RTP state that was stored in the demuxer's 'priv' field:
   RTPState* rtpState = (RTPState*)(demuxer->priv);
   if (rtpState == NULL) return;
+
+  teardownRTSPSession(rtpState);
+
   UsageEnvironment* env = NULL;
   TaskScheduler* scheduler = NULL;
   if (rtpState->mediaSession != NULL) {
@@ -296,76 +332,65 @@
 
 ////////// Extra routines that help implement the above interface functions:
 
-static void afterReading(void* clientData, unsigned frameSize,
-			 struct timeval presentationTime); // forward
-static void onSourceClosure(void* clientData); // forward
-
-static void scheduleNewBufferRead(ReadBufferQueue* bufferQueue) {
-  if (bufferQueue->readSource()->isCurrentlyAwaitingData()) return;
-      // a read from this source is already in progress
-
-  // Allocate a new packet buffer, and arrange to read into it:
-  unsigned const bufferSize = 30000; // >= the largest conceivable RTP packet
-  demux_packet_t* dp = new_demux_packet(bufferSize);
-  if (dp == NULL) return;
-  ReadBuffer* readBuffer = new ReadBuffer(bufferQueue, dp);
-
-  // Schedule the read operation:
-  bufferQueue->readSource()->getNextFrame(dp->buffer, bufferSize,
-					  afterReading, readBuffer,
-					  onSourceClosure, readBuffer);
-}
+#define MAX_RTP_FRAME_SIZE 50000
+    // >= the largest conceivable frame composed from one or more RTP packets
 
 static void afterReading(void* clientData, unsigned frameSize,
 			 struct timeval presentationTime) {
-  ReadBuffer* readBuffer = (ReadBuffer*)clientData;
-  ReadBufferQueue* bufferQueue = readBuffer->ourQueue();
+  if (frameSize >= MAX_RTP_FRAME_SIZE) {
+    fprintf(stderr, "Saw an input frame too large (>=%d).  Increase MAX_RTP_FRAME_SIZE in \"demux_rtp.cpp\".\n",
+	    MAX_RTP_FRAME_SIZE);
+  }
+  ReadBufferQueue* bufferQueue = (ReadBufferQueue*)clientData;
   demuxer_t* demuxer = bufferQueue->ourDemuxer();
   RTPState* rtpState = (RTPState*)(demuxer->priv);
 
   if (frameSize > 0) demuxer->stream->eof = 0;
 
-  demux_packet_t* dp = readBuffer->dp();
+  demux_packet_t* dp = bufferQueue->dp;
   dp->len = frameSize;
 
   // Set the packet's presentation time stamp, depending on whether or
   // not our RTP source's timestamps have been synchronized yet: 
-  {
-    Boolean hasBeenSynchronized
-      = bufferQueue->rtpSource()->hasBeenSynchronizedUsingRTCP();
-    if (hasBeenSynchronized) {
-      struct timeval* fst = &(rtpState->firstSyncTime); // abbrev
-      if (fst->tv_sec == 0 && fst->tv_usec == 0) {
-	*fst = presentationTime;
-      }
+  Boolean hasBeenSynchronized
+    = bufferQueue->rtpSource()->hasBeenSynchronizedUsingRTCP();
+  if (hasBeenSynchronized) {
+    if (verbose > 0 && !bufferQueue->prevPacketWasSynchronized) {
+      fprintf(stderr, "%s stream has been synchronized using RTCP \n",
+	      bufferQueue->tag());
+    }
 
-      // For the "pts" field, use the time differential from the first
-      // synchronized time, rather than absolute time, in order to avoid
-      // round-off errors when converting to a float:
-      dp->pts = presentationTime.tv_sec - fst->tv_sec
-	+ (presentationTime.tv_usec - fst->tv_usec)/1000000.0;
-    } else {
-      dp->pts = 0.0;
+    struct timeval* fst = &(rtpState->firstSyncTime); // abbrev
+    if (fst->tv_sec == 0 && fst->tv_usec == 0) {
+      *fst = presentationTime;
     }
+    
+    // For the "pts" field, use the time differential from the first
+    // synchronized time, rather than absolute time, in order to avoid
+    // round-off errors when converting to a float:
+    dp->pts = presentationTime.tv_sec - fst->tv_sec
+      + (presentationTime.tv_usec - fst->tv_usec)/1000000.0;
+    bufferQueue->prevPacketPTS = dp->pts;
+  } else {
+    if (verbose > 0 && bufferQueue->prevPacketWasSynchronized) {
+      fprintf(stderr, "%s stream is no longer RTCP-synchronized \n",
+	      bufferQueue->tag());
+    }
+
+    // use the previous packet's "pts" once again:
+    dp->pts = bufferQueue->prevPacketPTS;
   }
+  bufferQueue->prevPacketWasSynchronized = hasBeenSynchronized;
 
   dp->pos = demuxer->filepos;
   demuxer->filepos += frameSize;
-  if (!readBuffer->enqueue()) {
-    // The queue is full, so discard the buffer:
-    delete readBuffer;
-  }
 
   // Signal any pending 'doEventLoop()' call on this queue:
   bufferQueue->blockingFlag = ~0;
-
-  // Finally, arrange to do another read, if appropriate
-  scheduleNewBufferRead(bufferQueue);
 }
 
 static void onSourceClosure(void* clientData) {
-  ReadBuffer* readBuffer = (ReadBuffer*)clientData;
-  ReadBufferQueue* bufferQueue = readBuffer->ourQueue();
+  ReadBufferQueue* bufferQueue = (ReadBufferQueue*)clientData;
   demuxer_t* demuxer = bufferQueue->ourDemuxer();
 
   demuxer->stream->eof = 1;
@@ -374,90 +399,123 @@
   bufferQueue->blockingFlag = ~0;
 }
 
-static ReadBuffer* getBufferIfAvailable(ReadBufferQueue* bufferQueue) {
-  ReadBuffer* readBuffer = bufferQueue->dequeue();
+static demux_packet_t* getBuffer(demuxer_t* demuxer, demux_stream_t* ds,
+				 Boolean mustGetNewData,
+				 float& ptsBehind) {
+  // Begin by finding the buffer queue that we want to read from:
+  // (Get this from the RTP state, which we stored in
+  //  the demuxer's 'priv' field)
+  RTPState* rtpState = (RTPState*)(demuxer->priv);
+  ReadBufferQueue* bufferQueue = NULL;
+  if (ds == demuxer->video) {
+    bufferQueue = rtpState->videoBufferQueue;
+  } else if (ds == demuxer->audio) {
+    bufferQueue = rtpState->audioBufferQueue;
+  } else {
+    fprintf(stderr, "(demux_rtp)getBuffer: internal error: unknown stream\n");
+    return NULL;
+  }
+
+  if (bufferQueue == NULL || bufferQueue->readSource() == NULL) {
+    fprintf(stderr, "(demux_rtp)getBuffer failed: no appropriate RTP subsession has been set up\n");
+    return NULL;
+  }
+  
+  demux_packet_t* dp;
+  if (!mustGetNewData) {
+    // Check whether we have a previously-saved buffer that we can use:
+    dp = bufferQueue->getPendingBuffer();
+    if (dp != NULL) return dp;
+  }
 
-  // Arrange to read a new packet into this queue:
-  scheduleNewBufferRead(bufferQueue);
+  // Allocate a new packet buffer, and arrange to read into it:
+  dp = new_demux_packet(MAX_RTP_FRAME_SIZE);
+  bufferQueue->dp = dp;
+  if (dp == NULL) return NULL;
+
+  // Schedule the read operation:
+  bufferQueue->blockingFlag = 0;
+  bufferQueue->readSource()->getNextFrame(dp->buffer, MAX_RTP_FRAME_SIZE,
+					  afterReading, bufferQueue,
+					  onSourceClosure, bufferQueue);
+  // Block ourselves until data becomes available:
+  TaskScheduler& scheduler
+    = bufferQueue->readSource()->envir().taskScheduler();
+  scheduler.doEventLoop(&bufferQueue->blockingFlag);
 
-  return readBuffer;
+  // Set the "ptsBehind" result parameter:
+  if (bufferQueue->prevPacketPTS != 0.0 && *(bufferQueue->otherQueue) != NULL
+      && (*(bufferQueue->otherQueue))->prevPacketPTS != 0.0) {
+    ptsBehind = (*(bufferQueue->otherQueue))->prevPacketPTS
+		 - bufferQueue->prevPacketPTS;
+  } else {
+    ptsBehind = 0.0;
+  }
+
+  if (mustGetNewData) {
+    // Save this buffer for future reads:
+    bufferQueue->savePendingBuffer(dp);
+  }
+
+  return dp;
 }
 
-static ReadBuffer* getBuffer(ReadBufferQueue* bufferQueue,
-			     demuxer_t* demuxer) {
-  // Check whether there's a full buffer to deliver to the client:
-  bufferQueue->blockingFlag = 0;
-  ReadBuffer* readBuffer;
-  while ((readBuffer = getBufferIfAvailable(bufferQueue)) == NULL
-	 && !demuxer->stream->eof) {
-    // Because we weren't able to deliver a buffer to the client immediately,
-    // block myself until one comes available:
-    TaskScheduler& scheduler
-      = bufferQueue->readSource()->envir().taskScheduler();
-#if USAGEENVIRONMENT_LIBRARY_VERSION_INT >= 1038614400
-    scheduler.doEventLoop(&bufferQueue->blockingFlag);
-#else
-    scheduler.blockMyself(&bufferQueue->blockingFlag);
-#endif
+static void teardownRTSPSession(RTPState* rtpState) {
+  RTSPClient* rtspClient = rtpState->rtspClient;
+  MediaSession* mediaSession = rtpState->mediaSession;
+  if (rtspClient == NULL || mediaSession == NULL) return;
+
+  MediaSubsessionIterator iter(*mediaSession);
+  MediaSubsession* subsession;
+
+  while ((subsession = iter.next()) != NULL) {
+    rtspClient->teardownMediaSubsession(*subsession);
   }
-
-  return readBuffer;
 }
 
 ////////// "ReadBuffer" and "ReadBufferQueue" implementation:
 
-#define MAX_QUEUE_SIZE 5
-
-ReadBuffer::ReadBuffer(ReadBufferQueue* ourQueue, demux_packet_t* dp)
-  : next(NULL), fDP(dp), fOurQueue(ourQueue) {
-}
-
-Boolean ReadBuffer::enqueue() {
-  if (fOurQueue->counter >= MAX_QUEUE_SIZE) {
-    // This queue is full.  Clear out an old entry from it, so that
-    // this new one will fit:
-    while (fOurQueue->counter >= MAX_QUEUE_SIZE) {
-      delete fOurQueue->dequeue();
-    }
-  }
-
-  // Add ourselves to the tail of our queue:
-  if (fOurQueue->tail == NULL) {
-    fOurQueue->head = this;
-  } else {
-    fOurQueue->tail->next = this;
-  }
-  fOurQueue->tail = this;
-  ++fOurQueue->counter;
-
-  return True;
-}
-
-ReadBuffer::~ReadBuffer() {
-  free_demux_packet(fDP);
-  delete next;
-}
-
 ReadBufferQueue::ReadBufferQueue(MediaSubsession* subsession,
 				 demuxer_t* demuxer, char const* tag)
-  : head(NULL), tail(NULL), counter(0),
+  : prevPacketWasSynchronized(False), prevPacketPTS(0.0), otherQueue(NULL),
+    dp(NULL), pendingDPHead(NULL), pendingDPTail(NULL),
     fReadSource(subsession == NULL ? NULL : subsession->readSource()),
     fRTPSource(subsession == NULL ? NULL : subsession->rtpSource()),
     fOurDemuxer(demuxer), fTag(strdup(tag)) {
 } 
 
 ReadBufferQueue::~ReadBufferQueue() {
-  delete head;
   delete fTag;
+
+  // Free any pending buffers (that never got delivered):
+  demux_packet_t* dp = pendingDPHead;
+  while (dp != NULL) {
+    demux_packet_t* dpNext = dp->next;
+    dp->next = NULL;
+    free_demux_packet(dp);
+    dp = dpNext;
+  }
 }
 
-ReadBuffer* ReadBufferQueue::dequeue() {
-  ReadBuffer* readBuffer = head;
-  if (readBuffer != NULL) {
-    head = readBuffer->next;
-    if (head == NULL) tail = NULL; 
-    --counter;
-    readBuffer->next = NULL;
+void ReadBufferQueue::savePendingBuffer(demux_packet_t* dp) {
+  // Keep this buffer around, until MPlayer asks for it later:
+  if (pendingDPTail == NULL) {
+    pendingDPHead = pendingDPTail = dp;
+  } else {
+    pendingDPTail->next = dp;
+    pendingDPTail = dp;
   }
-  return readBuffer;
+  dp->next = NULL;
 }
+
+demux_packet_t* ReadBufferQueue::getPendingBuffer() {
+  demux_packet_t* dp = pendingDPHead;
+  if (dp != NULL) {
+    pendingDPHead = dp->next;
+    if (pendingDPHead == NULL) pendingDPTail = NULL; 
+
+    dp->next = NULL;
+  }
+
+  return dp;
+}
--- a/libmpdemux/demux_rtp_codec.cpp	Tue Mar 11 19:03:31 2003 +0000
+++ b/libmpdemux/demux_rtp_codec.cpp	Tue Mar 11 19:08:31 2003 +0000
@@ -6,6 +6,8 @@
 #include "stheader.h"
 }
 
+static void
+needVideoFrameRate(demuxer_t* demuxer, MediaSubsession* subsession); // forward
 static Boolean
 parseQTState_video(QuickTimeGenericRTPSource::QTState const& qtState,
 		   unsigned& fourcc); // forward
@@ -27,35 +29,38 @@
   demux_stream_t* d_video = demuxer->video;
   d_video->sh = sh_video; sh_video->ds = d_video;
   
-  // If we happen to know the subsession's video frame rate, set it,
-  // so that the user doesn't have to give the "-fps" option instead.
-  int fps = (int)(subsession->videoFPS());
-  if (fps != 0) sh_video->fps = fps;
-  
   // Map known video MIME types to the BITMAPINFOHEADER parameters
   // that this program uses.  (Note that not all types need all
   // of the parameters to be set.)
   if (strcmp(subsession->codecName(), "MPV") == 0 ||
       strcmp(subsession->codecName(), "MP1S") == 0 ||
       strcmp(subsession->codecName(), "MP2T") == 0) {
-    flags |= RTPSTATE_IS_MPEG;
+    flags |= RTPSTATE_IS_MPEG12_VIDEO;
   } else if (strcmp(subsession->codecName(), "H263") == 0 ||
 	     strcmp(subsession->codecName(), "H263-1998") == 0) {
     bih->biCompression = sh_video->format
       = mmioFOURCC('H','2','6','3');
+    needVideoFrameRate(demuxer, subsession);
   } else if (strcmp(subsession->codecName(), "H261") == 0) {
     bih->biCompression = sh_video->format
       = mmioFOURCC('H','2','6','1');
+    needVideoFrameRate(demuxer, subsession);
   } else if (strcmp(subsession->codecName(), "JPEG") == 0) {
     bih->biCompression = sh_video->format
       = mmioFOURCC('M','J','P','G');
-#if (LIVEMEDIA_LIBRARY_VERSION_INT < 1044662400)
-    fprintf(stderr, "WARNING: This video stream might not play correctly.  Please upgrade to version \"2003.02.08\" or later of the \"LIVE.COM Streaming Media\" libraries.\n");
-#endif
+    needVideoFrameRate(demuxer, subsession);
   } else if (strcmp(subsession->codecName(), "MP4V-ES") == 0) {
     bih->biCompression = sh_video->format
       = mmioFOURCC('m','p','4','v');
-    //flags |= RTPSTATE_IS_MPEG; // MPEG hdr checking in video.c doesn't work!
+    // For the codec to work correctly, it may need a 'VOL Header' to be
+    // inserted at the front of the data stream.  Construct this from the
+    // "config" MIME parameter, which was present (hopefully) in the
+    // session's SDP description:
+    unsigned configLen;
+    unsigned char* configData
+      = parseGeneralConfigStr(subsession->fmtp_config(), configLen);
+    insertRTPData(demuxer, demuxer->video, configData, configLen);
+    needVideoFrameRate(demuxer, subsession);
   } else if (strcmp(subsession->codecName(), "X-QT") == 0 ||
 	     strcmp(subsession->codecName(), "X-QUICKTIME") == 0) {
     // QuickTime generic RTP format, as described in
@@ -64,12 +69,13 @@
     // We can't initialize this stream until we've received the first packet
     // that has QuickTime "sdAtom" information in the header.  So, keep
     // reading packets until we get one:
-    unsigned char* packetData; unsigned packetDataLen;
+    unsigned char* packetData; unsigned packetDataLen; float pts;
     QuickTimeGenericRTPSource* qtRTPSource
       = (QuickTimeGenericRTPSource*)(subsession->rtpSource());
     unsigned fourcc;
     do {
-      if (!awaitRTPPacket(demuxer, 0 /*video*/, packetData, packetDataLen)) {
+      if (!awaitRTPPacket(demuxer, demuxer->video,
+			  packetData, packetDataLen, pts)) {
 	return;
       }
     } while (!parseQTState_video(qtRTPSource->qtState, fourcc));
@@ -94,6 +100,8 @@
   demux_stream_t* d_audio = demuxer->audio;
   d_audio->sh = sh_audio; sh_audio->ds = d_audio;
   
+  wf->nChannels = subsession->numChannels();
+
   // Map known audio MIME types to the WAVEFORMATEX parameters
   // that this program uses.  (Note that not all types need all
   // of the parameters to be set.)
@@ -105,44 +113,35 @@
     wf->wFormatTag = sh_audio->format = 0x55;
     // Note: 0x55 is for layer III, but should work for I,II also
     wf->nSamplesPerSec = 0; // sample rate is deduced from the data
-    flags |= RTPSTATE_IS_MPEG;
   } else if (strcmp(subsession->codecName(), "AC3") == 0) {
     wf->wFormatTag = sh_audio->format = 0x2000;
     wf->nSamplesPerSec = 0; // sample rate is deduced from the data
   } else if (strcmp(subsession->codecName(), "PCMU") == 0) {
     wf->wFormatTag = sh_audio->format = 0x7;
-    wf->nChannels = 1;
     wf->nAvgBytesPerSec = 8000;
     wf->nBlockAlign = 1;
     wf->wBitsPerSample = 8;
     wf->cbSize = 0;
   } else if (strcmp(subsession->codecName(), "PCMA") == 0) {
     wf->wFormatTag = sh_audio->format = 0x6;
-    wf->nChannels = 1;
     wf->nAvgBytesPerSec = 8000;
     wf->nBlockAlign = 1;
     wf->wBitsPerSample = 8;
     wf->cbSize = 0;
   } else if (strcmp(subsession->codecName(), "GSM") == 0) {
     wf->wFormatTag = sh_audio->format = mmioFOURCC('a','g','s','m');
-    wf->nChannels = 1;
     wf->nAvgBytesPerSec = 1650;
     wf->nBlockAlign = 33;
     wf->wBitsPerSample = 16;
     wf->cbSize = 0;
   } else if (strcmp(subsession->codecName(), "QCELP") == 0) {
     wf->wFormatTag = sh_audio->format = mmioFOURCC('Q','c','l','p');
-    // The following settings for QCELP don't quite work right #####
-    wf->nChannels = 1;
     wf->nAvgBytesPerSec = 1750;
     wf->nBlockAlign = 35;
     wf->wBitsPerSample = 16;
     wf->cbSize = 0;
   } else if (strcmp(subsession->codecName(), "MP4A-LATM") == 0) {
     wf->wFormatTag = sh_audio->format = mmioFOURCC('m','p','4','a');
-#if (LIVEMEDIA_LIBRARY_VERSION_INT < 1042761600)
-    fprintf(stderr, "WARNING: This audio stream might not play correctly.  Please upgrade to version \"2003.01.17\" or later of the \"LIVE.COM Streaming Media\" libraries.\n");
-#else
     // For the codec to work correctly, it needs "AudioSpecificConfig"
     // data, which is parsed from the "StreamMuxConfig" string that
     // was present (hopefully) in the SDP description:
@@ -151,8 +150,15 @@
       = parseStreamMuxConfigStr(subsession->fmtp_config(),
 				codecdata_len);
     sh_audio->codecdata_len = codecdata_len;
-#endif
-    flags |= RTPSTATE_IS_MPEG;
+  } else if (strcmp(subsession->codecName(), "MPEG4-GENERIC") == 0) {
+    wf->wFormatTag = sh_audio->format = mmioFOURCC('m','p','4','a');
+    // For the codec to work correctly, it needs "AudioSpecificConfig"
+    // data, which was present (hopefully) in the SDP description:
+    unsigned codecdata_len;
+    sh_audio->codecdata
+      = parseGeneralConfigStr(subsession->fmtp_config(),
+			      codecdata_len);
+    sh_audio->codecdata_len = codecdata_len;
   } else if (strcmp(subsession->codecName(), "X-QT") == 0 ||
 	     strcmp(subsession->codecName(), "X-QUICKTIME") == 0) {
     // QuickTime generic RTP format, as described in
@@ -161,12 +167,13 @@
     // We can't initialize this stream until we've received the first packet
     // that has QuickTime "sdAtom" information in the header.  So, keep
     // reading packets until we get one:
-    unsigned char* packetData; unsigned packetDataLen;
+    unsigned char* packetData; unsigned packetDataLen; float pts;
     QuickTimeGenericRTPSource* qtRTPSource
       = (QuickTimeGenericRTPSource*)(subsession->rtpSource());
     unsigned fourcc, numChannels;
     do {
-      if (!awaitRTPPacket(demuxer, 1 /*audio*/, packetData, packetDataLen)) {
+      if (!awaitRTPPacket(demuxer, demuxer->audio,
+			  packetData, packetDataLen, pts)) {
 	return;
       }
     } while (!parseQTState_audio(qtRTPSource->qtState, fourcc, numChannels));
@@ -180,6 +187,47 @@
   }
 }
 
+static void needVideoFrameRate(demuxer_t* demuxer,
+			       MediaSubsession* subsession) {
+  // For some codecs, MPlayer's decoding software can't (or refuses to :-)
+  // figure out the frame rate by itself, so (unless the user specifies
+  // it manually, using "-fps") we figure it out ourselves here, using the
+  // presentation timestamps in successive packets,
+  extern float force_fps; if (force_fps != 0.0) return; // user used "-fps"
+
+  demux_stream_t* d_video = demuxer->video;
+  sh_video_t* sh_video = (sh_video_t*)(demuxer->video->sh);
+
+  // If we already know the subsession's video frame rate, use it:
+  int fps = (int)(subsession->videoFPS());
+  if (fps != 0) {
+    sh_video->fps = fps;
+    return;
+  }
+  
+  // Keep looking at incoming frames until we see two with different,
+  // non-zero "pts" timestamps:
+  unsigned char* packetData; unsigned packetDataLen;
+  float lastPTS = 0.0, curPTS;
+  unsigned const maxNumFramesToWaitFor = 100;
+  for (unsigned i = 0; i < maxNumFramesToWaitFor; ++i) {
+    if (!awaitRTPPacket(demuxer, demuxer->video,
+			packetData, packetDataLen, curPTS)) break;
+
+    if (curPTS > lastPTS && lastPTS != 0.0) {
+      // Use the difference between these two "pts"s to guess the frame rate.
+      // (should really check that there were no missing frames inbetween)#####
+      // Guess the frame rate as an integer.  If it's not, use "-fps" instead.
+      fps = (int)(1/(curPTS-lastPTS) + 0.5); // rounding
+      fprintf(stderr, "demux_rtp: Guessed the video frame rate as %d frames-per-second.\n\t(If this is wrong, use the \"-fps <frame-rate>\" option instead.)\n", fps);
+      sh_video->fps = fps;
+      return;
+    }
+    lastPTS = curPTS;
+  }
+  fprintf(stderr, "demux_rtp: Failed to guess the video frame rate\n");
+}
+
 static Boolean
 parseQTState_video(QuickTimeGenericRTPSource::QTState const& qtState,
 		   unsigned& fourcc) {
--- a/libmpdemux/demux_rtp_internal.h	Tue Mar 11 19:03:31 2003 +0000
+++ b/libmpdemux/demux_rtp_internal.h	Tue Mar 11 19:08:31 2003 +0000
@@ -16,6 +16,10 @@
 #include <liveMedia.hh>
 #endif
 
+#if (LIVEMEDIA_LIBRARY_VERSION_INT < 1046649600)
+#error Please upgrade to version 2003.03.03 or later of the "LIVE.COM Streaming Media" libraries - available from <www.live.com/liveMedia/>
+#endif
+
 // Codec-specific initialization routines:
 void rtpCodecInitialize_video(demuxer_t* demuxer,
 			      MediaSubsession* subsession, unsigned& flags);
@@ -23,14 +27,19 @@
 			      MediaSubsession* subsession, unsigned& flags);
 
 // Flags that may be set by the above routines:
-#define RTPSTATE_IS_MPEG 0x1 // is an MPEG audio, video or transport stream
+#define RTPSTATE_IS_MPEG12_VIDEO 0x1 // is a MPEG-1 or 2 video stream
 
 // A routine to wait for the first packet of a RTP stream to arrive.
 // (For some RTP payload formats, codecs cannot be fully initialized until
 // we've started receiving data.)
-Boolean awaitRTPPacket(demuxer_t* demuxer, unsigned streamType,
-		       unsigned char*& packetData, unsigned& packetDataLen);
+Boolean awaitRTPPacket(demuxer_t* demuxer, demux_stream_t* ds,
+		       unsigned char*& packetData, unsigned& packetDataLen,
+		       float& pts);
     // "streamType": 0 => video; 1 => audio
     // This routine returns False if the input stream has closed
 
+// A routine for adding our own data to an incoming RTP data stream:
+Boolean insertRTPData(demuxer_t* demuxer, demux_stream_t* ds,
+		      unsigned char* data, unsigned dataLen);
+
 #endif