rtpreceiver.cpp

symbian平台S60_2nd_FP2_SC rtp实现

#include "vtypes.h"

#include "cpLog.h"
#include "NtpTime.h"
#include "rtpTypes.h"
#include "rtpTools.h"
#include "rtpCodec.h"
#include "RtcpReceiver.h"
/* ----------------------------------------------------------------- */
/* --- RtpReceiver Constructor ------------------------------------- */
/* ----------------------------------------------------------------- */

RtpReceiver::RtpReceiver (CSocketUdp* udp,
						  int /*localMinPort*/, 
						  int /*localMaxPort*/,
                          RtpPayloadType newApiFormat,
                          RtpPayloadType newNetworkFormat, 
						  int jitterNew)
{
    /// udp stack is a sendrecv stack
   // myStack = new UdpStack (NULL, localMinPort, localMaxPort) ;
	myStack = udp;
    freeStack = true;
    isEvent=false;
    m_event=DTMFEventNULL;
    constructRtpReceiver (newApiFormat, newNetworkFormat, jitterNew);
}

RtpReceiver::RtpReceiver (CSocketUdp* udp,
						  int /*localPort*/, 
						  RtpPayloadType newApiFormat,
                          RtpPayloadType newNetworkFormat,
                          int jitterNew)
{
    /// udp stack is a sendrecv stack
    //myStack = new UdpStack (NULL, localPort) ;
	myStack = udp;
    freeStack = true;
    isEvent=false;
    m_event=DTMFEventNULL;
    constructRtpReceiver (newApiFormat, newNetworkFormat, jitterNew);
}

RtpReceiver::RtpReceiver (CSocketUdp* udp, 
						  RtpPayloadType newApiFormat,
                          RtpPayloadType newNetworkFormat, 
						  int jitterNew)
{
    /// udp stack is a sendrecv stack
    myStack = udp;
    freeStack = false;

    constructRtpReceiver (newApiFormat, newNetworkFormat, jitterNew);
}


void RtpReceiver::constructRtpReceiver (RtpPayloadType newApiFormat,
                                        RtpPayloadType newNetworkFormat,
                                        int jitterNew)
{
    inPos = 0;
    playPos = 0;
    memset (inBuff, 0, IN_BUFFER_SIZE);

    // set format and baseSampleRate
    setApiFormat(newApiFormat, 160, 0, NULL, false);
    setNetworkFormat(newNetworkFormat, 160, 0, NULL, false);

    // set private variables
    jitterSeed = jitterNew;

    // no transmitter
    sourceSet = false;
    ssrc = 0;
    probationSet = false;
    srcProbation = 0;
    probation = -2;
    prevPacket = NULL;
    rtcpRecv = NULL;
    silenceCodec = 0;
    codecString[0] = '\0';

    cpLog(LOG_DEBUG_STACK, "Constructed receiver");
}


RtpReceiver::~RtpReceiver ()
{
    rtcpRecv = NULL;

    cpLog (LOG_DEBUG_STACK, "Close receiver");
}


/* --- receive packet functions ------------------------------------ */

