Home    News    Components    Get ViTooKi    Documentation    Community    Developers    @sf.net  

Info

Last update:
12 Feb 2007

Documentation

Doxygen LXR

SimpleRtp.cpp

#ifndef WINCE

#include <sched.h>
#include "SimpleRtp.hpp"
#include "Frame.hpp"
#include "global.hpp"
#include "SimpleStatistics.hpp"
#include "CompressedVideoFrame.hpp"
#include "CompressedAudioFrame.hpp"
#include "adaptors/MP4Decoder.hpp"
#include "adaptors/MP4audioDecoder.hpp"

// for stupid MSVC
#ifdef WIN32
static const int TTL = 16;
static const int MAX_PAYLOAD_SIZE = 65000;
static const int RTP_PAYLOAD_TYPE = 96;
static const int UPDATE_INTERVAL_MS=20;
static const int MINI_PREBUFFER_INTERVALL=500;
static const unsigned int MAX_NULL_FRAMES=5;
#endif

/* only sleep if can sleep at least 5 ms */
#define SIMPLERTP_STAYAHEAD 5
#define MAX_PAKETS_SENT_WITHOUT_SLEEP 5
#define SIMPLERTP_MAX_MS_BEHIND 50
#define SIMPLERTP_MAX_MS_AHEAD  10000

SimpleRtp::SimpleRtp(const char *uri, int clintprt, int locPort, const char *clintaddr, ESInfo * e,
         PacketizationLayer * p, bool w,
         const char* statsFileName, bool overwriteExistingStatFile,
         Delay delay, bool updateVopTimeIncrement,bool switching):IO(),
         clientPort(clintprt), writeOnly(w), updateESInfo(updateVopTimeIncrement),
         enableSwitching(switching), delayType(delay)
{
        assert(clintaddr && uri && e && p);
        dprintf_full("SimpleRtp::SimpleRtp: clientAddress %s clientPort %i localPort %i writeOnly: %s\n",
                clintaddr,clintprt,locPort,w?"YES":"NO");
        if(statsFileName) {
                // server runs in writeonly mode
                stats=new SimpleStatistics(statsFileName,e,w,overwriteExistingStatFile);
        }
        else
                stats=NULL;

        url = new char[strlen(uri) + 1];
        clientAddress = new char[strlen(clintaddr) + 1];
        es = e;
        strcpy(clientAddress, clintaddr);
        strcpy(url, uri);
        session = NULL;
        pLayer = p;
        localPort = locPort;
        firstFrame=false; // must use open to set to true
        resendFrameHeader=true;
        ctsForFirstPlay=0;
        // buffer at most 20001 packets (a packet odyssey ;-)
        packetBuffer=new FastPointerHashTable<AU*>(20001);
        resetPlayTime=true;
        packetInsertedDuringLastCallback=false;
        prefetchTimeInMs=1000;
        rtpSeqNrStart=4294967295u;
        rtpSeqNrEnd=0;
        overflow=0;
        lastCTSReturned=0;
        nullFrameCounter=0;
        bytesTransferred=0;
        frameToStart = 0;
        lockPacketBuffer.initialize("SimpleRtp-lockPacketBuffer");
};

SimpleRtp::~SimpleRtp()
{
    if (clientAddress) {
        delete clientAddress;
        clientAddress = NULL;
    }

    if (pLayer) {
        delete pLayer;
        pLayer = NULL;
    }
    if (state != CLOSING || state != CLOSED) {
        close();
    }
        if(session)     {
                rtp_send_bye(session);
                rtp_done(session);
                session=NULL;
        } // delete happens in rtp_done(..)
        lockPacketBuffer.lock();
        if(packetBuffer){
                delete packetBuffer;
                packetBuffer=NULL;
        }
        lockPacketBuffer.release();
        if(url)
                delete url;
        lockPacketBuffer.destroy();
        if(stats) {
                stats->teardown();
                delete stats;
        }
};

