sendstream.cpp

组播发送测试网络程序,服务器端

// SendStream.cpp: implementation of the CSendStream class.
//
//////////////////////////////////////////////////////////////////////

#include "stdafx.h"
#include "NetServer.h"
#include "SendStream.h"

#ifdef _DEBUG
#undef THIS_FILE
static char THIS_FILE[]=__FILE__;
#define new DEBUG_NEW
#endif

//////////////////////////////////////////////////////////////////////
// Construction/Destruction
//////////////////////////////////////////////////////////////////////

CSendStream::CSendStream()
{
	//int l=sizeof(BlockBuff);
    //int k=sizeof(DWORD);    
 memset(&BlockBuff,0,sizeof(BlockBuff));
 //SetPackageSize(DEFAULT_PACKAGE_SIZE);
 
m_fStreamBitrate=0;
m_dwPackageSize=0;


 sprintf(BlockBuff.pData,"this is chi test package ");
 m_bIsLimitedBitrate=TRUE;

 m_hIOMUTEX=CreateMutex(NULL, FALSE,NULL);
}

CSendStream::~CSendStream()
{
 Close();

 CloseHandle(m_hIOMUTEX);
}

void CSendStream::SetBitrate(DWORD dwbitrate)
{
    if(dwbitrate>MAX_NET_BITRATE)
		m_bIsLimitedBitrate=FALSE;
	else
		m_bIsLimitedBitrate=TRUE;

	m_fStreamBitrate=dwbitrate;
	

   CheckTransportPara();
}

DWORD CSendStream::GetRealBitrate()
{
 return m_dwRealbitrate;
}

DWORD CSendStream::StreamThread()
{   TRACE("send thead");
     int  nTimeCount=0;
     m_hTimerEvent = CreateEvent(NULL,TRUE,FALSE,NULL);
     UINT uDelay=BS_TIMER_DELAY ;
	DWORD dwRet=0,dwsize=0;  
    //BlockBuff.dwDataSize=DATA_BUFFER_LENGTH;
    mmTimerID = timeSetEvent(	uDelay ,
								uDelay ,
								(LPTIMECALLBACK)m_hTimerEvent,
								1,
								TIME_PERIODIC|TIME_CALLBACK_EVENT_SET);
 dwPreTime= timeGetTime();
HANDLE hMultipleHandles[4] = {m_hTimerEvent,m_hBeginPlay,
                            m_hStopPlay,m_hTerminateThread};
for(;;){
      	dwRet = WaitForMultipleObjects(4, hMultipleHandles,FALSE,INFINITE);  
		switch(dwRet){
		case  WAIT_OBJECT_0://time envent
			ResetEvent(m_hTimerEvent);
	 //	TRACE("time %d\n", dwPreTime);
           if(m_zStreamStatus!=S_S_Running)
				break;
		 //if(m_bIsLimitedBitrate&&!BitRateIsValid(dwPreTime,m_fStreamBitrate,m_dwRealSize))
		//			 break;	 // Bitrate is invalid
		  
		   if(--nTimeCount>0)
			   break;

			dwsize=SendMultiData();//Data((char *)&BlockBuff,sizeof(BlockBuff));
		 	TRACE("block index=%d spansend=%d each count=%d ttl=%d\n",
				BlockBuff.dwBlockIndex,
				m_dwBlockSpanTime,m_dwBlockCountEach,
				timeGetTime()-dwPreTime);
			m_dwNow=dwPreTime=timeGetTime();
			
		    nTimeCount= m_dwBlockSpanTime;
			
			if(dwsize>0)
				SetProInfo(dwsize);
			break;
		case WAIT_OBJECT_0+1://Begin play
			ResetEvent(m_hBeginPlay);
			TRACE(".............STart\n");
			if(m_zStreamStatus!=S_S_Running){
			 	dwCurMSGDataSize=0;	
				dwLastMSGTime=dwPreTime= timeGetTime();
			     nTimeCount=m_dwBlockSpanTime;
				m_zStreamStatus=S_S_Running;
			}
			break;
        case WAIT_OBJECT_0+2://stop play
			ResetEvent(m_hStopPlay);
            TRACE("Stop=============\n");  

			m_zStreamStatus=S_S_Stoped;
			
			break;
        case  WAIT_OBJECT_0+3:// terminated 
			timeKillEvent(mmTimerID);
			CloseHandle(m_hTimerEvent);
			TRACE("is close thread\n");
		    return 0;
			break;
		default:
          	return CLOSETHREADCODE;
		}// end switch
} // end for
 return 0;
}

void CSendStream::SetProInfo(int nSize)
{
  DWORD dwspan;
 // m_nDelayCount--;
   dwspan=m_dwNow-dwLastMSGTime;
   dwCurMSGDataSize+=nSize;
    if(dwspan>=MON_DELAY_TIME) {
		m_nDelayCount=MON_BLOCK_SPACE;
         m_dwRealbitrate=PubGetBitrate(dwspan,dwCurMSGDataSize);
		 TRACE("...datasize=%d ttl=%d\n",dwCurMSGDataSize,dwspan);
		 dwCurMSGDataSize=0;
        dwLastMSGTime=m_dwNow;
	    ::SendMessage(g_hMonWnd,MSG_SEND_PROGRESS,BlockBuff.dwBlockIndex,
		 m_dwRealbitrate);
	
	}// send mess
}

