⭐ 欢迎来到虫虫下载站！ | 📦 资源下载 📁 资源专辑 ℹ️ 关于我们
⭐ 虫虫下载站
📤 上传资源

📄 channel.cpp

📁 这是和p2p相关的一份源码
💻 CPP
📖 第 1 页 / 共 5 页
字号:
🔍
12345 
		for(int i=0; i<MAX_CHANNELS; i++)		{			Channel *c = &channels[i];			if (c->isPlaying())			{

				bool tracker = c->isBroadcasting();
				int ttl = (c->info.getUptime() / servMgr->relayBroadcast);	// 1 hop per N seconds				if (ttl < minTTL)					ttl = minTTL;				if (ttl > maxTTL)					ttl = maxTTL;				if (hit.initHit(sh,c,NULL,push,busy,stable,tracker,ttl))				{					int numOut=0;					numChans++;					if (serv)					{						serv->outputPacket(hit,false);						numOut++;					}					LOG_NETWORK("Sent ch.%d to %d servents, TTL %d",c->index,numOut,ttl);						}			}		}		//if (numChans)		//	LOG_NETWORK("Sent %d channels to %d servents",numChans,numOut);			}}// -----------------------------------
void ChanMgr::setUpdateInterval(unsigned int v)
{
	hostUpdateInterval = v;
}


// -----------------------------------
// message check
#if 0
				ChanPacket pack;
				MemoryStream mem(pack.data,sizeof(pack.data));
				AtomStream atom(mem);
				atom.writeParent(PCP_BCST,3);
					atom.writeChar(PCP_BCST_GROUP,PCP_BCST_GROUP_ALL);
					atom.writeBytes(PCP_BCST_FROM,servMgr->sessionID.id,16);
					atom.writeParent(PCP_MESG,1);
						atom.writeString(PCP_MESG_DATA,msg.cstr());

				mem.len = mem.pos;
				mem.rewind();
				pack.len = mem.len;

				GnuID noID;
				noID.clear();

				BroadcastState bcs;
				PCPStream::readAtom(atom,bcs);
				//int cnt = servMgr->broadcastPacketUp(pack,noID,servMgr->sessionID);
				//int cnt = servMgr->broadcastPacketDown(pack,noID,servMgr->sessionID);
				//int cnt = chanMgr->broadcastPacketUp(pack,noID,servMgr->sessionID);
				//LOG_DEBUG("Sent message to %d clients",cnt);
#endif
// -----------------------------------void ChanMgr::setBroadcastMsg(String &msg){
	if (!msg.isSame(broadcastMsg))	{		broadcastMsg = msg;		for(int i=0; i<MAX_CHANNELS; i++)		{			Channel *c = &channels[i];			if (c->isActive() && c->isBroadcasting())
			{
				ChanInfo newInfo = c->info;				newInfo.comment = broadcastMsg;				c->updateInfo(newInfo);
			}		}	}}

// -----------------------------------void ChanMgr::clearHitLists(){	for(int i=0; i<MAX_HITLISTS; i++)	{		peercastApp->delChannel(&hitlists[i].info);		hitlists[i].init();	}}// -----------------------------------Channel *ChanMgr::createChannel(ChanInfo &info, const char *mount){	lock.on();

	int i;
	Channel *nc=NULL;	for(i=0; i<MAX_CHANNELS; i++)	{		Channel *c = &channels[i];		if (!c->isActive())
		{
			nc=c;
			break;
		}
	}


	if (!nc)
	{
		for(i=0; i<MAX_CHANNELS; i++)
		{
			Channel *c = &channels[i];
			if (!c->isPlaying())
			{
				c->close();
				for(int j=0; j<100; j++)	// wait 10 seconds
				{
					if (!c->isActive())
					{
						nc = c;
						break;
					}
					sys->sleep(100);
				}
				if (nc)
					break;
			}
		}
	}



	if (nc)
	{		nc->info = info;		nc->info.lastPlayTime = 0;		nc->info.status = ChanInfo::S_UNKNOWN;		if (mount)			nc->mount.set(mount);		nc->index = i+1;		nc->setStatus(Channel::S_WAIT);		nc->type = Channel::T_ALLOCATED;

		LOG_CHANNEL("New channel (%d) created",nc->index);
		lock.off();		return nc;	}else
	{		LOG_ERROR("Unable to create channel");
		lock.off();		return NULL;
	}}// -----------------------------------ChanHitList *ChanMgr::findHitList(ChanInfo &info){
	for(int i=0; i<MAX_HITLISTS; i++)		if (hitlists[i].isUsed())			if (hitlists[i].info.matchNameID(info))				return &hitlists[i];	return NULL;}// -----------------------------------