RtpPacket* RtpReceiver::receive ()
{
    RtpPacket* p = NULL;
    int len = 0;
    int len1 = 0;
    int silencePatched = 0;
    bool faking = 0;


    // empty network que
    NtpTime arrival (0, 0);
    while (1) // network empty or time to play return packet
    {
        p = getPacket();
        if (p == NULL)
           break;

        RtpPayloadType lType = p->getPayloadType();
        arrival = getNtpTime();
        int packetTransit = 0;
        int delay = 0;

        rtp_ntohl(p);
		len = p->getPayloadUsage();
		if (probation < 0)
        {
            cpLog(LOG_ERR, "****Packet from invalid source");
            delete p;
            p = NULL;
            continue;
        }
		
        if (len <= 0 || len > 1500)
        {
		    cpLog(LOG_ERR, "Got an invalid packet size");
            delete p;
            p = NULL;
            continue;
        }

        // fix frame boundry
        if (len > networkFormat_payloadSize )
        {

			//modify for receive ,send must
            /*int lenold = len;
            len = ( len / networkFormat_payloadSize ) * networkFormat_payloadSize;*/
            p->setPayloadUsage( len );
            cpLog( LOG_DEBUG_STACK, "Fixing frame boundry to %d from %d", len, len );
            //network_pktSampleSize = (len / networkFormat_payloadSize) * network_pktSampleSize;
        }

        // bypass jitterBuffer
        if (jitterTime == -1)
        {
            cpLog( LOG_DEBUG_STACK, "Skipping jitter buffer" );

            // update counters
            prevSeqRecv = p->getSequence();
            prevSeqPlay = p->getSequence();

            // update packet received
            packetReceived++;
            payloadReceived += p->getPayloadUsage();

            // update jitter calculation
            packetTransit = arrival - rtp2ntp(p->getRtpTime());
            delay = packetTransit - transit;
            transit = packetTransit;
            if (delay < 0) delay = -delay;
            jitter += delay - ((jitter + 8) >> 4);

            return p;
        }


        // reordering the packets according to the seq no
        // leave gaps and copy the next packet in all the gap
        // when the late packets come copy it into the correct pos

        if (RtpSeqGreater(p->getSequence(), prevSeqRecv))
        {
            // insert packet at end, repeat filling buffer if sequence skipping
            // bool cycleEnd = false;
			// int w=0;

            while (RtpSeqGreater(p->getSequence(), prevSeqRecv))
            {
                // silence patching
				// prevPacketRtpTime += network_pktSampleSize;
                silencePatched = 0;
                faking = 0;
                while( RtpTimeGreater( p->getRtpTime() - network_pktSampleSize, prevPacketRtpTime ) && ((p->getSequence() - 1) == prevSeqRecv))
                {
					// Patching silence for packet
                    if( silenceCodec == 0 )
                    {
                        cpLog( LOG_DEBUG_STACK, "Patching silence" );
                        if ((p->getPayloadType() >= rtpPayloadDynMin) &&
                            (p->getPayloadType() <= rtpPayloadDynMax) &&
                            (codecString[0] != '\0'))
                        {
							silenceCodec = findSilenceCodecString(codecString, len);
                        }
                        else
                        {
							silenceCodec = findSilenceCodec( p->getPayloadType(), len );
                        }
                        if( silenceCodec == 0 )
                        {
                            if( len > rtpCodecInfo[ numRtpCodecInfo - 1 ].length )
                            {
                                cpLog( LOG_ERR, "Requested codec too big to fake %d", len );
                                assert( 0 );
                            }
                            cpLog( LOG_DEBUG_STACK, "Faking silence packet with 0x00" );
                            silenceCodec = (char*)&rtpCodecInfo[ numRtpCodecInfo - 1 ].silence;
                            faking = 1;
                        }
                    }
                  
                    if ((inPos + len) < IN_BUFFER_SIZE)
                    {   
                        memcpy (inBuff + inPos, silenceCodec, len);
                        inPos += len;
                        silencePatched++;
                    }
                    else
                    {
                        // circular memory copy 
                        len1 = IN_BUFFER_SIZE - inPos;
                        memcpy (inBuff + inPos, silenceCodec, len1);
                        memcpy (inBuff, silenceCodec + len1, len - len1);
                        inPos = len - len1;
                        silencePatched++;
                    }
                    prevPacketRtpTime += network_pktSampleSize;
                }// end while( RtpTimeGreater...

                if( prevPacketRtpTime != p->getRtpTime() - network_pktSampleSize)
                {
                    cpLog( LOG_DEBUG_STACK, "Silent patching fail to correct rtptime" );
                    cpLog( LOG_DEBUG_STACK,
                           "prevPacketRtpTime(%u), p->getRtpTime()(%u), networkPacketSize(%d)",
                           prevPacketRtpTime, p->getRtpTime(), network_pktSampleSize );
                    prevPacketRtpTime = p->getRtpTime() - network_pktSampleSize;
                }

				// 接受数据写入inBuff

                if ((inPos + len) < IN_BUFFER_SIZE)
                {
                    memcpy (inBuff + inPos, p->getPayloadLoc(), len);
                    inPos += len;
                }
                else
                {
                    // circular memory copy
                    len1 = IN_BUFFER_SIZE - inPos;
                    memcpy (inBuff + inPos, p->getPayloadLoc(), len1);
                    memcpy (inBuff, p->getPayloadLoc() + len1, len - len1);
                    inPos = len - len1;
                }

                // update counters
                RtpSeqNumber tSeq = prevSeqRecv;
                prevSeqRecv++;
		        if(prevSeqRecv > RTP_SEQ_MOD)
                {
                   prevSeqRecv = 0;
                }
                if (prevSeqRecv < tSeq)
                {
                    cpLog(LOG_DEBUG_STACK, "Recv cycle");
                    assert(prevSeqRecv == 0);
                    recvCycles += RTP_SEQ_MOD;
                }
            }//  end while(RtpSeqGreater(p->getSequence(), prevSeqRecv))

            prevPacketRtpTime = p->getRtpTime();
            if (silencePatched > 0)
                cpLog(LOG_DEBUG_STACK, "silencePatched = %d", silencePatched);
            if (faking) 
                silenceCodec = 0;
            if (p->getSequence() != prevSeqRecv)
            {
                cpLog(LOG_DEBUG_STACK, "Unequal packet:%d stack:%d",
                      prevSeqRecv, p->getSequence());
            }
        }
		
		//	packets come copy it into the correct pos
		//	natural order
        else
        {
            // insert packet in middle
            cpLog(LOG_DEBUG_STACK, "insert middle packet: %d prevSeq %d",
                  p->getSequence(), prevSeqRecv);
            RtpSeqNumber base_prevSeqRecv = prevSeqRecv;
            int inSeqRecv = 1;
            while (RtpSeqGreater(base_prevSeqRecv, p->getSequence()))
            {
                inSeqRecv++;
                base_prevSeqRecv--;
            }
            int inPosTemp = inPos - inSeqRecv * len;
            if (inPosTemp < 0) inPosTemp = IN_BUFFER_SIZE + inPosTemp;

            if ((inPosTemp + len) < IN_BUFFER_SIZE)
            {
                memcpy (inBuff + inPosTemp, p->getPayloadLoc(), len);
            }
            else
            {
                // circular memory copy
                len1 = IN_BUFFER_SIZE - inPosTemp;
                memcpy (inBuff + inPosTemp, p->getPayloadLoc(), len1);
                memcpy (inBuff, (p->getPayloadLoc()) + len1, len - len1);
            }
        }

        // update packet received
        packetReceived++;
        payloadReceived += len;

        // update jitter calculation
        packetTransit = arrival - rtp2ntp(p->getRtpTime());
        delay = packetTransit - transit;
        transit = packetTransit;
        if (delay < 0) delay = -delay;
        jitter += delay - ((jitter + 8) >> 4);

        if (p)
        {
            delete p;
            p = NULL;
        }
    } // end while

    int packetSize = apiFormat_payloadSize;

    // deque next packet
    if ( (inPos == 0) && (playPos == 0) )
    {
        //cpLog (LOG_ERR, "Recv buffer is empty");
        receiverError = recv_bufferEmpty;
        return NULL;
    }

    if (((inPos + IN_BUFFER_SIZE - playPos) % IN_BUFFER_SIZE) < packetSize)
    {
        cpLog (LOG_ERR,"Not enough data for a api packet size %d", packetSize);
        receiverError = recv_bufferEmpty;
        return NULL;
    }


    // create next packect
	// while loop 结束后，创建rtppacket
	// packet 大小为 apiFormat_payloadSize

    assert (!p);
    p = new RtpPacket (packetSize);
    if ( (playPos + packetSize) < IN_BUFFER_SIZE)
    {
        memcpy (p->getPayloadLoc(), inBuff + playPos, packetSize);
        playPos += packetSize;
    }
    else
    {
        // circular memory copy
        len1 = IN_BUFFER_SIZE - playPos;
        memcpy (p->getPayloadLoc(), inBuff + playPos, len1);
        memcpy (p->getPayloadLoc() + len1, inBuff, packetSize - len1);
        playPos = packetSize - len1;
    }

    // finish packet
    p->setSSRC (ssrc);
    p->setPayloadType (apiFormat);
    p->setPayloadUsage (packetSize);
    p->setRtpTime (prevRtpTime + api_pktSampleSize);
    p->setSequence (prevSeqPlay + 1);

    if (probation > 0)
       probation --;
    receiverError = recv_success;

    if( ( prevRtpTime + network_pktSampleSize ) != p->getRtpTime() )
    {
        network_pktSampleSize = p->getRtpTime() - prevRtpTime;
        cpLog( LOG_DEBUG_STACK, "Fixing network_pktSampelSize to %d", network_pktSampleSize );
    }

    prevRtpTime = p->getRtpTime();
    prevNtpTime = getNtpTime();
    gotime = rtp2ntp (p->getRtpTime() + api_pktSampleSize) + jitterTime;


    // update counters
    RtpSeqNumber sSeq = prevSeqPlay;
    prevSeqPlay++;
    if (prevSeqPlay < sSeq)
    {
        cpLog(LOG_DEBUG_STACK, "Play cycle");
        assert (prevSeqPlay == 0);
        playCycles += RTP_SEQ_MOD;
    }
    return p;
}


