emsocket.cpp

非常出名开源客户端下载的程序emule

			// Security: Check for buffer overflow (2MB)
			if(pendingPacket->size > sizeof(GlobalReadBuffer)) {
				delete pendingPacket;
				pendingPacket = NULL;
				OnError(ERR_TOOBIG);
				return;
			}

			// Init data buffer
			pendingPacket->pBuffer = new char[pendingPacket->size + 1];
			pendingPacketSize = 0;
		}

		// Bytes ready to be copied into packet's internal buffer
		ASSERT(rptr <= rend);
		uint32 toCopy = ((pendingPacket->size - pendingPacketSize) < (uint32)(rend - rptr)) ? 
			             (pendingPacket->size - pendingPacketSize) : (uint32)(rend - rptr);

		// Copy Bytes from Global buffer to packet's internal buffer
		memcpy(&pendingPacket->pBuffer[pendingPacketSize], rptr, toCopy);
		pendingPacketSize += toCopy;
		rptr += toCopy;
		
		// Check if packet is complet
		ASSERT(pendingPacket->size >= pendingPacketSize);
		if(pendingPacket->size == pendingPacketSize){
			#ifdef EMSOCKET_DEBUG
			EMTrace("CEMSocket::PacketReceived on %d, opcode=%X, realSize=%d", 
				    (SOCKET)this, pendingPacket->opcode, pendingPacket->GetRealPacketSize());
			#endif

			// Process packet
			PacketReceived(pendingPacket);
			delete pendingPacket;	
			pendingPacket = NULL;
			pendingPacketSize = 0;
		}
	}

	// Finally, if there is any data left over, save it for next time
	ASSERT(rptr <= rend);
	ASSERT(rend - rptr < PACKET_HEADER_SIZE);
	if(rptr != rend) {
		// Keep the partial head
		pendingHeaderSize = rend - rptr;
		memcpy(pendingHeader, rptr, pendingHeaderSize);
	}	
}

void CEMSocket::SetDownloadLimit(uint32 limit){	
	downloadLimit = limit;
	downloadLimitEnable = true;	
	
	// CPU load improvement
	if(limit > 0 && pendingOnReceive == true){
		OnReceive(0);
	}
}

void CEMSocket::DisableDownloadLimit(){
	downloadLimitEnable = false;

	// CPU load improvement
	if(pendingOnReceive == true){
		OnReceive(0);
	}
}

bool CEMSocket::SendPacket(Packet* packet, bool delpacket,bool controlpacket){
	//EMTrace("CEMSocket::OnSenPacked1 linked: %i, controlcount %i, standartcount %i, isbusy: %i",m_bLinkedPackets, controlpacket_queue.GetCount(), standartpacket_queue.GetCount(), IsBusy());
	if (!delpacket){
		ASSERT ( !packet->IsSplitted() );
		Packet* copy = new Packet(packet->opcode,packet->size);
		memcpy(copy->pBuffer,packet->pBuffer,packet->size);
		packet = copy;
	}			
	if ( ( (!IsConnected()) || IsBusy() ) || ( m_bLinkedPackets && controlpacket ) ){
		if (controlpacket){
			controlpacket_queue.AddTail(packet);
			return true;
		}
		else{
			standartpacket_queue.AddTail(packet);
			return true;
		}
	}
	bool bCheckControlQueue = false;
	if (packet->IsLastSplitted() ){
		m_bLinkedPackets = false;
		bCheckControlQueue = true;
	}
	else if (packet->IsSplitted())
		m_bLinkedPackets = true;
	else if (m_bLinkedPackets)
		ASSERT (false);
	//EMTrace("CEMSocket::OnSenPacked2 linked: %i, controlcount %i, standartcount %i, isbusy: %i",m_bLinkedPackets, controlpacket_queue.GetCount(), standartpacket_queue.GetCount(), IsBusy());
	Send(packet->DetachPacket(),packet->GetRealPacketSize());
	delete packet;
	if (!IsBusy() && bCheckControlQueue)
		OnSend(0);
	return true;
}

void CEMSocket::OnSend(int nErrorCode){
	if (nErrorCode){
		OnError(nErrorCode);
		return;
	}

	//EMTrace("CEMSocket::OnSend linked: %i, controlcount %i, standartcount %i, isbusy: %i",m_bLinkedPackets, controlpacket_queue.GetCount(), standartpacket_queue.GetCount(), IsBusy());

	if (byConnected == ES_DISCONNECTED)
		return;
	else
		byConnected = ES_CONNECTED;

	if (IsBusy())
		Send(0,0,0);
	if (IsBusy())
		return;
	while (controlpacket_queue.GetHeadPosition() != 0 && (!IsBusy()) && IsConnected() && !m_bLinkedPackets){
		Packet* cur_packet = controlpacket_queue.GetHead();
//		EMTrace("CEMSocket::OnSend sending control packet on %d, size=%u",(SOCKET)this, cur_packet->GetRealPacketSize());
		Send(cur_packet->DetachPacket(),cur_packet->GetRealPacketSize());
		controlpacket_queue.RemoveHead();
		delete cur_packet;
	}

	while (standartpacket_queue.GetHeadPosition() != 0 && (!IsBusy()) && IsConnected()){
		Packet* cur_packet = standartpacket_queue.GetHead();
		if (cur_packet->IsLastSplitted() )
			m_bLinkedPackets = false;
		else if (cur_packet->IsSplitted())
			m_bLinkedPackets = true;
		else if (m_bLinkedPackets)
			ASSERT (false);
//		EMTrace("CEMSocket::OnSend sending standart packet on %d, size=%u",(SOCKET)this, cur_packet->GetRealPacketSize());
		Send(cur_packet->DetachPacket(),cur_packet->GetRealPacketSize());
		standartpacket_queue.RemoveHead();
		delete cur_packet;
	}

	while (controlpacket_queue.GetHeadPosition() != 0 && (!IsBusy()) && IsConnected() && !m_bLinkedPackets){
		Packet* cur_packet = controlpacket_queue.GetHead();
//		EMTrace("CEMSocket::OnSend sending control packet on %d, size=%u",(SOCKET)this, cur_packet->GetRealPacketSize());
		Send(cur_packet->DetachPacket(),cur_packet->GetRealPacketSize());
		controlpacket_queue.RemoveHead();
		delete cur_packet;
	}

}

