ecsocket.cpp

电驴的MAC源代码

			if (m_rx_flags & EC_FLAG_ACCEPTS) {
				// Client sends its capabilities, update the internal mask.
				m_curr_rx_data->Read(&m_my_flags, sizeof(m_my_flags));
				m_my_flags = ENDIAN_NTOHL(m_my_flags);
				//printf("Reading accepts mask: %x\n", m_my_flags);
				wxASSERT(m_my_flags & 0x20);
				// There has to be 4 more bytes. THERE HAS TO BE, DAMN IT.
				m_curr_rx_data->ReadFromSocketAll(this, sizeof(m_curr_packet_len));
			}
			m_curr_rx_data->Read(&m_curr_packet_len, sizeof(m_curr_packet_len));
			m_curr_packet_len = ENDIAN_NTOHL(m_curr_packet_len);
			m_bytes_needed = m_curr_packet_len;
			// packet bigger that 16Mb looks more like broken request
			if (m_bytes_needed > 16*1024*1024) {
				CloseSocket();
				return;
			}
			size_t needed_size = m_bytes_needed + ((m_rx_flags & EC_FLAG_ACCEPTS) ? 12 : 8);
			if (!m_curr_rx_data.get() ||
			    m_curr_rx_data->GetLength() < needed_size) {
				m_curr_rx_data.reset(new CQueuedData(needed_size));
			}
			//#warning Kry TODO: Read packet?
		} else {
			//m_curr_rx_data->DumpMem();
			std::auto_ptr<const CECPacket> packet(ReadPacket());
			m_curr_rx_data->Rewind();
			if (packet.get()) {
				std::auto_ptr<const CECPacket> reply(OnPacketReceived(packet.get()));
				if (reply.get()) {
					SendPacket(reply.get());
				}
			}
			m_bytes_needed = 8;
			m_in_header = true;
		}
	}
}

void CECSocket::OnOutput()
{
	while (!m_output_queue.empty()) {
		CQueuedData* data = m_output_queue.front();
		data->WriteToSocket(this);
		if (!data->GetUnreadDataLength()) {
			m_output_queue.pop_front();
			delete data;
		}
		if (SocketError()) {
			if (WouldBlock()) {
				if ( m_use_events ) {
					return;
				} else {
					if ( !WaitSocketWrite(10, 0) ) {
						if (WouldBlock()) {
							continue;
						} else {
							OnError();
							break;
						}
	                		}
				}
			} else {
				OnError();
				return;
			}
		}
	}
}

//
// Socket I/O
//

size_t CECSocket::ReadBufferFromSocket(void *buffer, size_t required_len)
{
	wxASSERT(required_len);

	if (m_curr_rx_data->GetUnreadDataLength() < required_len) {
		// need more data that we have. Looks like nothing will help here
		return 0;
	}
	m_curr_rx_data->Read(buffer, required_len);
	return required_len;
}

void CECSocket::WriteBufferToSocket(const void *buffer, size_t len)
{
	unsigned char *wr_ptr = (unsigned char *)buffer;
	while ( len ) {
		size_t curr_free = m_curr_tx_data->GetRemLength();
		if ( len > curr_free ) {

			m_curr_tx_data->Write(wr_ptr, curr_free);
			len -= curr_free;
			wr_ptr += curr_free;
			m_output_queue.push_back(m_curr_tx_data.release());
			m_curr_tx_data.reset(new CQueuedData(EC_SOCKET_BUFFER_SIZE));
		} else {
			m_curr_tx_data->Write(wr_ptr, len);
			break;
		}
	}
}


//
// ZLib "error handler"
//

void ShowZError(int zerror, z_streamp strm)
{
	const char *p = NULL;

	switch (zerror) {
		case Z_STREAM_END: p = "Z_STREAM_END"; break;
		case Z_NEED_DICT: p = "Z_NEED_DICT"; break;
		case Z_ERRNO: p = "Z_ERRNO"; break;
		case Z_STREAM_ERROR: p = "Z_STREAM_ERROR"; break;
		case Z_DATA_ERROR: p = "Z_DATA_ERROR"; break;
		case Z_MEM_ERROR: p = "Z_MEM_ERROR"; break;
		case Z_BUF_ERROR: p = "Z_BUF_ERROR"; break;
		case Z_VERSION_ERROR: p = "Z_VERSION_ERROR"; break;
	}
	printf("ZLib operation returned %s\n", p);
	printf("ZLib error message: %s\n", strm->msg);
	printf("zstream state:\n\tnext_in=%p\n\tavail_in=%u\n\ttotal_in=%lu\n\tnext_out=%p\n\tavail_out=%u\n\ttotal_out=%lu\n",
		strm->next_in, strm->avail_in, strm->total_in, strm->next_out, strm->avail_out, strm->total_out);
}