bool SimpleRtp::open()
{
  if (state == OPEN)
                return true;
  state = OPENING;
        sendPacketCounter=0;
        firstPacket=true;
        u32 avgPacketsPerSecond=es->getAvgBandwidth()/(8*this->pLayer->getMaxPacketPayloadSize());
        u32 roundsPerSecond=avgPacketsPerSecond/MAX_PAKETS_SENT_WITHOUT_SLEEP;
        if(roundsPerSecond>0) {
                intermediateMiniSleep=1000/roundsPerSecond;
        }
        else
                intermediateMiniSleep=1;
        if(intermediateMiniSleep>5)
                intermediateMiniSleep=5;
        // init for read and write is the same
  if (clientAddress && !session) {
                session = ucl_rtp_init(clientAddress,
                                                                                                         localPort,
                                                                                                         clientPort,
                                                                                                         TTL, 100, SimpleRtp::RtpCallback, (uint8_t *) this,
                                                                                                         portGen.getSockFromPort(localPort),
                                                                                                         portGen.getSockFromPort(localPort+1));

        }
        resendFrameHeader=true;
        ctsForFirstPlay=0;nullFrameCounter=0;
        rtpSeqNrStart=4294967295u;
        rtpSeqNrEnd=0;
        overflow=0;
        resetPlayTime=true;

  if (session) {
                state = OPEN;
    rtp_set_option(session, RTP_OPT_WEAK_VALIDATION, FALSE);
    rtp_set_option(session, RTP_OPT_PROMISC, TRUE);

                firstFrame=true;
                update();
                return true;
  } else {
                state = CLOSED;
                return false;
  }
};

/***************************************************************/
int SimpleRtp::writeFrame(Frame * frm, ESInfo *out_es) {
  bool result = true;

  AU au;
  u8* newPayload=NULL;
  bool last = false;
        u8* currentPacketStart=NULL;;
        u8* lastPacketEnd=NULL;
        u32 nrBytes=0;
        u32 remainingBytes=0;

  if (!writeOnly || state != OPEN || !pLayer || !frm)
    return 0;
        lastCTSReturned=frm->getAU()->cts;
  dprintf_full("SimpleRtp::writeFrame CTS %i prio %i size %i\n",frm->getAU()->cts, frm->getAU()->prio, frm->getAU()->size);

        if (endFrameNumber>0 && frm->getAU()->cts / es->getVOPTimeIncrement() > endFrameNumber) {
                //this comes from unflushed DC senqueue, and is out of range, so drop it!
                dprintf_full("SimpleRtp::writeFrame nr %i TS %i exceeds endFrameNumber %i, ignoring!\n",frm->getAU()->cts / es->getVOPTimeIncrement(),frm->getAU()->cts, endFrameNumber );
                return frm->getAU()->size; //FIXME: fake return value for error
        }

        //HEADER DATA
  if ( (!firstFrame && resendFrameHeader )
       || (firstFrame &&
                                         !MP4Decoder::isValidDecoderConfig(frm->getAU()->payload)) ) { //HEADER already in first frame

                if(!MP4Decoder::isValidDecoderConfig(frm->getAU()->payload)     && resetPlayTime) {
                        // set the new lastCTS only for payload!
                        ctsForFirstPlay=frm->getAU()->cts;
                        gettimeofday(&lastTimeStamp,NULL);
                        resetPlayTime=false;
                }
    //send extra header!
    u8* esHeaderBuffer=NULL;
    u32 headerSize=es->getHeaders(&esHeaderBuffer);
    if(headerSize>0) {
      newPayload=new u8[headerSize];
      memcpy(newPayload, esHeaderBuffer,headerSize);

      au.payload=newPayload;
      au.size=headerSize;
                        if(firstFrame)
                                au.cts=au.dts=0;
                        else
                                au.cts=au.dts=frm->getAU()->cts;

      au.duration=0;
                        currentPacketStart=au.payload;
                        lastPacketEnd=au.payload+au.size;
                        nrBytes=pLayer->getMaxPacketPayloadSize();

      while(currentPacketStart<lastPacketEnd) {
                                remainingBytes=lastPacketEnd-currentPacketStart;
                                if(remainingBytes<=nrBytes) {
                                        nrBytes=remainingBytes;
                                        last=true;
                                }

                                dprintf_full("\nSimpleRtp::writeFrame: send cts %d instead of %d\n", au.cts - (es->getVOPTimeIncrement() * frameToStart), au.cts);
                                //result &= sendPacket(currentPacketStart,nrBytes,au.cts,last);
                                result &= sendPacket(currentPacketStart,nrBytes,au.cts - (es->getVOPTimeIncrement() * frameToStart),last);
                                currentPacketStart+=nrBytes;
      }
    }
    if(esHeaderBuffer) {
      delete[] esHeaderBuffer;
      esHeaderBuffer=NULL;
    }
                resendFrameHeader=false;
    firstFrame=false;
  }

  //FRAME DATA
  firstFrame=false;

  if (frm->getAU()->size==0) {
    return 0;
  }

        if(!MP4Decoder::isValidDecoderConfig(frm->getAU()->payload)     && resetPlayTime) {
                // set the new lastCTS only for payload!
                ctsForFirstPlay=frm->getAU()->cts;
                gettimeofday(&lastTimeStamp,NULL);
                resetPlayTime=false;
        }

  dprintf_full("SimpleRtp::writeFrame: FRAME first 4 bytes of Frame, CTS %u: %x %x %x %x\r\n",
                                                         frm->getAU()->cts, frm->getAU()->payload[0],
                                                         frm->getAU()->payload[1],
                                                         frm->getAU()->payload[2],frm->getAU()->payload[3]);
  last=false;
        int upd=0;
        currentPacketStart=frm->getAU()->payload;
        lastPacketEnd=currentPacketStart+frm->getAU()->size;
        nrBytes=pLayer->getMaxPacketPayloadSize();

  while(currentPacketStart<lastPacketEnd) {
                remainingBytes=lastPacketEnd-currentPacketStart;
                if(remainingBytes<=nrBytes) {
                        nrBytes=remainingBytes;
                        last=true;
                }
                upd++;
                dprintf_full("\nSimpleRtp::writeFrame sending cts %d instead of cts %d", frm->getAU()->cts - (es->getVOPTimeIncrement() * frameToStart), frm->getAU()->cts);
                //result &= sendPacket(currentPacketStart,nrBytes,frm->getAU()->cts,last);
                result &= sendPacket(currentPacketStart,nrBytes,frm->getAU()->cts - (es->getVOPTimeIncrement() * frameToStart),last);
                currentPacketStart+=nrBytes;
                if(upd>=MAX_PAKETS_SENT_WITHOUT_SLEEP) {
                        msleep(intermediateMiniSleep);
                        update();
                        msleep(1);
                        upd=0;
                }
  }

        // sync
#ifdef _POSIX_PRIORITY_SCHEDULING
        sched_yield();
#else
        // for windows pthread lib!
#ifdef WIN32
        sched_yield();
#endif
#endif
        syncToCts(frm->getAU()->cts);
        update();
  return (int)result;
}

