Merge pull request #3010 from nextcloud/feature/cfapi-handle-cancelation

CFAPI: Handle cancelation of hydration requests

src/libsync/propagatedownload.cpp

@@ -342,6 +342,15 @@ void GETFileJob::slotReadyRead()
     }
 }
 
+void GETFileJob::cancel()
+{
+    if (reply()->isRunning()) {
+        reply()->abort();
+    }
+
+    emit canceled();
+}
+
 void GETFileJob::onTimedOut()
 {
     qCWarning(lcGetJob) << "Timeout" << (reply() ? reply()->request().url() : path());

src/libsync/propagatedownload.h

@@ -83,6 +83,8 @@ public:
         }
     }
 
+    void cancel();
+
     void newReplyHook(QNetworkReply *reply) override;
 
     void setBandwidthManager(BandwidthManager *bwm);
@@ -108,6 +110,7 @@ public:
     void setExpectedContentLength(qint64 size) { _expectedContentLength = size; }
 
 signals:
+    void canceled();
     void finishedSignal();
     void downloadProgress(qint64, qint64);
 private slots:

src/libsync/vfs/cfapi/cfapiwrapper.cpp

@@ -58,7 +58,6 @@ void cfApiSendTransferInfo(const CF_CONNECTION_KEY &connectionKey, const CF_TRAN
     }
 }
 
-
 void CALLBACK cfApiFetchDataCallback(const CF_CALLBACK_INFO *callbackInfo, const CF_CALLBACK_PARAMETERS *callbackParameters)
 {
     qDebug(lcCfApiWrapper) << "Fetch data callback called. File size:" << callbackInfo->FileSize.QuadPart;
@@ -181,8 +180,27 @@ void CALLBACK cfApiFetchDataCallback(const CF_CALLBACK_INFO *callbackInfo, const
     }
 }
 
+void CALLBACK cfApiCancelFetchData(const CF_CALLBACK_INFO *callbackInfo, const CF_CALLBACK_PARAMETERS * /*callbackParameters*/)
+{
+    const auto path = QString(QString::fromWCharArray(callbackInfo->VolumeDosName) + QString::fromWCharArray(callbackInfo->NormalizedPath));
+
+    qInfo(lcCfApiWrapper) << "Cancel fetch data of" << path;
+
+    auto vfs = reinterpret_cast<OCC::VfsCfApi *>(callbackInfo->CallbackContext);
+    Q_ASSERT(vfs->metaObject()->className() == QByteArrayLiteral("OCC::VfsCfApi"));
+    const auto requestId = QString::number(callbackInfo->TransferKey.QuadPart, 16);
+
+    const auto invokeResult = QMetaObject::invokeMethod(
+        vfs, [=] { vfs->cancelHydration(requestId, path); }, Qt::QueuedConnection);
+    if (!invokeResult) {
+        qCritical(lcCfApiWrapper) << "Failed to cancel hydration for" << path << requestId;
+    }
+}
+
+
 CF_CALLBACK_REGISTRATION cfApiCallbacks[] = {
     { CF_CALLBACK_TYPE_FETCH_DATA, cfApiFetchDataCallback },
+    { CF_CALLBACK_TYPE_CANCEL_FETCH_DATA, cfApiCancelFetchData },
     CF_CALLBACK_REGISTRATION_END
 };
 

src/libsync/vfs/cfapi/hydrationjob.cpp

@@ -116,6 +116,15 @@ void OCC::HydrationJob::start()
     connect(_server, &QLocalServer::newConnection, this, &HydrationJob::onNewConnection);
 }
 
+void OCC::HydrationJob::cancel()
+{
+    if (!_job) {
+        return;
+    }
+
+    _job->cancel();
+}
+
 void OCC::HydrationJob::emitFinished(Status status)
 {
     _status = status;
@@ -131,6 +140,16 @@ void OCC::HydrationJob::emitFinished(Status status)
     }
 }
 
+void OCC::HydrationJob::emitCanceled()
+{
+    connect(_socket, &QLocalSocket::disconnected, this, [=] {
+        _socket->close();
+    });
+    _socket->disconnectFromServer();
+
+    emit canceled(this);
+}
+
 void OCC::HydrationJob::onNewConnection()
 {
     Q_ASSERT(!_socket);
@@ -140,9 +159,16 @@ void OCC::HydrationJob::onNewConnection()
     _socket = _server->nextPendingConnection();
     _job = new GETFileJob(_account, _remotePath + _folderPath, _socket, {}, {}, 0, this);
     connect(_job, &GETFileJob::finishedSignal, this, &HydrationJob::onGetFinished);
+    connect(_job, &GETFileJob::canceled, this, &HydrationJob::onGetCanceled);
     _job->start();
 }
 
