pcp.cpp

P2P应用 ： Peercast的源代码

// ------------------------------------------------
// File : pcp.cpp
// Date: 1-mar-2004
// Author: giles
//
// (c) 2002-4 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 "atom.h"
#include "pcp.h"
#include "peercast.h"
#include "version2.h"

// ------------------------------------------
void PCPStream::init(GnuID &rid)
{
	remoteID = rid;
	routeList.clear();

	lastPacketTime = 0;
	nextRootPacket = 0;	 // 0 seconds (never)

	inData.init();       //对象indata属于类ChanPacketBuffer，ChanPacketBuffer在cstream.h中定义
	inData.accept = ChanPacket::T_PCP;

	outData.init();
	outData.accept = ChanPacket::T_PCP;
}
// ------------------------------------------
void PCPStream::readVersion(Stream &in)
{
	int len = in.readInt();

	if (len != 4)
		throw StreamException("Invalid PCP");

	int ver = in.readInt();

	LOG_DEBUG("PCP ver: %d",ver);
}
// ------------------------------------------
void PCPStream::readHeader(Stream &in,Channel *)
{
//	AtomStream atom(in);

//	if (in.readInt() != PCP_CONNECT)
//		throw StreamException("Not PCP");

//	readVersion(in);
}
// ------------------------------------------
bool PCPStream::sendPacket(ChanPacket &pack,GnuID &destID)
{
	if (destID.isSet())
		if (!destID.isSame(remoteID))
			if (!routeList.contains(destID))
				return false;

	return outData.writePacket(pack);
}
// ------------------------------------------
void PCPStream::flush(Stream &in)
{
	ChanPacket pack;
	// send outward packets
	while (outData.numPending())
	{
		outData.readPacket(pack);
		pack.writeRaw(in);
	}
}
// ------------------------------------------
int PCPStream::readPacket(Stream &in,Channel *)
{
	BroadcastState bcs;
	return readPacket(in,bcs);
}
// ------------------------------------------
int PCPStream::readPacket(Stream &in,BroadcastState &bcs)
{
	int error = PCP_ERROR_GENERAL;
	try
	{
		AtomStream atom(in);

		ChanPacket pack;
		MemoryStream mem(pack.data,sizeof(pack.data));
		AtomStream patom(mem);


		// send outward packets
		error = PCP_ERROR_WRITE;
		if (outData.numPending())
		{
			outData.readPacket(pack);
			pack.writeRaw(in);
		}
		error = PCP_ERROR_GENERAL;

		if (outData.willSkip())
		{
			error = PCP_ERROR_WRITE+PCP_ERROR_SKIP;
			throw StreamException("Send too slow");
		}


		error = PCP_ERROR_READ;
		// poll for new downward packet
		if (in.readReady())
		{
			int numc,numd;
			ID4 id;

			id = atom.read(numc,numd);

			mem.rewind();
			pack.len = patom.writeAtoms(id, in, numc, numd);
			pack.type = ChanPacket::T_PCP;

			inData.writePacket(pack);
		}
		error = PCP_ERROR_GENERAL;

		// process downward packets
		if (inData.numPending())
		{
			inData.readPacket(pack);

			mem.rewind();

			int numc,numd;
			ID4 id = patom.read(numc,numd);

			error = PCPStream::procAtom(patom,id,numc,numd,bcs);
			
			if (error)
				throw StreamException("PCP exception");
		}

		error = 0;

	}catch(StreamException &e)
	{
		LOG_ERROR("PCP readPacket: %s (%d)",e.msg,error);
	}

	return error;
}

// ------------------------------------------
void PCPStream::readEnd(Stream &,Channel *)
{
}