void SimpleRtp::syncToCts(u32 cts)
{
        if(cts>ctsForFirstPlay) {
                long sleepMS=calcSleepTimeInMs(cts);
                dprintf_full("SimpleRtp::syncToCts: will sleep %li ms\n",sleepMS);
                if(sleepMS>SIMPLERTP_STAYAHEAD)
                        msleep(sleepMS-SIMPLERTP_STAYAHEAD);
        }
}

long SimpleRtp::calcSleepTimeInMs(u32 cts)
{
        long delay=0;

        switch(delayType) {
                case(CTS_BASED):
                        delay=calcSleepTimeCTSBased(cts);
                        break;
                case(BITRATE_BASED):
                        delay=calcSleepTimeBitRateBased(cts,115);
                        break;
                case(MIXED_CTS_BITRATE):
                        delay=calcSleepTimeMixedBased(cts,0);
                        break;
                case(AGGRESSIVE_MIXED_CTS_BITRATE):
                        delay=calcSleepTimeMixedBased(cts,prefetchTimeInMs);
                        break;
                case(NO_DELAY): default:
                        delay=0;
        }
        dprintf_full("SimpleRtp::calcSleepTimeInMs(%u) bytesTransf %u delay %li\n",cts,(u32)bytesTransferred,delay);
        return delay;

}