bool CECSocket::ReadNumber(void *buffer, size_t len)
{
	if (m_rx_flags & EC_FLAG_UTF8_NUMBERS) {
		unsigned char mb[6];
		uint32_t wc;
		if (!ReadBuffer(mb, 1)) return false;
		int remains = utf8_mb_remain(mb[0]);
		if (remains) if (!ReadBuffer(&(mb[1]), remains)) return false;
		if (utf8_mbtowc(&wc, mb, 6) == -1) return false;	// Invalid UTF-8 code sequence
		switch (len) {
			case 1: PokeUInt8( buffer,  wc ); break;
			case 2: RawPokeUInt16( buffer, wc ); break;
			case 4: RawPokeUInt32( buffer, wc ); break;
		}
	} else {
		if ( !ReadBuffer(buffer, len) ) {
			return false;
		}
		switch (len) {
			case 2:
				RawPokeUInt16( buffer, ENDIAN_NTOHS( RawPeekUInt16( buffer ) ) );
				break;
			case 4:
				RawPokeUInt32( buffer, ENDIAN_NTOHL( RawPeekUInt32( buffer ) ) );
				break;
		}
	}
	return true;
}

bool CECSocket::WriteNumber(const void *buffer, size_t len)
{
	if (m_tx_flags & EC_FLAG_UTF8_NUMBERS) {
		unsigned char mb[6];
		uint32_t wc = 0;
		int mb_len;
		switch (len) {
			case 1: wc = PeekUInt8( buffer ); break;
			case 2: wc = RawPeekUInt16( buffer ); break;
			case 4: wc = RawPeekUInt32( buffer ); break;
			default: return false;
		}
		if ((mb_len = utf8_wctomb(mb, wc, 6)) == -1) return false;	// Something is terribly wrong...
		return WriteBuffer(mb, mb_len);
	} else {
		char tmp[8];

		switch (len) {
			case 1: PokeUInt8( tmp, PeekUInt8( buffer ) ); break;
			case 2: RawPokeUInt16( tmp, ENDIAN_NTOHS( RawPeekUInt16( buffer ) ) ); break;
			case 4: RawPokeUInt32( tmp, ENDIAN_NTOHL( RawPeekUInt32( buffer ) ) ); break;
		}
		return WriteBuffer(tmp, len);
	}
}

bool CECSocket::ReadBuffer(void *buffer, size_t len)
{
	if (m_rx_flags & EC_FLAG_ZLIB) {
		if ( !m_z.avail_in ) {
			// no reason for this situation: all packet should be
			// buffered by now
			return false;
		}
		m_z.avail_out = (uInt)len;
		m_z.next_out = (Bytef*)buffer;
		int zerror = inflate(&m_z, Z_SYNC_FLUSH);
		if ((zerror != Z_OK) && (zerror != Z_STREAM_END)) {
			ShowZError(zerror, &m_z);
			return false;
		}
		return true;
	} else {
		// using uncompressed buffered i/o
		return ReadBufferFromSocket(buffer, len) == len;
	}
}

bool CECSocket::WriteBuffer(const void *buffer, size_t len)
{
	if (m_tx_flags & EC_FLAG_ZLIB) {

		unsigned char *rd_ptr = (unsigned char *)buffer;
		do {
			unsigned int remain_in = EC_SOCKET_BUFFER_SIZE - m_z.avail_in;
			if ( remain_in >= len ) {
				memcpy(m_z.next_in+m_z.avail_in, rd_ptr, len);
				m_z.avail_in += (uInt)len;
				len = 0;
			} else {
				memcpy(m_z.next_in+m_z.avail_in, rd_ptr, remain_in);
				m_z.avail_in += remain_in;
				len -= remain_in;
				rd_ptr += remain_in;
				// buffer is full, calling zlib
				do {
				    m_z.next_out = &m_out_ptr[0];
				    m_z.avail_out = EC_SOCKET_BUFFER_SIZE;
					int zerror = deflate(&m_z, Z_NO_FLUSH);
					if ( zerror != Z_OK ) {
						ShowZError(zerror, &m_z);
						return false;
					}
					WriteBufferToSocket(&m_out_ptr[0],
						EC_SOCKET_BUFFER_SIZE - m_z.avail_out);
				} while ( m_z.avail_out == 0 );
				// all input should be used by now
				wxASSERT(m_z.avail_in == 0);
				m_z.next_in = &m_in_ptr[0];
			}
		} while ( len );
		return true;
	} else {
		// using uncompressed buffered i/o
		WriteBufferToSocket(buffer, len);
		return true;
	}
}

bool CECSocket::FlushBuffers()
{
	if (m_tx_flags & EC_FLAG_ZLIB) {
		do {
		    m_z.next_out = &m_out_ptr[0];
		    m_z.avail_out = EC_SOCKET_BUFFER_SIZE;
			int zerror = deflate(&m_z, Z_FINISH);
			if ( zerror == Z_STREAM_ERROR ) {
				ShowZError(zerror, &m_z);
				return false;
			}
			WriteBufferToSocket(&m_out_ptr[0],
				EC_SOCKET_BUFFER_SIZE - m_z.avail_out);
		} while ( m_z.avail_out == 0 );
	}
	if ( m_curr_tx_data->GetDataLength() ) {
		m_output_queue.push_back(m_curr_tx_data.release());
		m_curr_tx_data.reset(new CQueuedData(EC_SOCKET_BUFFER_SIZE));
	}
	return true;
}

