P2P编程

开发平台：
Visual C++

Rerequester.cpp：源码内容
							// Rerequester.cpp: implementation of the CRerequester class.
//
//////////////////////////////////////////////////////////////////////
#include "stdafx.h"
#include "testBT.h"
#include "Rerequester.h"
#include "bdecode.h"
#include "BTFormat.h"
#include "Download.h"
#include "Connector.h"
#include "Measure.h"
#include "StorageWrapper.h"
#ifdef _DEBUG
#undef THIS_FILE
static char THIS_FILE[]=__FILE__;
#define new DEBUG_NEW
#endif
string quote(const char* pBuf, long length);
// void OpenUrl(string strUrl, memstream& buf);
//////////////////////////////////////////////////////////////////////
// Construction/Destruction
//////////////////////////////////////////////////////////////////////
CRerequester::CRerequester(long lConnectErrorMax, long lConnectIntval, long minpeers, long maxpeers)
{	
	m_lConnectErrorMax = lConnectErrorMax;
	m_lConnectIntval = lConnectIntval;
	m_minpeers = minpeers;
	m_maxpeers = maxpeers;
}
void CRerequester::SetConnectErrorMax(long lConnectErrorMax, long lConnectIntval, long minpeers, long maxpeers)
{
	m_lConnectErrorMax = lConnectErrorMax;
	m_lConnectIntval = lConnectIntval;
	m_minpeers = minpeers;
	m_maxpeers = maxpeers;
}
void CRerequester::Create(CDownloaderFeedback* pMain, CStorageWrapper* pStorageWrapper, CMeasure* pupmeasure,  
						  CMeasure* pdownmeasure, CConnector* pConnector,CEncrypter* pEncrypter, 
						  string strUrl, long port, string strIP, char* pMyId, char* pInfoHash,
						  long lInterval, 
						   long lTimeOut, HANDLE hevDone)
{
	m_pMain = pMain;
	m_pConnector = pConnector;
	m_pEncrypter = pEncrypter;
	m_pStorageWrapper = pStorageWrapper;
	m_pupmeasure = pupmeasure;
	m_pdownmeasure = pdownmeasure;
	m_hevDone = hevDone;
	m_lTimeOut = lTimeOut;
	m_lInterval = lInterval;
	m_lAannounce_interval = lInterval; //60*30;
	m_strTrackerid = "";
	m_lLast = 0;
	
	m_lFileCount = 0;
	m_hHttpDone = CreateEvent(0, false, false, 0);
	m_tLast = 0;
	
	m_bCommand = false;
	m_bStarted = false;
	m_never_succeeded = true;	
	m_pSingle = 0;
	// check to see if this is a reasonable URL
	CString strServerName;
	CString strObject;
	INTERNET_PORT nPort;
	DWORD dwServiceType;
	if (!AfxParseURL(strUrl.data(), dwServiceType, strServerName, strObject, nPort) ||
		dwServiceType != INTERNET_SERVICE_HTTP)
	{
		throw string("url is not correct");
	}
	char szport[100] = {0};
	sprintf(szport, "%d", port);
	m_strUrl = strUrl + "?info_hash=" + 
		quote(pInfoHash, 20) + "&peer_id=" + quote(pMyId, 20) + "&port=" + szport;
	if (!strIP.empty())
		m_strUrl += "&ip=" + quote(strIP.data(), strIP.size());
	// auto announce start.
	announce(REQ_STARTED);
	m_pMain->ShowSystemMessage("获取种子信息", CSystemMsg::eCmd);
	string strMsg = "开始连接服务器... : ";
	strMsg += m_strUrl;
	m_pMain->ShowSystemMessage(strMsg, CSystemMsg::eMsgOut);
	time(&m_tLast);
	time(&m_tStartLast);
}
CRerequester::~CRerequester()
{
	assert(!m_pSingle);
	CloseHandle(m_hHttpDone);
}
void CRerequester::Close()
{
	if (m_pSingle)
	{
		delete m_pSingle;
		m_pSingle = 0;
	}
	
	if (!m_never_succeeded)
	{
		bool bRet = false;
		int iCount = 20;
		HANDLE hSleep = CreateEvent(0, true, false, 0);
		while (iCount-- > 0)
		{
			announce(REQ_STOPPED);
			if (!m_pSingle)
			{
				CloseHandle(hSleep);
				assert(false);
				return;
			}
			string strErr;
			m_pSingle->WaitTerminate();			
			bRet = m_pSingle->GetResult(strErr) != 0;			
			delete m_pSingle;
			m_pSingle = 0;
			if (bRet) break;
			WaitForSingleObject(hSleep, m_lConnectIntval * CLOCKS_PER_SEC);
		}
		CloseHandle(hSleep);
		if (bRet)
			TRACE("***announce stop successedrn");
		else
			TRACE("***announce stop failedrn");
	}
}
HANDLE CRerequester::GetHttpEvent()
{
	return m_hHttpDone;
}
void CRerequester::AnnounceEx(long lEvent)
{
	assert(lEvent >= 0 && lEvent < 4);
	// cause memory leak.
	m_commands.push_back(lEvent);
	Excute();
}
void CRerequester::Excute()
{
	if (m_pSingle) return;
	time_t ltime = 0;
	time(&ltime);
	if (ltime < m_tLast)
	{
		assert(false);
		m_tLast = ltime;
	}
	
	if (ltime < m_tStartLast)
	{
		m_tStartLast = ltime;
		m_tLast = ltime;
	}
	
	bool bRet = false;
	if (m_commands.size() > 0)
	{
		bRet = true;
		m_bCommand = true;
		announce((*m_commands.begin()));
		TRACE("rnrnhttp Command : %d rnrn", (*m_commands.begin()));
	}
	else
	{
		m_bCommand = false;
		int lAannounce_interval = m_never_succeeded ? m_lConnectIntval : m_lAannounce_interval;
		
		if ((ltime - m_tStartLast) >= lAannounce_interval)
		{
			bRet = true;
			m_tStartLast = ltime;
			announce(REQ_STARTED);
			TRACE("nrnhttp Command REQ_STARTEDrnrn ");
		}
		else
		{
			int interval = m_lInterval;
			if (!m_bStarted)
			{
				m_bStarted = true;
				interval /= 2;
			}
			if ((ltime - m_tLast) > interval)
			{
				if ((m_pConnector->how_many_connections() < m_minpeers))
				{
					bRet = true;
					announce();
					TRACE("nrnhttp Command REQ_QUERYrnrn ");
				}
				m_tLast = ltime;
			}
		}
	}
	if (bRet)
	{
		string strMsg = "开始连接服务器... : ";
		// strMsg += m_strUrl;
		m_pMain->ShowSystemMessage(strMsg, CSystemMsg::eMsgOut);
	}
}
void CRerequester::PostExcute()
{
	assert(m_pSingle);
	string strErr;
	if (m_pSingle->GetResult(strErr))
	{
		if (m_bCommand)
		{
			assert(m_commands.size() > 0);			
			m_commands.erase(m_commands.begin());
		}
		
		if (postrequest())
			m_never_succeeded = false;
	}
	else
	{
		char szText[1024] = {0};
		sprintf (szText, "连接服务器失败 : (%s) : (%d)秒后重试...", strErr.data(), 
			m_never_succeeded ? m_lConnectIntval : m_lAannounce_interval);
		m_pMain->ShowSystemMessage(szText, CSystemMsg::eBad);
		if (m_never_succeeded)
		{
			if (++m_lFileCount>= m_lConnectErrorMax)
			{
				m_pMain->SetBadMsg("超过失败次数, 下载/上传停止");
			}
		}
	}
	delete m_pSingle;
	m_pSingle = 0;
	
	time(&m_tLast);
	m_tStartLast = m_tLast;
}
void CRerequester::announce(long lEvent) //(HANDLE hHttpDone, long lEvent)
{
	//
	// format url.
	//
	string strCurUrl = m_strUrl + "&uploaded=" + ltostring(m_pupmeasure->get_total()) + 
		"&downloaded=" + ltostring(m_pdownmeasure->get_total()) + 
		"&left=" + ltostring(m_pStorageWrapper->get_amount_left());
	if (m_lLast != 0)
	{
		char strLast[100] = {0};
		sprintf(strLast, "%d", m_lLast);
		strCurUrl += "&last=" + quote(strLast, strlen(strLast));
	}
	if (!m_strTrackerid.empty())
		strCurUrl += "&trackerid=" + quote(m_strTrackerid.data(), m_strTrackerid.size());
	if (m_pConnector->how_many_connections() >= m_maxpeers)
		strCurUrl += "&numwant=0";
	const string strEventArr[] = {"started", "completed", "stopped"};
	if (lEvent < 0 || lEvent > 3)
		throw string("announce() error: req event must be 0<x<3");
	if (lEvent != REQ_QUERY)
		strCurUrl += "&event=" + strEventArr[lEvent];
	
	TRACE("rnrnurl : %srnrn", strCurUrl.data());
	assert(!m_pSingle);
	m_pSingle = new CRerequesterSingle();
	m_pSingle->OpenUrlEx(strCurUrl, m_hHttpDone, lEvent);
}
bool CRerequester::postrequest()
{
	assert(m_pSingle);
	assert(m_pSingle->m_RetStream.size() > 0);
	try
	{
		CVal* pResponse = 0;
		CBdecode dec;
		pResponse = dec.bdecode((char*)m_pSingle->m_RetStream, m_pSingle->m_RetStream.size());
		if (!pResponse)
			throw string("postrequest() error : decode resonse from http fail");
		auto_ptr<CVal> aResponse(pResponse);		
		CBTFormat::check_peers(pResponse);
		if (HasKey(pResponse->pmapVal, "failure reason"))
		{
			// m_pMain->errorFunc(string("rejected by tracker - ") + (*pResponse)["failure reason"]->pstrVal);
			m_pMain->ShowSystemMessage(string("被服务器拒绝 - ") + (*pResponse)["failure reason"]->pstrVal, CSystemMsg::eBad);
		}
		else
		{
			if (HasKey(pResponse->pmapVal, "interval"))
				m_lAannounce_interval = (*pResponse)["interval"]->lVal;
			if (HasKey(pResponse->pmapVal, "min interval"))
				m_lInterval = (*pResponse)["min interval"]->lVal;
			if (HasKey(pResponse->pmapVal, "tracker id"))
				m_strTrackerid = (*pResponse)["tracker id"]->pstrVal;
			if (HasKey(pResponse->pmapVal, "last"))
				m_lLast = (*pResponse)["last"]->lVal;
			//*********************************
			// meaning of last unknown.wait ...
			//*********************************
			CVal* pPeers = (*pResponse)["peers"];
			long lps = pPeers->size() + m_pConnector->how_many_connections();
			if (lps < m_maxpeers)
			{
				long lNumPeers = 1000;
				if (HasKey(pResponse->pmapVal, "num peers"))
					lNumPeers = (*pResponse)["num peers"]->lVal;
				long lDonePeers = 0;
				if (HasKey(pResponse->pmapVal, "done peers"))
					lDonePeers = (*pResponse)["done peers"]->lVal;
				bool bDownloadDone = ( WaitForSingleObject(m_hevDone, 0) == WAIT_OBJECT_0);
				if (bDownloadDone)
				{
					if ((lNumPeers - lDonePeers) > (lps * 1.2))
						m_lLast = 0;
				}
				else
				{
					if (lNumPeers > (lps * 1.2) )
						m_lLast = 0;
				}
			}
			//
			// connect to peers.
			//
			TRACE("rnrnbegin...rn");
			for (int i=0; i<pPeers->size(); i++)
			{
				CVal* pPeer = (*pPeers)[i];
				string strIP = (*pPeer)["ip"]->pstrVal;
				long lPort = (*pPeer)["port"]->lVal;
				string strPeerId = (*pPeer)["peer id"]->pstrVal;
				// sprintf(strAll + strlen(strAll), "%s, %d, %srn", strIP.data(), lPort, quote(strPeerId.data(), (*pPeer)["peer id"]->lstrLen).data());
				char strAll[4096] = {0};
				// sprintf(strAll, "%s, %d, %srn", strIP.data(), lPort, quote(strPeerId.data(), (*pPeer)["peer id"]->lstrLen).data());
				sprintf(strAll, "%s, %d, %srn", strIP.data(), lPort, quote((*pPeer)["peer id"]->pstrVal, (*pPeer)["peer id"]->lstrLen).data());
				TRACE("%s", strAll);
				memstream memPeerID;
				memPeerID.write((*pPeer)["peer id"]->pstrVal, (*pPeer)["peer id"]->lstrLen);
				m_pEncrypter->start_connection(strIP, lPort, memPeerID);
			}
			TRACE("}rn");
			m_pMain->TrackerConnected();
			char strText[1024] = {0};
			sprintf(strText, "连接服务器成功, 返回(%d)条记录", pPeers->size());
			m_pMain->ShowSystemMessage(strText, CSystemMsg::eMsgIn);
			m_pMain->ShowSystemMessage("开始接收数据...", CSystemMsg::eCmd);
		}
	}
	catch (string& e)
	{
		e.data();
		m_pMain->ShowSystemMessage("连接服务器失败, 返回错误信息!等待重试...", CSystemMsg::eBad);
		return false;
	}
	return true;
}
/*
Replace special characters in string using the "%xx" escape. Letters, digits, 
and the characters "_,.-" are never quoted. The optional safe parameter 
specifies additional characters that should not be quoted -- its default value is '/'. 
Example: quote('/~connolly/') yields '/%7econnolly/'. 
*/
string quote(const char* pBuf, long length)
{
	string strRet;
	for (int i=0; i<length; i++)
	{
		unsigned char c = *(pBuf+i);
		if ((c > 64 && c<91) || (c > 96 && c < 123) || (c>47 && c<58) ||
			c== '_' || c == ',' || c=='.' || c=='-')
		{
			strRet += c;
		}
		else 
		{
			char temp[100] = {0};
			memset(temp, 0, sizeof(temp)/sizeof(char));
			if (c > 0xf)
				sprintf(temp, "%x", c);
			else
				sprintf(temp, "0%x", c);
			strRet += string("%") + temp;
		}
	}
	// Make the string upper case.
	// strRet = _strupr(_strdup(strRet.data()));
	return strRet;
}
/////////////////////////////////////////////////////////////////////////////////////////////
// CRerequesterSingle.
CRerequesterSingle::CRerequesterSingle()
{
	m_hDone = CreateEvent(0, true, true, 0);
	m_hHttpDone = 0;
	m_iResult = false;
}
CRerequesterSingle::~CRerequesterSingle()
{
	if (!IsEventSet(m_hDone))
	{
		// m_session.Close();
		WaitForSingleObject(m_hDone, INFINITE);
	}
	CloseHandle(m_hDone);
}
void CRerequesterSingle::WaitTerminate()
{
	WaitForSingleObject(m_hDone, INFINITE);
}
bool CRerequesterSingle::OpenUrlEx(string strUrl, HANDLE hHttpDone, long lEvent)
{
	m_lEvent = lEvent;
	m_hHttpDone = hHttpDone;
	ResetEvent(m_hHttpDone);
	ResetEvent(m_hDone);
	// check to see if this is a reasonable URL
	CString strServerName;
	CString strObject;
	INTERNET_PORT nPort;
	DWORD dwServiceType;
	if (!AfxParseURL(strUrl.data(), dwServiceType, strServerName, strObject, nPort) ||
		dwServiceType != INTERNET_SERVICE_HTTP)
	{
		return false;
	}
	m_strUrl = strUrl;
	DWORD dwThreadId = 0;
	::CreateThread(0, 0, SendRequestProc, this, 0, &dwThreadId);
	TRACE("rnSendRequestProc thread = (%x)", dwThreadId);
	return true;
}
DWORD WINAPI CRerequesterSingle::SendRequestProc(void *pParam)
{
	CRerequesterSingle* pRerequesterSingle = (CRerequesterSingle*) pParam;
	HANDLE m_hDone = pRerequesterSingle->m_hDone;
	HANDLE hHttpDone = pRerequesterSingle->m_hHttpDone;
	pRerequesterSingle->SendRequest();
	SetEvent(m_hDone);
	assert(hHttpDone);
	SetEvent(hHttpDone);
	return 0;
}
void CRerequesterSingle::SendRequest()
{
	m_iResult = true;
	try
	{
		m_RetStream.clear();
		OpenUrl(m_strUrl, m_RetStream);
	}
	catch (string& e)
	{
		e;
		m_iResult = false;
	}
	/*
	assert(m_hHttpDone);
	SetEvent(m_hHttpDone);
	//*/
}
void CRerequesterSingle::OpenUrl(string strUrl, memstream& buf)
{
	CStdioFile* pfile = 0;
	try
	{
		pfile = m_session.OpenURL(strUrl.data());
		if (!pfile)
			throw string("OpenURL() fail");
		auto_ptr<CStdioFile> af(pfile);		
	
		char* pBuf = new char[1024];
		auto_ptr<char> ab(pBuf);
		DWORD dwRead = 0;
		do
		{
			dwRead = pfile->Read(pBuf, 1024);
			buf.write(pBuf, dwRead);
		}
		while (dwRead > 0);
		pfile->Close();
	}
	catch (CInternetException* pEx)
	{
		TCHAR szErr[1024];
		pEx->GetErrorMessage(szErr, 1024);
		pEx->Delete();
		m_strErr = szErr;
		if (pfile) 
			pfile->Close();
		throw string(szErr);
	}
	catch (string e)
	{
		throw e;
	}
	catch (...)
	{
		assert(false);
	}
}
int CRerequesterSingle::GetResult(string& strErr)
{
	strErr = m_strErr;
	return m_iResult;
}

						