commonopt.cpp

跨操作系统的微型中间件

#include "CommonOpt.h"
#include "OS.h"
#include "MyAssert.h"

using namespace std;

#define  STREAM_JUMP_TIME_DELTA  10   // 10 s

void  CloseSocket(int fd)
{
#ifdef WIN32
	::closesocket(fd);
#else
	::close(fd);
#endif
	
}


std::string GetMovieName(const std::string &vURL)
{
	Int32 nPos = vURL.rfind("/");
	if(nPos != string::npos)
	{
		nPos+=1;
		return vURL.substr(nPos, vURL.length() - nPos);
	}
	else
	{
		nPos = vURL.rfind("\\");
		if(nPos != string::npos)
		{
			nPos+=1;
			return vURL.substr(nPos, vURL.length() - nPos);
		}
	}
	return string("");
}

Bool ParserURL(const std::string &vURL, std::string &outIP, UInt16 &outPort)
{
	Int32 nPos = vURL.find("://");
	if(nPos != string::npos)
	{
		Int32 nPos1 = vURL.find(":",nPos + 3);
		if(nPos1 != string::npos)
		{
			outIP = vURL.substr(nPos +3, nPos1 - nPos - 3);
			outPort = atoi(vURL.c_str() + nPos1 + 1);
		}
		else
		{
			nPos1 = vURL.find("/",nPos + 3);
			outIP = vURL.substr(nPos +3, nPos1 - nPos - 3);
			outPort = 0 ;
		}
		return TRUE;
	}
	return FALSE;
}

Bool ReplaceIpInURL(std::string &vURL, std::string inIP )
{
	Int32 nPos = vURL.find("://");
	if(nPos != string::npos)
	{
		Int32 nPos1 = vURL.find(":",nPos + 3);
		if(nPos1 != string::npos)
		{
			vURL.replace(nPos+3,nPos1-nPos-3,inIP);
		}
		else
		{
			nPos1 = vURL.find("/",nPos + 3);
			vURL.replace(nPos+3,nPos1-nPos-3,inIP);
		}
		return TRUE;
	}
	return FALSE;
}

// for paser rm stream,for test ,added by samsmith
Bool ParserRMPacket(char* inPacket,UInt32 inLen,UInt8* outFlag1,UInt16* outSeqNo,UInt8* outFlag2,UInt32* outTS,UInt32* outLen,UInt16 inPayload)
{
	*outFlag1=inPacket[0];  // get flag1 , contain flag + stream id
	if ((*outFlag1!=0x40)&&(*outFlag1!=0x42))
	{
		if(inPacket[2] == 0x06) {
			printf("Stream EOF detected\n");
			*outSeqNo=ntohs(*((UInt16*)(inPacket+3)));   //add by c05059
			*outTS = ntohl(*((UInt32*)(inPacket+5)));    //add by c05059
			return FALSE; // STREAM EOF
		}
		*outFlag1=inPacket[9];
		*outSeqNo=ntohs(*(UInt16*)(inPacket + 10));
		*outFlag2 = inPacket[12];
		*outTS = ntohl(*((UInt32*)(inPacket+13)));
	}
	else
	{
		*outSeqNo=ntohs(*(UInt16*)(inPacket + 10));
		*outFlag2 = inPacket[3];
		*outTS = ntohl(*((UInt32*)(inPacket+4)));
	}
	return TRUE;
}
Bool ParserRTPPacket(char* inPacket,UInt32 inLen,UInt16& outSeqNo,UInt32& outTS)
{
	if( !inPacket || inLen <= sizeof( tagRTPHEADER ) )
		return FALSE;
	
	tagRTPHEADER* pRtpHeader = (tagRTPHEADER*)inPacket;
	
	outSeqNo = ntohs(pRtpHeader->SeqNo);
	outTS = ntohl(pRtpHeader->TimeStamp);
	
	return TRUE;
	
}
Bool ModifyRTPPacket(char* inPacket,UInt32 inLen,UInt16 inSeqNo,UInt32 inTS)
{
	if( !inPacket || inLen <= sizeof( tagRTPHEADER ) )
		return FALSE;
	
	tagRTPHEADER* pRtpHeader = (tagRTPHEADER*)inPacket;
	
	pRtpHeader->SeqNo = htons( inSeqNo );
	pRtpHeader->TimeStamp = htonl( inTS );
	
	return TRUE;
}

#include "SocketUtils.h"
#include <iostream>
#include "LogInterface.h"
#include <time.h>
#include "Globle.h"

#include "MyAssert.h"

void CTimeStampExt::ConvertTimeStampToUI64(UInt64& TimeStamp)
{
	// ============================= 把时间戳转换为 64 位 =========================
	if( IsSetLastTimeStampFlag() )
	{
		UInt64 lastTimeStampExt = GetLastTimeStampExt( );
		UInt32 lastTimeStamp = GetLastTimeStamp( );
		SetLastTimeStamp( (UInt32)TimeStamp );
		Int32 Delta = (Int32)TimeStamp - (Int32)lastTimeStamp;
		
		TimeStamp = lastTimeStampExt + Delta;
		SetLastTimeStampExt( TimeStamp );
	}
	else
	{
		SetLastTimeStamp( (UInt32)TimeStamp );
		SetLastTimeStampExt( TimeStamp );
		SetLastTimeStampFlag();
	}
	// ------------------------------------------------------------------------------
}