long SimpleRtp::calcSleepTimeCTSBased(u32 cts)
{
        u64 ctsDifferenceInMs=(u64)(cts-ctsForFirstPlay); // in ticks

        if(ctsDifferenceInMs==0) // first frame after play
                return 0;

        ctsDifferenceInMs*=1000; // in ticks per millisec
        ctsDifferenceInMs/=(es->getMediaTimeScale()); // in ms


        // ctsDifferenceInMs calc the SHOULD difference (the play-out time in MS)
        // timeDiffInMs contains the observed difference
        u32 timeDiffInMs=getElapsedTimeInMSSinceLastPlay();
        long delay=(long)(ctsDifferenceInMs-timeDiffInMs);
        return delay;
}
long SimpleRtp::calcSleepTimeBitRateBased(u32 cts, u32 speedScaleFactor) {
        long delay=0;
        if(speedScaleFactor==0  || speedScaleFactor>400) {
                speedScaleFactor=115;
        }
        u32 timeDiffInMs=(u32)getElapsedTimeInMSSinceLastPlay();
        // constant netbandwidth streaming
        u64 bitsPerSecond=es->getAvgBandwidth();
        // send the video faster with a factor of 1.15
        // bitsPerSecond*timeDiffInMs gives bits with scale 100
        // divide through 8 to get byte, multiply with scale/100
        // divide through 1000 because we calculated with time in MS and bit rate was a per second value!
        // assume: max duration 3 hours=10,8x10^6ms max bit rate 10 Mbit=10^7bit, max scale is 400
        // largest intermediate result is approx 1,1x10^7*10^7*400=4.4*10^16 --> ceil(ld(10^17)), max 57 bits are used
        u64 shouldBytes=bitsPerSecond*timeDiffInMs*speedScaleFactor/800000u;
        u64 diffBytes=0;
        // transferredBytes contains the real transferred bytes for the WHOLE session
        if(bytesTransferred> shouldBytes) {
                diffBytes=bytesTransferred-shouldBytes;
                delay=(long)(diffBytes*8000/bitsPerSecond);
        }
        return delay;
};
long SimpleRtp::calcSleepTimeMixedBased(u32 cts,u32 clientPreBufferTime) {
        // try to keep client buffer between SIMPLERTP_MAX_MS_BEHIND and SIMPLERTP_MAX_MS_AHEAD
        // adjust scale factor depending on that information
        // decide periodically (3 seconds) on the new scale factor (1.0 < 1.15)
        // for the first 10 seconds use bit rate based streaming
        long delay=0;
        u32 timeDiffInMs=(u32)getElapsedTimeInMSSinceLastPlay();
        s32 clientBufferInMS=0;


        u64 bitsPerSecond=es->getAvgBandwidth();
        u64 ctsDifferenceInMs=(u64)(cts-ctsForFirstPlay); // in ticks
        ctsDifferenceInMs*=1000; // in ticks per millisec
        ctsDifferenceInMs/=(es->getMediaTimeScale()); // in ms
        dprintf_full("SimpleRtp::calcSleepTimeMixedBased(cts=%u): timeDiff %u ctsDiffMS %u preBuffer %u\n",
                cts,timeDiffInMs,(u32)ctsDifferenceInMs,clientPreBufferTime);
        if(ctsDifferenceInMs==0) // first frame
                return 0;
        // now calculate the client's buffer in MS
        // take the prebuffertime, add the streamout seconds, subtract the time elapsed since play
        // allow faster build up of buffer
        if(timeDiffInMs<clientPreBufferTime)
                clientPreBufferTime=timeDiffInMs;
        clientBufferInMS=clientPreBufferTime+ctsDifferenceInMs-timeDiffInMs;
        dprintf_full("SimpleRtp::calcSleepTimeMixedBased:  clientBuffer %i\n",clientBufferInMS);
        double scale=1.0;
        double maxScale=1.3,minScale=0.8;
        // a buffer of SIMPLERTP_MAX_MS_BEHIND+1000 equals to 1.3
        // a buffer of SIMPLERTP_MAX_MS_AHEAD-1000 equals to 0.8
        if(clientBufferInMS<SIMPLERTP_MAX_MS_BEHIND+1000) {
                scale=maxScale;
        }
        else if(clientBufferInMS>(SIMPLERTP_MAX_MS_AHEAD-1000)) {
                scale=minScale;
        }
        else {
                u32 offset=(u32)(clientBufferInMS-(SIMPLERTP_MAX_MS_BEHIND+1000));
                scale=maxScale-offset*(maxScale-minScale)/(SIMPLERTP_MAX_MS_AHEAD-2000-SIMPLERTP_MAX_MS_BEHIND);
        }


        u32 secElapsed=timeDiffInMs/1000;
        u32 seconds=1;
        if(secElapsed<seconds) {
                seconds=secElapsed+1;
        }
        u64 curMS=SimpleStatistics::getCurrentMilliSecond();

        curMS+=((seconds-1)*1000);
        u64 bytesToSend=(u64)(curMS*bitsPerSecond*scale/8000);
        u64 bytesSent=stats->getNumBytesOutLastNSeconds(seconds);
        long diff=(long)(bytesSent-bytesToSend);
        if(diff>0)
                delay=(long)(diff*8000/(seconds*scale*bitsPerSecond));
        else
                delay=0;
        dprintf_full("SimpleRtp::calcSleepTimeMixedBased: delay%li bytesToSend %u bytesSent %u scale %lf\n",delay,(u32)bytesToSend,(u32)bytesSent,scale);

        return delay;
}

long SimpleRtp::getElapsedTimeInMSSinceLastPlay() {
        struct timeval theTimeIsNow;
        gettimeofday(&theTimeIsNow,NULL);
        long timeDiffInMs=(theTimeIsNow.tv_sec-lastTimeStamp.tv_sec)*1000+
                (theTimeIsNow.tv_usec-lastTimeStamp.tv_usec)/1000;
        // should never happen but better catch this case
        if(timeDiffInMs<=0)
                timeDiffInMs=0;
        return timeDiffInMs;
};
bool SimpleRtp::sendPacket(u8* payload, u32 size,u32 rtpTimestamp, bool last)
{
  dprintf_full("SimpleRtp::sendPacket %i bytes CTS: %i \n",size,rtpTimestamp);
        if(stats)
                stats->insertPacket(rtpTimestamp,size,sendPacketCounter++,last);
        bytesTransferred+=size;
    int rc = rtp_send_data(session,
                           rtpTimestamp,
                           RTP_PAYLOAD_TYPE, last, 0, NULL,
                           (uint8_t *) payload,
                           size,
                           NULL, 0, 0);
    return rc >= 0;
};