ChanHitList *ChanMgr::findHitListByID(GnuID &id)
{
	for(int i=0; i<MAX_HITLISTS; i++)
		if (hitlists[i].isUsed())
			if (hitlists[i].info.id.isSame(id))
				return &hitlists[i];
	return NULL;
}
// -----------------------------------int ChanMgr::numHitLists(){	int num=0;	for(int i=0; i<MAX_HITLISTS; i++)		if (hitlists[i].isUsed())			num++;	return num;}// -----------------------------------ChanHitList *ChanMgr::addHitList(ChanInfo &info){	ChanHitList *hl = NULL;	for(int i=0; i<MAX_HITLISTS; i++)		if (!hitlists[i].isUsed())		{
			char idstr[64];
			info.id.toStr(idstr);
			LOG_DEBUG("Created new hitlist: %s",idstr);			hl = &hitlists[i];			break;		}	if (hl)	{		hl->info = info;		peercastApp->addChannel(&hl->info);
	}	return hl;}// -----------------------------------void ChanMgr::clearDeadHits(bool clearTrackers){	unsigned int interval;
	
	if (servMgr->isRoot)
		interval = 1200;		// mainly for old 0.119 clients
	else
		interval = hostUpdateInterval+30;

	for(int i=0; i<MAX_HITLISTS; i++)		if (hitlists[i].isUsed())
		{
			if (hitlists[i].clearDeadHits(interval,clearTrackers) == 0)			{
				if (!isBroadcasting(hitlists[i].info.id))
				{
					if (!chanMgr->findChannelByID(hitlists[i].info.id))
					{						LOG_DEBUG("Deleting hitlist");
						peercastApp->delChannel(&hitlists[i].info);						hitlists[i].init();
					}
				}			}
		}}// -----------------------------------
bool	ChanMgr::isBroadcasting(GnuID &id)
{
	Channel *ch = findChannelByID(id);
	if (ch)
		return ch->isBroadcasting();

	return false;
}
// -----------------------------------
bool	ChanMgr::isBroadcasting()
{
	for(int i=0; i<MAX_CHANNELS; i++)
		if (channels[i].isActive())
			if (channels[i].isBroadcasting())
				return true;
	return false;
}
// -----------------------------------int ChanMgr::numChannels(){	int tot = 0;	for(int i=0; i<MAX_CHANNELS; i++)		if (channels[i].isActive())			tot++;	return tot;}// -----------------------------------int ChanMgr::numRelayed(){	int tot = 0;	for(int i=0; i<MAX_CHANNELS; i++)		if (channels[i].isActive())			if (channels[i].isPlaying())				if (channels[i].numRelays()>0)					tot++;	return tot;}// -----------------------------------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;}// -----------------------------------
void ChanMgr::deadHit(GnuID &id, ChanHit &hit)
{
	ChanHitList *chl = findHitListByID(id);
	if (chl)
		chl->deadHit(hit);
}// -----------------------------------ChanHit	*ChanMgr::addHit(GnuID &id, ChanHit &h){	if (searchActive)		lastHit = sys->getTime();

	ChanHitList *hl=NULL;

	hl = findHitListByID(id);
	if (!hl)
	{
		ChanInfo info;
		info.id = id;
		hl = addHitList(info);
	}
	if (hl)	{
		return hl->addHit(h);	}else		return NULL;}// -----------------------------------class ChanFindInfo : public ThreadInfo{public:	ChanInfo	info;	bool	keep;};// -----------------------------------THREAD_PROC findAndPlayChannelProc(ThreadInfo *th){	ChanFindInfo *cfi = (ChanFindInfo *)th;	ChanInfo info;	info = cfi->info;
	Channel *ch = chanMgr->findChannelByNameID(info);
	chanMgr->currFindAndPlayChannel = info.id;
	if (!ch)		ch = chanMgr->findAndRelay(info);	if (ch)	{
		// check that a different channel hasn`t be selected already.
		if (chanMgr->currFindAndPlayChannel.isSame(ch->info.id))
			chanMgr->playChannel(ch->info);
		if (cfi->keep)			ch->stayConnected = cfi->keep;	}	delete cfi;	return 0;}// -----------------------------------void ChanMgr::findAndPlayChannel(ChanInfo &info, bool keep){	ChanFindInfo *cfi = new ChanFindInfo;	cfi->info = info;	cfi->keep = keep;	cfi->func = findAndPlayChannelProc;

	sys->startThread(cfi);}