CTSConvert::CTSConvert()
{
	m_uiBaseTrackID = INVALID_VALUE;
	
};

Bool  CTSConvert::AddTrack(UInt32 uiTrackID,UInt32 uiTS,Bool bBase,UInt32* pSameTrackArray,UInt32 uiSameNum)
{
	MAPTRACK::iterator it = m_mapTrack.find(uiTrackID);
	if( it != m_mapTrack.end() ) // this track have added 
	{
		return FALSE;
	}
	
	CTSConvert::TRACK  track;
	track.uiTimeScale = uiTS;
	if( pSameTrackArray !=0 && uiSameNum > 0 )
	{
		for( track.uiSameTrackNum = 0;track.uiSameTrackNum< uiSameNum;track.uiSameTrackNum++)
		{
			track.uiSameTrack[track.uiSameTrackNum] = pSameTrackArray[track.uiSameTrackNum];
		}
	}
	
	
	if( bBase && m_uiBaseTrackID==INVALID_VALUE )// only set once
	{
		m_uiBaseTrackID = uiTrackID;
	}
	m_mapTrack[uiTrackID] = track;
	
	return TRUE;
}
void  CTSConvert::ResetParam()
{
	for( MAPTRACK::iterator it = m_mapTrack.begin(); it != m_mapTrack.end();it++ )
	{
		TRACK* pTrack = &(it->second);
		pTrack->ResetParam();
	}
}

OS_Error  CTSConvert::ConvertRTP(UInt32 uiTrackID,char* pRtpData,UInt16 uiLen,PACKET_TYPE dt,UInt64& ms)
{
	Assert( m_uiBaseTrackID != INVALID_VALUE);// must select one track as base track.
	Assert( uiLen > 4 );
	
	MAPTRACK::iterator it = m_mapTrack.find(uiTrackID);
	if( it == m_mapTrack.end() ) 
	{
		Assert(FALSE);
	}
	
	TRACK* pTrack = &(it->second);
	
	return AdjustRtpTimestamp(uiTrackID,pTrack,pRtpData,dt,ms);
}

// 功能：把各路流的时间戳转换成64位，然后根据几路流的偏差来调整时间戳，并且获得毫妙时间。
OS_Error CTSConvert::AdjustRtpTimestamp(UInt32 vTrackID,TRACK *pTrackInfo,char *pRtpData,PACKET_TYPE dt,UInt64& rstpTimeStamp)
{
	Assert( pRtpData && pTrackInfo );
	
	UInt64 ts = 0;
	
	if( dt == DTYPE_RDT )
	{ 
		UInt8  flag1,flag2;
		UInt16 seqno;
		UInt32 timestamp,len;
		ParserRMPacket(pRtpData,0,&flag1,&seqno,&flag2,&timestamp,&len);
		ts = timestamp;
	}
	else
	{
		tagRTPHEADER* pRtpHeader = (tagRTPHEADER*)pRtpData;
		ts = ntohl(pRtpHeader->TimeStamp);
	}
	UInt64 test = ts;
	
	pTrackInfo->m_TsConverters.ConvertTimeStampToUI64(ts);
	
	if( pTrackInfo->uiFstTStamp == INVALID_VALUE ) // 该路流未接收到第一个包.
	{
		// 设置与之具有相同PAYLOAD的流的第一包的时间戳 .
		for( int i=0;i< pTrackInfo->uiSameTrackNum;i++)
		{
			TRACK* pTrack = FindTrackInfo(pTrackInfo->uiSameTrack[i]);
			pTrack->uiFstTStamp = ts;
		}
		pTrackInfo->uiFstTStamp = ts;
		std::cout << "IN RTPLIVE MODULE--->Get Track:" <<vTrackID << "data.The timescale of the first RTP packet is:" 
			<< (UInt32)ts << std::endl;
	}
	
	if( pTrackInfo->uiTimeOffset == INVALID_VALUE ) // 该路流未计算出与基准流的 时间 偏差.
	{
		TRACK * pBaseTrack = FindTrackInfo(m_uiBaseTrackID); // 如果基准流有SAME PAYLOAD 的TRACK,那么,就使用那个流 .
		if( pBaseTrack==0 )
			return OS_Err;
		if( pBaseTrack->uiFstTStamp != INVALID_VALUE )
		{
			pTrackInfo->uiTimeOffset = pBaseTrack->uiFstTStamp*1000/pBaseTrack->uiTimeScale - 
				pTrackInfo->uiFstTStamp*1000/pTrackInfo->uiTimeScale;
			std::cout << "IN RTPLIVE MODULE--->Calculate the time offset of track:" <<vTrackID<< "is:" 
				<< (UInt32)pTrackInfo->uiTimeOffset << std::endl;
		}
	}
	
	if( pTrackInfo->uiTimeOffset != INVALID_VALUE ) // 该路流未计算出与基准流的 时间 偏差.
	{
		Int64 tmp = pTrackInfo->uiTimeOffset;
		tmp *= pTrackInfo->uiTimeScale;		tmp /= 1000;
		
		tmp +=  ts;
		
		UInt32 ts32 = (UInt32)tmp;
		
		tmp *= 1000;
		rstpTimeStamp = tmp;
		rstpTimeStamp /= pTrackInfo->uiTimeScale;
		
		ts32 = htonl(ts32);
		*(UInt32*)(pRtpData+4) = ts32;
		
		return OS_NoErr;
	}
	
	return OS_Err;
}