bool SimpleRtp::setToFrameNumber(u32 frameNumber)
{
dprintf_full("SimpleRtp::setToFrameNumber(%d)\n", frameNumber);
        frameToStart = frameNumber; //added by klschoef on 02/03/2005
        resetPlayTime=true;
        lockPacketBuffer.lock();
        if(packetBuffer)
                packetBuffer->freeAllElements();
        lockPacketBuffer.release();
        return true;
};

bool SimpleRtp::close(bool immediate)
{
        if(state==OPEN || state==OPENING) {
                state = CLOSING;
                // can not do this, client can send a play again
/*              rtp_send_bye(session);
                rtp_done(session); */
                state = CLOSED;
                // empty the buffer
                lockPacketBuffer.lock();
                packetBuffer->freeAllElements();
                lockPacketBuffer.release();
        }
    return true;
};

Frame *SimpleRtp::getFrame()
{
        if (writeOnly || state != OPEN)
                return NULL;
        if(lastCTSReturned == es->getDuration()-es->getVOPTimeIncrement()) {
                //EOF
                close(); // locks lockpacketBuffer
                dprintf_full("SimpleRtp::getFrame: REACHED STREAMEOF\n");
                return NULL;
        }

        dprintf_full("SimpleRtp::getFrame() %u\n",rtpSeqNrStart);
        // otherwise prebuffer!
        if(firstFrame) {
                preBufferPackets(prefetchTimeInMs);
                firstFrame=false;
        }
        // now check if we have a complete frame available in the buffer
        // we start with rtp_seq_nr
        if(rtpSeqNrStart>=rtpSeqNrEnd)
                preBufferPackets(MINI_PREBUFFER_INTERVALL);
        else {
                //fetch all cached packets which are CURRENTLY available
                preBufferPackets(0);
        }
        lockPacketBuffer.release();
        Frame* frm=extractCompleteFrame(rtpSeqNrStart);
        if(frm) {
                // no more accesses to packetBuffer
                lockPacketBuffer.release();
                nullFrameCounter=0;
                if(resetPlayTime) {
                        ctsForFirstPlay=frm->getAU()->cts;
                        gettimeofday(&lastTimeStamp,NULL);
                        resetPlayTime=false;
                }

                if(frm->getAU()->cts==0 && MP4Decoder::isValidDecoderConfig(frm->getAU()->payload)) {
                        //check if we have payload too! if yes remove the header from the frame
                        // to avoid that we have duplicate headers when writing to file!
                        stripHeaderFromFrame(frm);
                }
                if(MP4Decoder::isValidDecoderConfig(frm->getAU()->payload)) {
                        if(es->isVisualStream() && enableSwitching)
                                es->setDecoderConfig(frm->getAU()->payload,frm->getAU()->size);
                        delete frm;
                        frm=getFrame();
                }
                else if(MP4audioDecoder::isValidDecoderConfig(frm->getAU()->payload)) {
                        if(es->isAudioStream() && enableSwitching)
                                es->setDecoderConfig(frm->getAU()->payload,frm->getAU()->size);
                        delete frm;
                        frm=getFrame();
                }
        }
        else {
                nullFrameCounter++;
                dprintf_full("SimpleRtp::getFrame() NULL %i\n",nullFrameCounter);
                if(lastCTSReturned >
                this->es->getDuration() - es->getMediaTimeScale()) {
                        // if 2 null frames in last second close!
                        if(nullFrameCounter>1) {
                                lockPacketBuffer.release();
                                close(); // locks lockpacketBuffer
                                dprintf_full("SimpleRtp::getFrame: REACHED STREAMEOF\n");
                        }
                        else
                                lockPacketBuffer.release();
                }
                else {
                        //give 2nd chance to retrieve the frame (>1)
                        if(nullFrameCounter>1) {
                                removeIncompleteFrame(rtpSeqNrStart); //access packetBuffer
                                lockPacketBuffer.release();
                        }
                        else if(nullFrameCounter>MAX_NULL_FRAMES){
                                lockPacketBuffer.release();
                                close(); // locks lockpacketBuffer
                                dprintf_full("SimpleRtp::getFrame: PACKET LOSS TOO HIGH! ENDING SimpleRtp\n");
                        }
                        else
                                lockPacketBuffer.release();
                }
        }
        if(frm)
                lastCTSReturned=frm->getAU()->cts;

        // sync
#ifdef _POSIX_PRIORITY_SCHEDULING
        sched_yield();
#else
        // for windows pthread lib!
#ifdef WIN32
        sched_yield();
#endif
#endif
        if(frm)
                syncToCts(frm->getAU()->cts);
        update();
        return frm;
}

