ppffsrc.cpp

linux下的一款播放器

     *	In the case of a BYE, we use OPTION B to reschedule the transmission of 
     *	BYE if necessary.
     */
    if (lEvent == RTCP_BYE)
    {
        if (m_pRTCPInterval)
	{

#ifdef XXXGo_DEBUG
	if (m_pLogFile)
	{
	    fprintf(m_pLogFile, "\t\tCurNumMembers %u, NumSenders %u\n",
					m_pRTCPInterval->GetCurNumMembers(), 
					m_pRTCPInterval->GetNumSenders());                		    
	    fflush(m_pLogFile);
	}	    
#endif

	    ulInterval = (UINT32)(m_pRTCPInterval->GetRTCPInterval(
					m_pRTCPInterval->GetRTCPBW() * COMPENSATION,
					m_pRTCPInterval->GetCurNumMembers(), 
					m_pRTCPInterval->GetNumSenders())                
        		 * 1000.0);

	    if (ulLastTime + ulInterval <= ulCurTime)
	    {
	    	SendBye();
	    }
	    else
	    {
#ifdef XXXGo_DEBUG
	if (m_pLogFile)
	{
	    fprintf(m_pLogFile, "\t\t%u + %u (%u)> %u\n", ulLastTime, ulInterval, 
	    ulLastTime + ulInterval, ulCurTime);
	    fflush(m_pLogFile);
	}	    
#endif
	    	// reschdule this for later
	    	Schedule(ulLastTime + ulInterval - ulCurTime , RTCP_BYE);	    
	    }    
	}	    

	return;
    }    

    HX_ASSERT(lEvent != RTCP_BYE);

    if ((m_algorithm == ALGORITHM_A) ||
	((m_algorithm == ALGORITHM_C) && !m_pRTCPInterval->IsInitial()))
    {
	ulInterval = (UINT32) 
		(m_pRTCPInterval->GetRTCPInterval(
		    m_pRTCPInterval->GetRTCPBW(),
		    ulNumMembers, ulNumSenders) * 1000.0);

	if (ulNumMembers + ulNumSenders <= m_pRTCPInterval->GetLastNumMembers())
	{
	    bSendIt = TRUE;
	}
	else
	{
	    if (ulLastTime + ulInterval <= ulCurTime)
	    {
		bSendIt = TRUE;
	    }
	}
    }
    else if ((m_algorithm == ALGORITHM_B) ||
	     ((m_algorithm == ALGORITHM_C) && m_pRTCPInterval->IsInitial()))
    {
	ulInterval = (UINT32) 
		(m_pRTCPInterval->GetRTCPInterval(
			m_pRTCPInterval->GetRTCPBW() * COMPENSATION,
			ulNumMembers, ulNumSenders) * 1000.0);

	if (ulLastTime + ulInterval <= ulCurTime)
	{
	    bSendIt = TRUE;
	}
    }

    if (bSendIt)
    {
	UINT32 ulPktSize = SendRecvReport();
	if (ulPktSize)
	    m_pRTCPInterval->UpdateAvgRTCPSize(ulPktSize);
	m_pRTCPInterval->SetLastRTCPTime(ulCurTime);
    }
    else
    {
	Schedule(ulInterval, RTCP_SR);	
    }

    m_pRTCPInterval->SetLastNumMembers(ulNumMembers + ulNumSenders);	
}


// Don't delete the map itself.  Just entries in the map!
void
CPurePlaySource::CleanupMemberTable()
{
    CHXSimpleList* pBin = NULL;
    if (SUCCEEDED(m_MemberTimeoutBins.GetOldestBin(pBin)))
    {
    	if (!pBin->IsEmpty())
    	{
    	    CHXSimpleList::Iterator i;
	    for (i = pBin->Begin(); i != pBin->End(); ++i)
            {
		// this will remove this entry from m_SsrcToMember and
		// delete the obj.
		RemoveMember(((CMember*)(*i))->m_ulSSRC, FALSE);
	    }
	    pBin->RemoveAll(); 
    	}    
    }

    m_MemberTimeoutBins.UpdateBins();
}