int CEMSocket::Send(char* lpBuf,int nBufLen,int nFlags){
	//EMTrace("CEMSocket::Send linked: %i, controlcount %i, standartcount %i, isbusy: %i",m_bLinkedPackets, controlpacket_queue.GetCount(), standartpacket_queue.GetCount(), IsBusy());
	ASSERT (sendbuffer == NULL || lpBuf == NULL );
	if (lpBuf){
		sendbuffer = lpBuf;
		sendblen = nBufLen;
		sent = 0;
	}
	while (true){
		uint32 tosend = sendblen-sent;
		if (tosend > MAXFRAGSIZE)
			tosend = MAXFRAGSIZE;
		ASSERT (tosend != 0);
		uint32 result = CAsyncSocket::Send(sendbuffer+sent,tosend);
		if (result == (uint32)SOCKET_ERROR){
			uint32 error = GetLastError();
			if (error == WSAEWOULDBLOCK){
				break;
			}
			else{
				//OnError(error);
				return -1;
			}
		}
		sent += result;
		ASSERT (sent <= sendblen);
		if (sent == sendblen){
			delete[] sendbuffer;
			sendbuffer = 0;
			sent = 0;
			sendblen = 0;
			break;
		}
	}
	return 0;
}

// pach2:
// written this overriden Receive to handle transparently FIN notifications coming from calls to recv()
// This was maybe(??) the cause of a lot of socket error, notably after a brutal close from peer
// also added trace so that we can debug after the fact ...
int CEMSocket::Receive(void* lpBuf, int nBufLen, int nFlags)
{
//	EMTrace("CEMSocket::Receive on %d, maxSize=%d",(SOCKET)this,nBufLen);
	int recvRetCode = CAsyncSocket::Receive(lpBuf,nBufLen,nFlags);
	switch (recvRetCode) {
	case 0:
		//EMTrace("CEMSocket::##Received FIN on %d, maxSize=%d",(SOCKET)this,nBufLen);
		// FIN received on socket // Connection is being closed by peer
		//ASSERT (false);
		if ( 0 == AsyncSelect(FD_CLOSE|FD_WRITE) ) { // no more READ notifications ...
			int waserr = GetLastError(); // oups, AsyncSelect failed !!!
			ASSERT(false);
		}
		return 0;
	case SOCKET_ERROR:
		switch(GetLastError()) {
		case WSANOTINITIALISED:
			ASSERT(false);
			EMTrace("CEMSocket::OnReceive:A successful AfxSocketInit must occur before using this API.");
			break;
		case WSAENETDOWN:
			ASSERT(true);
			EMTrace("CEMSocket::OnReceive:The socket %d received a net down error",(SOCKET)this);
			break;
		case WSAENOTCONN: // The socket is not connected. 
			EMTrace("CEMSocket::OnReceive:The socket %d is not connected",(SOCKET)this);
			break;
		case WSAEINPROGRESS:   // A blocking Windows Sockets operation is in progress. 
			EMTrace("CEMSocket::OnReceive:The socket %d is blocked",(SOCKET)this);
			break;
		case WSAEWOULDBLOCK:   // The socket is marked as nonblocking and the Receive operation would block. 
			EMTrace("CEMSocket::OnReceive:The socket %d would block",(SOCKET)this);
			break;
		case WSAENOTSOCK:   // The descriptor is not a socket. 
			EMTrace("CEMSocket::OnReceive:The descriptor %d is not a socket (may have been closed or never created)",(SOCKET)this);
			break;
		case WSAEOPNOTSUPP:  // MSG_OOB was specified, but the socket is not of type SOCK_STREAM. 
			break;
		case WSAESHUTDOWN:   // The socket has been shut down; it is not possible to call Receive on a socket after ShutDown has been invoked with nHow set to 0 or 2. 
			EMTrace("CEMSocket::OnReceive:The socket %d has been shut down",(SOCKET)this);
			break;
		case WSAEMSGSIZE:   // The datagram was too large to fit into the specified buffer and was truncated. 
			EMTrace("CEMSocket::OnReceive:The datagram was too large to fit and was truncated (socket %d)",(SOCKET)this);
			break;
		case WSAEINVAL:   // The socket has not been bound with Bind. 
			EMTrace("CEMSocket::OnReceive:The socket %d has not been bound",(SOCKET)this);
			break;
		case WSAECONNABORTED:   // The virtual circuit was aborted due to timeout or other failure. 
			EMTrace("CEMSocket::OnReceive:The socket %d has not been bound",(SOCKET)this);
			break;
		case WSAECONNRESET:   // The virtual circuit was reset by the remote side. 
			EMTrace("CEMSocket::OnReceive:The socket %d has not been bound",(SOCKET)this);
			break;
		default:
			EMTrace("CEMSocket::OnReceive:Unexpected socket error %x on socket %d",GetLastError(),(SOCKET)this);
			break;
		}
		return SOCKET_ERROR;
	default:
//		EMTrace("CEMSocket::OnReceive on %d, receivedSize=%d",(SOCKET)this,recvRetCode);
		return recvRetCode;
	}
	return SOCKET_ERROR;
}