// -----------------------------------
void ChanMgr::playChannel(ChanInfo &info)
{

	char str[128],fname[256],idStr[128];

	sprintf(str,"http://localhost:%d",servMgr->serverHost.port);
	info.id.toStr(idStr);

	PlayList::TYPE type;


	if ((info.contentType == ChanInfo::T_WMA) || (info.contentType == ChanInfo::T_WMV))
	{
		type = PlayList::T_ASX;
		// WMP seems to have a bug where it doesn`t re-read asx files if they have the same name
		// so we prepend the channel id to make it unique - NOTE: should be deleted afterwards.
		sprintf(fname,"%s/%s.asx",peercastApp->getPath(),idStr);	
	}else
	{
		type = PlayList::T_SCPLS;
		sprintf(fname,"%s/play.pls",peercastApp->getPath());
	}


	PlayList *pls = new PlayList(type,1);
	pls->addChannel(str,info);


	LOG_DEBUG("Writing %s",fname);
	FileStream file;
	file.openWriteReplace(fname);
	pls->write(file);
	file.close();


	LOG_DEBUG("Executing: %s",fname);
	sys->executeFile(fname);
	delete pls;

}
// -----------------------------------void ChanHitList::init(){	info.init();	memset(hits,0,sizeof(ChanHit)*MAX_HITS);	lastHitTime = 0;}
// -----------------------------------
ChanHit ChanHitList::findHit(GnuID &sid, bool tracker)
{
	ChanHit hit;
	hit.init();

	for(int i=0; i<MAX_HITS; i++)
		if (hits[i].host.ip)
			if (hits[i].sessionID.isSame(sid))
				if (hits[i].tracker == tracker)
				{
					hit = hits[i];
					break;
				}

	return hit;
}
// -----------------------------------
void ChanHit::pickNearestIP(Host &h)
{
	for(int i=0; i<2; i++)
	{
		if (h.classType() == rhost[i].classType())
		{
			host = rhost[i];
			break;
		}
	}
}

// -----------------------------------
void ChanHit::init()
{
	host.init();

	rhost[0].init();
	rhost[1].init();

	numListeners = 0;
	numRelays = 0;
	firewalled = busy = stable = false;
	hops = 0;
	index = 0;
	time = upTime = 0;
	agentStr[0]=0;
	numSkips=0;
	packetID.clear();
	maxPreviewTime=0;
	tracker=false;
	lastContact = 0;
	recv = true;
	yp=false;
}

