recvagent.cpp

组播网络程序测试,对于学习组播技术有帮助

// RecvAgent.cpp: implementation of the CRecvAgent class.
//
//////////////////////////////////////////////////////////////////////

#include "stdafx.h"
#include "NetRecv.h"
#include "RecvAgent.h"
#include "RBaseStream.h"
#ifdef _DEBUG
#undef THIS_FILE
static char THIS_FILE[]=__FILE__;
#define new DEBUG_NEW
#endif

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

CRecvAgent::CRecvAgent()
{  char szbuff[MAX_PATH];
	m_bIsSave=false;
  fp=NULL;
  m_nSaveBlockCount=MAX_SAVE_BLOCK_NUM;
  m_nDelayCount=MON_BLOCK_SPACE;
  m_lRecvPackageCount= m_lLossPackageCount=0;
  m_nMaxBlockListSize=10;
  m_dwPackageSize=m_dwPreBlockIndex=0;
  m_dwPreRecvBlockIndex=0;
  GetCurrentDirectory(MAX_PATH,szbuff);
  sprintf(m_strOutFile,"%s\\%s",szbuff,OUT_FILE);  
}

CRecvAgent::~CRecvAgent()
{
 
}

DWORD CRecvAgent::RStreamThread()
{  DWORD dwRet;
   int nSize=0;
   //long lBlockIndexOff; 
  // int noff=0;
   //TRACE("in tt");
   DWORD dwFirstTime=dwLastMSGTime=timeGetTime();	
   HANDLE hMultipleEvent[2] = {m_hRecvEvent,m_hTerminateThread};
   for(;;){
	   dwRet=WaitForMultipleObjects(2,hMultipleEvent,FALSE, INFINITE);
	   switch(dwRet){
	   case  WAIT_OBJECT_0:// recv event
		   ResetEvent(m_hRecvEvent);
           nSize=RecvData(&databuff,sizeof(databuff));
		   if(m_dwPreRecvBlockIndex!=0&&m_dwPreRecvBlockIndex>databuff.dwBlockIndex)
                 CheckBlockListError();

		  m_dwPreRecvBlockIndex=databuff.dwBlockIndex;
		   if(nSize<=0)
			   break;
		   AddBlockInBlockList(&m_listBlock,&databuff);
		   TRACE("Recv index=%d\n",databuff.dwBlockIndex);
		 
		  if(!PopHeaderBlock(&m_listBlock,&databuff,m_nMaxBlockListSize))
			  break;
		 TRACE("Get Block index=%d count=%d\n",databuff.dwBlockIndex,m_nMaxBlockListSize);
		  m_lRecvPackageCount++;
		  CheckIsLoss();
		  dwPreTime=m_dwNow=timeGetTime();
          
		   if(m_bIsSave)
			   SaveData();
            SetProInfo(nSize);
		   break;
       case WAIT_OBJECT_0+1://TERMINATE THREAD
		   return 0;
		   break;
       default:// other
		   return CLOSETHREADCODE;
		   break;
	   }// end switch
   }// end for
 
	return 0;
}

void CRecvAgent::SetSave(bool bType)
{
 m_bIsSave=bType;
}

void CRecvAgent::Init(char *strIP, DWORD dwPort,bool bissave)
{ 
  m_bIsSave=bissave;
  SOCKADDR_IN m_HostAddr,  multiAddr;
  m_HostAddr.sin_family=multiAddr.sin_family=AF_INET;
  m_HostAddr.sin_addr.s_addr=INADDR_ANY;
  multiAddr.sin_addr.s_addr=inet_addr(strIP);
  multiAddr.sin_port=m_HostAddr.sin_port=dwPort;
  CRBaseStream::Create(m_HostAddr,multiAddr);
}

void CRecvAgent::SaveData()
{
	TRACE("Save %d\n",m_nSaveBlockCount);
	if(fp==NULL){
	if((fp=fopen(m_strOutFile,"wb"))==NULL){
		TRACE("open file error!!\n");
		return;
	}
	}// create file
	int nsize=fwrite(databuff.pData,sizeof(char),databuff.dwDataSize,fp);
    m_nSaveBlockCount--;
	if(m_nSaveBlockCount<=0){
	    fclose(fp);
		fp=NULL;
		TRACE("..................Clear all......................");
		m_nSaveBlockCount=MAX_SAVE_BLOCK_NUM;
       
	}
	
}

