servent.cpp.svn-base

来自「这是和p2p相关的一份源码」· SVN-BASE 代码 · 共 2,288 行 · 第 1/4 页

		}catch(TimeoutException &e)
		{
			LOG_ERROR("Outgoing to %s: timeout (%s)",ipStr,e.msg);
			sv->setStatus(S_TIMEOUT);
		}catch(StreamException &e)
		{
			LOG_ERROR("Outgoing to %s: %s",ipStr,e.msg);
			sv->setStatus(S_REFUSED);
		}

		try
		{
			if (sv->sock)
			{
				sv->sock->close();
				delete sv->sock;
				sv->sock = NULL;
			}

		}catch(StreamException &) {}

		sys->sleepIdle();
	}

	sv->kill();
	return 0;
}
// -----------------------------------
int Servent::incomingProc(ThreadInfo *thread)
{
	thread->lock();

	Servent *sv = (Servent*)thread->data;
	
	char ipStr[64];
	sv->sock->host.toStr(ipStr);

	try 
	{
		sv->handshakeIncoming();
	}catch(HTTPException &e)
	{
		try
		{
			sv->sock->writeLine(e.msg);
			if (e.code == 401)
				sv->sock->writeLine("WWW-Authenticate: Basic realm=\"PeerCast\"");
			sv->sock->writeLine("");
		}catch(StreamException &){}
		LOG_ERROR("Incoming from %s: %s",ipStr,e.msg);
	}catch(StreamException &e)
	{
		LOG_ERROR("Incoming from %s: %s",ipStr,e.msg);
	}


	sv->kill();
	return 0;
}
// -----------------------------------
void Servent::processServent()
{
	setStatus(S_HANDSHAKE);

	PROTOCOL proto = handshakeIn();

	if (!sock)
		throw StreamException("Servent has no socket");


	if (proto == P_PCP)
	{
		pcpStream.process(*sock,remoteID);
	}else if (proto == P_GNUTELLA06)
		processGnutella();

}

// -----------------------------------
void Servent::processStream(bool doneHandshake,ChanInfo &chanInfo,Channel *ch)
{	

	if (!doneHandshake)
	{
		setStatus(S_HANDSHAKE);

		if (!handshakeStream(chanInfo,ch))
			ch = NULL;
	}

	if (ch)
	{

		chanID = ch->info.id;

		LOG_CHANNEL("Ch. %d: Sending %s ",ch->index,ChanInfo::getProtocolStr(chanInfo.srcProtocol));

		unsigned int ctime=sys->getTime();
		while (!ch->isPlaying())
		{
			if ((sys->getTime()-ctime) > 30)
				throw StreamException("Timeout");
			sys->sleepIdle();
		}


		servMgr->totalStreams++;

		Host host = sock->host;
		host.port = 0;	// force to 0 so we ignore the incoming port

		if (chanInfo.srcProtocol == ChanInfo::SP_HTTP)
		{
			// we have agent string now, so check again.
			if (!canPreview())
				throw StreamException("Preview disallowed");

			if ((addMetadata) && (chanMgr->icyMetaInterval))
				sendRawMetaChannel(ch,chanMgr->icyMetaInterval);
			else
				sendRawChannel(ch);
		}else if (chanInfo.srcProtocol == ChanInfo::SP_PCP)
		{
			sendPCPChannel(ch);

		} else if (chanInfo.srcProtocol == ChanInfo::SP_PEERCAST)
		{
			sendChannel(ch);
		}
	}

	setStatus(S_CLOSING);
}

// -----------------------------------------
#if 0
// debug
		FileStream file;
		file.openReadOnly("c://test.mp3");

		LOG_DEBUG("raw file read");
		char buf[4000];
		int cnt=0;
		while (!file.eof())
		{
			LOG_DEBUG("send %d",cnt++);
			file.read(buf,sizeof(buf));
			sock->write(buf,sizeof(buf));

		}
		file.close();
		LOG_DEBUG("raw file sent");

	return;
