gnutella.cpp

P2P应用 ： Peercast的源代码

// ------------------------------------------------
// File : gnutella.cpp
// Date: 4-apr-2002
// Author: giles
// Desc: 
//		GnuPacket is a Gnutella protocol packet.
//		GnuStream is a Stream that reads/writes GnuPackets
//
//
// (c) 2002 peercast.org
// 
// ------------------------------------------------
// 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.
// ------------------------------------------------

#include "gnutella.h"
#include "stream.h"
#include "common.h"
#include "servent.h"
#include "servmgr.h"
#include "stats.h"
#include <stdlib.h>

// ---------------------------
const char *GNU_FUNC_STR(int func)
{
	switch (func)
	{
		case GNU_FUNC_PING: return "PING"; break;
		case GNU_FUNC_PONG: return "PONG"; break;
		case GNU_FUNC_QUERY: return "QUERY"; break;
		case GNU_FUNC_HIT: return "HIT"; break;
		case GNU_FUNC_PUSH: return "PUSH"; break;
		default: return "UNKNOWN"; break;
	}
}

// ---------------------------
const char *GnuStream::getRouteStr(R_TYPE r)
{
	switch(r)
	{
		case R_PROCESS: return "PROCESS"; break;
		case R_DEAD: return "DEAD"; break;
		case R_DISCARD: return "DISCARD"; break;
		case R_ACCEPTED: return "ACCEPTED"; break;
		case R_BROADCAST: return "BROADCAST"; break;
		case R_ROUTE: return "ROUTE"; break;
		case R_DUPLICATE: return "DUPLICATE"; break;
		case R_BADVERSION: return "BADVERSION"; break;
		case R_DROP: return "DROP"; break;
		default: return "UNKNOWN"; break;
	}
}
// ---------------------------
void GnuPacket::makeChecksumID()
{
	for(unsigned int i=0; i<len; i++)
		id.id[i%16] += data[i];	
}

// ---------------------------
void GnuPacket::initPing(int t)
{
	func = GNU_FUNC_PING;
	ttl = t;
	hops = 0;
	len = 0;

	id.generate();
}
// ---------------------------
void GnuPacket::initPong(Host &h, bool ownPong, GnuPacket &ping)
{
	func = GNU_FUNC_PONG;
	ttl = ping.hops;
	hops = 0;
	len = 14;
	id = ping.id;

	MemoryStream pk(data,len);

	pk.writeShort(h.port);		// port
	pk.writeLong(SWAP4(h.ip));	// ip
	if (ownPong)
	{
		pk.writeLong(chanMgr->numChannels());	// cnt
		pk.writeLong(servMgr->totalOutput(false));	// total
	}else{
		pk.writeLong(0);				// cnt
		pk.writeLong(0);				// total
	}


}
// ---------------------------
void GnuPacket::initPush(ChanHit &ch, Host &sh)
{
#if 0
	func = GNU_FUNC_PUSH;
	ttl = ch.numHops;
	hops = 0;
	len = 26;
	id.generate();

	MemoryStream data(data,len);

	// ID of Hit packet
	data.write(ch.packetID.id,16);

	// index of channel
	data.writeLong(ch.index);

	data.writeLong(SWAP4(sh.ip));	// ip
	data.writeShort(sh.port);		// port
#endif
}


// ---------------------------
bool GnuPacket::initHit(Host &h, Channel *ch, GnuPacket *query, bool push, bool busy, bool stable, bool tracker, int maxttl)
{
	if (!ch)
		return false;

	func = GNU_FUNC_HIT;
	hops = 0;
	id.generate();

	ttl = maxttl;


	MemoryStream mem(data,MAX_DATA);

	mem.writeChar(1);			// num hits
	mem.writeShort(h.port);		// port
	mem.writeLong(SWAP4(h.ip));	// ip

	if (query)
		mem.writeLong(0);			// speed - route
	else
		mem.writeLong(1);			// broadcast

	
	//mem.writeLong(ch->index);				// index
	mem.writeLong(0);				// index
	mem.writeShort(ch->getBitrate());	// bitrate
	mem.writeShort(ch->localListeners());		// num listeners

	mem.writeChar(0);	// no name

	XML xml;
	XML::Node *cn = ch->info.createChannelXML();
	cn->add(ch->info.createTrackXML());
	xml.setRoot(cn);
	xml.writeCompact(mem);

	mem.writeChar(0);							// extra null 


	// QHD
	mem.writeLong('PCST');				// vendor ID		
	mem.writeChar(2);					// public sector length 

	int f1 = 0, f2 = 0;

	f1 = 1 | 4 | 8 | 32 | 64;	// use push | busy | stable | broadcast | tracker 

	if (push) f2 |= 1;			
	if (busy) f2 |= 4;
	if (stable) f2 |= 8;
	if (!query) f2 |= 32;
	if (tracker) f2 |= 64;
	
	mem.writeChar(f1);
	mem.writeChar(f2);

	{
		// write private sector
		char pbuf[256];
		MemoryStream pmem(pbuf,sizeof(pbuf));
		XML xml;
		XML::Node *pn = servMgr->createServentXML();
		xml.setRoot(pn);
		xml.writeCompact(pmem);
		pmem.writeChar(0);			// add null terminator
		if (pmem.pos <= 255)
		{
			mem.writeChar(pmem.pos);
			mem.write(pmem.buf,pmem.pos);
		}else
			mem.writeChar(0);
	}


	// queryID/not used
	if (query)
		mem.write(query->id.id,16);					
	else
		mem.write(id.id,16);					

	len = mem.pos;

	LOG_NETWORK("Created Hit packet: %d bytes",len);

	if (len >= MAX_DATA)
		return false;

//	servMgr->addReplyID(id);
	return true;
}


