/* * Copyright (C) by Olivier Goffart * * This program is free software; you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation; either version 2 of the License, or * (at your option) any later version. * * This program is distributed in the hope that it will be useful, but * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License * for more details. */ #ifndef OWNCLOUDPROPAGATOR_H #define OWNCLOUDPROPAGATOR_H #include #include #include #include #include #include #include #include #include #include "syncfileitem.h" #include "syncjournaldb.h" #include "bandwidthmanager.h" struct hbf_transfer_s; struct ne_session_s; struct ne_decompress_s; typedef struct ne_prop_result_set_s ne_prop_result_set; namespace Mirall { class Account; class SyncJournalDb; class OwncloudPropagator; class PropagatorJob : public QObject { Q_OBJECT protected: OwncloudPropagator *_propagator; public: enum JobState { NotYetStarted, Running, Finished }; enum JobParallelism { FullParallelism, WaitForFinished, WaitForFinishedInParentDirectory }; JobState _state; explicit PropagatorJob(OwncloudPropagator* propagator) : _propagator(propagator), _state(NotYetStarted) {} public slots: virtual JobParallelism parallelism() { return FullParallelism; } virtual void abort() {} virtual bool scheduleNextJob() = 0; signals: /** * Emitted when the job is fully finished */ void finished(SyncFileItem::Status); /** * Emitted when one item has been completed within a job. */ void completed(const SyncFileItem &); /** * Emitted when all the sub-jobs have been scheduled and * we are ready and more jobs might be started * This signal is not always emitted. */ void ready(); void progress(const SyncFileItem& item, quint64 bytes); }; /* * Propagate a directory, and all its sub entries. */ class PropagateDirectory : public PropagatorJob { Q_OBJECT public: // e.g: create the directory QScopedPointer_firstJob; // all the sub files or sub directories. QVector _subJobs; SyncFileItem _item; int _current; // index of the current running job int _runningNow; // number of subJob running now SyncFileItem::Status _hasError; // NoStatus, or NormalError / SoftError if there was an error explicit PropagateDirectory(OwncloudPropagator *propagator, const SyncFileItem &item = SyncFileItem()) : PropagatorJob(propagator) , _firstJob(0), _item(item), _current(-1), _runningNow(0), _hasError(SyncFileItem::NoStatus) { } virtual ~PropagateDirectory() { qDeleteAll(_subJobs); } void append(PropagatorJob *subJob) { _subJobs.append(subJob); } virtual bool scheduleNextJob() Q_DECL_OVERRIDE; virtual JobParallelism parallelism() Q_DECL_OVERRIDE; virtual void abort() Q_DECL_OVERRIDE { if (_firstJob) _firstJob->abort(); foreach (PropagatorJob *j, _subJobs) j->abort(); } private slots: bool possiblyRunNextJob(PropagatorJob *next) { if (next->_state == NotYetStarted) { connect(next, SIGNAL(finished(SyncFileItem::Status)), this, SLOT(slotSubJobFinished(SyncFileItem::Status)), Qt::QueuedConnection); connect(next, SIGNAL(completed(SyncFileItem)), this, SIGNAL(completed(SyncFileItem))); connect(next, SIGNAL(progress(SyncFileItem,quint64)), this, SIGNAL(progress(SyncFileItem,quint64))); connect(next, SIGNAL(ready()), this, SIGNAL(ready())); _runningNow++; } return next->scheduleNextJob(); } void slotSubJobFinished(SyncFileItem::Status status); }; /* * Abstract class to propagate a single item */ class PropagateItemJob : public PropagatorJob { Q_OBJECT protected: void done(SyncFileItem::Status status, const QString &errorString = QString()); bool checkForProblemsWithShared(int httpStatusCode, const QString& msg); /* * set a custom restore job message that is used if the restore job succeeded. * It is displayed in the activity view. */ QString restoreJobMsg() const { return _item._isRestoration ? _item._errorString : QString(); } void setRestoreJobMsg( const QString& msg = QString() ) { _item._isRestoration = true; _item._errorString = msg; } SyncFileItem _item; protected slots: void slotRestoreJobCompleted(const SyncFileItem& ); private: QScopedPointer _restoreJob; public: PropagateItemJob(OwncloudPropagator* propagator, const SyncFileItem &item) : PropagatorJob(propagator), _item(item) {} bool scheduleNextJob() Q_DECL_OVERRIDE { if (_state != NotYetStarted) { return false; } _state = Running; start(); return true; } virtual void start() = 0; }; // Dummy job that just mark it as completed and ignored. class PropagateIgnoreJob : public PropagateItemJob { Q_OBJECT public: PropagateIgnoreJob(OwncloudPropagator* propagator,const SyncFileItem& item) : PropagateItemJob(propagator, item) {} void start() Q_DECL_OVERRIDE { SyncFileItem::Status status = _item._status; done(status == SyncFileItem::NoStatus ? SyncFileItem::FileIgnored : status, _item._errorString); } }; class OwncloudPropagator : public QObject { Q_OBJECT PropagateItemJob *createJob(const SyncFileItem& item); QScopedPointer _rootJob; bool useLegacyJobs(); public: /* 'const' because they are accessed by the thread */ QThread* _neonThread; ne_session_s * const _session; const QString _localDir; // absolute path to the local directory. ends with '/' const QString _remoteDir; // path to the root of the remote. ends with '/' (include remote.php/webdav) const QString _remoteFolder; // folder. (same as remoteDir but without remote.php/webdav) SyncJournalDb * const _journal; bool _finishedEmited; // used to ensure that finished is only emit once BandwidthManager _bandwidthManager; public: OwncloudPropagator(ne_session_s *session, const QString &localDir, const QString &remoteDir, const QString &remoteFolder, SyncJournalDb *progressDb, QThread *neonThread) : _neonThread(neonThread) , _session(session) , _localDir((localDir.endsWith(QChar('/'))) ? localDir : localDir+'/' ) , _remoteDir((remoteDir.endsWith(QChar('/'))) ? remoteDir : remoteDir+'/' ) , _remoteFolder((remoteFolder.endsWith(QChar('/'))) ? remoteFolder : remoteFolder+'/' ) , _journal(progressDb) , _finishedEmited(false) , _bandwidthManager(this) , _activeJobs(0) , _anotherSyncNeeded(false) { } void start(const SyncFileItemVector &_syncedItems); QAtomicInt _downloadLimit; QAtomicInt _uploadLimit; QAtomicInt _abortRequested; // boolean set by the main thread to abort. /* The number of currently active jobs */ int _activeJobs; /** We detected that another sync is required after this one */ bool _anotherSyncNeeded; /* The maximum number of active job in parallel */ int maximumActiveJob(); bool isInSharedDirectory(const QString& file); bool localFileNameClash(const QString& relfile); QString getFilePath(const QString& tmp_file_name) const; void abort() { _abortRequested.fetchAndStoreOrdered(true); if (_rootJob) { _rootJob->abort(); } emitFinished(); } // timeout in seconds static int httpTimeout(); private slots: /** Emit the finished signal and make sure it is only emit once */ void emitFinished() { if (!_finishedEmited) emit finished(); _finishedEmited = true; } void scheduleNextJob(); signals: void completed(const SyncFileItem &); void progress(const SyncFileItem&, quint64 bytes); void finished(); /** * Called when we detect that the total number of bytes changes (because a download or upload * turns out to be bigger or smaller than what was initially computed in the update phase */ void adjustTotalTransmissionSize( qint64 adjust ); }; // Job that wait for all the poll jobs to be completed class CleanupPollsJob : public QObject { Q_OBJECT QVector< SyncJournalDb::PollInfo > _pollInfos; Account *_account; SyncJournalDb *_journal; QString _localPath; public: explicit CleanupPollsJob(const QVector< SyncJournalDb::PollInfo > &pollInfos, Account *account, SyncJournalDb *journal, const QString &localPath, QObject* parent = 0) : QObject(parent), _pollInfos(pollInfos), _account(account), _journal(journal), _localPath(localPath) {} void start(); signals: void finished(); void aborted(const QString &error); private slots: void slotPollFinished(); }; } #endif