Restructure the downloader into a generic task list/queue.
Yay for bringing paper and pencil on long train rides. Nothing to do but design ... and possibly chat with random strangers ;)
This commit is contained in:
Petr Mrázek
parent
030e03e134
commit
84298d621d
594
asset_test.cpp
594
asset_test.cpp
@ -3,26 +3,324 @@
|
||||
#include <QDebug>
|
||||
#include <QtXml/QtXml>
|
||||
|
||||
enum DlStatus
|
||||
enum JobStatus
|
||||
{
|
||||
Dl_NotStarted,
|
||||
Dl_InProgress,
|
||||
Dl_Finished,
|
||||
Dl_Failed
|
||||
Job_NotStarted,
|
||||
Job_InProgress,
|
||||
Job_Finished,
|
||||
Job_Failed
|
||||
};
|
||||
|
||||
class JobList;
|
||||
|
||||
class Job : public QObject
|
||||
{
|
||||
Q_OBJECT
|
||||
protected:
|
||||
explicit Job(): QObject(0){};
|
||||
public:
|
||||
virtual ~Job() {};
|
||||
signals:
|
||||
void finish();
|
||||
void fail();
|
||||
void progress(qint64 current, qint64 total);
|
||||
public slots:
|
||||
virtual void start() = 0;
|
||||
};
|
||||
typedef QSharedPointer<Job> JobPtr;
|
||||
|
||||
/**
|
||||
* A list of jobs, to be processed one by one.
|
||||
*/
|
||||
class JobList : public QObject
|
||||
{
|
||||
friend class JobListQueue;
|
||||
Q_OBJECT
|
||||
public:
|
||||
|
||||
JobList() : QObject(0)
|
||||
{
|
||||
m_status = Job_NotStarted;
|
||||
current_job_idx = 0;
|
||||
}
|
||||
JobStatus getStatus()
|
||||
{
|
||||
return m_status;
|
||||
}
|
||||
void add(JobPtr dlable)
|
||||
{
|
||||
if(m_status == Job_NotStarted)
|
||||
m_jobs.append(dlable);
|
||||
//else there's a bug. TODO: catch the bugs
|
||||
}
|
||||
JobPtr getFirstJob()
|
||||
{
|
||||
if(m_jobs.size())
|
||||
return m_jobs[0];
|
||||
else
|
||||
return JobPtr();
|
||||
}
|
||||
void start()
|
||||
{
|
||||
current_job_idx = 0;
|
||||
auto job = m_jobs[current_job_idx];
|
||||
|
||||
connect(job.data(), SIGNAL(progress(qint64,qint64)), SLOT(currentJobProgress(qint64,qint64)));
|
||||
connect(job.data(), SIGNAL(finish()), SLOT(currentJobFinished()));
|
||||
connect(job.data(), SIGNAL(fail()), SLOT(currentJobFailed()));
|
||||
job->start();
|
||||
emit started();
|
||||
}
|
||||
private slots:
|
||||
void currentJobFinished()
|
||||
{
|
||||
if(current_job_idx == m_jobs.size() - 1)
|
||||
{
|
||||
m_status = Job_Finished;
|
||||
emit finished();
|
||||
}
|
||||
else
|
||||
{
|
||||
current_job_idx++;
|
||||
auto job = m_jobs[current_job_idx];
|
||||
connect(job.data(), SIGNAL(progress(qint64,qint64)), SLOT(currentJobProgress(qint64,qint64)));
|
||||
connect(job.data(), SIGNAL(finish()), SLOT(currentJobFinished()));
|
||||
connect(job.data(), SIGNAL(fail()), SLOT(currentJobFailed()));
|
||||
job->start();
|
||||
}
|
||||
}
|
||||
void currentJobFailed()
|
||||
{
|
||||
m_status = Job_Failed;
|
||||
emit failed();
|
||||
}
|
||||
void currentJobProgress(qint64 current, qint64 total)
|
||||
{
|
||||
if(!total)
|
||||
return;
|
||||
|
||||
int total_jobs = m_jobs.size();
|
||||
|
||||
if(!total_jobs)
|
||||
return;
|
||||
|
||||
float job_chunk = 1000.0 / float(total_jobs);
|
||||
float cur = current;
|
||||
float tot = total;
|
||||
float last_chunk = (cur / tot) * job_chunk;
|
||||
|
||||
float list_total = job_chunk * current_job_idx + last_chunk;
|
||||
emit progress(qint64(list_total), 1000LL);
|
||||
}
|
||||
private:
|
||||
QVector<JobPtr> m_jobs;
|
||||
/// The overall status of this job list
|
||||
JobStatus m_status;
|
||||
int current_job_idx;
|
||||
signals:
|
||||
void progress(qint64 current, qint64 total);
|
||||
void started();
|
||||
void finished();
|
||||
void failed();
|
||||
};
|
||||
typedef QSharedPointer<JobList> JobListPtr;
|
||||
|
||||
|
||||
/**
|
||||
* A queue of job lists! The job lists fail or finish as units.
|
||||
*/
|
||||
class JobListQueue : public QObject
|
||||
{
|
||||
Q_OBJECT
|
||||
public:
|
||||
JobListQueue(QObject *p = 0):
|
||||
QObject(p),
|
||||
nam(new QNetworkAccessManager()),
|
||||
currentIndex(0),
|
||||
is_running(false){}
|
||||
|
||||
void enqueue(JobListPtr job)
|
||||
{
|
||||
jobs.enqueue(job);
|
||||
|
||||
// finish or fail, we should catch that and start the next one
|
||||
connect(job.data(),SIGNAL(finished()), SLOT(startNextJob()));
|
||||
connect(job.data(),SIGNAL(failed()), SLOT(startNextJob()));
|
||||
|
||||
if(!is_running)
|
||||
{
|
||||
QTimer::singleShot(0, this, SLOT(startNextJob()));
|
||||
}
|
||||
}
|
||||
|
||||
private slots:
|
||||
void startNextJob()
|
||||
{
|
||||
if (jobs.isEmpty())
|
||||
{
|
||||
currentJobList.clear();
|
||||
currentIndex = 0;
|
||||
is_running = false;
|
||||
emit finishedAllJobs();
|
||||
return;
|
||||
}
|
||||
|
||||
currentJobList = jobs.dequeue();
|
||||
is_running = true;
|
||||
currentIndex = 0;
|
||||
currentJobList->start();
|
||||
}
|
||||
|
||||
signals:
|
||||
void finishedAllJobs();
|
||||
|
||||
private:
|
||||
JobListPtr currentJobList;
|
||||
QQueue<JobListPtr> jobs;
|
||||
QSharedPointer<QNetworkAccessManager> nam;
|
||||
unsigned currentIndex;
|
||||
bool is_running;
|
||||
};
|
||||
|
||||
/**
|
||||
* A single file for the downloader/cache to process.
|
||||
*/
|
||||
struct Downloadable
|
||||
class DownloadJob : public Job
|
||||
{
|
||||
Downloadable(QUrl url, QString rel_target_path = QString(), QString expected_md5 = QString())
|
||||
:m_url(url), m_rel_target_path(rel_target_path), m_expected_md5(expected_md5)
|
||||
friend class Downloader;
|
||||
Q_OBJECT
|
||||
private:
|
||||
DownloadJob(QUrl url, QString rel_target_path = QString(), QString expected_md5 = QString())
|
||||
:Job()
|
||||
{
|
||||
m_url = url;
|
||||
m_rel_target_path = rel_target_path;
|
||||
m_expected_md5 = expected_md5;
|
||||
|
||||
m_check_md5 = m_expected_md5.size();
|
||||
m_save_to_file = m_rel_target_path.size();
|
||||
status = Dl_NotStarted;
|
||||
m_status = Job_NotStarted;
|
||||
};
|
||||
public:
|
||||
static JobPtr create(QUrl url, QString rel_target_path = QString(), QString expected_md5 = QString())
|
||||
{
|
||||
return JobPtr(new DownloadJob(url, rel_target_path, expected_md5));
|
||||
}
|
||||
public slots:
|
||||
virtual void start()
|
||||
{
|
||||
m_manager.reset(new QNetworkAccessManager());
|
||||
if(m_save_to_file)
|
||||
{
|
||||
QString filename = m_rel_target_path;
|
||||
m_output_file.setFileName(filename);
|
||||
// if there already is a file and md5 checking is in effect
|
||||
if(m_output_file.exists() && m_check_md5)
|
||||
{
|
||||
// and it can be opened
|
||||
if(m_output_file.open(QIODevice::ReadOnly))
|
||||
{
|
||||
// check the md5 against the expected one
|
||||
QString hash = QCryptographicHash::hash(m_output_file.readAll(), QCryptographicHash::Md5).toHex().constData();
|
||||
m_output_file.close();
|
||||
// skip this file if they match
|
||||
if(hash == m_expected_md5)
|
||||
{
|
||||
qDebug() << "Skipping " << m_url.toString() << ": md5 match.";
|
||||
emit finish();
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
QFileInfo a(filename);
|
||||
QDir dir;
|
||||
if(!dir.mkpath(a.path()))
|
||||
{
|
||||
/*
|
||||
* error when making the folder structure
|
||||
*/
|
||||
emit fail();
|
||||
return;
|
||||
}
|
||||
if (!m_output_file.open(QIODevice::WriteOnly))
|
||||
{
|
||||
/*
|
||||
* Can't open the file... the job failed
|
||||
*/
|
||||
emit fail();
|
||||
return;
|
||||
}
|
||||
}
|
||||
qDebug() << "Downloading " << m_url.toString();
|
||||
QNetworkRequest request(m_url);
|
||||
QNetworkReply * rep = m_manager->get(request);
|
||||
m_reply = QSharedPointer<QNetworkReply>(rep, &QObject::deleteLater);
|
||||
connect(rep, SIGNAL(downloadProgress(qint64,qint64)), SLOT(downloadProgress(qint64,qint64)));
|
||||
connect(rep, SIGNAL(finished()), SLOT(downloadFinished()));
|
||||
connect(rep, SIGNAL(error(QNetworkReply::NetworkError)), SLOT(downloadError(QNetworkReply::NetworkError)));
|
||||
connect(rep, SIGNAL(readyRead()), SLOT(downloadReadyRead()));
|
||||
};
|
||||
private slots:
|
||||
void downloadProgress(qint64 bytesReceived, qint64 bytesTotal)
|
||||
{
|
||||
emit progress(bytesReceived, bytesTotal);
|
||||
};
|
||||
|
||||
void downloadError(QNetworkReply::NetworkError error)
|
||||
{
|
||||
// error happened during download.
|
||||
// TODO: log the reason why
|
||||
m_status = Job_Failed;
|
||||
}
|
||||
|
||||
void downloadFinished()
|
||||
{
|
||||
// if the download succeeded
|
||||
if(m_status != Job_Failed)
|
||||
{
|
||||
// nothing went wrong...
|
||||
m_status = Job_Finished;
|
||||
// save the data to the downloadable if we aren't saving to file
|
||||
if(!m_save_to_file)
|
||||
{
|
||||
m_data = m_reply->readAll();
|
||||
}
|
||||
else
|
||||
{
|
||||
m_output_file.close();
|
||||
}
|
||||
|
||||
//TODO: check md5 here!
|
||||
m_reply.clear();
|
||||
emit finish();
|
||||
return;
|
||||
}
|
||||
// else the download failed
|
||||
else
|
||||
{
|
||||
if(m_save_to_file)
|
||||
{
|
||||
m_output_file.close();
|
||||
m_output_file.remove();
|
||||
}
|
||||
m_reply.clear();
|
||||
emit fail();
|
||||
return;
|
||||
}
|
||||
}
|
||||
void downloadReadyRead()
|
||||
{
|
||||
if(m_save_to_file)
|
||||
{
|
||||
m_output_file.write( m_reply->readAll());
|
||||
}
|
||||
};
|
||||
|
||||
public:
|
||||
/// the associated network manager
|
||||
QSharedPointer<QNetworkAccessManager> m_manager;
|
||||
/// the network reply
|
||||
QSharedPointer<QNetworkReply> m_reply;
|
||||
/// source URL
|
||||
QUrl m_url;
|
||||
|
||||
@ -36,259 +334,14 @@ struct Downloadable
|
||||
bool m_save_to_file;
|
||||
/// if saving to file, use the one specified in this string
|
||||
QString m_rel_target_path;
|
||||
/// this is the output file, if any
|
||||
QFile m_output_file;
|
||||
/// if not saving to file, downloaded data is placed here
|
||||
QByteArray data;
|
||||
QByteArray m_data;
|
||||
|
||||
|
||||
/// The file's status
|
||||
DlStatus status;
|
||||
};
|
||||
typedef QSharedPointer<Downloadable> DownloadablePtr;
|
||||
|
||||
class Downloader;
|
||||
|
||||
/**
|
||||
* A downloader job, which can have multiple files
|
||||
* the user of the downloader is responsible for creating it
|
||||
* and connecting its signals to his own slots.
|
||||
*/
|
||||
class DLJob : public QObject
|
||||
{
|
||||
friend class Downloader;
|
||||
Q_OBJECT
|
||||
public:
|
||||
|
||||
DLJob() : QObject(0) {}
|
||||
DLJob (QUrl url, QString rel_target_path = QString(), QString expected_md5 = QString() ) : QObject(0)
|
||||
{
|
||||
m_status = Dl_NotStarted;
|
||||
append(url, rel_target_path, expected_md5);
|
||||
}
|
||||
DlStatus getStatus()
|
||||
{
|
||||
return m_status;
|
||||
}
|
||||
void append (QUrl url, QString target = QString(), QString md5 = QString())
|
||||
{
|
||||
Downloadable * dlable = new Downloadable(url, target, md5);
|
||||
m_downloads.append(DownloadablePtr(dlable));
|
||||
}
|
||||
QByteArray getFirstFileData()
|
||||
{
|
||||
if(!m_downloads.size())
|
||||
return QByteArray();
|
||||
else return m_downloads[0]->data;
|
||||
}
|
||||
void add(DownloadablePtr dlable)
|
||||
{
|
||||
m_downloads.append(dlable);
|
||||
}
|
||||
private:
|
||||
void emitStart()
|
||||
{
|
||||
m_status = Dl_InProgress;
|
||||
emit started();
|
||||
}
|
||||
void emitFail()
|
||||
{
|
||||
m_status = Dl_Failed;
|
||||
emit failed();
|
||||
}
|
||||
void emitFinish()
|
||||
{
|
||||
m_status = Dl_Finished;
|
||||
emit finished();
|
||||
}
|
||||
private:
|
||||
QVector<DownloadablePtr> m_downloads;
|
||||
/// The job's status
|
||||
DlStatus m_status;
|
||||
signals:
|
||||
void started();
|
||||
void finished();
|
||||
void failed();
|
||||
};
|
||||
typedef QSharedPointer<DLJob> DLJobPtr;
|
||||
|
||||
/**
|
||||
* The downloader itself. Make one, use it.
|
||||
* User is responsible for keeping it around until all the jobs either finish or fail.
|
||||
*/
|
||||
class Downloader : public QObject
|
||||
{
|
||||
Q_OBJECT
|
||||
public:
|
||||
Downloader(QObject *p = 0):
|
||||
QObject(p),
|
||||
nam(new QNetworkAccessManager(this)),
|
||||
currentReply(0),
|
||||
currentIndex(0){}
|
||||
|
||||
void enqueue(DLJobPtr job)
|
||||
{
|
||||
if(jobs.empty())
|
||||
{
|
||||
qDebug() << "NEXT JOB!";
|
||||
QTimer::singleShot(0, this, SLOT(startNextJob()));
|
||||
}
|
||||
jobs.enqueue(job);
|
||||
}
|
||||
|
||||
private slots:
|
||||
void startNextJob()
|
||||
{
|
||||
if (jobs.isEmpty())
|
||||
{
|
||||
currentJob.clear();
|
||||
currentIndex = 0;
|
||||
emit finishedAllJobs();
|
||||
return;
|
||||
}
|
||||
|
||||
currentJob = jobs.dequeue();
|
||||
currentIndex = 0;
|
||||
currentJob->emitStart();
|
||||
|
||||
startNextDownload();
|
||||
}
|
||||
|
||||
void startNextDownload()
|
||||
{
|
||||
// we finished the current job. Good job.
|
||||
if(currentIndex >= currentJob->m_downloads.size())
|
||||
{
|
||||
currentJob->emitFinish();
|
||||
qDebug() << "NEXT JOB!";
|
||||
QTimer::singleShot(0, this, SLOT(startNextJob()));
|
||||
return;
|
||||
}
|
||||
|
||||
DownloadablePtr dlable = currentJob->m_downloads[currentIndex];
|
||||
if(dlable->m_save_to_file)
|
||||
{
|
||||
QString filename = dlable->m_rel_target_path;
|
||||
currentOutput.setFileName(filename);
|
||||
// if there already is a file and md5 checking is in effect
|
||||
if(currentOutput.exists() && dlable->m_check_md5)
|
||||
{
|
||||
// and it can be opened
|
||||
if(currentOutput.open(QIODevice::ReadOnly))
|
||||
{
|
||||
// check the md5 against the expected one
|
||||
QString hash = QCryptographicHash::hash(currentOutput.readAll(), QCryptographicHash::Md5).toHex().constData();
|
||||
currentOutput.close();
|
||||
// skip this file if they match
|
||||
if(hash == dlable->m_expected_md5)
|
||||
{
|
||||
currentIndex++;
|
||||
QTimer::singleShot(0, this, SLOT(startNextDownload()));
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
QFileInfo a(filename);
|
||||
QDir dir;
|
||||
if(!dir.mkpath(a.path()))
|
||||
{
|
||||
/*
|
||||
* TODO: error when making the folder structure
|
||||
*/
|
||||
currentJob->emitFail();
|
||||
currentJob.clear();
|
||||
currentIndex = 0;
|
||||
QTimer::singleShot(0, this, SLOT(startNextJob()));
|
||||
qDebug() << "NEXT JOB!";
|
||||
return;
|
||||
}
|
||||
if (!currentOutput.open(QIODevice::WriteOnly))
|
||||
{
|
||||
/*
|
||||
* TODO: Can't open the file... the job failed
|
||||
*/
|
||||
currentJob->emitFail();
|
||||
currentJob.clear();
|
||||
currentIndex = 0;
|
||||
QTimer::singleShot(0, this, SLOT(startNextJob()));
|
||||
qDebug() << "NEXT JOB!";
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
QNetworkRequest request(dlable->m_url);
|
||||
QNetworkReply * rep = nam->get(request);
|
||||
currentReply = QSharedPointer<QNetworkReply>(rep, &QObject::deleteLater);
|
||||
connect(rep, SIGNAL(downloadProgress(qint64,qint64)), SLOT(downloadProgress(qint64,qint64)));
|
||||
connect(rep, SIGNAL(finished()), SLOT(downloadFinished()));
|
||||
connect(rep, SIGNAL(error(QNetworkReply::NetworkError)), SLOT(downloadError(QNetworkReply::NetworkError)));
|
||||
connect(rep, SIGNAL(readyRead()), SLOT(downloadReadyRead()));
|
||||
}
|
||||
void downloadProgress(qint64 bytesReceived, qint64 bytesTotal)
|
||||
{
|
||||
// eventually, progress bars. yeah.
|
||||
};
|
||||
|
||||
void downloadError(QNetworkReply::NetworkError error)
|
||||
{
|
||||
// error happened during download. :<
|
||||
DownloadablePtr dlable = currentJob->m_downloads[currentIndex];
|
||||
//TODO: log the reason why
|
||||
dlable->status = Dl_Failed;
|
||||
}
|
||||
|
||||
void downloadFinished()
|
||||
{
|
||||
DownloadablePtr dlable = currentJob->m_downloads[currentIndex];
|
||||
// if the download succeeded
|
||||
if(dlable->status != Dl_Failed)
|
||||
{
|
||||
// nothing went wrong...
|
||||
dlable->status = Dl_Finished;
|
||||
// save the data to the downloadable if we aren't saving to file
|
||||
if(!dlable->m_save_to_file)
|
||||
{
|
||||
dlable->data = currentReply->readAll();
|
||||
}
|
||||
else
|
||||
{
|
||||
currentOutput.close();
|
||||
}
|
||||
|
||||
//TODO: check md5 here!
|
||||
|
||||
// continue with the next download, if any
|
||||
currentIndex ++;
|
||||
QTimer::singleShot(0, this, SLOT(startNextDownload()));
|
||||
return;
|
||||
}
|
||||
// else the download failed
|
||||
else
|
||||
{
|
||||
if(dlable->m_save_to_file)
|
||||
{
|
||||
currentOutput.close();
|
||||
currentOutput.remove();
|
||||
}
|
||||
}
|
||||
}
|
||||
void downloadReadyRead()
|
||||
{
|
||||
DownloadablePtr dlable = currentJob->m_downloads[currentIndex];
|
||||
if(dlable->m_save_to_file)
|
||||
{
|
||||
currentOutput.write(currentReply->readAll());
|
||||
}
|
||||
};
|
||||
|
||||
signals:
|
||||
void finishedAllJobs();
|
||||
|
||||
public slots:
|
||||
|
||||
private:
|
||||
QSharedPointer<QNetworkAccessManager> nam;
|
||||
DLJobPtr currentJob;
|
||||
QSharedPointer<QNetworkReply> currentReply;
|
||||
QQueue<DLJobPtr> jobs;
|
||||
QFile currentOutput;
|
||||
unsigned currentIndex;
|
||||
JobStatus m_status;
|
||||
};
|
||||
|
||||
inline QDomElement getDomElementByTagName(QDomElement parent, QString tagname)
|
||||
@ -311,7 +364,9 @@ public slots:
|
||||
|
||||
void fetchFinished()
|
||||
{
|
||||
QByteArray ba = index_job->getFirstFileData();
|
||||
JobPtr firstJob = index_job->getFirstJob();
|
||||
auto DlJob = firstJob.dynamicCast<DownloadJob>();
|
||||
QByteArray ba = DlJob->m_data;
|
||||
|
||||
QString xmlErrorMsg;
|
||||
QDomDocument doc;
|
||||
@ -323,7 +378,7 @@ public slots:
|
||||
QRegExp etag_match(".*([a-f0-9]{32}).*");
|
||||
QDomNodeList contents = doc.elementsByTagName("Contents");
|
||||
|
||||
DLJob *job = new DLJob();
|
||||
JobList *job = new JobList();
|
||||
connect(job, SIGNAL(finished()), SLOT(filesFinished()));
|
||||
|
||||
for (int i = 0; i < contents.length(); i++)
|
||||
@ -353,11 +408,7 @@ public slots:
|
||||
QString trimmedEtag = etagStr.remove('"');
|
||||
QString prefix("http://s3.amazonaws.com/Minecraft.Resources/");
|
||||
QString fprefix("assets/");
|
||||
Downloadable * d = new Downloadable(QUrl(prefix + keyStr),fprefix + keyStr, trimmedEtag);
|
||||
|
||||
job->add(DownloadablePtr(d));
|
||||
|
||||
//qDebug() << keyStr << " " << lastModStr << " " << etagStr << sizeStr;
|
||||
job->add(DownloadJob::create(QUrl(prefix + keyStr),fprefix + keyStr, trimmedEtag));
|
||||
}
|
||||
files_job.reset(job);
|
||||
dl.enqueue(files_job);
|
||||
@ -369,15 +420,16 @@ public slots:
|
||||
public:
|
||||
void start()
|
||||
{
|
||||
DLJob *job = new DLJob(QUrl("http://s3.amazonaws.com/Minecraft.Resources/"));
|
||||
JobList *job = new JobList();
|
||||
job->add(DownloadJob::create(QUrl("http://s3.amazonaws.com/Minecraft.Resources/")));
|
||||
connect(job, SIGNAL(finished()), SLOT(fetchFinished()));
|
||||
connect(job, SIGNAL(started()), SLOT(fetchStarted()));
|
||||
index_job.reset(job);
|
||||
dl.enqueue(index_job);
|
||||
}
|
||||
Downloader dl;
|
||||
DLJobPtr index_job;
|
||||
DLJobPtr files_job;
|
||||
JobListQueue dl;
|
||||
JobListPtr index_job;
|
||||
JobListPtr files_job;
|
||||
};
|
||||
|
||||
int main(int argc, char *argv[])
|
||||
|
||||
Loading…
Reference in New Issue
Block a user