void SimpleRtp::removeIncompleteFrame(u32 seqNr)
{
        bool cont=true;
        bool firstPacket=true;
        u32 cts=0;
        dprintf_full("SimpleRtp::removeIncompleteFrame: start with seqNr%u (cts should be %u)\n",seqNr,lastCTSReturned);

        while(cont && seqNr<=rtpSeqNrEnd) {

                AU* p=packetBuffer->get(seqNr);
                u32 size=(p?p->size:0);
                if(p) {
                        if(firstPacket) {
                                cts=p->cts;
                        }
                        if(p->sampleFlags>0) //end of frame detected
                                cont=false;
                        if(cts==p->cts) {
                                if(stats) {
                                        stats->dropInsertedPacket(size,seqNr,cts,firstPacket);
                                }
                                packetBuffer->freeElement(seqNr);
                                p=NULL;
                        }
                        else {
                                cont=false;
                                seqNr--; // put back the last packet, belongs to next frame
                        }
                        if(firstPacket)
                                firstPacket=false;
                }
                else {
                        // a NULL packet returned -> lost!!!
                        stats->notifyLostPacket(seqNr);
                }
                seqNr++;
        }

        dprintf_full("SimpleRtp::removeIncompleteFrame: last deleted seqNr%u\n",seqNr-1);
        rtpSeqNrStart=seqNr;
}

Frame* SimpleRtp::extractCompleteFrame(u32 seqNr)
{
        std::list<AU*> packets;
        std::list<AU*>::iterator li;
        Frame* frm=NULL;
        dprintf_full("SimpleRtp::extractCompleteFrame(u32 seqNr=%u)\n",seqNr);

        u32 rtpTimeStamp=0; // the cts value
        if(rtpSeqNrStart>=rtpSeqNrEnd) // empty buffer
                return NULL;
        // extract packets until empty packet or end of frame
        // or new cts! no need to care about seqNr here, HashTable guarantees
        // ordering!
        // seqNr is a unique id!!! Every list has exactly ONE packet stored
        bool cont=true;
        bool complete=false;
        u32 tmpSeqNr=seqNr;
        u32 frmSize=0;
        while(cont && tmpSeqNr<=rtpSeqNrEnd && !complete) {
                AU* p=packetBuffer->get(tmpSeqNr);
                if(p==NULL) {
                        cont=false;
                        // don't do notify here. to catch every lost packet do it in dropIncompleteFrame
                        // stats->notifyLostPacket(tmpSeqNr);
                }
                else {
                        if(packets.empty()) // if no other packet in the result list
                                rtpTimeStamp=p->cts; // the cts value
                        if(p->cts==rtpTimeStamp) {
                                packets.push_back(p);
                                frmSize+=p->size;
                                complete=(p->sampleFlags>0?true:false);
                        }
                        else {  // no endofframe but different cts,
                                cont=false;
                                // don't do notify here. to catch every lost packet do it in dropIncompleteFrame
                                // stats->notifyLostPacket(tmpSeqNr);
                                if(updateESInfo) {
                                        if(p->cts-rtpTimeStamp>0) {
                                                es->setVOPTimeIncrement(p->cts-rtpTimeStamp);
                                                updateESInfo=false;
                                        }
                                }

                        }
                }
                tmpSeqNr++;

        }
        if(complete) {
                // check next frame to calc voptimeincrement
                if(updateESInfo) {
                        AU* tmpp=packetBuffer->get(tmpSeqNr);
                        if(tmpp) {
                                if(tmpp->cts-rtpTimeStamp>0) {
                                        es->setVOPTimeIncrement(tmpp->cts-rtpTimeStamp);
                                        updateESInfo=false;
                                }
                        }
                }
                // rebuild the frame
                dprintf_full("SimpleRtp::extractCompleteFrame: Using packets %u to %u to rebuild frame(%i packets)\n",
                        seqNr,tmpSeqNr-1,packets.size());
                if(es->isAudioStream())
                        frmSize-=packets.size()*4;
                AU* au=new AU();
                au->size=frmSize;
                au->payload=new u8[frmSize];
                if(stats) {
                        stats->getFrame(rtpTimeStamp,frmSize,seqNr,tmpSeqNr-1);
                }
                // FIXME: handle header correctly
                // for now just concatenate the payload of the packets
                // free the packets from seqNr to including tmpSeqNr
                u32 offset=0;
                u32 dataOffset=0;
                // remove header for audio
                if(es->isAudioStream())
                        dataOffset+=4;
                for(li=packets.begin();li!=packets.end();++li) {
                        if(li==packets.begin()) {
                                // set cts for first packet
                                // FIXME: dts is wrong
                                au->cts=au->dts=(*li)->cts;
                        }
                        AU* tmp=(*li);
                        memcpy(au->payload+offset,tmp->payload+dataOffset,tmp->size-dataOffset);
                        offset+=tmp->size-dataOffset;
                }
                // free memory from seqNr to excl tmpSeqNr
                for(u32 sNr=seqNr;sNr<tmpSeqNr;sNr++)
                        packetBuffer->freeElement(sNr);

                if(es->isAudioStream())
                        frm=new CompressedAudioFrame(Frame::NN_VOP);
                else
                        frm=new CompressedVideoFrame(Frame::NN_VOP,((VideoESInfo*)es)->getWidth(),((VideoESInfo*)es)->getHeight());
                frm->setAU(au);
                rtpSeqNrStart=tmpSeqNr; // the next packet to read
        }
        else {
                dprintf_full("SimpleRtp::extractCompleteFrame: incomplete frame CTS %u\n",rtpTimeStamp);
        }
        return frm;
};