// -----------------------------------
void ChanHit::initLocal(int numl,int numr,int nums,int uptm,bool connected,bool b)
{
	init();
	busy = b;
	firewalled = (servMgr->getFirewall() != ServMgr::FW_OFF);
	numListeners = numl;
	numRelays = numr;
	numSkips = nums;
	upTime = uptm;
	stable = servMgr->totalStreams>0;
	strcpy(agentStr,PCX_AGENT);
	sessionID = servMgr->sessionID;
	recv = connected;

	host = servMgr->serverHost;

	rhost[0] = Host(host.ip,host.port);
	rhost[1] = Host(ClientSocket::getIP(NULL),host.port);

	if (firewalled)
		rhost[0].port = 0;
}

// -----------------------------------
void ChanHit::writeAtoms(AtomStream &atom,bool tryHost, GnuID &chanID)
{
	bool addChan=chanID.isSet();	

	if (tryHost)
	{
		atom.writeParent(PCP_HOST,5 + (addChan?1:0));
			if (addChan)
				atom.writeBytes(PCP_HOST_CHANID,chanID.id,16);
			atom.writeInt(PCP_HOST_IP,rhost[0].ip);
			atom.writeShort(PCP_HOST_PORT,rhost[0].port);
			atom.writeInt(PCP_HOST_IP,rhost[1].ip);
			atom.writeShort(PCP_HOST_PORT,rhost[1].port);
			atom.writeChar(PCP_HOST_TRACKER,tracker);
	}else
	{
		atom.writeParent(PCP_HOST,13  + (addChan?1:0));
			if (addChan)
				atom.writeBytes(PCP_HOST_CHANID,chanID.id,16);
			atom.writeBytes(PCP_HOST_ID,sessionID.id,16);
			atom.writeInt(PCP_HOST_IP,rhost[0].ip);
			atom.writeShort(PCP_HOST_PORT,rhost[0].port);
			atom.writeInt(PCP_HOST_IP,rhost[1].ip);
			atom.writeShort(PCP_HOST_PORT,rhost[1].port);
			atom.writeChar(PCP_HOST_BUSY,busy);
			atom.writeChar(PCP_HOST_PUSH,firewalled);
			atom.writeChar(PCP_HOST_RECV,recv);
			atom.writeInt(PCP_HOST_NUML,numListeners);
			atom.writeInt(PCP_HOST_NUMR,numRelays);
			atom.writeString(PCP_HOST_AGENT,agentStr);
			atom.writeInt(PCP_HOST_UPTIME,upTime);
			atom.writeChar(PCP_HOST_TRACKER,tracker);
	}

}
// -----------------------------------
bool	ChanHit::writeVariable(Stream &out, const String &var)
{
	char buf[1024];

	if (var == "rhost0")
		rhost[0].toStr(buf);
	else if (var == "rhost1")
		rhost[1].toStr(buf);
	else if (var == "numHops")
		sprintf(buf,"%d",hops);
	else if (var == "numListeners")
		sprintf(buf,"%d",numListeners);
	else if (var == "numRelays")
		sprintf(buf,"%d",numRelays);
	else if (var == "uptime")
	{
		String timeStr;
		timeStr.setFromStopwatch(upTime);
		strcpy(buf,timeStr.cstr());
	}else if (var == "update")
	{
		String timeStr;
		if (timeStr)
			timeStr.setFromStopwatch(sys->getTime()-time);
		else
			timeStr.set("-");
		strcpy(buf,timeStr.cstr());
	}else if (var == "isFirewalled")
		sprintf(buf,"%d",firewalled?1:0);
	else if (var == "isBusy")
		sprintf(buf,"%d",busy?1:0);
	else if (var == "isTracker")
		sprintf(buf,"%d",tracker?1:0);
	else if (var == "agent")
		strcpy(buf,agentStr);
	else
		return false;

	out.writeString(buf);
	return true;
}

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

// -----------------------------------
int ChanHitList::contactTrackers(bool connected, int numl, int nums, int uptm)
{
	return 0;
}
12345

⌨️ 快捷键说明

复制代码 Ctrl + C
搜索代码 Ctrl + F
全屏模式 F11
切换主题 Ctrl + Shift + D
显示快捷键 ?
增大字号 Ctrl + =
减小字号 Ctrl + -