ecsocket.cpp

电驴的MAC源代码

//
// This file is part of the aMule Project.
//
// Copyright (c) 2004-2008 aMule Team ( [e-mail] / http://www.amule.org )
// Copyright (c) 2004-2008 Angel Vidal Veiga ( [e-mail] )
//
// Any parts of this program derived from the xMule, lMule or eMule project,
// or contributed by third-party developers are copyrighted by their
// respective authors.
//
// This program is free software; you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation; either version 2 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
// 
// You should have received a copy of the GNU General Public License
// along with this program; if not, write to the Free Software
// Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301, USA
//

#include "ECSocket.h"

#include <sstream>
#include <iostream>
#include <algorithm>

using namespace std;

#include "ECPacket.h"		// Needed for CECPacket

#define EC_COMPRESSION_LEVEL	Z_BEST_COMPRESSION
#define EC_MAX_UNCOMPRESSED	1024

#ifndef __GNUC__
#define __attribute__(x)
#endif

// If your compiler gives errors on these lines, just remove them.
int utf8_mbtowc(wchar_t *p, const unsigned char *s, int n) __attribute__((__visibility__("internal")));
int utf8_wctomb(unsigned char *s, wchar_t wc, int maxlen) __attribute__((__visibility__("internal")));
int utf8_mb_remain(char c) __attribute__((__pure__));

/*----------=> Import from the Linux kernel <=----------*/
/*
 * linux/fs/nls_base.c
 */

/*
 * Sample implementation from Unicode home page.
 * http://www.stonehand.com/unicode/standard/fss-utf.html
 */
struct utf8_table {
	int     cmask;
	int     cval;
	int     shift;
	uint32_t  lmask;
	uint32_t  lval;
};

static struct utf8_table utf8_table[] =
{
    {0x80,  0x00,   0*6,    0x7F,           0,         /* 1 byte sequence */},
    {0xE0,  0xC0,   1*6,    0x7FF,          0x80,      /* 2 byte sequence */},
    {0xF0,  0xE0,   2*6,    0xFFFF,         0x800,     /* 3 byte sequence */},
    {0xF8,  0xF0,   3*6,    0x1FFFFF,       0x10000,   /* 4 byte sequence */},
    {0xFC,  0xF8,   4*6,    0x3FFFFFF,      0x200000,  /* 5 byte sequence */},
    {0xFE,  0xFC,   5*6,    0x7FFFFFFF,     0x4000000, /* 6 byte sequence */},
    {0,     0,      0,      0,              0,         /* end of table    */}
};

int utf8_mbtowc(uint32_t *p, const unsigned char *s, int n)
{
	uint32_t l;
	int c0, c, nc;
	struct utf8_table *t;

	nc = 0;
	c0 = *s;
	l = c0;
	for (t = utf8_table; t->cmask; t++) {
		nc++;
		if ((c0 & t->cmask) == t->cval) {
			l &= t->lmask;
			if (l < t->lval)
				return -1;
			*p = l;
			return nc;
		}
		if (n <= nc)
			return -1;
		s++;
		c = (*s ^ 0x80) & 0xFF;
		if (c & 0xC0)
			return -1;
		l = (l << 6) | c;
	}
	return -1;
}

int utf8_wctomb(unsigned char *s, uint32_t wc, int maxlen)
{
	uint32_t l;
	int c, nc;
	struct utf8_table *t;

	l = wc;
	nc = 0;
	for (t = utf8_table; t->cmask && maxlen; t++, maxlen--) {
		nc++;
		if (l <= t->lmask) {
			c = t->shift;
			*s = t->cval | (l >> c);
			while (c > 0) {
				c -= 6;
				s++;
				*s = 0x80 | ((l >> c) & 0x3F);
			}
			return nc;
		}
	}
	return -1;
}
/*----------=> End of Import <=----------*/

int utf8_mb_remain(char c)
{
	int i;
	for (i = 0; i < 5; ++i) {
		if ((c & utf8_table[i].cmask) == utf8_table[i].cval) break;
	}
	return i;
}


