queuemanager.cpp

p2p技术C源代码.rar

		}

		s = q->addSource(aUser, aFile);
		
		if(newItem && (p != QueueItem::DEFAULT)) {
			q->setPriority(p);
		}

		if(q->getPriority() == QueueItem::PAUSED) {
			wantConnection = false;
		}

		if(aUser->isSet(User::PASSIVE) && (SETTING(CONNECTION_TYPE) != SettingsManager::CONNECTION_ACTIVE) ) {
			q->removeSource(aUser, QueueItem::Source::FLAG_PASSIVE);
			wantConnection = false;
		} else if(q->getStatus() != QueueItem::STATUS_RUNNING) {
			userQueue.add(q, aUser);
		} else {
			wantConnection = false;
		}

		if(newItem) {
			if(!q->isSet(QueueItem::FLAG_USER_LIST)) {
				if(aTempTarget.empty()) {
					if(!SETTING(TEMP_DOWNLOAD_DIRECTORY).empty() && (File::getSize(q->getTarget()) == -1)) {
						q->setTempTarget(SETTING(TEMP_DOWNLOAD_DIRECTORY) + getTempName(q->getTargetFileName()));
					}
				} else {
					q->setTempTarget(aTempTarget);
				}
			}

			if(lastInsert == queue.end()) {
				lastInsert = queue.insert(make_pair(q->getTarget(), q)).first;
			} else {
				lastInsert = queue.insert(lastInsert, make_pair(q->getTarget(), q));
			}
			fire(QueueManagerListener::ADDED, q);

		}

		fire(QueueManagerListener::SOURCES_UPDATED, q);
		setDirty();
	}

	if(wantConnection && aUser->isOnline())
		ConnectionManager::getInstance()->getDownloadConnection(aUser);
}

QueueItem* QueueManager::getQueueItem(const string& aTarget, int64_t aSize, bool aResume, bool& newItem) throw(QueueException, FileException) {
	QueueItem* q;
	if((lastInsert != queue.end()) && (lastInsert->first == aTarget)) {
		q = lastInsert->second;
	} else {
		q = findByTarget(aTarget);
	}
		
	if(q == NULL) {
		newItem = true;

		q = new QueueItem(aTarget, aSize, (aSize <= 16*1024) ? QueueItem::HIGHEST : QueueItem::NORMAL, aResume);

	} else {
		newItem = false;
		if(q->getSize() != aSize) {
			throw QueueException(STRING(FILE_WITH_DIFFERENT_SIZE));
		}
	}

	return q;
}

void QueueManager::addDirectory(const string& aDir, User::Ptr& aUser, const string& aTarget, QueueItem::Priority p /* = QueueItem::DEFAULT */) throw() {
	bool needList;
	{
		Lock l(cs);
		
		DirectoryItem::DirectoryPair dp = directories.equal_range(aUser);
		
		for(DirectoryItem::DirectoryIter i = dp.first; i != dp.second; ++i) {
			if(Util::stricmp(aTarget.c_str(), i->second->getName().c_str()) == 0)
				return;
		}
		
		// Unique directory, fine...
		directories.insert(make_pair(aUser, new DirectoryItem(aUser, aDir, aTarget, p)));
		needList = (dp.first == dp.second);
		setDirty();
	}

	if(needList) {
		try {
			addList(aUser, true);
		} catch(const Exception&) {
			// Ignore, we don't really care...
		}
	}
}

typedef vector<pair<DirectoryListing::File*, string> > MatchList;
typedef MatchList::iterator MatchIter;

static void matchFile(const string& name, const int64_t& size, 
					  MatchList& f, DirectoryListing::Directory* dir, const string& target) throw() {

	for(DirectoryListing::File::Iter i = dir->files.begin(); i != dir->files.end(); ++i) {
		DirectoryListing::File* df = *i;
		if(df->getSize() == size && Util::stricmp(df->getName(), name) == 0) {
			f.push_back(make_pair(df, target));
		}
	}

	for(DirectoryListing::Directory::Iter j = dir->directories.begin(); j != dir->directories.end(); ++j) {
		if(!(*j)->getAdls())
			matchFile(name, size, f, *j, target);
	}
}