// debug
#endif
// -----------------------------------
bool Servent::waitForChannelHeader(Channel *ch)
{
	for(int i=0; i<30*10; i++)
	{
		if (ch->isPlaying() && (ch->rawData.writePos>0))
			return true;

		if (!thread.active || !sock->active())
			break;
		sys->sleep(100);
	}
	return false;
}
// -----------------------------------
bool Servent::checkPreview(unsigned int connectTime)
{
	Host h = sock->host;
	h.port = 0;			// ignore incoming port number

	if (!servMgr->isFiltered(ServFilter::F_DIRECT,h))
		return false;

	if (isPrivate())	// always allow private clients
		return true;

	return false;
}
// -----------------------------------
bool Servent::canPreview()
{	
	// probably not needed, but just in case we bind to the actual IP address and not localhost
	if (sock->host.ip == servMgr->serverHost.ip)
		return true;

	Host h = getHost();
	h.port = 0;

	if (!servMgr->isFiltered(ServFilter::F_DIRECT,h))
		return false;

	if (isPrivate())	// always allow private clients
		return true;

	if (agent.contains("PeerCast"))		// allow connections from peercast clients (direct relays etc..)
		return true;

	return true;
}
// -----------------------------------
void Servent::sendRawChannel(Channel *ch)
{
	ch->numListeners++;
	outputBitrate = ch->getBitrate();

	try
	{

		sock->timeout = 10000;
		setStatus(S_CONNECTED);

		LOG_DEBUG("Starting Raw stream: %s",ch->info.name.cstr());

		if (!waitForChannelHeader(ch))
			throw StreamException("Channel not ready");

		// pre-send HEAD data, and set current position to the packet after that
		LOG_DEBUG("Send %d bytes header ",ch->headPack.len);
		ch->headPack.write(*sock);


		unsigned int connectTime=sys->getTime();

		ChanPacket pack;

		unsigned int chanPos = ch->rawData.findPacket(ch->headPack.pos + ch->headPack.len);

		while ((thread.active) && sock->active())
		{
			if (!checkPreview(connectTime))
				throw StreamException("Preview time limit reached");

			if (!ch->isActive())
				throw StreamException("Channel closed");

			chanPos = ch->rawData.peekPacket(chanPos,pack);

			if ((pack.type == ChanPacket::T_DATA) || (pack.type == ChanPacket::T_HEAD))
				sock->write(pack.data,pack.len);

		}
	}catch(StreamException &e)
	{
		LOG_ERROR("Stream channel: %s",e.msg);
	}
	if (ch->numListeners)
		ch->numListeners--;
}