void CRecvAgent::SetProInfo(int nSize)
{
 DWORD dwspan;
  //m_nDelayCount--;
    dwspan=m_dwNow-dwLastMSGTime; 
    dwCurMSGDataSize+=nSize;
    if(dwspan>=MON_DELAY_TIME) {
	//	m_nDelayCount=MON_BLOCK_SPACE;
  //      dwspan=m_dwNow-dwLastMSGTime;
        m_dwRealbitrate=GetBitrate(dwspan,dwCurMSGDataSize);
		 //TRACE("...datasize=%d ttl=%d\n",dwCurMSGDataSize,dwspan);
		 dwCurMSGDataSize=0;
        dwLastMSGTime=m_dwNow;
		 m_dwPackageSize=nSize;
	 ::PostMessage(g_hMonWnd,MSG_RECV_PROGRESS,databuff.dwBlockIndex,
		 m_dwRealbitrate);
	
	}// send mess
}

void CRecvAgent::Close()
{
  CRBaseStream::CloseStream();
  
  m_nSaveBlockCount=MAX_SAVE_BLOCK_NUM;
  m_nDelayCount=MON_BLOCK_SPACE;
  m_lRecvPackageCount= m_lLossPackageCount=0;
  m_nMaxBlockListSize=10;
  m_dwPackageSize=m_dwPreBlockIndex=0;
  m_dwPreRecvBlockIndex=0;
  memset(&databuff,0,sizeof(databuff));
   m_dwRealbitrate=0;
  ::PostMessage(g_hMonWnd,MSG_RECV_PROGRESS,databuff.dwBlockIndex,
		 m_dwRealbitrate);
	
	//if(IsClose())
	//	TRACE("");
	if(fp!=NULL)
	{fclose(fp);
	fp=NULL;
	DeleteFile(m_strOutFile);
	}


}

long CRecvAgent::GetLossCount()
{ 
	return m_lLossPackageCount;
}

void CRecvAgent::CheckIsLoss()
{int nLossCount;
 if(m_dwPreBlockIndex==0)
 {m_dwPreBlockIndex=databuff.dwBlockIndex;
 return;}
 if((nLossCount=databuff.dwBlockIndex-m_dwPreBlockIndex)!=1)
 { m_lLossPackageCount+=nLossCount-1;
   TRACE("nlosscount=%d\n",nLossCount);
   zBlockInfo.nFlag=BLOCK_ERROR_LOSSBLOCK ;
   zBlockInfo.dwBlockIndex=databuff.dwBlockIndex;
   zBlockInfo.dwPreBlockIndex=m_dwPreBlockIndex;
   zBlockInfo.lIndexOff=nLossCount-1;
   ::PostMessage(g_hMonWnd,MSG_LIST_ERROR,0,(LPARAM)&zBlockInfo);
 } 
 m_dwPreBlockIndex=databuff.dwBlockIndex;
}

long CRecvAgent::GetRecvPackageCount()
{ return m_lRecvPackageCount;

}

long CRecvAgent::GetPackageSize()
{return m_dwPackageSize;

}

void CRecvAgent::SetBlockListSize(long lcount)
{ //char szbuff[32];
  m_nMaxBlockListSize=lcount;
  
 // sprintf(szbuff,"%d",m_nMaxBlockListSize);
  //AfxMessageBox(szbuff);
}

void CRecvAgent::CheckBlockListError()
{
 zBlockInfo.dwBlockIndex=databuff.dwBlockIndex;
 zBlockInfo.dwPreBlockIndex=m_dwPreRecvBlockIndex;

 zBlockInfo.lIndexOff=zBlockInfo.dwPreBlockIndex-zBlockInfo.dwBlockIndex;

 zBlockInfo.lTimeOff=zBlockInfo.lIndexOff*(timeGetTime()-dwPreTime);
 zBlockInfo.nFlag=BLOCK_ERROR_LIST;
 ::PostMessage(g_hMonWnd,MSG_LIST_ERROR,0,(LPARAM)&zBlockInfo);
}

char* CRecvAgent::GetPackageInfo(char *pbuffer, int len)
{
 if(databuff.dwDataSize==0)
	 sprintf(pbuffer,"pacakge is empty!!");
 else
 { 
	 memcpy(pbuffer,databuff.pData,len);
	 pbuffer[len]='\0';
 } 
	 return pbuffer;
}