connectionmanager.cpp

p2p技术C源代码.rar

		uc->setNick(aNick);
		uc->setState(UserConnection::STATE_CONNECT);
		uc->connect(aServer, aPort);
	} catch(const SocketException&) {
		if(uc)
			putConnection(uc);
	}
}

void ConnectionManager::onConnected(UserConnection* aSource) throw() {
	dcassert(aSource->getState() == UserConnection::STATE_CONNECT);
	aSource->myNick(aSource->getNick());
	aSource->lock(CryptoManager::getInstance()->getLock(), CryptoManager::getInstance()->getPk());
	aSource->setState(UserConnection::STATE_NICK);
}

/**
 * Nick received. If it's a downloader, fine, otherwise it must be an uploader.
 */
void ConnectionManager::onMyNick(UserConnection* aSource, const string& aNick) throw() {

	if(aSource->getState() != UserConnection::STATE_NICK) {
		// Already got this once, ignore...
		dcdebug("CM::onMyNick %p sent nick twice\n", aSource);
		return;
	}

	dcassert(aNick.size() > 0);
	dcdebug("ConnectionManager::onMyNick %p, %s\n", aSource, aNick.c_str());
	dcassert(!aSource->getUser());

	// First, we try looking in the pending downloads...hopefully it's one of them...
	{
		Lock l(cs);
		for(ConnectionQueueItem::TimeIter i = pendingDown.begin(); i != pendingDown.end(); ++i) {
			if(i->first->getUser()->getNick() == aNick) {
				aSource->setUser(i->first->getUser());
				// Indicate that we're interested in this file...
				aSource->setFlag(UserConnection::FLAG_DOWNLOAD);
			}
		}
	}

	if(!aSource->getUser()) {
		// Make sure we know who it is, i e that he/she is connected...
		if(!ClientManager::getInstance()->isOnline(aNick)) {
			dcdebug("CM::onMyNick Incoming connection from unknown user %s\n", aNick.c_str());
			putConnection(aSource);
			return;
		}

		aSource->setUser(ClientManager::getInstance()->getUser(aNick));
		// We don't need this connection for downloading...make it an upload connection instead...
		aSource->setFlag(UserConnection::FLAG_UPLOAD);
	}

	if( aSource->isSet(UserConnection::FLAG_INCOMING) ) {
		aSource->myNick(aSource->getUser()->getClientNick()); 
		aSource->lock(CryptoManager::getInstance()->getLock(), CryptoManager::getInstance()->getPk());
	}

	aSource->setState(UserConnection::STATE_LOCK);
}

void ConnectionManager::onLock(UserConnection* aSource, const string& aLock, const string& aPk) throw() {
	if(aSource->getState() != UserConnection::STATE_LOCK) {
		dcdebug("CM::onLock %p received lock twice, ignoring\n", aSource);
		return;
	}
	
	if( CryptoManager::getInstance()->isExtended(aLock) ) {
		// Alright, we have an extended protocol, set a user flag for this user and refresh his info...
		if( (aPk.find("DCPLUSPLUS") != string::npos) && aSource->getUser()) {
			aSource->getUser()->setFlag(User::DCPLUSPLUS);
			User::updated(aSource->getUser());
		}
		aSource->supports(features);
	}

	aSource->setState(UserConnection::STATE_DIRECTION);
	aSource->direction(aSource->getDirectionString(), aSource->getNumber());
	aSource->key(CryptoManager::getInstance()->makeKey(aLock));
}

void ConnectionManager::onDirection(UserConnection* aSource, const string& dir, const string& num) throw() {
	if(aSource->getState() != UserConnection::STATE_DIRECTION) {
		dcdebug("CM::onDirection %p received direction twice, ignoring\n", aSource);
		return;
	}

	dcassert(aSource->isSet(UserConnection::FLAG_DOWNLOAD) ^ aSource->isSet(UserConnection::FLAG_UPLOAD));
	if(dir == "Upload") {
		// Fine, the other fellow want's to send us data...make sure we really want that...
		if(aSource->isSet(UserConnection::FLAG_UPLOAD)) {
			// Huh? Strange...disconnect...
			putConnection(aSource);
			return;
		}
	} else {
		if(aSource->isSet(UserConnection::FLAG_DOWNLOAD)) {
			int number = Util::toInt(num);
			// Damn, both want to download...the one with the highest number wins...
			if(aSource->getNumber() < number) {
				// Damn! We lost!
				aSource->unsetFlag(UserConnection::FLAG_DOWNLOAD);
				aSource->setFlag(UserConnection::FLAG_UPLOAD);
			} else if(aSource->getNumber() == number) {
				putConnection(aSource);
				return;
			}
		}
	}

	dcassert(aSource->isSet(UserConnection::FLAG_DOWNLOAD) ^ aSource->isSet(UserConnection::FLAG_UPLOAD));

	aSource->setState(UserConnection::STATE_KEY);
}