UINT32
CPurePlaySource::SendRecvReport(void)
{
    UINT32	ulPktSize = 0;
    IHXBuffer*	pUDPPacket = NULL;
    RTCPPacket	    pktRR;
    RTCPPacket	    pktSDES;
    CSDES	    Sdes(m_pCName, m_pName, m_pTool, m_pEmail);
    

    // wait until admission is closed
    if (!m_pPlugin->IsSourceAdmissionClosed())
    {
	goto bail;
    }

/*
* Form the RR and SDES compound packet then write it out
* to the socket
    */

    if (HXR_OK != MakeRR(pktRR))
    {
    	// we can't send RTCP pkt w/o RR
    	goto bail;
    }
    
    
    if (HXR_OK != Sdes.MakeSDES(pktSDES, m_ulThisSrcID))
    {
    	// no SDES, no RTCP pkt
    	goto bail;
    }

    
    pUDPPacket = new CHXBuffer;
    pUDPPacket->AddRef();

    pUDPPacket->SetSize(((pktRR.length + 1) * 4) + ((pktSDES.length + 1) * 4));
    
    unsigned char*   pBuf;
    pBuf = pUDPPacket->GetBuffer();

    UINT32  ulLenRR;
    UINT32  ulLenSDES;
    
    pBuf = pktRR.pack(pBuf, ulLenRR);
    pBuf = pktSDES.pack(pBuf, ulLenSDES);
    
    HX_ASSERT(ulLenRR == (UINT32)(pktRR.length + 1) * 4);
    HX_ASSERT(ulLenSDES == (UINT32)(pktSDES.length + 1) * 4);
    
    m_pRTCPSock->WriteTo(m_ulMultiAddr, m_uMultiPort + 1, pUDPPacket);
    m_pRTCPInterval->SetInitial(FALSE);
    m_bRTCPSent = TRUE;
    ulPktSize = pUDPPacket->GetSize();  

#ifdef XXXGo_DEBUG
	if (m_pLogFile)
	{
	    UINT32 ulTime = GetCurrentSchedulerTimeInMsec(m_pScheduler);
//	    HXTimeval lTime = m_pScheduler->GetCurrentSchedulerTime();
//	    UINT32 ulTime = lTime.tv_sec * 1000 + lTime.tv_usec / 1000;
	    fprintf(m_pLogFile, "\t\tSendRevReport (%u)\n", ulTime);
	    fflush(m_pLogFile);
	}	    
#endif
    

bail:
    /*
    * Schedule the next status report
    */
    UINT32  ulNextTime = (UINT32) 
		(m_pRTCPInterval->GetRTCPInterval(
			  m_pRTCPInterval->GetRTCPBW(),
			  m_SsrcToMember.GetCount(),  // #members
			  GetTrueNumSrc())                // #senders
        	 * 1000.0);
    Schedule(ulNextTime, RTCP_SR);

    HX_RELEASE(pUDPPacket);    

    return ulPktSize;
}

BOOL	
CPurePlaySource::MakeBye(REF(IHXBuffer*) pPkt)
{
    pPkt = new CHXBuffer();
    if (!pPkt)
    {
	return FALSE;
    }
    pPkt->AddRef();

    
    RTCPPacket pktRR;
    if (HXR_OK != MakeRR(pktRR))
    {
	HX_RELEASE(pPkt);
    	return FALSE;
    }
    
    RTCPPacket pktBye; 

    pktBye.version_flag = 0x02;	   
    pktBye.padding_flag = 0;	   
    pktBye.packet_type = RTCP_BYE;   
    pktBye.length = 1;		   // len in 32-bits words minus one
    pktBye.count = 1;

    // use access function
    pktBye.SetByeSrc(&m_ulThisSrcID, pktBye.count);
    
    pPkt->SetSize(((pktRR.length + 1) * 4) + ((pktBye.length + 1) * 4));
    
    unsigned char*   pBuf = pPkt->GetBuffer();
    UINT32  ulLenRR;
    UINT32  ulLenBye;

    pBuf = pktRR.pack(pBuf, ulLenRR);
    pBuf = pktBye.pack(pBuf, ulLenBye);

    HX_ASSERT(ulLenRR = (UINT32)(pktRR.length + 1) * 4);
    HX_ASSERT(ulLenBye == (UINT32)(pktBye.length + 1) * 4);        

    return TRUE;
}