void CQueuedData::Write(const void *data, size_t len)
{
	const size_t canWrite = std::min(GetRemLength(), len);
	wxASSERT(len == canWrite);

	memcpy(m_wr_ptr, data, canWrite);
	m_wr_ptr += canWrite;
}


void CQueuedData::WriteAt(const void *data, size_t len, size_t offset)
{
	wxASSERT(len + offset <= m_data.size());
	if (offset > m_data.size()) {
		return;
	} else if (offset + len > m_data.size()) {
		len = m_data.size() - offset;
	}

	memcpy(&m_data[0] + offset, data, len);
}


void CQueuedData::Read(void *data, size_t len)
{
	const size_t canRead = std::min(GetUnreadDataLength(), len);
	wxASSERT(len == canRead);

	memcpy(data, m_rd_ptr, canRead);
	m_rd_ptr += canRead;
}


void CQueuedData::WriteToSocket(CECSocket *sock)
{
	wxCHECK_RET(m_rd_ptr < m_wr_ptr,
		wxT("Reading past written data in WriteToSocket"));
	
	sock->SocketWrite(m_rd_ptr, GetUnreadDataLength());
	m_rd_ptr += sock->GetLastCount();
}


void CQueuedData::ReadFromSocket(CECSocket *sock, size_t len)
{
	const size_t canWrite = std::min(GetRemLength(), len);
	wxASSERT(len == canWrite);

	sock->SocketRead(m_wr_ptr, canWrite);
	m_wr_ptr += sock->GetLastCount();
}


size_t CQueuedData::ReadFromSocketAll(CECSocket *sock, size_t len)
{
	size_t read_rem = std::min(GetRemLength(), len);
	wxASSERT(read_rem == len);

	//  We get here when socket is truly blocking
	do {
		// Give socket a 10 sec chance to recv more data.
		if ( !sock->WaitSocketRead(10, 0) ) {
			break;
		}

		wxASSERT(m_wr_ptr + read_rem <= &m_data[0] + m_data.size());
		sock->SocketRead(m_wr_ptr, read_rem);
		m_wr_ptr += sock->GetLastCount();
		read_rem -= sock->GetLastCount();

		if (sock->SocketError() && !sock->WouldBlock()) {
				break;
		}
	} while (read_rem);

	return len - read_rem;
}


size_t CQueuedData::GetLength() const
{
	return m_data.size();
}


size_t CQueuedData::GetDataLength() const
{
	const size_t len = m_wr_ptr - &m_data[0];
	wxCHECK_MSG(len <= m_data.size(), m_data.size(),
		wxT("Write-pointer past end of buffer"));

	return len;
}


size_t CQueuedData::GetRemLength() const
{
	return m_data.size() - GetDataLength();
}


size_t CQueuedData::GetUnreadDataLength() const
{
	wxCHECK_MSG(m_wr_ptr >= m_rd_ptr, 0,
		wxT("Read position past write position."));

	return m_wr_ptr - m_rd_ptr;
}



//
// CECSocket API - User interface functions
//

CECSocket::CECSocket(bool use_events)
:
m_use_events(use_events),
m_output_queue(),
m_in_ptr(EC_SOCKET_BUFFER_SIZE),
m_out_ptr(EC_SOCKET_BUFFER_SIZE),
m_curr_rx_data(new CQueuedData(EC_SOCKET_BUFFER_SIZE)),
m_curr_tx_data(new CQueuedData(EC_SOCKET_BUFFER_SIZE)),
m_rx_flags(0),
m_tx_flags(0),
m_my_flags(0x20 | EC_FLAG_ZLIB | EC_FLAG_UTF8_NUMBERS | EC_FLAG_ACCEPTS),
// setup initial state: 4 flags + 4 length
m_bytes_needed(8),
m_in_header(true)
{
	
}

CECSocket::~CECSocket()
{
	while (!m_output_queue.empty()) {
		CQueuedData *data = m_output_queue.front();
		m_output_queue.pop_front();
		delete data;
	}
}