void SimpleRtp::preBufferPackets(long millisecs)
{
        struct timeval tv;
        struct timeval preBufferStart;
        struct timeval preBufferEnd;
        gettimeofday(&preBufferStart,NULL);
        packetInsertedDuringLastCallback=true;
        long timeDiffInMs=0;
        int ret=0;
        dprintf_full("SimpleRtp::preBufferPackets %li\n",millisecs);
        while(timeDiffInMs<=millisecs) {
          tv.tv_sec  = 0;
    tv.tv_usec = 1; //rtp_recv returns one single packet, wait nearly no time
                if(packetInsertedDuringLastCallback) {
                        // when we read a packet the last time, try to read again
                        packetInsertedDuringLastCallback=false;
                        ret=rtp_recv(session, &tv, rtpSeqNrStart);      // only returns one packet
                        // will set above var. to true if data avail.
                        // if no data is available it is still false and in case millisecs is zero
                        // exit immediately
                        if(!packetInsertedDuringLastCallback && millisecs==0) {
                                update();
                                return;
                        }
                }
                else if(millisecs-timeDiffInMs>=UPDATE_INTERVAL_MS) {
                        msleep(UPDATE_INTERVAL_MS);
                        packetInsertedDuringLastCallback=true; // read again after sleep
                }
                else if(millisecs-timeDiffInMs>0) {
                        msleep(millisecs-timeDiffInMs);
                        packetInsertedDuringLastCallback=true; // read again after sleep
                }
                // calc prebuffer
                gettimeofday(&preBufferEnd,NULL);
                timeDiffInMs=(preBufferEnd.tv_sec-preBufferStart.tv_sec)*1000+
                        (preBufferEnd.tv_usec-preBufferStart.tv_usec)/1000;

        }
        dprintf_full("SimpleRtp::preBufferPackets spent %li\n",timeDiffInMs);
        update();
}

char *SimpleRtp::getSSRC()
{
    if (state != OPEN)
        return NULL;

    char *result = new char[9];
    sprintf(result, "%08x", rtp_my_ssrc(session));
    return result;
};

void SimpleRtp::update()
{
        if(state==IO::OPEN) {
                if (session) {
                        rtp_update(session);
                        rtp_send_ctrl(session, 0, NULL,RTCP_NORMAL);
                }
        }
};

IO::State SimpleRtp::play(double prefetchTime)
{
        resetPlayTime=true;
        prefetchTimeInMs=(long)prefetchTime*1000;
        // prebuffer after pause
        if(!writeOnly) {
                firstFrame=true; // prebuffer!
                ctsForFirstPlay=0;nullFrameCounter=0;
                rtpSeqNrStart=4294967295u;
                rtpSeqNrEnd=0;
        }
        bytesTransferred=0;
        if(stats)
                stats->play(0,true);
        return IO::play(prefetchTime);
}

IO::State SimpleRtp::pause()
{
        resetPlayTime=true;
        if(stats)
                stats->pause();
        return IO::pause();
}