int QueueManager::matchListing(DirectoryListing* dl) throw() {
	MatchList f;
	{
		Lock l(cs);
		for(QueueItem::StringIter i = queue.begin(); i != queue.end(); ++i) {
			QueueItem* qi = i->second;
			if(qi->getSize() != -1) {
				matchFile(qi->getTargetFileName(), qi->getSize(), f, dl->getRoot(), qi->getTarget());
			}
		}
	}
	int totalItems = (int)f.size();
	for(MatchIter i = f.begin(); i != f.end(); ++i) {
		DirectoryListing::File* df = i->first;
		try {
			add(dl->getPath(df) + df->getName(), df->getSize(), dl->getUser(), i->second);
		} catch(const Exception&) {
			totalItems--;
		}
	}
	return totalItems;
}

void QueueManager::move(const string& aSource, const string& aTarget) throw() {
	if(Util::stricmp(aSource.c_str(), aTarget.c_str()) == 0)
		return;

	QueueItem* copy = NULL;

	Lock l(cs);
	QueueItem* qs = findByTarget(aSource);
	if(qs != NULL) {
		// Don't move running downloads
		if(qs->getStatus() == QueueItem::STATUS_RUNNING) {
			return;
		}
		// Don't move file lists
		if(qs->isSet(QueueItem::FLAG_USER_LIST))
			return;

		// Let's see if the target exists...then things get complicated...
		QueueItem* qt = findByTarget(aTarget);
		if(qt == NULL) {
			// Good, update the target and move in the queue...
			queue.erase(aSource);
			qs->setTarget(aTarget);
			lastInsert = queue.insert(make_pair(qs->getTarget(), qs)).first;
			fire(QueueManagerListener::MOVED, qs);
			setDirty();
		} else {
			// Don't move to target of different size
			if(qs->getSize() != qt->getSize())
				return;

			copy = new QueueItem(*qs);
		}
	}
	if(copy != NULL) {
		remove(aSource);
		// Damn...difficult...we add all sources then remove the old download...
		for(QueueItem::Source::Iter i = copy->getSources().begin(); i != copy->getSources().end(); ++i) {
			try {
				QueueItem::Source* s = *i;
				add(s->getPath(), copy->getSize(), s->getUser(), aTarget);
			} catch(const Exception&) {
				// Blah, ignore...
			}
		}
		delete copy;
	}
}


Download* QueueManager::getDownload(User::Ptr& aUser) throw() {
	Lock l(cs);

	QueueItem* q = userQueue.getNext(aUser);

	if(q == NULL)
		return NULL;

	userQueue.setRunning(q, aUser);

	fire(QueueManagerListener::STATUS_UPDATED, q);
	
	return new Download(q);
}