+void OCC::HydrationJob::onGetCanceled()
+{
+    qCInfo(lcHydration) << "GETFileJob canceled" << _requestId << _folderPath << _job->reply()->error();
+    emitCanceled();
+}
+
 void OCC::HydrationJob::onGetFinished()
 {
     qCInfo(lcHydration) << "GETFileJob finished" << _requestId << _folderPath << _job->reply()->error();

src/libsync/vfs/cfapi/hydrationjob.h

@@ -57,15 +57,19 @@ public:
     Status status() const;
 
     void start();
+    void cancel();
 
 signals:
     void finished(HydrationJob *job);
+    void canceled(HydrationJob *job);
 
 private:
     void emitFinished(Status status);
+    void emitCanceled();
 
     void onNewConnection();
     void onGetFinished();
+    void onGetCanceled();
 
     AccountPtr _account;
     QString _remotePath;

src/libsync/vfs/cfapi/vfs_cfapi.cpp

@@ -264,6 +264,19 @@ Vfs::AvailabilityResult VfsCfApi::availability(const QString &folderPath)
     return AvailabilityError::NoSuchItem;
 }
 
+void VfsCfApi::cancelHydration(const QString &requestId, const QString & /*path*/)
+{
+    // Find matching hydration job for request id
+    const auto hydrationJobsIter = std::find_if(d->hydrationJobs.cbegin(), d->hydrationJobs.cend(), [&](const HydrationJob *job) {
+        return job->requestId() == requestId;
+    });
+
+    // If found, cancel it
+    if (hydrationJobsIter != d->hydrationJobs.cend()) {
+        (*hydrationJobsIter)->cancel();
+    }
+}
+
 void VfsCfApi::requestHydration(const QString &requestId, const QString &path)
 {
     qCInfo(lcCfApi) << "Received request to hydrate" << path << requestId;
@@ -331,6 +344,7 @@ void VfsCfApi::scheduleHydrationJob(const QString &requestId, const QString &fol
     job->setRequestId(requestId);
     job->setFolderPath(folderPath);
     connect(job, &HydrationJob::finished, this, &VfsCfApi::onHydrationJobFinished);
+    connect(job, &HydrationJob::canceled, this, &VfsCfApi::onHydrationJobCanceled);
     d->hydrationJobs << job;
     job->start();
     emit hydrationRequestReady(requestId);
@@ -347,6 +361,22 @@ void VfsCfApi::onHydrationJobFinished(HydrationJob *job)
     }
 }
 
+void VfsCfApi::onHydrationJobCanceled(HydrationJob *job)
+{
+    const auto folderPath = job->localPath();
+    const auto folderRelativePath = job->folderPath();
+
+    // Remove placeholder file because there might be already pumped
+    // some data into it
+    QFile::remove(folderPath + folderRelativePath);
+
+    // Create a new placeholder file
+    SyncJournalFileRecord record;
+    params().journal->getFileRecord(folderRelativePath, &record);
+    const auto item = SyncFileItem::fromSyncJournalFileRecord(record);
+    createPlaceholder(*item);
+}
+
 VfsCfApi::HydratationAndPinStates VfsCfApi::computeRecursiveHydrationAndPinStates(const QString &folderPath, const Optional<PinState> &basePinState)
 {
     Q_ASSERT(!folderPath.endsWith('/'));

src/libsync/vfs/cfapi/vfs_cfapi.h

@@ -53,6 +53,8 @@ public:
     Optional<PinState> pinState(const QString &folderPath) override;
     AvailabilityResult availability(const QString &folderPath) override;
 
+    void cancelHydration(const QString &requestId, const QString &path);
+
 public slots:
     void requestHydration(const QString &requestId, const QString &path);
     void fileStatusChanged(const QString &systemFileName, SyncFileStatus fileStatus) override;
@@ -68,6 +70,7 @@ protected:
 private:
     void scheduleHydrationJob(const QString &requestId, const QString &folderPath);
     void onHydrationJobFinished(HydrationJob *job);
+    void onHydrationJobCanceled(HydrationJob *job);
 
     struct HasHydratedDehydrated {
         bool hasHydrated = false;