// -----------------------------------
void Servent::sendRawMetaChannel(Channel *ch, int interval)
{
	ch->numListeners++;
	outputBitrate = ch->getBitrate();

	try
	{

		sock->timeout = 10000;
		setStatus(S_CONNECTED);

		LOG_DEBUG("Starting Raw Meta stream: %s (metaint: %d)",ch->info.name.cstr(),interval);

		if (!waitForChannelHeader(ch))
			throw StreamException("Channel not ready");


		String lastTitle,lastURL;

		int		lastMsgTime=sys->getTime();
		bool	showMsg=true;

		ChanPacket pack;

		char buf[16384];
		int bufPos=0;

		if ((interval > sizeof(buf)) || (interval < 1))
			throw StreamException("Bad ICY Meta Interval value");

		unsigned int chanPos = 0;

		unsigned int connectTime = sys->getTime();

		while ((thread.active) && sock->active())
		{
			if (!checkPreview(connectTime))
				throw StreamException("Preview time limit reached");

			if (!ch->isActive())
				throw StreamException("Channel closed");

			chanPos = ch->rawData.peekPacket(chanPos,pack);
			MemoryStream mem(pack.data,pack.len);


			if (pack.type == ChanPacket::T_DATA)
			{

				int len = pack.len;
				char *p = pack.data;
				while (len)
				{
					int rl = len;
					if ((bufPos+rl) > interval)
						rl = interval-bufPos;
					memcpy(&buf[bufPos],p,rl);
					bufPos+=rl;
					p+=rl;
					len-=rl;

					if (bufPos >= interval)
					{
						bufPos = 0;	
						sock->write(buf,interval);

						if (chanMgr->broadcastMsgInterval)
							if ((sys->getTime()-lastMsgTime) >= chanMgr->broadcastMsgInterval)
							{
								showMsg ^= true;
								lastMsgTime = sys->getTime();
							}

						String *metaTitle = &ch->info.track.title;
						if (!ch->info.comment.isEmpty() && (showMsg))
							metaTitle = &ch->info.comment;


						if (!metaTitle->isSame(lastTitle) || !ch->info.url.isSame(lastURL))
						{

							char tmp[1024];
							String title,url;

							title = *metaTitle;
							url = ch->info.url;

							title.convertTo(String::T_META);
							url.convertTo(String::T_META);

							sprintf(tmp,"StreamTitle='%s';StreamUrl='%s';\0",title.cstr(),url.cstr());
							int len = ((strlen(tmp) + 15+1) / 16);
							sock->writeChar(len);
							sock->write(tmp,len*16);

							lastTitle = *metaTitle;
							lastURL = ch->info.url;

							LOG_DEBUG("StreamTitle: %s, StreamURL: %s",lastTitle.cstr(),lastURL.cstr());

						}else
						{
							sock->writeChar(0);					
						}

					}
				}
			}

		}
	}catch(StreamException &e)
	{
		LOG_ERROR("Stream channel: %s",e.msg);
	}
	if (ch->numListeners)
		ch->numListeners--;
}
// -----------------------------------
void Servent::sendChannel(Channel *ch)
{
#if 0
	downData.accept = ChanPacket::T_HEAD|ChanPacket::T_DATA|ChanPacket::T_META;

	ch->numRelays++;
	outputBitrate = ch->getBitrate();


	try
	{


		sock->timeout = 10000;
		setStatus(S_CONNECTED);

		LOG_DEBUG("Starting PeerCast stream: %s",ch->info.name.cstr());


		sock->writeTag("PCST");

		ChanPacket pack;

		ch->headPack.write(*sock);

		pack.init(ChanPacket::T_META,ch->insertMeta.data,ch->insertMeta.len,ch->syncPos,ch->streamPos);
		pack.write(*sock);
		

		while ((thread.active) && sock->active())
		{
			if (!ch->isActive())
				throw StreamException("Channel closed");

			unsigned int np = downData.readPacket(syncPos,pack);
			if ((np-syncPos) > 1)
				LOG_DEBUG("sendChannel skip: %d",np-syncPos);
			syncPos = np;

			pack.write(*sock);
		}

	}catch(StreamException &e)
	{
		LOG_ERROR("Stream channel: %s",e.msg);
	}

	if (ch->numRelays)
		ch->numRelays--;
#endif
}
// -----------------------------------
void Servent::sendPCPChannel(Channel *ch)
{
	ch->numRelays++;
	outputBitrate = ch->getBitrate();

	AtomStream atom(*sock);

	try
	{

		LOG_DEBUG("Starting PCP stream of channel %d at %d/%d",ch->index,streamPos,ch->streamPos);


		sock->timeout = 10000;
		setStatus(S_CONNECTED);

		atom.writeParent(PCP_CHAN,4);
			atom.writeBytes(PCP_CHAN_ID,ch->info.id.id,16);
			ch->info.writeInfoAtoms(atom);
			ch->info.writeTrackAtoms(atom);
			atom.writeParent(PCP_CHAN_PKT,3);
				atom.writeInt(PCP_CHAN_PKT_TYPE,PCP_CHAN_PKT_HEAD);
				atom.writeInt(PCP_CHAN_PKT_POS,ch->headPack.pos);
				atom.writeBytes(PCP_CHAN_PKT_DATA,ch->headPack.data,ch->headPack.len);


		LOG_DEBUG("Sent %d bytes header",ch->headPack.len);

		unsigned int chanPos = ch->rawData.findPacket(streamPos);
		while (thread.active)
		{

			ChanPacket rawPack;

			if (ch->rawData.pollRead(chanPos))
			{
				chanPos = ch->rawData.peekPacket(chanPos,rawPack);

				streamPos = rawPack.pos;

				bool hasKey=rawPack.key.isSet();

				if (rawPack.type == ChanPacket::T_HEAD)
				{
					atom.writeParent(PCP_CHAN,2);
						atom.writeBytes(PCP_CHAN_ID,ch->info.id.id,16);
						atom.writeParent(PCP_CHAN_PKT,3 + (hasKey?1:0));
							atom.writeInt(PCP_CHAN_PKT_TYPE,PCP_CHAN_PKT_HEAD);
							atom.writeInt(PCP_CHAN_PKT_POS,rawPack.pos);
							atom.writeBytes(PCP_CHAN_PKT_DATA,rawPack.data,rawPack.len);
							if (hasKey)
								atom.writeBytes(PCP_CHAN_PKT_KEY,rawPack.key.id,16);


				}else if (rawPack.type == ChanPacket::T_DATA)
				{
					atom.writeParent(PCP_CHAN,2);
						atom.writeBytes(PCP_CHAN_ID,ch->info.id.id,16);
						atom.writeParent(PCP_CHAN_PKT,3 + (hasKey?1:0));
							atom.writeInt(PCP_CHAN_PKT_TYPE,PCP_CHAN_PKT_DATA);
							atom.writeInt(PCP_CHAN_PKT_POS,rawPack.pos);
							atom.writeBytes(PCP_CHAN_PKT_DATA,rawPack.data,rawPack.len);
							if (hasKey)
								atom.writeBytes(PCP_CHAN_PKT_KEY,rawPack.key.id,16);

				}
			}
			BroadcastState bcs;
			bcs.srcID = remoteID;
			pcpStream.readPacket(*sock,bcs);

			sys->sleepIdle();

		}



		LOG_DEBUG("PCP stream of channel %d closed.",ch->index);

	}catch(StreamException &e)
	{
		LOG_ERROR("Stream channel: %s",e.msg);
	}

	try
	{
		atom.writeInt(PCP_QUIT,0);
	}catch(StreamException &) {}


	if (ch->numRelays)
		ch->numRelays--;
}

// -----------------------------------
int Servent::serverProc(ThreadInfo *thread)
{
	thread->lock();

	Servent *sv = (Servent*)thread->data;

	try 
	{
		if (!sv->sock)
			throw StreamException("Server has no socket");

		sv->setStatus(S_LISTENING);
		//LOG4("Listening on port %d",sv->sock->host.port);


		char servIP[64];
		sv->sock->host.toStr(servIP);

		if (servMgr->isRoot)
			LOG_DEBUG("Root Server started: %s",servIP);
		else
			LOG_DEBUG("Server started: %s",servIP);
		

		while ((thread->active) && (sv->sock->active()))
		{
			ClientSocket *cs = sv->sock->accept();
			if (cs)
			{		
				Servent *ns = servMgr->allocServent();
				if (ns)
				{
					ns->networkID = servMgr->networkID;
					ns->initIncoming(cs,sv->allow);
				}else
					LOG_ERROR("Out of servents");
			}
		}
	}catch(StreamException &e)
	{
		LOG_ERROR("Server Error: %s:%d",e.msg,e.err);
	}

	
	LOG_DEBUG("Server stopped");

	sv->kill();
	return 0;
}