// ------------------------------------------
void PCPStream::readPushAtoms(AtomStream &atom, int numc,BroadcastState &bcs)
{
	Host host;
	GnuID	chanID;

	chanID.clear();

	for(int i=0; i<numc; i++)
	{
		int c,d;
		ID4 id = atom.read(c,d);

		if (id == PCP_PUSH_IP)
			host.ip = atom.readInt();
		else if (id == PCP_PUSH_PORT)
			host.port = atom.readShort();
		else if (id == PCP_PUSH_CHANID)
			atom.readBytes(chanID.id,16);
		else
		{
			LOG_DEBUG("PCP skip: %s,%d,%d",id.getString().str(),c,d);
			atom.skip(c,d);
		}
	}


	if (bcs.forMe)
	{
		char ipstr[64];
		host.toStr(ipstr);

		Servent *s = NULL;

		if (chanID.isSet())
		{
			Channel *ch = chanMgr->findChannelByID(chanID);
			if (ch)
				if (ch->isBroadcasting() || !ch->isFull() && !servMgr->relaysFull() && ch->info.id.isSame(chanID))
					s = servMgr->allocServent();
		}else{
			s = servMgr->allocServent();
		}

		if (s)
		{
			LOG_DEBUG("GIVing to %s",ipstr);
			s->initGIV(host,chanID);
		}
	}

}
// ------------------------------------------
void PCPStream::readRootAtoms(AtomStream &atom, int numc,BroadcastState &bcs)
{
	String url;

	for(int i=0; i<numc; i++)
	{
		int c,d;
		ID4 id = atom.read(c,d);

		if (id == PCP_ROOT_UPDINT)
		{
			int si = atom.readInt();

			chanMgr->setUpdateInterval(si);
			LOG_DEBUG("PCP got new host update interval: %ds",si);
		}else if (id == PCP_ROOT_URL)
		{
			url = "http://www.peercast.org/";
			String loc;
			atom.readString(loc.data,sizeof(loc.data),d);
			url.append(loc);

		}else if (id == PCP_ROOT_CHECKVER)
		{
			unsigned int newVer = atom.readInt();
			if (newVer > PCP_CLIENT_VERSION)
			{
				strcpy(servMgr->downloadURL,url.cstr());
				peercastApp->notifyMessage(ServMgr::NT_UPGRADE,"There is a new version of PeerCast available, please click here to upgrade your client.");
			}
			LOG_DEBUG("PCP got version check: %d / %d",newVer,PCP_CLIENT_VERSION);

		}else if (id == PCP_ROOT_NEXT)
		{
			unsigned int time = atom.readInt();

			if (time)
			{
				unsigned int ctime = sys->getTime();
				nextRootPacket = ctime+time;
				LOG_DEBUG("PCP expecting next root packet in %ds",time);
			}else
			{
				nextRootPacket = 0;
			}

		}else if (id == PCP_ROOT_UPDATE)
		{
			atom.skip(c,d);

			chanMgr->broadcastTrackerUpdate(remoteID,true);

		}else if ((id == PCP_MESG_ASCII) || (id == PCP_MESG))			// PCP_MESG_ASCII to be depreciated 
		{
			String newMsg;

			atom.readString(newMsg.data,sizeof(newMsg.data),d);
			if (!newMsg.isSame(servMgr->rootMsg.cstr()))
			{
				servMgr->rootMsg = newMsg;
				LOG_DEBUG("PCP got new root mesg: %s",servMgr->rootMsg.cstr());
				peercastApp->notifyMessage(ServMgr::NT_PEERCAST,servMgr->rootMsg.cstr());
			}
		}else
		{
			LOG_DEBUG("PCP skip: %s,%d,%d",id.getString().str(),c,d);
			atom.skip(c,d);
		}
	}
}

// ------------------------------------------
void PCPStream::readPktAtoms(Channel *ch,AtomStream &atom,int numc,BroadcastState &bcs)
{
	ChanPacket pack;
	ID4 type;


	for(int i=0; i<numc; i++)
	{
		int c,d;
		ID4 id = atom.read(c,d);

		if (id == PCP_CHAN_PKT_TYPE)
		{
			type = atom.readID4();

			if (type == PCP_CHAN_PKT_HEAD)
				pack.type = ChanPacket::T_HEAD;
			else if (type == PCP_CHAN_PKT_DATA)
				pack.type = ChanPacket::T_DATA;
			else
				pack.type = ChanPacket::T_UNKNOWN;

		}else if (id == PCP_CHAN_PKT_POS)
		{
			pack.pos = atom.readInt();


		}else if (id == PCP_CHAN_PKT_DATA)
		{
			pack.len = d;
			atom.readBytes(pack.data,pack.len);
		}
		else
		{
			LOG_DEBUG("PCP skip: %s,%d,%d",id.getString().str(),c,d);
			atom.skip(c,d);
		}
	}

	if (ch)
	{

		int diff = pack.pos - ch->streamPos;
		if (diff)
			LOG_DEBUG("PCP skipping %s%d (%d -> %d)",(diff>0)?"+":"",diff,ch->streamPos,pack.pos);


		if (pack.type == ChanPacket::T_HEAD)
		{
			LOG_DEBUG("New head packet at %d",pack.pos);

			// check for stream restart
			if (pack.pos == 0)		
			{
				LOG_CHANNEL("PCP resetting stream");
				ch->streamIndex++;
				ch->rawData.init();
			}

			ch->headPack = pack;

			ch->rawData.writePacket(pack,true);
			ch->streamPos = pack.pos+pack.len;

		}else if (pack.type == ChanPacket::T_DATA)
		{
			ch->rawData.writePacket(pack,true);
			ch->streamPos = pack.pos+pack.len;
		}

	}

	// update this parent packet stream position
	if ((pack.pos) && (!bcs.streamPos || (pack.pos < bcs.streamPos)))
		bcs.streamPos = pack.pos;

}
// -----------------------------------
void PCPStream::readHostAtoms(AtomStream &atom, int numc, BroadcastState &bcs)
{
	ChanHit hit;