void SimpleRtp::RtpCallback(struct rtp *session, rtp_event * e) {
        // Currently we ignore RTCP packets
        // Just do our required housekeeping
        rtp_packet      *packet;
  SimpleRtp *caller = NULL;

  if (e->ssrc == rtp_my_ssrc(session))
    return;

        switch(e->type) {
        case RX_RTP:
                // we must deep-copy the packets, they are freed in the ucl lib!
                packet = (rtp_packet*)e->data;
                caller=(SimpleRtp*)rtp_get_userdata(session);
                if(!packet) {
                        dprintf_err("SimpleRtp::RtpCallback NULL packet?");
                        break;
                }
                if(packet->pd.rtp_pd_data_len==0 || packet->pd.rtp_pd_data==NULL) {
                        dprintf_err("SimpleRtp::RtpCallback empty packet?");
                        xfree(packet);
                        break;
                }
                if (caller->isInput()) { //only client handles incoming data!
                        AU* pClone=new AU();
                        pClone->cts=packet->ph.ph_ts;
                        pClone->payload=new u8[packet->pd.rtp_pd_data_len];
                        pClone->size=packet->pd.rtp_pd_data_len;
                        pClone->sampleFlags=(packet->ph.ph_m?1:0);
                        memcpy(pClone->payload,packet->pd.rtp_pd_data,pClone->size);
                        caller->insertPacket(pClone,packet->ph.ph_seq);
                }
                else
                        dprintf_full("SimpleRtp::RtpCallback: Not supposed to get data?\n");

                xfree(packet); /* xfree() is mandatory to release RTP packet data */
                break;
  case RX_SDES:
    dprintf_full("RX_SDES SSRC = 0x%08x\n", e->ssrc);
    break;
  case RX_BYE:
    dprintf_full("RX_BYE SSRC = 0x%08x\n", e->ssrc);
    break;
        case SOURCE_CREATED:
    dprintf_full("SimpleRtp::RtpCallback: New source created, SSRC = 0x%08x\n", e->ssrc);
    break;
  case SOURCE_DELETED:
                caller=(SimpleRtp*)rtp_get_userdata(session);
                caller->close();
                break;
  case RX_APP:
    break;
  default:
    break;
    }
};

void SimpleRtp::insertPacket(AU* p, u32 origSeqNr)
{
        u32 seqNr= origSeqNr+overflow*65536u;
        if(firstPacket) {
                rtpSeqNrStart=rtpSeqNrEnd=seqNr;
                firstPacket=false;
        }
        // howto detect overflow?
        // the difference between the largest and the next will exceed 60000!
        if((seqNr+65536)-rtpSeqNrEnd<2000u) {
                overflow++;
                dprintf_full("SimpleRtp::insertPacket detected rtpSeqNr Overflow\n");
                seqNr+=65536;
        }
        // now seq Nr is correct
        if(stats) {
                stats->insertPacket(p->cts,p->size,seqNr,(p->sampleFlags==1) );
        }
        lockPacketBuffer.lock();
        if(packetBuffer)
                packetBuffer->put(p,seqNr); //use rtp timestamp as key
        lockPacketBuffer.release();
        // check for a new max/min seqNr
        if(seqNr<rtpSeqNrStart)
                rtpSeqNrStart=seqNr;
        if(seqNr>rtpSeqNrEnd)
                rtpSeqNrEnd=seqNr;
        packetInsertedDuringLastCallback=true;
        dprintf_full("SimpleRtp::insertPacket(%u/%u)[%u,%u] CTS=%u Size=%u\n",origSeqNr,seqNr,rtpSeqNrStart,rtpSeqNrEnd,p->cts,p->size);
};

void SimpleRtp::stripHeaderFromFrame(Frame* frm)
{
        assert(frm&&frm->getAU());
        if(!frm)
                return;
        if(!frm->getAU())
                return;

        u8* payload=frm->getAU()->payload;
        u32 size=frm->getAU()->size;
        u8* payloadEnd=payload+size;
        if(!payload || size==0)
                return;
        if(!MP4Decoder::isValidDecoderConfig(payload))
                return;

  bool stop=false;
        u8* tmp=payload+4;

  // we received a header
        // with or without payload?
  while(!stop && tmp < payloadEnd &&
                                (tmp=(u8*)memchr(tmp,0xb6,payloadEnd-tmp))) {
                if(tmp==NULL)
                        tmp=payloadEnd;
                else if(tmp[-1] == 1 && tmp[-2] == 0 && tmp[-3] == 0) {
                        stop=true;
                        tmp-=3;
                } else
                        tmp++;

  }
  if (stop && tmp && enableSwitching)
                es->setDecoderConfig(payload,tmp-payload);
  else if(enableSwitching)
                es->setDecoderConfig(payload,(int)size);

        if(stop && tmp) {
                // remove the header from the frame
                u32 newSize=size-(tmp-payload);
                u8* newPayload=new u8[newSize];
                memcpy(newPayload,tmp,newSize);
                AU* tmp=frm->getAU();
                delete[] tmp->payload;
                tmp->payload=newPayload;
                tmp->size=newSize;
                frm->detectFrameType();
        }
}


#endif //wince

© ViTooKi 2003 :: [Lite version] Created with Doxygen