channel.cpp.svn-base

这是和p2p相关的一份源码

// -----------------------------------
int ChanMgr::numListeners()
{
	int tot = 0;
	for(int i=0; i<MAX_CHANNELS; i++)
		if (channels[i].isActive())
			if (channels[i].isPlaying())
				if (channels[i].numListeners>0)
					tot++;
	return tot;
}
// -----------------------------------
int ChanMgr::numIdle()
{
	int tot = 0;
	for(int i=0; i<MAX_CHANNELS; i++)
		if (channels[i].isActive())
			if (channels[i].status == Channel::S_IDLE)
				tot++;
	return tot;
}

// -----------------------------------
unsigned int ChanMgr::totalInput()
{
	unsigned int tot = 0;
	for(int i=0; i<MAX_CHANNELS; i++)
		if (channels[i].isActive())
			if (channels[i].isPlaying())
				if (channels[i].input)
					tot+=channels[i].input->bytesInPerSec;

	return tot;
}
// -----------------------------------
ChanHit	*ChanMgr::addHit(ChanInfo &info, ChanHit &h)
{
	if (searchActive)
		lastHit = sys->getTime();

	// if channel is waiting for update then give it
	Channel *ch = findChannelByID(info.id);
	if (ch)
	{
		if (!ch->isBroadcasting())
			ch->info.update(info);
	}


	// otherwise add to list of channels
	ChanHitList *hl = findHitListByID(info.id);
	if (!hl)
		hl = addHitList(info);

	if (hl)
	{
		hl->info.update(info);
		return hl->addHit(h);
	}else
		return NULL;
}

// -----------------------------------
void ChanMgr::playChannels(Channel **cl, int num)
{
	if (num)
	{
		char str[128];
		sprintf(str,"http://localhost:%d",servMgr->serverHost.port);

		PlayList *pls = new PlayList(PlayList::T_SCPLS,num);
		pls->addChannels(str,cl,num);


		FileStream file;
		file.openWriteReplace("play.pls");
		pls->write(file);
		file.close();

		sys->executeFile("play.pls");

		delete pls;
	}

}
// -----------------------------------
class ChanFindInfo : public ThreadInfo
{
public:
	GnuID	id;
	bool	keep;
};
// -----------------------------------
THREAD_PROC playChannelProc(ThreadInfo *th)
{
	ChanFindInfo *cfi = (ChanFindInfo *)th;

	ChanInfo info;
	info.id = cfi->id;

	peercastApp->notifyMessage(ServMgr::NT_PEERCAST,"Connecting to channel, please wait...");

	Channel *ch;
	int num = chanMgr->findChannels(info,&ch,1);

	if (!num)
		num = chanMgr->findAndRelay(info,&ch,1);

	if (num)
	{
		ch->prefetchCnt = 20;	// prefetch for 20 packets before giving up and closing channel
		chanMgr->playChannels(&ch,1);
		ch->stayConnected = cfi->keep;
	}

	delete cfi;
	return 0;
}
// -----------------------------------
void ChanMgr::playChannel(GnuID &id, bool keep)
{
	ChanFindInfo *cfi = new ChanFindInfo;
	cfi->id = id;
	cfi->keep = keep;
	cfi->func = playChannelProc;

	sys->startThread(cfi);
}
// -----------------------------------
void ChanHitList::init()
{
	info.init();
	memset(hits,0,sizeof(ChanHit)*MAX_HITS);
	lastHitTime = 0;
	locked = false;
}
// -----------------------------------
bool	ChanHitList::isAvailable() 
{
	return (numHits()-numBusy())>0;
}

// -----------------------------------
ChanHit *ChanHitList::addHit(ChanHit &h)
{
	int i;

	lastHitTime = sys->getTime();
	h.time = lastHitTime;

	for(i=0; i<MAX_HITS; i++)
		if (hits[i].host.ip == h.host.ip)
			if (hits[i].host.port == h.host.port)
			{
				hits[i] = h;
				return &hits[i];
			}


	for(i=0; i<MAX_HITS; i++)
		if (hits[i].host.ip == 0)
		{
			hits[i] = h;
			return &hits[i];
		}

	return NULL;
}


// -----------------------------------
int	ChanHitList::clearDeadHits(unsigned int timeout)
{
	int cnt=0;
	unsigned int ctime = sys->getTime();
	for(int i=0; i<MAX_HITS; i++)
		if (hits[i].host.ip)
		{
			if ((ctime-hits[i].time) > timeout)
				hits[i].init();
			else
				cnt++;
		}
	return cnt;
}