void
CPurePlaySource::SendBye(void)
{
#ifdef XXXGo_DEBUG
	if (m_pLogFile)
	{
	    fprintf(m_pLogFile, "\tSendBye()\n");
	    fflush(m_pLogFile);
	}	    
#endif

    if (!m_bRTCPSent)
    {
	// if no RTCP packet has been sent, don't send the BYE pkt.
	goto bail;
    }

    HX_ASSERT(m_pRTCPInterval);
    if (!m_pRTCPSock)
    {
	// no RTCP packet has been sent
	goto bail;
    }

    if (!m_pByePkt)
    {
	if(!MakeBye(m_pByePkt))
	{
	    HX_ASSERT(m_pByePkt);
	    goto bail;
	}
    }

    HX_ASSERT(m_pByePkt);

    m_pRTCPSock->WriteTo(m_ulMultiAddr, m_uMultiPort + 1, m_pByePkt);

#ifdef XXXGo_DEBUG
	if (m_pLogFile)
	{
	    fprintf(m_pLogFile, "\tSending()\n");
	    fflush(m_pLogFile);
	}	    
#endif

bail:

    Cleanup();
}

BOOL
CPurePlaySource::ScheduleBye(void)
{
#ifdef XXXGo_DEBUG
	if (m_pLogFile)
	{
	    fprintf(m_pLogFile, "ScheduleBye %u\n", m_SsrcToMember.GetCount() + GetTrueNumSrc());
	    fflush(m_pLogFile);
	}	    
#endif

    if (!m_pRTCPInterval)
    {
	HX_ASSERT(FALSE);
	return FALSE;
    }

    if (m_pRTCPInterval->IsInitial() || NULL == m_pRTCPSock)
    {
	// no RTCP packet has been sent
	return FALSE;
    }
   
    // if there are more than 50 participants, don't send BYE pkt right away 
    // since doing so may flood the network.
    if (m_SsrcToMember.GetCount() + GetTrueNumSrc() < 50)
    {
	// just send it
	SendBye();
	return TRUE;
    }

    if (!m_pScheduler)
    {
	return FALSE;
    }

    if (!m_pByePkt && !MakeBye(m_pByePkt))         
    {
	HX_ASSERT(!m_pByePkt);	
	return FALSE;
    }
    HX_ASSERT(m_pByePkt);
    
    // we have to do some work.
    UINT32 ulCurTime = GetCurrentSchedulerTimeInMsec(m_pScheduler);
    
    m_pRTCPInterval->SetLastRTCPTime(ulCurTime);
    m_pRTCPInterval->SetCurNumMembers(1);
    m_pRTCPInterval->SetLastNumMembers(1);
    m_pRTCPInterval->SetInitial(TRUE);
    m_pRTCPInterval->SetNumSenders(1);
    m_pRTCPInterval->SetAvgRtcpSize(m_pByePkt->GetSize());
    
    Schedule((UINT32)(m_pRTCPInterval->GetRTCPInterval(0.0, 1, 1) * 1000.0), 
	     RTCP_BYE);

    return TRUE;	     
}


BOOL
CPurePlaySource::GetStreamBySSRC(UINT32 ulSsrc, CStream*& pStrm)
{
    // get the right source from ssrc
    if (m_SsrcToStream.Lookup(ulSsrc, (void*&)pStrm))
    {
        // it's all good
        HX_ASSERT(ulSsrc == pStrm->m_ulSSRC);
        return TRUE;
    }

    pStrm = NULL; 
    return FALSE;
}

BOOL
CPurePlaySource::GetMember(UINT32 ulSsrc, CMember*& pMember)
{
    // get the right source from ssrc
    if (m_SsrcToMember.Lookup(ulSsrc, (void*&)pMember))
    {
        // it's all good
        HX_ASSERT(ulSsrc == pMember->m_ulSSRC);
        return TRUE;
    }
    else
    {
        pMember = NULL; 
        return FALSE;
    }
}

BOOL
CPurePlaySource::RemoveMember(UINT32 ulSsrc, BOOL bRemoveFromBin)
{
    BOOL bTheErr = FALSE;
    CMember*	 pMember = NULL;
    if (m_SsrcToMember.Lookup(ulSsrc, (void*&)pMember))
    {
	HX_ASSERT(ulSsrc == pMember->m_ulSSRC);

	if (bRemoveFromBin)
	{
	    // remove this entry from bin.
	    m_MemberTimeoutBins.RemoveEntry(pMember->m_binInfo);
	}	    

	bTheErr = m_SsrcToMember.RemoveKey((ULONG32)ulSsrc);
	HX_DELETE(pMember);
    }
    return bTheErr;
}