void QueueManager::putDownload(Download* aDownload, bool finished /* = false */) throw() {
	User::List getConn;
	string file;
	User::Ptr up;
	bool isBZ = false;

	{
		Lock l(cs);
		QueueItem* q = findByTarget(aDownload->getTarget());

		if(q != NULL) {
			if(finished) {
				dcassert(q->getStatus() == QueueItem::STATUS_RUNNING);
				userQueue.remove(q);
				queue.erase(q->getTarget());
				lastInsert = queue.end();
				if(aDownload->isSet(Download::FLAG_USER_LIST) && aDownload->getSource() == "MyList.bz2") {
					q->setFlag(QueueItem::FLAG_BZLIST);
					isBZ = true;
				}
				fire(QueueManagerListener::FINISHED, q);
				fire(QueueManagerListener::REMOVED, q);
				// Now, let's see if this was a directory download filelist...
				if(q->isSet(QueueItem::FLAG_DIRECTORY_DOWNLOAD) && directories.find(q->getCurrent()->getUser()) != directories.end()) {
					try {
						string fname = q->isSet(QueueItem::FLAG_BZLIST) ? aDownload->getTarget().substr(0, aDownload->getTarget().length() - 5) + "bz2" : aDownload->getTarget();
						file = File(fname, File::READ, File::OPEN).read();
						up = q->getCurrent()->getUser();
					} catch(const FileException&) {
						// ...
					}
				}
				delete q;
				setDirty();
			} else {
				if(q->getPriority() != QueueItem::PAUSED) {
					for(QueueItem::Source::Iter j = q->getSources().begin(); j != q->getSources().end(); ++j) {
						if((*j)->getUser()->isOnline()) {
							getConn.push_back((*j)->getUser());
						}
					}
				}

				// This might have been set to wait by removesource already...
				if(q->getStatus() == QueueItem::STATUS_RUNNING) {
					userQueue.setWaiting(q);
					fire(QueueManagerListener::STATUS_UPDATED, q);
				}
				if(q->isSet(QueueItem::FLAG_USER_LIST)) {
					// Blah...no use keeping an unfinished file list...
					string fname = q->isSet(QueueItem::FLAG_BZLIST) ? aDownload->getTarget().substr(0, aDownload->getTarget().length() - 5) + "bz2" : aDownload->getTarget();
					File::deleteFile(fname);
				}
			}
		} else if(!aDownload->getTempTarget().empty() && aDownload->getTempTarget() != aDownload->getTarget()) {
			File::deleteFile(aDownload->getTempTarget());
		}
		aDownload->setUserConnection(NULL);
		delete aDownload;
	}

	for(User::Iter i = getConn.begin(); i != getConn.end(); ++i) {
		ConnectionManager::getInstance()->getDownloadConnection(*i);
	}

	if(!file.empty()) {
		string userList;
		try {
			if(isBZ) {
				CryptoManager::getInstance()->decodeBZ2((u_int8_t*)file.c_str(), file.size(), userList);
			} else {
				CryptoManager::getInstance()->decodeHuffman((u_int8_t*)file.c_str(), userList);
			}
		} catch(const CryptoException&) {
			addList(up, true);
			return;
		}

		DirectoryItem::List dl;
		{
			Lock l(cs);
			DirectoryItem::DirectoryPair dp = directories.equal_range(up);
			for(DirectoryItem::DirectoryIter i = dp.first; i != dp.second; ++i) {
				dl.push_back(i->second);
			}
			directories.erase(up);
		}
		DirectoryListing dirList(up);
		dirList.load(userList);

		for(DirectoryItem::Iter i = dl.begin(); i != dl.end(); ++i) {
			DirectoryItem* di = *i;
			dirList.download(di->getName(), di->getTarget());
			delete di;
		}
	}
}

void QueueManager::remove(const string& aTarget) throw() {
	string x;
	{
		Lock l(cs);

		QueueItem* q = findByTarget(aTarget);
		if(q != NULL) {
			if(q->isSet(QueueItem::FLAG_DIRECTORY_DOWNLOAD)) {
				dcassert(q->getSources().size() == 1);
				DirectoryItem::DirectoryPair dp = directories.equal_range(q->getSources()[0]->getUser());
				for(DirectoryItem::DirectoryIter i = dp.first; i != dp.second; ++i) {
					delete i->second;
				}
				directories.erase(q->getSources()[0]->getUser());
			}

			queue.erase(q->getTarget());
			lastInsert = queue.end();
			if(q->getStatus() == QueueItem::STATUS_RUNNING) {
				x = q->getTarget();
			} else if(!q->getTempTarget().empty() && q->getTempTarget() != q->getTarget()) {
				File::deleteFile(q->getTempTarget());
			}

			userQueue.remove(q);

			fire(QueueManagerListener::REMOVED, q);
			delete q;
			setDirty();
		}
	}
	if(!x.empty()) {
		DownloadManager::getInstance()->abortDownload(x);
	}
}

void QueueManager::removeSource(const string& aTarget, User::Ptr& aUser, int reason, bool removeConn /* = true */) throw() {
	Lock l(cs);
	QueueItem* q = findByTarget(aTarget);
	string x;
	if(q != NULL) {
		dcassert(q->isSource(aUser));
		if(q->isSet(QueueItem::FLAG_USER_LIST)) {
			remove(q->getTarget());
			return;
		}
		if(reason == QueueItem::Source::FLAG_CRC_WARN) {
			// Already flagged?
			QueueItem::Source* s = *q->getSource(aUser);
			if(s->isSet(QueueItem::Source::FLAG_CRC_WARN)) {
				reason = QueueItem::Source::FLAG_CRC_FAILED;
			} else {
				s->setFlag(reason);
				return;
			}
		}
		if((q->getStatus() == QueueItem::STATUS_RUNNING) && q->getCurrent()->getUser() == aUser) {
			if(removeConn)
				x = q->getTarget();
			userQueue.setWaiting(q);
			userQueue.remove(q, aUser);
		} else if(q->getStatus() == QueueItem::STATUS_WAITING) {
			userQueue.remove(q, aUser);