// -----------------------------------
void	ChanHitList::deadHit(ChanHit &h)
{
	for(int i=0; i<MAX_HITS; i++)
		if (hits[i].host.ip == h.host.ip)
			if (hits[i].host.port == h.host.port)
				hits[i].init();
}
// -----------------------------------
int	ChanHitList::numHits()
{
	int cnt=0;
	for(int i=0; i<MAX_HITS; i++)
		if (hits[i].host.ip)
			cnt++;

	return cnt;
}
// -----------------------------------
int	ChanHitList::numListeners()
{
	int cnt=0;
	for(int i=0; i<MAX_HITS; i++)
		if (hits[i].host.ip)
			cnt += hits[i].numListeners;

	return cnt;
}
// -----------------------------------
int	ChanHitList::numBusy()
{
	int cnt=0;
	for(int i=0; i<MAX_HITS; i++)
		if (hits[i].host.ip)
			cnt += hits[i].busy?1:0;

	return cnt;
}
// -----------------------------------
int	ChanHitList::numStable()
{
	int cnt=0;
	for(int i=0; i<MAX_HITS; i++)
		if (hits[i].host.ip)
			cnt += hits[i].stable?1:0;

	return cnt;
}
// -----------------------------------
int	ChanHitList::numFirewalled()
{
	int cnt=0;
	for(int i=0; i<MAX_HITS; i++)
		if (hits[i].host.ip)
			cnt += hits[i].firewalled?1:0;

	return cnt;
}
// -----------------------------------
int	ChanHitList::closestHit()
{
	int hop=10000;
	for(int i=0; i<MAX_HITS; i++)
		if (hits[i].host.ip)
			if (hits[i].hops < hop)
				hop = hits[i].hops;

	return hop;
}
// -----------------------------------
int	ChanHitList::furthestHit()
{
	int hop=0;
	for(int i=0; i<MAX_HITS; i++)
		if (hits[i].host.ip)
			if (hits[i].hops > hop)
				hop = hits[i].hops;

	return hop;
}
// -----------------------------------
unsigned int	ChanHitList::newestHit()
{
	unsigned int time=0;
	for(int i=0; i<MAX_HITS; i++)
		if (hits[i].host.ip)
			if (hits[i].time > time)
				time = hits[i].time;

	return time;
}
// -----------------------------------
ChanHit	ChanHitList::getHit(bool allowPush)
{
	int minHops = 100;
	unsigned int maxUpTime = 0;

	ChanHit ch;
	ch.init();

	for(int i=0; i<MAX_HITS; i++)	
	{
		ChanHit *c = &hits[i];
		if (c->host.ip)
			if (!c->busy)
				if (c->upTime >= maxUpTime)
				{
					bool firewallOK;

					firewallOK = (allowPush && c->firewalled) || (!c->firewalled) || (c->host.localIP());

					if (c->firewalled)
						if (((c->hops > chanMgr->maxPushHops) || (c->hops > minHops)))
							firewallOK = false;
					
					if (firewallOK)
					{
						ch = *c;
						minHops = c->hops;
						maxUpTime = c->upTime;
					}
				}
	}
	return ch;
}