// ---------------------------
void GnuPacket::initFind(const char *str, XML *xml, int maxTTL)
{

	func = GNU_FUNC_QUERY;
	ttl = maxTTL;
	hops = 0;
	id.generate();

	MemoryStream mem(data,MAX_DATA);

	mem.writeShort(0);		// min speed

	if (str)
	{
		int slen = strlen(str);
		mem.write((void *)str,slen+1);	// string
	}else
		mem.writeChar(0);		// null string

	
	if (xml)
		xml->writeCompact(mem);

	len = mem.pos;
}

// ---------------------------
void GnuStream::ping(int ttl)
{
	GnuPacket ping;
	ping.initPing(ttl);
//	servMgr->addReplyID(ping.id);
	sendPacket(ping);
	LOG_NETWORK("ping out %02x%02x%02x%02x",ping.id.id[0],ping.id.id[1],ping.id.id[2],ping.id.id[3]);
}

// ---------------------------
void GnuStream::sendPacket(GnuPacket &p)
{
	try 
	{
		lock.on();
		packetsOut++;
		stats.add(Stats::NUMPACKETSOUT);

		switch(p.func)
		{
			case GNU_FUNC_PING: stats.add(Stats::NUMPINGOUT); break;
			case GNU_FUNC_PONG: stats.add(Stats::NUMPONGOUT); break;
			case GNU_FUNC_QUERY: stats.add(Stats::NUMQUERYOUT); break;
			case GNU_FUNC_HIT: stats.add(Stats::NUMHITOUT); break;
			case GNU_FUNC_PUSH: stats.add(Stats::NUMPUSHOUT); break;
			default: stats.add(Stats::NUMOTHEROUT); break;
		}


		write(p.id.id,16);
		writeChar(p.func);	// ping func
		writeChar(p.ttl);	// ttl
		writeChar(p.hops);	// hops
		writeLong(p.len);	// len

		if (p.len)
			write(p.data,p.len);

		stats.add(Stats::PACKETDATAOUT,23+p.len);

		lock.off();
	}catch(StreamException &e)
	{		
		lock.off();
		throw e;
	}
}
// ---------------------------
bool GnuStream::readPacket(GnuPacket &p)
{
	try 
	{
		lock.on();
		packetsIn++;
		stats.add(Stats::NUMPACKETSIN);

		read(p.id.id,16);
		p.func = readChar();
		p.ttl = readChar();
		p.hops = readChar();
		p.len = readLong();


		if ((p.hops >= 1) && (p.hops <= 10))
			stats.add((Stats::STAT)((int)Stats::NUMHOPS1+p.hops-1));

		stats.add(Stats::PACKETDATAIN,23+p.len);

		switch(p.func)
		{
			case GNU_FUNC_PING: stats.add(Stats::NUMPINGIN); break;
			case GNU_FUNC_PONG: stats.add(Stats::NUMPONGIN); break;
			case GNU_FUNC_QUERY: stats.add(Stats::NUMQUERYIN); break;
			case GNU_FUNC_HIT: stats.add(Stats::NUMHITIN); break;
			case GNU_FUNC_PUSH: stats.add(Stats::NUMPUSHIN); break;
			default: stats.add(Stats::NUMOTHERIN); break;
		}


		if (p.len)
		{
			if (p.len > GnuPacket::MAX_DATA)
			{
				while (p.len--)
					readChar();
				lock.off();
				return false;
			}
			read(p.data,p.len);
		}

		lock.off();
		return true;
	}catch(StreamException &e)
	{		
		lock.off();
		throw e;
	}
}

// ---------------------------
GnuStream::R_TYPE GnuStream::processPacket(GnuPacket &in, Servent *serv, GnuID &routeID)
{

	R_TYPE ret = R_DISCARD;

	MemoryStream data(in.data,in.len);

	Host remoteHost = serv->getHost();

	in.ttl--;
	in.hops++;

	routeID = in.id;



	switch(in.func)
	{
		case GNU_FUNC_PING: // ping
			{
				LOG_NETWORK("ping: from %d.%d.%d.%d : %02x%02x%02x%02x",
							remoteHost.ip>>24&0xff,remoteHost.ip>>16&0xff,remoteHost.ip>>8&0xff,remoteHost.ip&0xff,
							in.id.id[0],in.id.id[1],in.id.id[2],in.id.id[3]
					);