bool CECSocket::ConnectSocket(uint32_t ip, uint16_t port)
{
	bool res;
#if wxCHECK_VERSION(2, 8, 8)
	res = InternalConnect(ip, port, !m_use_events);
#else
	res = InternalConnect(ip, port, false);
	if ( !m_use_events ) {
		res = WaitSocketConnect(10, 0) && InternalIsConnected();
		if ( res ) {
			OnConnect();
		} else {
			OnLost();
		}
	}
#endif
	return !SocketError() && res;
}

void CECSocket::SendPacket(const CECPacket *packet)
{
	WritePacket(packet);
	OnOutput();
}

const CECPacket *CECSocket::SendRecvPacket(const CECPacket *packet)
{
	SendPacket(packet);
	m_curr_rx_data->ReadFromSocketAll(this, 2 * sizeof(uint32_t));
	if (SocketError() && !WouldBlock()) {
		OnError();
		return 0;
	}

	m_curr_rx_data->Read(&m_rx_flags, sizeof(m_rx_flags));
	m_rx_flags = ENDIAN_NTOHL(m_rx_flags);
	m_curr_rx_data->Read(&m_curr_packet_len, sizeof(m_curr_packet_len));
	m_curr_packet_len = ENDIAN_NTOHL(m_curr_packet_len);
	
	if ( m_curr_rx_data->GetLength() < (m_curr_packet_len+2*sizeof(uint32_t)) ) {
		m_curr_rx_data.reset(new CQueuedData(m_curr_packet_len));
	}
	m_curr_rx_data->ReadFromSocketAll(this, m_curr_packet_len);
	if (SocketError() && !WouldBlock()) {
		OnError();
		return 0;
	}
	const CECPacket *reply = ReadPacket();
	m_curr_rx_data->Rewind();
	return reply;
}

std::string CECSocket::GetLastErrorMsg()
{
	int code = InternalGetLastError();
	switch(code) {
		case EC_ERROR_NOERROR:
			return "No error happened";
		case EC_ERROR_INVOP:
			return "Invalid operation";
		case EC_ERROR_IOERR:
			return "Input/Output error";
		case EC_ERROR_INVADDR:
			return "Invalid address passed to wxSocket";
		case EC_ERROR_INVSOCK:
			return "Invalid socket (uninitialized)";
		case EC_ERROR_NOHOST:
			return "No corresponding host";
		case EC_ERROR_INVPORT:
			return "Invalid port";
		case EC_ERROR_WOULDBLOCK:
			return "The socket is non-blocking and the operation would block";
		case EC_ERROR_TIMEDOUT:
			return "The timeout for this operation expired";
		case EC_ERROR_MEMERR:
			return "Memory exhausted";
		case EC_ERROR_DUMMY:
			return "Dummy code - should not happen";
	}
	ostringstream error_string;
	error_string << "Error code " << code <<  " unknown.";
	return error_string.str();
}

void CECSocket::OnError()
{
#ifdef __DEBUG__
	cout << GetLastErrorMsg() << endl;
#endif
}

void CECSocket::OnLost()
{
}

//
// Event handlers
//
void CECSocket::OnConnect()
{
}

void CECSocket::OnInput()
{
	size_t bytes_rx = 0;
	do {
		if (m_curr_rx_data.get()) {
			m_curr_rx_data->ReadFromSocket(this, m_bytes_needed);
		} else {
			return;
		}
		if (SocketError() && !WouldBlock()) {
			OnError();
			// socket already disconnected in this point
			m_curr_rx_data.reset(0);
			return;
		}
		bytes_rx = GetLastCount();
		m_bytes_needed -= bytes_rx;
	} while (m_bytes_needed && bytes_rx);
	
	if (!m_bytes_needed) {
		if (m_in_header) {
			m_in_header = false;
			m_curr_rx_data->Read(&m_rx_flags, sizeof(m_rx_flags));
			m_rx_flags = ENDIAN_NTOHL(m_rx_flags);