// -----------------------------------
void ChanHit::init()
{
	host.ip = 0;
	host.port = 0;
	numListeners = 0;
	firewalled = busy = stable = false;
	hops = 0;
	index = 0;
	time = upTime = 0;
	agentStr[0]=0;
	numSkips=0;
	packetID.clear();
	maxPreviewTime=0;
}
// -----------------------------------
const char *ChanInfo::getTypeStr(int t)
{
	switch (t)
	{
		case T_OGG: return "OGG";
		case T_RAW: return "RAW";
		case T_MP3: return "MP3";
		case T_MOV: return "MOV";
		case T_MPG: return "MPG";
		case T_NSV: return "NSV";
		case T_PEERCAST: return "PEERCAST";
		default: return "Unknown";
	}
}
// -----------------------------------
const char *ChanInfo::getTypeExt(int t)
{
	switch(t)
	{
		case ChanInfo::T_OGG:
			return "ogg";
		case ChanInfo::T_MP3:
			return "mp3";
		case ChanInfo::T_MOV:
			return "mov";
		case ChanInfo::T_NSV:
			return "nsv";
		case ChanInfo::T_PEERCAST:
			return "channel";
		default:
			return "stream";
	}
}
// -----------------------------------
ChanInfo::TYPE ChanInfo::getTypeFromStr(const char *str)
{
	if (stricmp(str,"MP3")==0)
		return T_MP3;
	else if (stricmp(str,"OGG")==0)
		return T_OGG;
	else if (stricmp(str,"RAW")==0)
		return T_RAW;
	else if (stricmp(str,"NSV")==0)
		return T_NSV;
	else if (stricmp(str,"PEERCAST")==0)
		return T_PEERCAST;
	else if (stricmp(str,"PLS")==0)
		return T_PLAYLIST;
	else if (stricmp(str,"M3U")==0)
		return T_PLAYLIST;
	else 
		return T_UNKNOWN;
}
// -----------------------------------
bool	ChanInfo::match(ChanInfo &inf)
{
	bool matchAny=true;

	if (inf.status != S_UNKNOWN)
	{
		if (status != inf.status)
			return false;
	}

	if (inf.bitrate != 0)
	{
		if (bitrate == inf.bitrate)
			return true;
		matchAny = false;
	}

	if (inf.id.isSet())
	{
		if (id.isSame(inf.id))
			return true;
		matchAny = false;
	}

	if (inf.contentType != T_UNKNOWN)
	{
		if (contentType == inf.contentType)
			return true;
		matchAny = false;
	}

	if (!inf.name.isEmpty())
	{
		if (name.contains(inf.name))
			return true;
		matchAny = false;
	}

	if (!inf.genre.isEmpty())
	{
		if (genre.contains(inf.genre))
			return true;
		matchAny = false;
	}

	return matchAny;
}

// -----------------------------------
void ChanInfo::update(ChanInfo &info)
{
	// only update from chaninfo that has full name etc..
	if (info.name.isEmpty())
		return;


	if (contentType == T_UNKNOWN)
	{
		bitrate = info.bitrate;
		contentType = info.contentType;
		name = info.name;
	}

	track = info.track;
	comment = info.comment;
	genre = info.genre;
	url = info.url;
}
// -----------------------------------
void ChanInfo::initNameID(const char *n)
{
	init();
	id.fromStr(n);
	if (!id.isSet())
		name.set(n);
}

// -----------------------------------
void ChanInfo::init()
{
	status = S_UNKNOWN;
	name.clear();
	bitrate = 0;
	contentType = T_UNKNOWN;
	srcType = T_UNKNOWN;
	id.clear();
	url.clear();
	genre.clear();
	comment.clear();
	track.clear();
	lastPlay = 0;
	numSkips = 0;

}
// -----------------------------------
void ChanInfo::readTrackXML(XML::Node *n)
{
	track.clear();
	readXMLString(track.title,n,"title");
	readXMLString(track.contact,n,"contact");
	readXMLString(track.artist,n,"artist");
	readXMLString(track.album,n,"album");
	readXMLString(track.genre,n,"genre");
}
// -----------------------------------
XML::Node *ChanInfo::createChannelXML()
{
	char idStr[64];

	String nameHTML = name;
	nameHTML.convertTo(String::T_HTML);
	String urlHTML = url;
	urlHTML.convertTo(String::T_HTML);
	String genreHTML = genre;
	genreHTML.convertTo(String::T_HTML);
	String descHTML = desc;
	descHTML.convertTo(String::T_HTML);
	String commentHTML;
	commentHTML = comment;
	commentHTML.convertTo(String::T_HTML);


	id.toStr(idStr);


	return new XML::Node("channel name=\"%s\" id=\"%s\" bitrate=\"%d\" type=\"%s\" genre=\"%s\" desc=\"%s\" url=\"%s\" uptime=\"%d\" comment=\"%s\" skips=\"%d\"",
		nameHTML.cstr(),
		idStr,
		bitrate,
		getTypeStr(contentType),
		genreHTML.cstr(),
		descHTML.cstr(),
		urlHTML.cstr(),
		lastPlay?(sys->getTime()-lastPlay):0,
		commentHTML.cstr(),
		numSkips
		);	
}