void CSendStream::Init(char *strIP, char * strlocalIP,DWORD dwPort,int ntype)
{
 
	CBaseStream::SetSocketType(ntype);
	CBaseStream::Create(strIP,strlocalIP,dwPort,g_nTTL);

}

DWORD CSendStream::SendMultiData()
{
  int  dwSize=0;
  
 // WaitForSingleObject(m_hIOMUTEX,INFINITE );
  for(register int  i=0;i<m_dwBlockCountEach;i++)
  {	dwSize+=SendData((char *)&BlockBuff,m_dwPackageSize);
     TRACE("index=%d size=%d\n",i,dwSize);
    BlockBuff.dwBlockIndex++;
  }  
 // ReleaseMutex(m_hIOMUTEX);
  return dwSize;
}

void CSendStream::SetPackageSize(DWORD dwPackageSize)
{
 
 m_dwPackageSize=dwPackageSize;
// BlockBuff.dwDataSize=dwPackageSize-8;
 
 CheckTransportPara();
//m_dwPerSendCount=(MAX_UDP_PACKAGE_SIZE)/dwPackageSize;
 
//m_dwPerSendCount=1;

//m_dwRealSize=m_dwPerSendCount*m_dwPackageSize;

//sprintf(BlockBuff.pData,"ilovetisgamek");
}

void CSendStream::FillBlock(char *pinfo)
{
 int nItemLen=strlen(pinfo);
  int nAllCopySize=0;
  int buffersize=DATA_BUFFER_LENGTH;
  int nBlockCount=buffersize/nItemLen;
  int nCopyLen=nItemLen;

  TRACE("Start fill block\n");
  WaitForSingleObject(m_hIOMUTEX,INFINITE );

  for(int i=0;i<nBlockCount;i++){
     if(nAllCopySize+nCopyLen>buffersize)
		 nCopyLen= buffersize-nAllCopySize;

	 memcpy(BlockBuff.pData+nAllCopySize,pinfo,nCopyLen);
	 
	 nAllCopySize+=nCopyLen;

    //  if(i<3000)
	//	  continue;
	// TRACE("index=%d allsize =%d len=%d\n",
	//	 i,nAllCopySize,nCopyLen);
  }// end for 
  BlockBuff.pData[nAllCopySize]='\0'; 
  
  ReleaseMutex(m_hIOMUTEX);

  TRACE("End fill\n");
}

void CSendStream::CheckTransportPara()
{
  if(m_fStreamBitrate==0||m_dwPackageSize==0) return;
   
  float ftempbitrate,fblocksizenew;
  
  	if(m_dwPackageSize<1024)
		m_dwPackageSize=1024;
  
  WaitForSingleObject(m_hIOMUTEX,INFINITE );
   CalculatePara(m_fStreamBitrate,m_dwPackageSize);
   ftempbitrate=CalculateBitrate(); 
   TRACE("size=%d bitrate=%.2f\n",
	   m_dwPackageSize,ftempbitrate);
   m_dwPackageSize=CalculateBlockSize(m_fStreamBitrate);
   ftempbitrate=CalculateBitrate();

   TRACE("size=%d bitrate=%.2f dettt=%d count=%d\n",
	   (DWORD)m_dwPackageSize,ftempbitrate,
	   m_dwBlockSpanTime,m_dwBlockCountEach);
   // CalculatePara(ftempbitrate,fblocksizenew);
  
   BlockBuff.dwDataSize=m_dwPackageSize-8;
  ReleaseMutex(m_hIOMUTEX);

}

DWORD CSendStream::GetPackageSize()
{
 return m_dwPackageSize;
}

void CSendStream::CalculatePara(float fBitrate, float fPackageSize)
{
 if(fBitrate==0||fBitrate==0) return;
   
  float ftemp= fBitrate/fPackageSize/8000;

  
   if(ftemp<=1)
   { 
    m_dwBlockSpanTime=1/ftemp;
    m_dwBlockCountEach=1;
   }
   else
   {
     m_dwBlockSpanTime=1;
     m_dwBlockCountEach=ftemp;
   }


}

DWORD  CSendStream::CalculateBitrate()
{
 return (float)m_dwPackageSize*(float)m_dwBlockCountEach*1000/(float)m_dwBlockSpanTime*8;
}

DWORD  CSendStream::CalculateBlockSize(float  dwbitrate)
{

 return (DWORD)(double)dwbitrate*m_dwBlockSpanTime/(double)m_dwBlockCountEach/8000;
}

DWORD CSendStream::GetCountEach()
{
 return m_dwBlockCountEach;
}

DWORD CSendStream::GetSpanTime()
{
 return m_dwBlockSpanTime;
}

void CSendStream::Close()
{
   memset(&BlockBuff,0,sizeof(BlockBuff));
   m_fStreamBitrate=0;
   m_dwPackageSize=0;

   CBaseStream::Close();
}