gnutella.cpp

P2P应用 ： Peercast的源代码

				Host sh = servMgr->serverHost;
				if (sh.isValid())
				{
					if ((servMgr->getFirewall() != ServMgr::FW_ON) && (!servMgr->pubInFull()))
					{
						GnuPacket pong;
						pong.initPong(sh,true,in);
						if (serv->outputPacket(pong,true))
							LOG_NETWORK("pong out");
					}
					ret = R_BROADCAST;			
				}
			}
			break;
		case GNU_FUNC_PONG: // pong
			{
				{
					int ip,port,cnt,total;
					port = data.readShort();
					ip = data.readLong();
					cnt = data.readLong();
					total = data.readLong();

					ip = SWAP4(ip);


					Host h;
					h.ip = ip;
					h.port = port;

					char sIP[64],rIP[64];
					h.toStr(sIP);
					remoteHost.toStr(rIP);
					
					LOG_NETWORK("pong: %s via %s : %02x%02x%02x%02x",sIP,ip,rIP,in.id.id[0],in.id.id[1],in.id.id[2],in.id.id[3]);


					ret = R_DISCARD;

					if (h.isValid())
					{

						#if 0
						// accept if this pong is a reply from one of our own pings, otherwise route back
						if (servMgr->isReplyID(in.id))
						{
							servMgr->addHost(h,ServHost::T_SERVENT,sys->getTime());
							ret = R_ACCEPTED;
						}else
							ret = R_ROUTE;
						#endif
					}

				}
			}
			break;
		case GNU_FUNC_QUERY: // query
			ret = R_BROADCAST;

			{
				Host sh = servMgr->serverHost;
				if (!sh.isValid())
					sh.ip = 127<<24|1;

				char words[256];
				short spd = data.readShort();
				data.readString(words,sizeof(words));
				words[sizeof(words)-1] = 0;

				MemoryStream xm(&data.buf[data.pos],data.len-data.pos);
				xm.buf[xm.len] = 0;

				Channel *hits[16];
				int numHits=0;

				if (strncmp(xm.buf,"<?xml",5)==0)
				{
					XML xml;
					xml.read(xm);
					XML::Node *cn = xml.findNode("channel");
					if (cn)
					{
						ChanInfo info;
						info.init(cn);
						info.status = ChanInfo::S_PLAY;
						numHits = chanMgr->findChannels(info,hits,16);
					}
					LOG_NETWORK("query XML: %s : found %d",xm.buf,numHits);
				}else{
					ChanInfo info;
					info.name.set(words);
					info.genre.set(words);
					info.id.fromStr(words);
					info.status = ChanInfo::S_PLAY;
					numHits = chanMgr->findChannels(info,hits,16);
					LOG_NETWORK("query STR: %s : found %d",words,numHits);
				}

			

				for(int i=0; i<numHits; i++)
				{
					bool push = (servMgr->getFirewall()!=ServMgr::FW_OFF);
					bool busy = (servMgr->pubInFull() && servMgr->outFull()) || servMgr->relaysFull();
					bool stable = servMgr->totalStreams>0;
					bool tracker = 	hits[i]->isBroadcasting();

					GnuPacket hit;
					if (hit.initHit(sh,hits[i],&in,push,busy,stable,tracker,in.hops))
						serv->outputPacket(hit,true);
				}
			}
			break;
		case GNU_FUNC_PUSH:	// push
			{

				GnuID pid;
				data.read(pid.id,16);
				
				//LOG("push serv= %02x%02x%02x%02x",servMgr->id[0],servMgr->id[1],servMgr->id[2],servMgr->id[3]);
				//LOG("pack = %02x%02x%02x%02x",id[0],id[1],id[2],id[3]);


				int index = data.readLong();
				int ip = data.readLong();
				int port = data.readShort();

			
				ip = SWAP4(ip);

				Host h(ip,port);
				char hostName[64];
				h.toStr(hostName);

#if 0
				if (servMgr->isReplyID(pid))
				{
#if 0
					Channel *c = chanMgr->findChannelByIndex(index);

					if (!c)
					{
						LOG_NETWORK("push 0x%x to %s: Not found",index,hostName);
					}else
					{
						if (!c->isFull() && !servMgr->streamFull())
						{
							LOG_NETWORK("push: 0x%x to %s: OK",index,hostName);

							Servent *s = servMgr->allocServent();
							if (s)
								s->initGIV(h,c->info.id);
						}else
							LOG_NETWORK("push: 0x%x to %s: FULL",index,hostName);
					}
#endif
					ret = R_ACCEPTED;
				}else{
					LOG_NETWORK("push: 0x%x to %s: ROUTE",index,hostName);
					routeID = pid;
					ret = R_ROUTE;
				}
#endif		
			}
			break;
		case GNU_FUNC_HIT: // hit
			{
				ret = R_DISCARD;

				ChanHit hit;
				if (readHit(data,hit,in.hops,in.id))
				{

					char flstr[64];
					flstr[0]=0;
					if (hit.firewalled) strcat(flstr,"Push,");
					if (hit.tracker) strcat(flstr,"Tracker,");
					
#if 0
					if ((spd == 0) && (!isBroadcastHit))
					{
						if (servMgr->isReplyID(queryID))
						{
							ret = R_ACCEPTED;
							LOG_NETWORK("self-hit: %s 0x%02x %s %d chan",hostName,f2,flstr,num);
						}else
						{
							routeID = queryID;
							ret = R_ROUTE;
							LOG_NETWORK("route-hit: %s 0x%02x %s %d chan",hostName,f2,flstr,num);
						}
					}else
					{
						ret = R_BROADCAST;
						LOG_NETWORK("broadcast-hit: %s 0x%02x %s %d chan",hostName,f2,flstr,num);
					}
#else
					ret = R_BROADCAST;
					LOG_NETWORK("broadcast-hit: %s",flstr);
#endif
				}
			}
			break;
		default:
			LOG_NETWORK("packet: %d",in.func);
			break;
	}

	
	if ((in.ttl > 10) || (in.hops > 10) || (in.ttl==0))
		if ((ret == R_BROADCAST) || (ret == R_ROUTE))
			ret = R_DEAD;

	return ret;
}