// -----------------------------------
XML::Node *ChanInfo::createQueryXML()
{
	char buf[512];
	char idStr[64];


	String nameHTML = name;
	nameHTML.convertTo(String::T_HTML);
	String genreHTML = genre;
	genreHTML.convertTo(String::T_HTML);

	buf[0]=0;
	if (!nameHTML.isEmpty())
	{
		strcat(buf," name=\"");
		strcat(buf,nameHTML.cstr());
		strcat(buf,"\"");
	}

	if (!genreHTML.isEmpty())
	{
		strcat(buf," genre=\"");
		strcat(buf,genreHTML.cstr());
		strcat(buf,"\"");
	}

	if (id.isSet())
	{
		id.toStr(idStr);
		strcat(buf," id=\"");
		strcat(buf,idStr);
		strcat(buf,"\"");
	}
		

	return new XML::Node("channel %s",buf);
}

// -----------------------------------
XML::Node *ChanInfo::createRelayChannelXML()
{
	char idStr[64];

	id.toStr(idStr);


	return new XML::Node("channel id=\"%s\" uptime=\"%d\" skips=\"%d\"",
		idStr,
		lastPlay?(sys->getTime()-lastPlay):0,
		numSkips
		);	
}// -----------------------------------
XML::Node *ChanInfo::createTrackXML()
{
	String titleHTML,contactHTML,albumHTML,genreHTML,artistHTML;
	titleHTML = track.title;
	titleHTML.convertTo(String::T_HTML);
	artistHTML = track.artist;
	artistHTML.convertTo(String::T_HTML);
	albumHTML = track.album;
	albumHTML.convertTo(String::T_HTML);
	genreHTML = track.genre;
	genreHTML.convertTo(String::T_HTML);
	contactHTML = track.contact;
	contactHTML.convertTo(String::T_HTML);


	return new XML::Node("track title=\"%s\" artist=\"%s\" album=\"%s\" genre=\"%s\" contact=\"%s\"",
		titleHTML.cstr(),
		artistHTML.cstr(),
		albumHTML.cstr(),
		genreHTML.cstr(),
		contactHTML.cstr()
		);
}

// -----------------------------------
void ChanInfo::init(XML::Node *n)
{
	init();

	updateFromXML(n);
}
// -----------------------------------
void ChanInfo::updateFromXML(XML::Node *n)
{
	String typeStr,idStr;

	readXMLString(name,n,"name");
	readXMLString(genre,n,"genre");
	readXMLString(url,n,"url");
	readXMLString(desc,n,"desc");


	int br = n->findAttrInt("bitrate");
	if (br)
		bitrate = br;

	readXMLString(typeStr,n,"type");
	if (!typeStr.isEmpty())
		contentType = getTypeFromStr(typeStr.cstr());


	readXMLString(idStr,n,"id");
	if (!idStr.isEmpty())
		id.fromStr(idStr.cstr());

	readXMLString(comment,n,"comment");

	XML::Node *tn = n->findNode("track");
	if (tn)
		readTrackXML(tn);

}

// -----------------------------------
void ChanInfo::init(const char *n, GnuID &cid, TYPE tp, int br)
{
	init();

	name.set(n);
	bitrate = br;
	contentType = tp;
	id = cid;
}

// -----------------------------------
void ChanInfo::init(const char *fn)
{
	init();

	if (fn)
		name.set(fn);
}

// -----------------------------------
void PlayList::writeSCPLS(Stream &out)
{
	out.writeLine("[playlist]");
	out.writeLine("");
	out.writeLine("NumberOfEntries=%d",numURLs);

	for(int i=0; i<numURLs; i++)
	{
		out.writeLine("File%d=%s",i+1,urls[i].cstr());
		out.writeLine("Title%d=%s",i+1,titles[i].cstr());
		out.writeLine("Length%d=-1",i+1);
	}
	out.writeLine("Version=2");
}
// -----------------------------------
void PlayList::writePLS(Stream &out)
{
	for(int i=0; i<numURLs; i++)
		out.writeLine("%s",urls[i].cstr());
}

// -----------------------------------
void PlayList::addChannels(const char *path, Channel **cl, int num)
{
	String url;

	for(int i=0; i<num; i++)
	{
		char idStr[64];

		cl[i]->getIDStr(idStr);

		sprintf(url.cstr(),"%s/stream/%s.%s",path,idStr,ChanInfo::getTypeExt(cl[i]->info.contentType));
		addURL(url.cstr(),cl[i]->info.name);

	}	
}
// -----------------------------------
void PlayList::addChannel(const char *path, ChanInfo &info)
{
	String url;

	char idStr[64];

	info.id.toStr(idStr);

	sprintf(url.cstr(),"%s/stream/%s.%s",path,idStr,ChanInfo::getTypeExt(info.contentType));
	addURL(url.cstr(),info.name);
}