void ConnectionManager::onKey(UserConnection* aSource, const string&/* aKey*/) throw() {
	if(aSource->getState() != UserConnection::STATE_KEY) {
		dcdebug("CM::onKey Bad state, ignoring");
		return;
	}
	// We don't want any messages while the Up/DownloadManagers are working...
	aSource->removeListener(this);
	dcassert(aSource->getUser());

	{
		Lock l(cs);

		// Only one connection / user & direction...
		for(ConnectionQueueItem::Iter k = active.begin(); k != active.end(); ++k) {
			bool sameDirection = (*k)->getConnection()->isSet(UserConnection::FLAG_UPLOAD) == aSource->isSet(UserConnection::FLAG_UPLOAD);
			if( sameDirection && (*k == aSource->getUser()) ) {
				putConnection(aSource);
				return;
			}
		}
		
		ConnectionQueueItem* cqi = NULL;

		if(aSource->isSet(UserConnection::FLAG_DOWNLOAD)) {
			// See if we have a matching user in the pending connections...
			ConnectionQueueItem::TimeIter i;
			for(i = pendingDown.begin(); i != pendingDown.end(); ++i) {
				if(i->first->getUser() == aSource->getUser())
					break;
			}

			if(i == pendingDown.end()) {
				putConnection(aSource);
				return;
			}
			cqi = i->first;
			pendingDown.erase(i);
			cqi->setConnection(aSource);
		} else {
			dcassert(aSource->isSet(UserConnection::FLAG_UPLOAD));
			cqi = new ConnectionQueueItem(aSource->getUser());
			cqi->setConnection(aSource);
			fire(ConnectionManagerListener::ADDED, cqi);			
		}

		aSource->setCQI(cqi);

		dcassert(find(active.begin(), active.end(), cqi) == active.end());
		active.push_back(cqi);

		fire(ConnectionManagerListener::CONNECTED, cqi);

		if(aSource->isSet(UserConnection::FLAG_DOWNLOAD)) {
			dcdebug("ConnectionManager::onKey, leaving to downloadmanager\n");
			DownloadManager::getInstance()->addConnection(aSource);
		} else {
			dcassert(aSource->isSet(UserConnection::FLAG_UPLOAD));
			dcdebug("ConnectionManager::onKey, leaving to uploadmanager\n");
			UploadManager::getInstance()->addConnection(aSource);
		}
	}
}

void ConnectionManager::onFailed(UserConnection* aSource, const string& /*aError*/) throw() {
	if(aSource->isSet(UserConnection::FLAG_DOWNLOAD) && aSource->getCQI()) {
		{
			Lock l(cs);
			
			for(ConnectionQueueItem::Iter i = downPool.begin(); i != downPool.end(); ++i) {
				dcassert((*i)->getConnection());
				if((*i)->getConnection() == aSource) {
					dcdebug("ConnectionManager::onError Removing connection %p to %s from active pool\n", aSource, aSource->getUser()->getNick().c_str());
					downPool.erase(i);
					break;
				}
			}
		}
	}
	putConnection(aSource);
}

void ConnectionManager::removeConnection(const User::Ptr& aUser, int isDownload) {
	{
		Lock l(cs);
		for(UserConnection::Iter i = userConnections.begin(); i != userConnections.end(); ++i) {
			UserConnection* uc = *i;
			if(uc->getUser() == aUser && uc->isSet(isDownload ? UserConnection::FLAG_DOWNLOAD : UserConnection::FLAG_UPLOAD)) {
				uc->disconnect();
				break;
			}
		}
	}
}

void ConnectionManager::shutdown() {
	shuttingDown = true;
	socket.removeListener(this);
	socket.disconnect();
	{
		Lock l(cs);
		for(UserConnection::Iter j = userConnections.begin(); j != userConnections.end(); ++j) {
			(*j)->disconnect();
		}
	}
	// Wait until all connections have died out...
	while(true) {
		{
			Lock l(cs);
			if(userConnections.empty()) {
				break;
			}
		}
		Thread::sleep(50);
	}
}		

// ServerSocketListener
void ConnectionManager::onAction(ServerSocketListener::Types type) throw() {
	switch(type) {
	case ServerSocketListener::INCOMING_CONNECTION:
		onIncomingConnection();
	}
}

// UserConnectionListener
void ConnectionManager::onAction(UserConnectionListener::Types type, UserConnection* conn) throw() {
	switch(type) {
	case UserConnectionListener::CONNECTED:
		onConnected(conn); break;
	default:
		break;
	}
}
void ConnectionManager::onAction(UserConnectionListener::Types type, UserConnection* conn, const string& line) throw() {
	switch(type) {
	case UserConnectionListener::MY_NICK:
		onMyNick(conn, line); break;
	case UserConnectionListener::KEY:
		onKey(conn, line); break;
	case UserConnectionListener::FAILED:
		onFailed(conn, line); break;
	default:
		break;
	}
}
void ConnectionManager::onAction(UserConnectionListener::Types type, UserConnection* conn, const string& line1, const string& line2) throw() {
	switch(type) {
	case UserConnectionListener::C_LOCK:
		onLock(conn, line1, line2); break;
	case UserConnectionListener::DIRECTION:
		onDirection(conn, line1, line2); break;
	default:
		break;
	}
}
// UserConnectionListener
void ConnectionManager::onAction(UserConnectionListener::Types type, UserConnection* conn, const StringList& feat) throw() {
	switch(type) {
	case UserConnectionListener::SUPPORTS:
		{
			for(StringList::const_iterator i = feat.begin(); i != feat.end(); ++i) {
				if(*i == "BZList")
					conn->setFlag(UserConnection::FLAG_SUPPORTS_BZLIST);
#ifdef WITH_GETZBLOCK
				else if(*i == "GetZBlock")
					conn->setFlag(UserConnection::FLAG_SUPPORTS_GETZBLOCK);
#endif
			}
		}
		break;
	default:
		break;
	}
}

// TimerManagerListener
void ConnectionManager::onAction(TimerManagerListener::Types type, u_int32_t aTick) throw() {
	switch(type) {
	case TimerManagerListener::SECOND: onTimerSecond(aTick); break;
	case TimerManagerListener::MINUTE: onTimerMinute(aTick); break;
	}
}

/**
 * @file
 * $Id: ConnectionManager.cpp,v 1.61 2003/07/15 14:53:10 arnetheduck Exp $
 */