// ---------------------------
bool GnuStream::readHit(Stream &data, ChanHit &ch,int hops,GnuID &id)
{
	int i;
	int num = data.readChar();	// hits
	int port = data.readShort();		// port
	int ip = data.readLong();		// ip
	ip = SWAP4(ip);
	int spd = data.readLong();		// speed/broadcast

	Host h(ip,port);
	char hostName[64];

	h.IPtoStr(hostName);

	bool dataValid=true;

	ChanHit *hits[100];
	int numHits=0;

	for(i=0; i<num; i++)
	{
		int index,bitrate,listeners;


		index = data.readLong();		// index
		bitrate = data.readShort();		// bitrate
		listeners = data.readShort();	// listeners

		// read name .. not used.
		char fname[256];
		data.readString(fname,sizeof(fname));
		fname[sizeof(fname)-1] = 0;

		ch.init();
		ch.firewalled = false;		// default to NO as we dont get the info until the next section.
		ch.host = h;
		ch.numListeners = listeners;
		ch.numHops = hops;
		ch.rhost[0] = ch.host;

		ChanInfo info;


		{
			char xmlData[4000];
			int xlen = data.readString(xmlData,sizeof(xmlData));

			if ((strncmp(xmlData,"<?xml",5)==0) && (xlen < GnuPacket::MAX_DATA))
			{
				//LOG_NETWORK("Hit XML: %s",xmlData);

				MemoryStream xm(xmlData,xlen);
				XML xml;
				xml.read(xm);
				XML::Node *n = xml.findNode("channel");
				if (n)
				{
					info.init(n);
					char idStr[64];
					info.id.toStr(idStr);
					LOG_NETWORK("got hit %s %s",idStr,info.name.cstr());

					ch.upTime = n->findAttrInt("uptime");

				}else
					LOG_NETWORK("Missing Channel node");
			}else
			{
				LOG_NETWORK("Missing XML data");
				//LOG_NETWORK("%s",xmlData);
				dataValid = false;
			}
		}

		if (info.id.isSet())
		{
			if (!chanMgr->findHitList(info))
				chanMgr->addHitList(info);

			ch.recv = true;
			ch.chanID = info.id;
			ChanHit *chp = chanMgr->addHit(ch);		

			if ((chp) && (numHits<100))
				hits[numHits++] = chp;
		}

	}


	int vendor = data.readLong();	// vendor ID

	int pubLen = data.readChar();	// public sec length - should be 2

	int f1 = data.readChar() & 0xff; // flags 1
	int f2 = data.readChar() & 0xff; // flags 2

	pubLen -= 2;
	while (pubLen-->0)
		data.readChar();


	char agentStr[16];
	agentStr[0]=0;
	int maxPreviewTime=0;

	// read private sector with peercast servant specific info
	int privLen = data.readChar();	

	if (privLen)
	{
		char privData[256];
		data.read(privData,privLen);
		if (strncmp(privData,"<?xml",5)==0)
		{
			MemoryStream xm(privData,privLen);
			XML xml;
			xml.read(xm);
			XML::Node *sn = xml.findNode("servent");
			if (sn)
			{
				char *ag = sn->findAttr("agent");
				if (ag)
				{
					strncpy(agentStr,ag,16);
					agentStr[15]=0;
				}
				maxPreviewTime = sn->findAttrInt("preview");
			}

		}
	}


	// not used anymore
	GnuID queryID;
	data.read(queryID.id,16);

	bool isBroadcastHit=false;
	if (f1 & 32)
		isBroadcastHit = (f2 & 32)!=0;

	for(i=0; i<numHits; i++)
	{
		if (f1 & 1)
			hits[i]->firewalled = (f2 & 1)!=0;

		if (f1 & 64)
			hits[i]->tracker = (f2 & 64)!=0;

	}

	return dataValid;
}