CTSConvert::TRACK* CTSConvert::FindTrackInfo(UInt32 vTrackID)
{
	TRACK* pTrack = 0;
	
	MAPTRACK::iterator it = m_mapTrack.find( vTrackID );
	if( it !=m_mapTrack.end() )
	{
		pTrack = &(it->second);
	}
	
	return pTrack;
}


/////////////////////////////////////////////////////////////////////////////////////////////////////

#define   MIN_VALUE   0x00001000
#define   MAX_VALUE   0x01000000
CRtpSortContainer::~CRtpSortContainer()
{
	reset();
}
void CRtpSortContainer::ClearBuf()
{
	RTP_SORT_VEC::iterator it = m_vecRtp.begin();
	printf("recaliam %d\n",m_vecRtp.size());
	for(;it!=m_vecRtp.end();it++)
	{
		m_alloctor.Reclaim(*it); 
	}
	
	m_vecRtp.clear();
	
	m_bRecvFirstPacket = FALSE;
}

void CRtpSortContainer::reset()
{
	ClearBuf();
	
	m_uiBufLen = INVALID_VALUE;
	
}

OS_Error CRtpSortContainer::Flow(const char* pData,UInt16 len,PACKET_TYPE dt,RTP_SORT*pItem)
{
	Assert(pData && m_uiBufLen!=INVALID_VALUE);
	
	OS_Error  ret = PushPacket(pData,len,dt);
	if( OS_NoErr != ret )
		return 	ret;
	
	return PopPacket(pItem);
}

OS_Error CRtpSortContainer::PushPacket(const char* pData,UInt16 len,PACKET_TYPE dt)
{
	UInt64 ts = 0;
	if( dt == DTYPE_RDT )
	{
		UInt8  flag1,flag2;
		UInt16 seqno;
		UInt32 timestamp,uiLen;
		ParserRMPacket((char*)pData,len,&flag1,&seqno,&flag2,&timestamp,&uiLen,0);
		ts = (UInt64)timestamp; // get timestamp
	}
	else
	{
		tagRTPHEADER* pRtpHeader = (tagRTPHEADER*)pData;
		ts = (UInt64)ntohl(pRtpHeader->TimeStamp); // get timestamp
	}
	
	if( m_bRecvFirstPacket )
	{
		Int32 delta = (Int32)( (Int64)ts - (Int64)GetMaxTimeStamp() );
		// 两个包的时间差大于2倍 PACKET BUF长度时,则判断流断续,发生跳越. 
		if( abs(delta) >= STREAM_JUMP_TIME_DELTA*m_timescale )
		{
			const char* pLastData = GetBiggestPacketData();
			UInt32 ts = 0;
			UInt16 seqno = 0;
			if( pLastData )
			{
				tagRTPHEADER* pRtpHeader = (tagRTPHEADER*)pLastData;
				ts = ntohl(pRtpHeader->TimeStamp);
				seqno = ntohs( pRtpHeader->SeqNo );
			}
			UInt32 nowTS;
			UInt16 nowSeqno;
			tagRTPHEADER* pRtpHeader = (tagRTPHEADER*)pData;
			nowTS = ntohl(pRtpHeader->TimeStamp);
			nowSeqno = ntohs( pRtpHeader->SeqNo );
			
			char log[1024];
			sprintf(log,"IN RTPSORTCONTAINER  MODULE--->delta:%d,buflen:%u,timscale:%u,interval:%u\n!",delta,m_uiBufLen,m_timescale,m_uiBufLen+STREAM_JUMP_TIME_DELTA*m_timescale);
			RUN_LOGGER(0,0,log);
			
			sprintf(log,"IN RTPSORTCONTAINER  MODULE--->stream %u jumped!the last packet[seqno:%u,timestamp:%u],current packet[seqno:%u,timestamp:%u]!",m_uiStreamType,seqno,ts,nowSeqno,nowTS);
			RUN_LOGGER(0,0,log);
			
			m_bRecvFirstPacket = FALSE;
			
			return OS_ERR_STREAM_JUMP;
		}
	}
	else
	{
		m_bRecvFirstPacket = TRUE;
	}
	
	RTP_SORT item;
	item.pData = new char[len];