P2P编程

开发平台：
Visual C++

EncryptedConnection.cpp：源码内容
							// EncryptedConnection.cpp: implementation of the CEncryptedConnection class.
//
//////////////////////////////////////////////////////////////////////
#include "stdafx.h"
#include "testbt.h"
#include "EncryptedConnection.h"
#include "Encrypter.h"
#include "Connector.h"
#include "Connection.h"
#include "download.h"
#ifdef _DEBUG
#undef THIS_FILE
static char THIS_FILE[]=__FILE__;
#define new DEBUG_NEW
#endif
const string strProtocolName = "BitTorrent protocol";
const long lHeaderLen[7] = {1, 
							strlen(strProtocolName.data()), 
							8, 20, 20, 4, 1};
char* tobinary(long lnum, char* pBuf, long length=4);
//////////////////////////////////////////////////////////////////////
// Construction/Destruction
//////////////////////////////////////////////////////////////////////
CEncryptedConnection::CEncryptedConnection(CDownloaderFeedback* pMain, SOCKET hsocket,CConnector* pconnector, CEncrypter* pEncrypter, memstream& memPeerid, long lConnectionID)
{
	m_pMain = pMain;
	m_pconnector = pconnector;
	m_pEncrypter = pEncrypter;
	m_hsocket = hsocket;
	m_memPeerid = memPeerid;
	m_bLocallyInitiated = (m_memPeerid.size() > 0) ? true:false;
	time(&m_lLastHit);
	memset(m_ReadBufferTemp, 0, READ_BUFFER_LEN );
	// select event.
	m_hReadEvent = CreateEvent(0, false, false, 0);
	assert(m_hReadEvent);
	int iRet = WSAEventSelect(hsocket, m_hReadEvent , FD_READ|FD_WRITE|FD_CLOSE);
	if (iRet == SOCKET_ERROR) assert(false);	
	m_bCompleted = false;
	m_eStep = eread_header_len;
	m_lReadMessageLength = 0;	
	//
	// get ip or id.
	//
	m_lConnectionID = lConnectionID;
	sockaddr_in addr = {0};
	int len = sizeof(addr);
	iRet = getpeername(m_hsocket, (sockaddr*)&addr, &len);
	if (iRet != 0)
	{
		string strErr = WSAShowError();
		assert(false);
	}
	m_lAddr = addr.sin_addr.S_un.S_addr;
	m_sPort = addr.sin_port;
}
CEncryptedConnection::~CEncryptedConnection()
{
	CloseHandle(m_hReadEvent);
	closesocket(m_hsocket);
}
bool CEncryptedConnection::GetIP(long& lAddr, short& sPort)
{
	lAddr = m_lAddr;
	sPort = m_sPort;
	return true;
}
string CEncryptedConnection::GetIP()
{
	assert(m_hsocket);
	sockaddr_in addr = {0};
	int len = sizeof(addr);
	int iRet = getpeername(m_hsocket, (sockaddr*)&addr, &len);
	if (iRet != 0)
	{
		string strErr = WSAShowError();
		assert(false);
		return strErr;
	}
	string strRet(inet_ntoa(addr.sin_addr));
	strRet += ":" + ltostring(ntohs(addr.sin_port));
	return strRet;
}
void CEncryptedConnection::Shutdown(int how)
{
	assert(m_hsocket);
	int iRet = shutdown(m_hsocket, how);
	if (iRet != 0)
		throw string ("shutdown socket error");
}
bool CEncryptedConnection::IsFlush()
{
	return m_WriteBuffer.size() <= 0;
}
void CEncryptedConnection::Send(char *pBuf, long lBufLen)
{
	assert(m_hsocket);
	m_WriteBuffer.write(pBuf, lBufLen);
	if (m_WriteBuffer.size() > 0)
		try_send();
}
void CEncryptedConnection::try_send()
{
	while (m_WriteBuffer.size() > 0)
	{
		int iRet = send(m_hsocket, (char*)m_WriteBuffer, m_WriteBuffer.size(), 0);
		if (iRet == SOCKET_ERROR)
		{
			if (::WSAGetLastError() == WSAEWOULDBLOCK)
				break;
			else
			{
				string e = WSAShowError();				
				// m_pEncrypter->m_connections_died.push_back(this);
				Close();
				return;
			}
		}
		else // succeed.
		{
			m_WriteBuffer.TrimLeft(iRet);
		}
	}
	/*
	int iRet = 0;
	if (m_WriteBuffer.size() > 0)
		iRet = WSAEventSelect(m_hsocket, m_hReadEvent , FD_READ|FD_WRITE|FD_CLOSE);
	else
		iRet = WSAEventSelect(m_hsocket, m_hReadEvent , FD_READ|FD_CLOSE);
	if (iRet == SOCKET_ERROR) 
	{
		assert(false);
	}
	//*/
}
void CEncryptedConnection::handle_events()
{	
	// string strPeerIp = GetIP();
	WSANETWORKEVENTS events;
	int iRet = WSAEnumNetworkEvents(m_hsocket, m_hReadEvent, &events);
	if (iRet == SOCKET_ERROR) 
	{
		string e = WSAShowError();
		assert(false);
		return;
	}
	
	if (events.lNetworkEvents & FD_CLOSE)
	{
		switch(events.iErrorCode[FD_CLOSE_BIT])
		{
		case WSAENETDOWN:
			break;
		case WSAECONNRESET:
			break;
		case WSAECONNABORTED:
			break;
		}
		int i = 0;
		
		Close();
		return;
	}
	if (events.lNetworkEvents & FD_WRITE)
	{
		// TRACE("try_send() rn");
		try_send();
		if (IsFlush())
		{
			if (m_bCompleted)
				m_pconnector->connection_flushed(this);
		}
		/*
		iRet = send(m_hsocket, (char*)m_WriteBuffer, m_WriteBuffer.size(), 0);
		if (iRet == SOCKET_ERROR)
		{
			Close();
			string e = WSAShowError();
			assert(false);
			return;
		}
		else // succeed.
		{
			m_WriteBuffer.TrimLeft(iRet);
		}
		//*/
	}
	if (events.lNetworkEvents & FD_READ)
	{		
		iRet = recv(m_hsocket, m_ReadBufferTemp, READ_BUFFER_LEN, 0);
		if (iRet == SOCKET_ERROR)
		{
			Close();
			assert(false);
			return;
		}
		else if (iRet == 0) // connection closed.
		{
			Close();
			return;
		}
		else // succeed.
		{			
			data_came_in(m_ReadBufferTemp, iRet);		
		}
	}
	
}
///////////////////////////////////////////////////////////////////////////
//
// data transfer.
//
///////////////////////////////////////////////////////////////////////////
void CEncryptedConnection::connection_init()
{
	//
	// format header and send.
	// 
	char zero[8] = {0};
	memset(zero, 0, 8);
	memstream memHeader;
	char cprosize = strProtocolName.size();
	memHeader.write(&cprosize, 1);
	memHeader.write(strProtocolName.data(), strProtocolName.size());
	memHeader.write(zero, sizeof(zero));
	memHeader.write((char*)m_pEncrypter->m_memInfohash, m_pEncrypter->m_memInfohash.size());
	memHeader.write((char*)m_pEncrypter->m_memMyId, m_pEncrypter->m_memMyId.size());
	// string strPeer = GetIP();
	Send((char*)memHeader, memHeader.size());
}
void CEncryptedConnection::CloseDied()
{	
	m_pMain->ShowEncryptedMessage("CEncryptedConnection CloseDied()");
	if (m_bCompleted)
		m_pconnector->connection_lost(this);
	vector<CEncryptedConnection*>::iterator it = 
		find(m_pEncrypter->m_connections.begin(), m_pEncrypter->m_connections.end(), this);
	
	if (it != m_pEncrypter->m_connections.end())
	{
		m_pEncrypter->m_connections.erase(it);
		delete this;
	}
	else assert(false);
}
void CEncryptedConnection::Close()
{
	vector<CEncryptedConnection*>::iterator it = 
		find(m_pEncrypter->m_connections.begin(), m_pEncrypter->m_connections.end(), this);
	assert(it != m_pEncrypter->m_connections.end());
	// if the connectionalready in deletes, skip it.	
	it = find(m_pEncrypter->m_connections_died.begin(), m_pEncrypter->m_connections_died.end(), this);
	if (it != m_pEncrypter->m_connections_died.end())
	{
		// assert(false);
		TRACE("if the connectionalready in deletes, skip it.rn");
		return;
	}
	m_pEncrypter->m_connections_died.push_back(this);
	return;
	/*
	ShowMessage(this, "CEncryptedConnection close()");
	if (m_bCompleted)
	{
		m_pconnector->connection_lost(this);
	}
	vector<CEncryptedConnection*>::iterator it = 
		find(m_pEncrypter->m_connections.begin(), m_pEncrypter->m_connections.end(), this);
	
	if (it != m_pEncrypter->m_connections.end())
	{
		m_pEncrypter->m_connections.erase(it);
		delete this;
	}
	else assert(false);
	//*/
}
#define FORMAT_ERR 2
void CEncryptedConnection::data_came_in(char *pBuf, long length)
{
	// string strPeerIp = GetIP();
	m_ReadBuffer.write(pBuf, length);
	int bRet = true;
	while (bRet)
	{
		switch(m_eStep)
		{
		case eread_header_len:
			bRet = read_header_len();
			break;
		case eread_header:
			bRet = read_header();
			break;
		case eread_reserved:
			bRet = read_reserved();
			break;
		case eread_download_id:
			bRet = read_download_id();
			break;
		case eread_peer_id:
			bRet = read_peer_id();
			break;
		case eread_len:
			bRet = read_len();
			break;
		case  eread_message:
			bRet = read_message();
			break;
		}
		
		if (bRet == (int)true)
		{
			if (eread_message == m_eStep)
				m_eStep --;
			else
				m_eStep ++;
		}
		else if (bRet == FORMAT_ERR)
		{
			m_pMain->ShowEncryptedMessage(string("close HeadData format error peers : ") + GetIP());
			Close();
			return;
		}
	}
}
int  CEncryptedConnection::read_header_len()
{
	if (m_ReadBuffer.size() < 1)
		return false;
	
	char* pbuf = m_ReadBuffer;
	char c = *pbuf;
	m_ReadBuffer.TrimLeft(1);
	if (c != strlen(strProtocolName.data()))
		return FORMAT_ERR;
	m_pMain->ShowEncryptedMessage(string("HeadData begin read peers : ") + GetIP());
	return true;
}
int   CEncryptedConnection::read_header()
{
	if (m_ReadBuffer.size() < strlen(strProtocolName.data()))
		return false;
	char* pbuf = new char[strlen(strProtocolName.data()) + 1];
	memset(pbuf, 0, strlen(strProtocolName.data()) + 1);
	memcpy(pbuf, m_ReadBuffer, strlen(strProtocolName.data()));
	m_ReadBuffer.TrimLeft(strlen(strProtocolName.data()));
	if (string(pbuf) != strProtocolName)
	{
		delete pbuf;
		return FORMAT_ERR;
	}
	
	delete pbuf;
	return true;
}
int  CEncryptedConnection::read_reserved()
{
	if (m_ReadBuffer.size() < 8)
		return false;
	m_ReadBuffer.TrimLeft(8);
	return true;
}
int   CEncryptedConnection::read_download_id()
{
	if (m_ReadBuffer.size() < 20)
		return false;
	char pbuf [20];
	memcpy(pbuf, m_ReadBuffer, 20);
	m_ReadBuffer.TrimLeft(20);
	char* pInfohash = (char*)m_pEncrypter->m_memInfohash;
	for (int i=0; i<20; i++)
	{
		if (pbuf[i] != *(pInfohash+i))
		{
			return FORMAT_ERR;
		}
	}
	return true;
}
int   CEncryptedConnection::read_peer_id()
{
	if (m_ReadBuffer.size() < 20)
		return false;
	if (m_memPeerid.size() <= 0) // connected by peer, peerid unknown.
		m_memPeerid.write(m_ReadBuffer, 20);
	char pbuf [20];
	memcpy(pbuf, m_ReadBuffer, 20);
	m_ReadBuffer.TrimLeft(20);
	char* pPeerid = (char*)m_memPeerid;
	for (int i=0; i<20; i++)
	{
		if (pbuf[i] != *(pPeerid+i))
		{
			return FORMAT_ERR;
		}
	}
	m_pconnector->connection_made(this);
	m_bCompleted = true;
	m_pMain->ShowEncryptedMessage(string("HeadData format succed peers : ") + GetIP());
	return true;
}
int   CEncryptedConnection::read_len()
{
	if (m_ReadBuffer.size() < 4)
		return false;
	unsigned char len[4] = {0};	
	memcpy(len, m_ReadBuffer, 4);
	m_lReadMessageLength = len[0]*0x1000000 + len[1]*0x10000 + len[2]*0x100 + len[3];
	m_ReadBuffer.TrimLeft(4);
	if (m_lReadMessageLength > m_pEncrypter->GetMaxLen())
		return FORMAT_ERR;
	return true;
}
int   CEncryptedConnection::read_message()
{
	if (m_ReadBuffer.size() < m_lReadMessageLength)
		return false;
	if (m_lReadMessageLength <= 0)
		return true;
	
	memstream membuf;
	membuf.write(m_ReadBuffer, m_lReadMessageLength);
	m_ReadBuffer.TrimLeft(m_lReadMessageLength);
	// m_pconnector->got_message(this, membuf, membuf.size());
	m_pconnector->got_message(this, membuf);
	return true;
}
void CEncryptedConnection::send_message(char *pBuf, long length)
{
	long len = length;
	
	char lenBuf[4] = {0};
	tobinary(len, lenBuf);
	memstream mBuf;
	mBuf.write(lenBuf, sizeof(long));
	mBuf.write(pBuf, length);
	Send(mBuf, mBuf.size());
}
bool CEncryptedConnection::is_locally_initiated()
{
	return m_bLocallyInitiated;
}
memstream& CEncryptedConnection::GetID()
{
	return m_memPeerid;
}
string CEncryptedConnection::GetInfohashString()
{	
	// assert(m_memPeerid.size() == 20);
	string strRet;
	char szText[20] = {0};
	for (int i=0; i<m_memPeerid.size(); i++)
	{
		if (((unsigned char) m_memPeerid[i] )< 0x10)
			sprintf(szText, "0%x", (unsigned char) m_memPeerid[i]);
		else
			sprintf(szText, "%x", (unsigned char) m_memPeerid[i]);
		strRet += szText;
	}
	return strRet;
}

						