/*
*   There are two maps that we have to take care of SSRC map and ID map
*   (i.e. m_SsrcToStream && m_StrmIdToStrm)
*   SSRC map:
*   - In any case, remove this key from SSRC map.
* 
*   ID map:
*   - If it didn't make the source admission cut, this source is not in this 
*     map.  Just delete the obj
*   - If it did make the cut:
*     - !(RA && WantClientStats) -> remove this key and delete obj
*     - RA && WantClientStats -> this is the only case in which we want to keep
*	this obj around.  SO leave this key in ID map, and don't delete it
*/
BOOL
CPurePlaySource::RemoveSource(UINT32 ulSsrc)
{
    BOOL bTheErr = FALSE;
    CStream* pStrm = NULL;
    if (m_SsrcToStream.Lookup(ulSsrc, (void*&)pStrm))
    {	
	HX_ASSERT(pStrm->m_ulSSRC == ulSsrc);

	// first, remove this key from SSRC map
	bTheErr = m_SsrcToStream.RemoveKey((LONG32)ulSsrc);

	if (pStrm->m_bMadeCut)
	{
	    // m_unStreamNum should be correct
	    UINT16 nStrmNum = pStrm->m_unStreamNum;
	    
	    if (!(m_pPlugin->WantClientStats() &&
	          m_pPlugin->IsRAStream(nStrmNum)))
	    {
		bTheErr = m_StrmIdToStrm.RemoveKey(nStrmNum);
		HX_DELETE(pStrm);
	    }
	    
	    // this stream is done
    	    m_pPlugin->StreamDone(HXR_OK, nStrmNum);	    
	}
	else
	{
#ifdef _DEBUG
	    // make sure this obj is not in ID map
	    for (CHXMapLongToObj::Iterator i  = m_StrmIdToStrm.Begin();
		 i != m_StrmIdToStrm.End();
    	         ++i)
	    {
		if (((CStream*)(*i)) == pStrm)
		{
		    // this can't happen
		    HX_ASSERT(!"Source obj which didnt make the cut is in ID map");
		}
	    }
	    
#endif
	    HX_DELETE(pStrm);
	}
    } 

    return bTheErr;
}

BOOL
CPurePlaySource::AddNewEntryToMap(UINT32 ulSsrc, CStream*& pStrm)
{
    BOOL bMadeCut;
    BOOL bAddAsMasterSync = FALSE;

#ifdef XXXGo_DEBUG
    if (m_pLogFile)
    {
	fprintf(m_pLogFile, "NewSrc #%u\n", ulSsrc);
	fflush(m_pLogFile);
    }    
    if (m_pPlugin->IsRMPresentation())
    {
	if (GetTrueNumSrc() > 1)
	{
	    // if this is RM presentation, there shouldn't be more
	    // than one stream here!
	    HX_ASSERT(FALSE);
	}
    }
#endif

    bMadeCut = (!m_pPlugin->IsSourceAdmissionClosed());

#ifdef _USE_MASTER_STREAM_SYNC_SCHEME
    bAddAsMasterSync = (bMadeCut &&
		        (m_StreamType == CStream::STREAM_TYPE_AUDIO) &&
		        (!m_pPlugin->HasMasterSyncStream()));
#endif	// _USE_MASTER_STREAM_SYNC_SCHEME

    pStrm = new CStream(this, 
			ulSsrc, 
			0, 
			m_ulHXFactor,
			m_ulRTPFactor,
			m_pPlugin->m_pClassFactory,
			bAddAsMasterSync); 
					
    if (!pStrm)
    {
        return FALSE;
    }

    pStrm->m_bMadeCut = bMadeCut;
    if (bMadeCut)
    {
	if (m_bStartTimeDistributed)
	{
	    pStrm->m_Syncer.SetStartTime(m_ulStartTime, TRUE);
	}
	if (m_bSyncAnchorDistributed)
	{
	    pStrm->m_Syncer.AnchorSync(m_ulHXAnchorTime,
				       m_ulNTPHXTime);
	}
	if (m_bSyncDistributed)
	{
	    pStrm->m_Syncer.HandleMasterSync(m_ulHXMasterTime,
					     m_lHXOffsetToMaster);
	}
    }

    m_SsrcToStream.SetAt(ulSsrc, (void*)pStrm);

    if (bAddAsMasterSync)
    {
	m_pPlugin->SetMasterSyncStream(pStrm);
    }
	
    return TRUE;
}

BOOL
CPurePlaySource::AddNewMember(UINT32 ulSsrc, CMembe