//
// Packet I/O
//

void CECSocket::WritePacket(const CECPacket *packet)
{
	if (SocketError() && !WouldBlock()) {
		OnError();
		return;
	}

	uint32_t flags = 0x20;

	if ( packet->GetPacketLength() > EC_MAX_UNCOMPRESSED ) {
		flags |= EC_FLAG_ZLIB;
	} else {
		flags |= EC_FLAG_UTF8_NUMBERS;
	}

	flags &= m_my_flags;
	m_tx_flags = flags;
	
	if (flags & EC_FLAG_ZLIB) {
		m_z.zalloc = Z_NULL;
		m_z.zfree = Z_NULL;
		m_z.opaque = Z_NULL;
		m_z.avail_in = 0;
		m_z.next_in = &m_in_ptr[0];
		int zerror = deflateInit(&m_z, EC_COMPRESSION_LEVEL);
		if (zerror != Z_OK) {
			// don't use zlib if init failed
			flags &= ~EC_FLAG_ZLIB;
			ShowZError(zerror, &m_z);
		}
	}

	uint32_t tmp_flags = ENDIAN_HTONL(flags/* | EC_FLAG_ACCEPTS*/);
	WriteBufferToSocket(&tmp_flags, sizeof(uint32));
	
/*	uint32_t tmp_accepts_flags = ENDIAN_HTONL(m_my_flags);
	WriteBufferToSocket(&tmp_accepts_flags, sizeof(uint32));*/

	// preallocate 4 bytes in buffer for packet length
	uint32_t packet_len = 0;
	WriteBufferToSocket(&packet_len, sizeof(uint32));
	
	packet->WritePacket(*this);

	FlushBuffers();

	// now calculate actual size of data
	wxASSERT(m_curr_tx_data->GetDataLength() < 0xFFFFFFFF);
	packet_len = (uint32_t)m_curr_tx_data->GetDataLength();
	for(std::deque<CQueuedData*>::iterator i = m_output_queue.begin(); i != m_output_queue.end(); i++) {
		wxASSERT(( packet_len + m_curr_tx_data->GetDataLength()) < 0xFFFFFFFF);
		packet_len += (uint32_t)(*i)->GetDataLength();
	}
	// 4 flags and 4 length are not counted
	packet_len -= 8;
	// now write actual length @ offset 4
	packet_len = ENDIAN_HTONL(packet_len);
	
	CQueuedData *first_buff = m_output_queue.front();
	if ( !first_buff ) first_buff = m_curr_tx_data.get();
	first_buff->WriteAt(&packet_len, sizeof(uint32_t), sizeof(uint32_t));
	
	if (flags & EC_FLAG_ZLIB) {
		int zerror = deflateEnd(&m_z);
		if ( zerror != Z_OK ) {
			ShowZError(zerror, &m_z);
			return;
		}
	}
}


const CECPacket *CECSocket::ReadPacket()
{
	CECPacket *packet = 0;

	uint32_t flags = m_rx_flags;
	
	if ( ((flags & 0x60) != 0x20) || (flags & EC_FLAG_UNKNOWN_MASK) ) {
		// Protocol error - other end might use an older protocol
		cout << "ReadPacket: packet have invalid flags " << flags << endl;
		CloseSocket();
		return 0;
	}

	if (flags & EC_FLAG_ZLIB) {

	    m_z.zalloc = Z_NULL;
	    m_z.zfree = Z_NULL;
	    m_z.opaque = Z_NULL;
	    m_z.avail_in = 0;
	    m_z.next_in = 0;
	    
		int zerror = inflateInit(&m_z);
		if (zerror != Z_OK) {
			ShowZError(zerror, &m_z);
			cout << "ReadPacket: failed zlib init" << endl;
			CloseSocket();
			return 0;
		}
	}

	m_curr_rx_data->ToZlib(m_z);
	packet = new CECPacket(*this);
	packet->ReadFromSocket(*this);
	
	if (packet->m_error != 0) {
		cout << "ReadPacket: error " << packet->m_error << "in packet read" << endl;
		delete packet;
		packet = NULL;
		CloseSocket();
	}

	if (flags & EC_FLAG_ZLIB) {
		int zerror = inflateEnd(&m_z);
		if ( zerror != Z_OK ) {
			ShowZError(zerror, &m_z);
			cout << "ReadPacket: failed zlib free" << endl;
			CloseSocket();
		}
	}

	return packet;
}

const CECPacket *CECSocket::OnPacketReceived(const CECPacket *)
{
	return 0;
}
// File_checked_for_headers