RtpPacket* RtpReceiver::getPacket ()
{
/*ZHAOSONG

    // check for network activity
    fd_set netFD;
    FD_ZERO (&netFD);
    FD_SET (myStack->getSocketFD(), &netFD);
    struct timeval timeout;
    timeout.tv_sec = 0;
    timeout.tv_usec = 0;
    int selret = select (myStack->getSocketFD() + 1, &netFD, NULL, NULL, &timeout);
    if (selret <= 0)
    {
        // select error or no network activity
        if (selret < 0) // 出错
           cpLog(LOG_ERR, "Select loop error");
        else  if(selret ==0) // 等待超时
        {
            cpLog(LOG_ERR, "RtpReceiver receives %d pkts", packetReceived);
            cpLog(LOG_ERR, "UdpStack receives %d pkts", myStack->getPacketsReceived());
            return NULL;
        }
    }
*/
/*
struct timeval* timeout是select的超时时间，这个参数至关重要，
它可以使select处于三种状态，
第一，若将NULL以形参传入，即不传入时间结构，就是将select置于阻塞状态，
一定等到监视文件描述符集合中某个文件描述符发生变化为止；
第二，若将时间值设为0秒0毫秒，就变成一个纯粹的非阻塞函数，
不管文件描述符是否有变化，都立刻返回继续执行，文件无变化返回0，有变化返回一个正值；
第三，timeout的值大于0，这就是等待的超时时间，即select在timeout时间内阻塞，
超时时间之内有事件到来就返回了，否则在超时后不管怎样一定返回，返回值同上述。

返回值：
负值：select错误 正值：某些文件可读写或出错 0：等待超时，没有可读写或错误的文件
*/