libmpdemux/demux_rtp.cpp

   1 ////////// Routines (with C-linkage) that interface between "MPlayer"
   2 ////////// and the "LIVE555 Streaming Media" libraries:
   3
   4 extern "C" {
   5 // on MinGW, we must include windows.h before the things it conflicts
   6 #ifdef __MINGW32__    // with.  they are each protected from
   7 #include <windows.h>  // windows.h, but not the other way around.
   8 #endif
   9 #include "demux_rtp.h"
  10 #include "stheader.h"
  11 #include "options.h"
  12 }
  13 #include "demux_rtp_internal.h"
  14
  15 #include "BasicUsageEnvironment.hh"
  16 #include "liveMedia.hh"
  17 #include "GroupsockHelper.hh"
  18 #include <unistd.h>
  19
  20 // A data structure representing input data for each stream:
  21 class ReadBufferQueue {
  22 public:
  23   ReadBufferQueue(MediaSubsession* subsession, demuxer_t* demuxer,
  24                   char const* tag);
  25   virtual ~ReadBufferQueue();
  26
  27   FramedSource* readSource() const { return fReadSource; }
  28   RTPSource* rtpSource() const { return fRTPSource; }
  29   demuxer_t* ourDemuxer() const { return fOurDemuxer; }
  30   char const* tag() const { return fTag; }
  31
  32   char blockingFlag; // used to implement synchronous reads
  33
  34   // For A/V synchronization:
  35   Boolean prevPacketWasSynchronized;
  36   float prevPacketPTS;
  37   ReadBufferQueue** otherQueue;
  38
  39   // The 'queue' actually consists of just a single "demux_packet_t"
  40   // (because the underlying OS does the actual queueing/buffering):
  41   demux_packet_t* dp;
  42
  43   // However, we sometimes inspect buffers before delivering them.
  44   // For this, we maintain a queue of pending buffers:
  45   void savePendingBuffer(demux_packet_t* dp);
  46   demux_packet_t* getPendingBuffer();
  47
  48   // For H264 over rtsp using AVParser, the next packet has to be saved
  49   demux_packet_t* nextpacket;
  50
  51 private:
  52   demux_packet_t* pendingDPHead;
  53   demux_packet_t* pendingDPTail;
  54
  55   FramedSource* fReadSource;
  56   RTPSource* fRTPSource;
  57   demuxer_t* fOurDemuxer;
  58   char const* fTag; // used for debugging
  59 };
  60
  61 // A structure of RTP-specific state, kept so that we can cleanly
  62 // reclaim it:
  63 typedef struct RTPState {
  64   char const* sdpDescription;
  65   RTSPClient* rtspClient;
  66   SIPClient* sipClient;
  67   MediaSession* mediaSession;
  68   ReadBufferQueue* audioBufferQueue;
  69   ReadBufferQueue* videoBufferQueue;
  70   unsigned flags;
  71   struct timeval firstSyncTime;
  72 };
  73
  74 extern "C" char* network_username;
  75 extern "C" char* network_password;
  76 static char* openURL_rtsp(RTSPClient* client, char const* url) {
  77   // If we were given a user name (and optional password), then use them:
  78   if (network_username != NULL) {
  79     char const* password = network_password == NULL ? "" : network_password;
  80     return client->describeWithPassword(url, network_username, password);
  81   } else {
  82     return client->describeURL(url);
  83   }
  84 }
  85
  86 static char* openURL_sip(SIPClient* client, char const* url) {
  87   // If we were given a user name (and optional password), then use them:
  88   if (network_username != NULL) {
  89     char const* password = network_password == NULL ? "" : network_password;
  90     return client->inviteWithPassword(url, network_username, password);
  91   } else {
  92     return client->invite(url);
  93   }
  94 }
  95
  96 int rtspStreamOverTCP = 0;
  97 extern int rtsp_port;
  98
  99 extern "C" demuxer_t* demux_open_rtp(demuxer_t* demuxer) {
 100   struct MPOpts *opts = demuxer->opts;
 101   Boolean success = False;
 102   do {
 103     TaskScheduler* scheduler = BasicTaskScheduler::createNew();
 104     if (scheduler == NULL) break;
 105     UsageEnvironment* env = BasicUsageEnvironment::createNew(*scheduler);
 106     if (env == NULL) break;
 107
 108     RTSPClient* rtspClient = NULL;
 109     SIPClient* sipClient = NULL;
 110
 111     if (demuxer == NULL || demuxer->stream == NULL) break;  // shouldn't happen
 112     demuxer->stream->eof = 0; // just in case
 113
 114     // Look at the stream's 'priv' field to see if we were initiated
 115     // via a SDP description:
 116     char* sdpDescription = (char*)(demuxer->stream->priv);
 117     if (sdpDescription == NULL) {
 118       // We weren't given a SDP description directly, so assume that
 119       // we were given a RTSP or SIP URL:
 120       char const* protocol = demuxer->stream->streaming_ctrl->url->protocol;
 121       char const* url = demuxer->stream->streaming_ctrl->url->url;
 122       extern int verbose;
 123       if (strcmp(protocol, "rtsp") == 0) {
 124         rtspClient = RTSPClient::createNew(*env, verbose, "MPlayer");
 125         if (rtspClient == NULL) {
 126           fprintf(stderr, "Failed to create RTSP client: %s\n",
 127                   env->getResultMsg());
 128           break;
 129         }
 130         sdpDescription = openURL_rtsp(rtspClient, url);
 131       } else { // SIP
 132         unsigned char desiredAudioType = 0; // PCMU (use 3 for GSM)
 133         sipClient = SIPClient::createNew(*env, desiredAudioType, NULL,
 134                                          verbose, "MPlayer");
 135         if (sipClient == NULL) {
 136           fprintf(stderr, "Failed to create SIP client: %s\n",
 137                   env->getResultMsg());
 138           break;
 139         }
 140         sipClient->setClientStartPortNum(8000);
 141         sdpDescription = openURL_sip(sipClient, url);
 142       }
 143
 144       if (sdpDescription == NULL) {
 145         fprintf(stderr, "Failed to get a SDP description from URL \"%s\": %s\n",
 146                 url, env->getResultMsg());
 147         break;
 148       }
 149     }
 150
 151     // Now that we have a SDP description, create a MediaSession from it:
 152     MediaSession* mediaSession = MediaSession::createNew(*env, sdpDescription);
 153     if (mediaSession == NULL) break;
 154
 155
 156     // Create a 'RTPState' structure containing the state that we just created,
 157     // and store it in the demuxer's 'priv' field, for future reference:
 158     RTPState* rtpState = new RTPState;
 159     rtpState->sdpDescription = sdpDescription;
 160     rtpState->rtspClient = rtspClient;
 161     rtpState->sipClient = sipClient;
 162     rtpState->mediaSession = mediaSession;
 163     rtpState->audioBufferQueue = rtpState->videoBufferQueue = NULL;
 164     rtpState->flags = 0;
 165     rtpState->firstSyncTime.tv_sec = rtpState->firstSyncTime.tv_usec = 0;
 166     demuxer->priv = rtpState;
 167
 168     int audiofound = 0, videofound = 0;
 169     // Create RTP receivers (sources) for each subsession:
 170     MediaSubsessionIterator iter(*mediaSession);
 171     MediaSubsession* subsession;
 172     unsigned desiredReceiveBufferSize;
 173     while ((subsession = iter.next()) != NULL) {
 174       // Ignore any subsession that's not audio or video:
 175       if (strcmp(subsession->mediumName(), "audio") == 0) {
 176         if (audiofound) {
 177           fprintf(stderr, "Additional subsession \"audio/%s\" skipped\n", subsession->codecName());
 178           continue;
 179         }
 180         desiredReceiveBufferSize = 100000;
 181       } else if (strcmp(subsession->mediumName(), "video") == 0) {
 182         if (videofound) {
 183           fprintf(stderr, "Additional subsession \"video/%s\" skipped\n", subsession->codecName());
 184           continue;
 185         }
 186         desiredReceiveBufferSize = 2000000;
 187       } else {
 188         continue;
 189       }
 190
 191       if (rtsp_port)
 192           subsession->setClientPortNum (rtsp_port);
 193
 194       if (!subsession->initiate()) {
 195         fprintf(stderr, "Failed to initiate \"%s/%s\" RTP subsession: %s\n", subsession->mediumName(), subsession->codecName(), env->getResultMsg());
 196       } else {
 197         fprintf(stderr, "Initiated \"%s/%s\" RTP subsession on port %d\n", subsession->mediumName(), subsession->codecName(), subsession->clientPortNum());
 198
 199         // Set the OS's socket receive buffer sufficiently large to avoid
 200         // incoming packets getting dropped between successive reads from this
 201         // subsession's demuxer.  Depending on the bitrate(s) that you expect,
 202         // you may wish to tweak the "desiredReceiveBufferSize" values above.
 203         int rtpSocketNum = subsession->rtpSource()->RTPgs()->socketNum();
 204         int receiveBufferSize
 205           = increaseReceiveBufferTo(*env, rtpSocketNum,
 206                                     desiredReceiveBufferSize);
 207         if (verbose > 0) {
 208           fprintf(stderr, "Increased %s socket receive buffer to %d bytes \n",
 209                   subsession->mediumName(), receiveBufferSize);
 210         }
 211
 212         if (rtspClient != NULL) {
 213           // Issue a RTSP "SETUP" command on the chosen subsession:
 214           if (!rtspClient->setupMediaSubsession(*subsession, False,
 215                                                 rtspStreamOverTCP)) break;
 216           if (!strcmp(subsession->mediumName(), "audio"))
 217             audiofound = 1;
 218           if (!strcmp(subsession->mediumName(), "video"))
 219             videofound = 1;
 220         }
 221       }
 222     }
 223
 224     if (rtspClient != NULL) {
 225       // Issue a RTSP aggregate "PLAY" command on the whole session:
 226       if (!rtspClient->playMediaSession(*mediaSession)) break;
 227     } else if (sipClient != NULL) {
 228       sipClient->sendACK(); // to start the stream flowing
 229     }
 230
 231     // Now that the session is ready to be read, do additional
 232     // MPlayer codec-specific initialization on each subsession:
 233     iter.reset();
 234     while ((subsession = iter.next()) != NULL) {
 235       if (subsession->readSource() == NULL) continue; // not reading this
 236
 237       unsigned flags = 0;
 238       if (strcmp(subsession->mediumName(), "audio") == 0) {
 239         rtpState->audioBufferQueue
 240           = new ReadBufferQueue(subsession, demuxer, "audio");
 241         rtpState->audioBufferQueue->otherQueue = &(rtpState->videoBufferQueue);
 242         rtpCodecInitialize_audio(demuxer, subsession, flags);
 243       } else if (strcmp(subsession->mediumName(), "video") == 0) {
 244         rtpState->videoBufferQueue
 245           = new ReadBufferQueue(subsession, demuxer, "video");
 246         rtpState->videoBufferQueue->otherQueue = &(rtpState->audioBufferQueue);
 247         rtpCodecInitialize_video(demuxer, subsession, flags);
 248       }
 249       rtpState->flags |= flags;
 250     }
 251     success = True;
 252   } while (0);
 253   if (!success) return NULL; // an error occurred
 254
 255   // Hack: If audio and video are demuxed together on a single RTP stream,
 256   // then create a new "demuxer_t" structure to allow the higher-level
 257   // code to recognize this:
 258   if (demux_is_multiplexed_rtp_stream(demuxer)) {
 259     stream_t* s = new_ds_stream(demuxer->video);
 260     demuxer_t* od = demux_open(opts, s, DEMUXER_TYPE_UNKNOWN,
 261                                opts->audio_id, opts->video_id, opts->sub_id,
 262                                NULL);
 263     demuxer = new_demuxers_demuxer(od, od, od);
 264   }
 265
 266   return demuxer;
 267 }
 268
 269 extern "C" int demux_is_mpeg_rtp_stream(demuxer_t* demuxer) {
 270   // Get the RTP state that was stored in the demuxer's 'priv' field:
 271   RTPState* rtpState = (RTPState*)(demuxer->priv);
 272
 273   return (rtpState->flags&RTPSTATE_IS_MPEG12_VIDEO) != 0;
 274 }
 275
 276 extern "C" int demux_is_multiplexed_rtp_stream(demuxer_t* demuxer) {
 277   // Get the RTP state that was stored in the demuxer's 'priv' field:
 278   RTPState* rtpState = (RTPState*)(demuxer->priv);
 279
 280   return (rtpState->flags&RTPSTATE_IS_MULTIPLEXED) != 0;
 281 }
 282
 283 static demux_packet_t* getBuffer(demuxer_t* demuxer, demux_stream_t* ds,
 284                                  Boolean mustGetNewData,
 285                                  float& ptsBehind); // forward
 286
 287 extern "C" int demux_rtp_fill_buffer(demuxer_t* demuxer, demux_stream_t* ds) {
 288   // Get a filled-in "demux_packet" from the RTP source, and deliver it.
 289   // Note that this is called as a synchronous read operation, so it needs
 290   // to block in the (hopefully infrequent) case where no packet is
 291   // immediately available.
 292
 293   while (1) {
 294     float ptsBehind;
 295     demux_packet_t* dp = getBuffer(demuxer, ds, False, ptsBehind); // blocking
 296     if (dp == NULL) return 0;
 297
 298     if (demuxer->stream->eof) return 0; // source stream has closed down
 299
 300     // Before using this packet, check to make sure that its presentation
 301     // time is not far behind the other stream (if any).  If it is,
 302     // then we discard this packet, and get another instead.  (The rest of
 303     // MPlayer doesn't always do a good job of synchronizing when the
 304     // audio and video streams get this far apart.)
 305     // (We don't do this when streaming over TCP, because then the audio and
 306     // video streams are interleaved.)
 307     // (Also, if the stream is *excessively* far behind, then we allow
 308     // the packet, because in this case it probably means that there was
 309     // an error in the source's timestamp synchronization.)
 310     const float ptsBehindThreshold = 1.0; // seconds
 311     const float ptsBehindLimit = 60.0; // seconds
 312     if (ptsBehind < ptsBehindThreshold ||
 313         ptsBehind > ptsBehindLimit ||
 314         rtspStreamOverTCP) { // packet's OK
 315       ds_add_packet(ds, dp);
 316       break;
 317     }
 318
 319 #ifdef DEBUG_PRINT_DISCARDED_PACKETS
 320     RTPState* rtpState = (RTPState*)(demuxer->priv);
 321     ReadBufferQueue* bufferQueue = ds == demuxer->video ? rtpState->videoBufferQueue : rtpState->audioBufferQueue;
 322     fprintf(stderr, "Discarding %s packet (%fs behind)\n", bufferQueue->tag(), ptsBehind);
 323 #endif
 324     free_demux_packet(dp); // give back this packet, and get another one
 325   }
 326
 327   return 1;
 328 }
 329
 330 Boolean awaitRTPPacket(demuxer_t* demuxer, demux_stream_t* ds,
 331                        unsigned char*& packetData, unsigned& packetDataLen,
 332                        float& pts) {
 333   // Similar to "demux_rtp_fill_buffer()", except that the "demux_packet"
 334   // is not delivered to the "demux_stream".
 335   float ptsBehind;
 336   demux_packet_t* dp = getBuffer(demuxer, ds, True, ptsBehind); // blocking
 337   if (dp == NULL) return False;
 338
 339   packetData = dp->buffer;
 340   packetDataLen = dp->len;
 341   pts = dp->pts;
 342
 343   return True;
 344 }
 345
 346 static void teardownRTSPorSIPSession(RTPState* rtpState); // forward
 347
 348 extern "C" void demux_close_rtp(demuxer_t* demuxer) {
 349   // Reclaim all RTP-related state:
 350
 351   // Get the RTP state that was stored in the demuxer's 'priv' field:
 352   RTPState* rtpState = (RTPState*)(demuxer->priv);
 353   if (rtpState == NULL) return;
 354
 355   teardownRTSPorSIPSession(rtpState);
 356
 357   UsageEnvironment* env = NULL;
 358   TaskScheduler* scheduler = NULL;
 359   if (rtpState->mediaSession != NULL) {
 360     env = &(rtpState->mediaSession->envir());
 361     scheduler = &(env->taskScheduler());
 362   }
 363   Medium::close(rtpState->mediaSession);
 364   Medium::close(rtpState->rtspClient);
 365   Medium::close(rtpState->sipClient);
 366   delete rtpState->audioBufferQueue;
 367   delete rtpState->videoBufferQueue;
 368   delete rtpState->sdpDescription;
 369   delete rtpState;
 370
 371   env->reclaim(); delete scheduler;
 372 }
 373
 374 ////////// Extra routines that help implement the above interface functions:
 375
 376 #define MAX_RTP_FRAME_SIZE 5000000
 377     // >= the largest conceivable frame composed from one or more RTP packets
 378
 379 static void afterReading(void* clientData, unsigned frameSize,
 380                          unsigned /*numTruncatedBytes*/,
 381                          struct timeval presentationTime,
 382                          unsigned /*durationInMicroseconds*/) {
 383   int headersize = 0;
 384   if (frameSize >= MAX_RTP_FRAME_SIZE) {
 385     fprintf(stderr, "Saw an input frame too large (>=%d).  Increase MAX_RTP_FRAME_SIZE in \"demux_rtp.cpp\".\n",
 386             MAX_RTP_FRAME_SIZE);
 387   }
 388   ReadBufferQueue* bufferQueue = (ReadBufferQueue*)clientData;
 389   demuxer_t* demuxer = bufferQueue->ourDemuxer();
 390   RTPState* rtpState = (RTPState*)(demuxer->priv);
 391
 392   if (frameSize > 0) demuxer->stream->eof = 0;
 393
 394   demux_packet_t* dp = bufferQueue->dp;
 395
 396   if (bufferQueue->readSource()->isAMRAudioSource())
 397     headersize = 1;
 398   else if (bufferQueue == rtpState->videoBufferQueue &&
 399       ((sh_video_t*)demuxer->video->sh)->format == mmioFOURCC('H','2','6','4')) {
 400     dp->buffer[0]=0x00;
 401     dp->buffer[1]=0x00;
 402     dp->buffer[2]=0x01;
 403     headersize = 3;
 404   }
 405
 406   resize_demux_packet(dp, frameSize + headersize);
 407
 408   // Set the packet's presentation time stamp, depending on whether or
 409   // not our RTP source's timestamps have been synchronized yet:
 410   Boolean hasBeenSynchronized
 411     = bufferQueue->rtpSource()->hasBeenSynchronizedUsingRTCP();
 412   if (hasBeenSynchronized) {
 413     if (verbose > 0 && !bufferQueue->prevPacketWasSynchronized) {
 414       fprintf(stderr, "%s stream has been synchronized using RTCP \n",
 415               bufferQueue->tag());
 416     }
 417
 418     struct timeval* fst = &(rtpState->firstSyncTime); // abbrev
 419     if (fst->tv_sec == 0 && fst->tv_usec == 0) {
 420       *fst = presentationTime;
 421     }
 422
 423     // For the "pts" field, use the time differential from the first
 424     // synchronized time, rather than absolute time, in order to avoid
 425     // round-off errors when converting to a float:
 426     dp->pts = presentationTime.tv_sec - fst->tv_sec
 427       + (presentationTime.tv_usec - fst->tv_usec)/1000000.0;
 428     bufferQueue->prevPacketPTS = dp->pts;
 429   } else {
 430     if (verbose > 0 && bufferQueue->prevPacketWasSynchronized) {
 431       fprintf(stderr, "%s stream is no longer RTCP-synchronized \n",
 432               bufferQueue->tag());
 433     }
 434
 435     // use the previous packet's "pts" once again:
 436     dp->pts = bufferQueue->prevPacketPTS;
 437   }
 438   bufferQueue->prevPacketWasSynchronized = hasBeenSynchronized;
 439
 440   dp->pos = demuxer->filepos;
 441   demuxer->filepos += frameSize + headersize;
 442
 443   // Signal any pending 'doEventLoop()' call on this queue:
 444   bufferQueue->blockingFlag = ~0;
 445 }
 446
 447 static void onSourceClosure(void* clientData) {
 448   ReadBufferQueue* bufferQueue = (ReadBufferQueue*)clientData;
 449   demuxer_t* demuxer = bufferQueue->ourDemuxer();
 450
 451   demuxer->stream->eof = 1;
 452
 453   // Signal any pending 'doEventLoop()' call on this queue:
 454   bufferQueue->blockingFlag = ~0;
 455 }
 456
 457 static demux_packet_t* getBuffer(demuxer_t* demuxer, demux_stream_t* ds,
 458                                  Boolean mustGetNewData,
 459                                  float& ptsBehind) {
 460   // Begin by finding the buffer queue that we want to read from:
 461   // (Get this from the RTP state, which we stored in
 462   //  the demuxer's 'priv' field)
 463   RTPState* rtpState = (RTPState*)(demuxer->priv);
 464   ReadBufferQueue* bufferQueue = NULL;
 465   int headersize = 0;
 466   TaskToken task;
 467
 468   if (demuxer->stream->eof) return NULL;
 469
 470   if (ds == demuxer->video) {
 471     bufferQueue = rtpState->videoBufferQueue;
 472     if (((sh_video_t*)ds->sh)->format == mmioFOURCC('H','2','6','4'))
 473       headersize = 3;
 474   } else if (ds == demuxer->audio) {
 475     bufferQueue = rtpState->audioBufferQueue;
 476     if (bufferQueue->readSource()->isAMRAudioSource())
 477       headersize = 1;
 478   } else {
 479     fprintf(stderr, "(demux_rtp)getBuffer: internal error: unknown stream\n");
 480     return NULL;
 481   }
 482
 483   if (bufferQueue == NULL || bufferQueue->readSource() == NULL) {
 484     fprintf(stderr, "(demux_rtp)getBuffer failed: no appropriate RTP subsession has been set up\n");
 485     return NULL;
 486   }
 487
 488   demux_packet_t* dp = NULL;
 489   if (!mustGetNewData) {
 490     // Check whether we have a previously-saved buffer that we can use:
 491     dp = bufferQueue->getPendingBuffer();
 492     if (dp != NULL) {
 493       ptsBehind = 0.0; // so that we always accept this data
 494       return dp;
 495     }
 496   }
 497
 498   // Allocate a new packet buffer, and arrange to read into it:
 499     if (!bufferQueue->nextpacket) {
 500   dp = new_demux_packet(MAX_RTP_FRAME_SIZE);
 501   bufferQueue->dp = dp;
 502   if (dp == NULL) return NULL;
 503     }
 504
 505 #ifdef CONFIG_LIBAVCODEC
 506   extern AVCodecParserContext * h264parserctx;
 507   int consumed, poutbuf_size = 1;
 508   const uint8_t *poutbuf = NULL;
 509   float lastpts = 0.0;
 510
 511   do {
 512     if (!bufferQueue->nextpacket) {
 513 #endif
 514   // Schedule the read operation:
 515   bufferQueue->blockingFlag = 0;
 516   bufferQueue->readSource()->getNextFrame(&dp->buffer[headersize], MAX_RTP_FRAME_SIZE - headersize,
 517                                           afterReading, bufferQueue,
 518                                           onSourceClosure, bufferQueue);
 519   // Block ourselves until data becomes available:
 520   TaskScheduler& scheduler
 521     = bufferQueue->readSource()->envir().taskScheduler();
 522   int delay = 10000000;
 523   if (bufferQueue->prevPacketPTS * 1.05 > rtpState->mediaSession->playEndTime())
 524     delay /= 10;
 525   task = scheduler.scheduleDelayedTask(delay, onSourceClosure, bufferQueue);
 526   scheduler.doEventLoop(&bufferQueue->blockingFlag);
 527   scheduler.unscheduleDelayedTask(task);
 528   if (demuxer->stream->eof) {
 529     free_demux_packet(dp);
 530     return NULL;
 531   }
 532
 533   if (headersize == 1) // amr
 534     dp->buffer[0] =
 535         ((AMRAudioSource*)bufferQueue->readSource())->lastFrameHeader();
 536 #ifdef CONFIG_LIBAVCODEC
 537     } else {
 538       bufferQueue->dp = dp = bufferQueue->nextpacket;
 539       bufferQueue->nextpacket = NULL;
 540     }
 541     if (headersize == 3 && h264parserctx) { // h264
 542       consumed = h264parserctx->parser->parser_parse(h264parserctx,
 543                                NULL,
 544                                &poutbuf, &poutbuf_size,
 545                                dp->buffer, dp->len);
 546
 547       if (!consumed && !poutbuf_size)
 548         return NULL;
 549
 550       if (!poutbuf_size) {
 551         lastpts=dp->pts;
 552         free_demux_packet(dp);
 553         bufferQueue->dp = dp = new_demux_packet(MAX_RTP_FRAME_SIZE);
 554       } else {
 555         bufferQueue->nextpacket = dp;
 556         bufferQueue->dp = dp = new_demux_packet(poutbuf_size);
 557         memcpy(dp->buffer, poutbuf, poutbuf_size);
 558         dp->pts=lastpts;
 559       }
 560     }
 561   } while (!poutbuf_size);
 562 #endif
 563
 564   // Set the "ptsBehind" result parameter:
 565   if (bufferQueue->prevPacketPTS != 0.0
 566       && bufferQueue->prevPacketWasSynchronized
 567       && *(bufferQueue->otherQueue) != NULL
 568       && (*(bufferQueue->otherQueue))->prevPacketPTS != 0.0
 569       && (*(bufferQueue->otherQueue))->prevPacketWasSynchronized) {
 570     ptsBehind = (*(bufferQueue->otherQueue))->prevPacketPTS
 571                  - bufferQueue->prevPacketPTS;
 572   } else {
 573     ptsBehind = 0.0;
 574   }
 575
 576   if (mustGetNewData) {
 577     // Save this buffer for future reads:
 578     bufferQueue->savePendingBuffer(dp);
 579   }
 580
 581   return dp;
 582 }
 583
 584 static void teardownRTSPorSIPSession(RTPState* rtpState) {
 585   MediaSession* mediaSession = rtpState->mediaSession;
 586   if (mediaSession == NULL) return;
 587   if (rtpState->rtspClient != NULL) {
 588     rtpState->rtspClient->teardownMediaSession(*mediaSession);
 589   } else if (rtpState->sipClient != NULL) {
 590     rtpState->sipClient->sendBYE();
 591   }
 592 }
 593
 594 ////////// "ReadBuffer" and "ReadBufferQueue" implementation:
 595
 596 ReadBufferQueue::ReadBufferQueue(MediaSubsession* subsession,
 597                                  demuxer_t* demuxer, char const* tag)
 598   : prevPacketWasSynchronized(False), prevPacketPTS(0.0), otherQueue(NULL),
 599     dp(NULL), nextpacket(NULL),
 600     pendingDPHead(NULL), pendingDPTail(NULL),
 601     fReadSource(subsession == NULL ? NULL : subsession->readSource()),
 602     fRTPSource(subsession == NULL ? NULL : subsession->rtpSource()),
 603     fOurDemuxer(demuxer), fTag(strdup(tag)) {
 604 }
 605
 606 ReadBufferQueue::~ReadBufferQueue() {
 607   delete fTag;
 608
 609   // Free any pending buffers (that never got delivered):
 610   demux_packet_t* dp = pendingDPHead;
 611   while (dp != NULL) {
 612     demux_packet_t* dpNext = dp->next;
 613     dp->next = NULL;
 614     free_demux_packet(dp);
 615     dp = dpNext;
 616   }
 617 }
 618
 619 void ReadBufferQueue::savePendingBuffer(demux_packet_t* dp) {
 620   // Keep this buffer around, until MPlayer asks for it later:
 621   if (pendingDPTail == NULL) {
 622     pendingDPHead = pendingDPTail = dp;
 623   } else {
 624     pendingDPTail->next = dp;
 625     pendingDPTail = dp;
 626   }
 627   dp->next = NULL;
 628 }
 629
 630 demux_packet_t* ReadBufferQueue::getPendingBuffer() {
 631   demux_packet_t* dp = pendingDPHead;
 632   if (dp != NULL) {
 633     pendingDPHead = dp->next;
 634     if (pendingDPHead == NULL) pendingDPTail = NULL;
 635
 636     dp->next = NULL;
 637   }
 638
 639   return dp;
 640 }
 641
 642 static int demux_rtp_control(struct demuxer *demuxer, int cmd, void *arg) {
 643   double endpts = ((RTPState*)demuxer->priv)->mediaSession->playEndTime();
 644
 645   switch(cmd) {
 646     case DEMUXER_CTRL_GET_TIME_LENGTH:
 647       if (endpts <= 0)
 648         return DEMUXER_CTRL_DONTKNOW;
 649       *((double *)arg) = endpts;
 650       return DEMUXER_CTRL_OK;
 651
 652     case DEMUXER_CTRL_GET_PERCENT_POS:
 653       if (endpts <= 0)
 654         return DEMUXER_CTRL_DONTKNOW;
 655       *((int *)arg) = (int)(((RTPState*)demuxer->priv)->videoBufferQueue->prevPacketPTS*100/endpts);
 656       return DEMUXER_CTRL_OK;
 657
 658     default:
 659       return DEMUXER_CTRL_NOTIMPL;
 660     }
 661 }
 662
 663 demuxer_desc_t demuxer_desc_rtp = {
 664   "LIVE555 RTP demuxer",
 665   "live555",
 666   "",
 667   "Ross Finlayson",
 668   "requires LIVE555 Streaming Media library",
 669   DEMUXER_TYPE_RTP,
 670   0, // no autodetect
 671   NULL,
 672   demux_rtp_fill_buffer,
 673   demux_open_rtp,
 674   demux_close_rtp,
 675   NULL,
 676   demux_rtp_control
 677 };