Use a compositejob for receiving files

core/CMakeLists.txt

@@ -33,6 +33,7 @@ set(kdeconnectcore_SRCS
     dbushelper.cpp
     networkpacket.cpp
     filetransferjob.cpp
+    compositefiletransferjob.cpp
     daemon.cpp
     device.cpp
     core_debug.cpp

core/compositefiletransferjob.cpp

@@ -0,0 +1,159 @@
+/**
+ * Copyright 2019 Nicolas Fella <nicolas.fella@gmx.de>
+ *
+ * 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) version 3 or any later version
+ * accepted by the membership of KDE e.V. (or its successor approved
+ * by the membership of KDE e.V.), which shall act as a proxy
+ * defined in Section 14 of version 3 of the license.
+ *
+ * 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.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "compositefiletransferjob.h"
+#include <core_debug.h>
+#include <KLocalizedString>
+#include <kio/global.h>
+#include <KJobTrackerInterface>
+#include <daemon.h>
+#include "filetransferjob.h"
+
+CompositeFileTransferJob::CompositeFileTransferJob(const QString& deviceId)
+    : KCompositeJob()
+    , m_deviceId(deviceId)
+    , m_running(false)
+    , m_currentJobNum(1)
+    , m_totalJobs(0)
+    , m_currentJobSendPayloadSize(0)
+    , m_totalSendPayloadSize(0)
+    , m_totalPayloadSize(0)
+    , m_currentJob(nullptr)
+    , m_prevElapsedTime(0)
+{
+    setCapabilities(Killable);
+}
+
+bool CompositeFileTransferJob::isRunning() const
+{
+    return m_running;
+}
+
+void CompositeFileTransferJob::start()
+{
+    QMetaObject::invokeMethod(this, "startNextSubJob", Qt::QueuedConnection);
+    m_running = true;
+}
+
+void CompositeFileTransferJob::startNextSubJob()
+{
+    m_currentJob = qobject_cast<FileTransferJob*>(subjobs().at(0));
+    m_currentJobSendPayloadSize = 0;
+    emitDescription(m_currentJob->destination().toString());
+    m_currentJob->start();
+    connect(m_currentJob, QOverload<KJob*,KJob::Unit,qulonglong>::of(&FileTransferJob::processedAmount), this, &CompositeFileTransferJob::slotProcessedAmount);
+}
+
+bool CompositeFileTransferJob::addSubjob(KJob* job)
+{
+    if (FileTransferJob *uploadJob = qobject_cast<FileTransferJob*>(job)) {
+
+        const NetworkPacket* np = uploadJob->networkPacket();
+
+        if (np->has(QStringLiteral("totalPayloadSize"))) {
+            m_totalPayloadSize = np->get<quint64>(QStringLiteral("totalPayloadSize"));
+            setTotalAmount(Bytes, m_totalPayloadSize);
+        }
+
+        if (np->has(QStringLiteral("numberOfFiles"))) {
+            m_totalJobs = np->get<int>(QStringLiteral("numberOfFiles"));
+        }
+
+        QString filename = np->get<QString>(QStringLiteral("filename"));
+        emitDescription(filename);
+
+        if (!hasSubjobs()) {
+            QMetaObject::invokeMethod(this, "startNextSubJob", Qt::QueuedConnection);
+        }
+
+        return KCompositeJob::addSubjob(job);
+    } else {
+        qCDebug(KDECONNECT_CORE) << "CompositeFileTransferJob::addSubjob() - you can only add FileTransferJob's, ignoring";
+        return false;
+    }
+    return true;
+}
+
+bool CompositeFileTransferJob::doKill()
+{
+    m_running = false;
+    if (m_currentJob) {
+        return m_currentJob->kill();
+    }
+    return true;
+}
+
+void CompositeFileTransferJob::slotProcessedAmount(KJob *job, KJob::Unit unit, qulonglong amount)
+{
+    Q_UNUSED(job);
+    m_currentJobSendPayloadSize = amount;
+    quint64 uploaded = m_totalSendPayloadSize + m_currentJobSendPayloadSize;
+
+    if (uploaded == m_totalPayloadSize || m_prevElapsedTime == 0 || m_timer.elapsed() - m_prevElapsedTime >= 100) {
+        m_prevElapsedTime = m_timer.elapsed();
+        setProcessedAmount(unit, uploaded);
+
+        const auto elapsed = m_timer.elapsed();
+        if (elapsed > 0) {
+            emitSpeed((1000 * uploaded) / elapsed);
+        }
+    }
+}
+
+void CompositeFileTransferJob::slotResult(KJob *job)
+{
+    //Copies job error and errorText and emits result if job is in error otherwise removes job from subjob list
+    KCompositeJob::slotResult(job);
+
+    if (error() || !m_running) {
+        return;
+    }
+
+    m_totalSendPayloadSize += m_currentJobSendPayloadSize;
+
+    setProcessedAmount(Files, m_currentJobNum);
+
+    if (m_currentJobNum < m_totalJobs) {
+        m_currentJobNum++;
+        if (!subjobs().empty()) {
+            startNextSubJob();
+        }
+    } else {
+        emitResult();
+    }
+}
+
+void CompositeFileTransferJob::emitDescription(const QString& currentFileName)
+{
+    QPair<QString, QString> field2;
+
+    const QUrl fileUrl(currentFileName);
+    const QString fileName = fileUrl.toDisplayString(QUrl::PreferLocalFile);
+
+    if (m_totalJobs > 1) {
+        field2.first = i18n("Progress");
+        field2.second = i18n("Receiving file %1 of %2", m_currentJobNum, m_totalJobs);
+    }
+
+    Q_EMIT description(this, i18np("Receiving file from %2", "Receiving %1 files from %2", m_totalJobs, Daemon::instance()->getDevice(this->m_deviceId)->name()),
+                       { i18n("File"), fileName }, field2
+    );
+}
+

core/compositefiletransferjob.h

@@ -0,0 +1,67 @@
+/**
+ * Copyright 2019 Nicolas Fella <nicolas.fella@gmx.de>
+ *
+ * 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) version 3 or any later version
+ * accepted by the membership of KDE e.V. (or its successor approved
+ * by the membership of KDE e.V.), which shall act as a proxy
+ * defined in Section 14 of version 3 of the license.
+ *
+ * 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.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef COMPOSITEFILETRANSFERJOB_H
+#define COMPOSITEFILETRANSFERJOB_H
+
+#include "kdeconnectcore_export.h"
+#include <KCompositeJob>
+#include <QElapsedTimer>
+
+class FileTransferJob;
+
+class KDECONNECTCORE_EXPORT CompositeFileTransferJob
+    : public KCompositeJob
+{
+    Q_OBJECT
+
+public:
+    explicit CompositeFileTransferJob(const QString& deviceId);
+
+    void start() override;
+    bool isRunning() const;
+    bool addSubjob(KJob* job) override;
+
+protected:
+    bool doKill() override;
+
+private Q_SLOTS:
+    void slotProcessedAmount(KJob *job, KJob::Unit unit, qulonglong amount);
+    void slotResult(KJob *job) override;
+    void startNextSubJob();
+
+private:
+    void emitDescription(const QString& currentFileName);
+
+    QString m_deviceId;
+    bool m_running;
+    int m_currentJobNum;
+    int m_totalJobs;
+    quint64 m_currentJobSendPayloadSize;
+    quint64 m_totalSendPayloadSize;
+    quint64 m_totalPayloadSize;
+    FileTransferJob *m_currentJob;
+    QElapsedTimer m_timer;
+    quint64 m_prevElapsedTime;
+
+};
+
+#endif //COMPOSITEFILETRANSFERJOB_H
+

core/filetransferjob.cpp

@@ -29,15 +29,16 @@
 
 #include <KLocalizedString>
 
-FileTransferJob::FileTransferJob(const QSharedPointer<QIODevice>& origin, qint64 size, const QUrl& destination)
+FileTransferJob::FileTransferJob(const NetworkPacket* np, const QUrl& destination)
     : KJob()
-    , m_origin(origin)
+    , m_origin(np->payload())
     , m_reply(nullptr)
     , m_from(QStringLiteral("KDE Connect"))
     , m_destination(destination)
     , m_speedBytes(0)
     , m_written(0)
-    , m_size(size)
+    , m_size(np->payloadSize())
+    , m_np(np)
 {
     Q_ASSERT(m_origin);
     //Disabled this assert: QBluetoothSocket doesn't report "->isReadable() == true" until it's connected
@@ -48,7 +49,7 @@ FileTransferJob::FileTransferJob(const QSharedPointer<QIODevice>& origin, qint64
     }
 
     setCapabilities(Killable);
-    qCDebug(KDECONNECT_CORE) << "FileTransferJob Downloading payload to" << destination << "size:" << size;
+    qCDebug(KDECONNECT_CORE) << "FileTransferJob Downloading payload to" << destination << "size:" << m_size;
 }
 
 void FileTransferJob::start()
@@ -59,10 +60,6 @@ void FileTransferJob::start()
 
 void FileTransferJob::doStart()
 {
-    Q_EMIT description(this, i18n("Receiving file over KDE Connect"),
-        { i18nc("File transfer origin", "From"), m_from }
-    );
-
     if (m_destination.isLocalFile() && QFile::exists(m_destination.toLocalFile())) {
         setError(2);
         setErrorText(i18n("Filename already present"));
@@ -82,10 +79,6 @@ void FileTransferJob::startTransfer()
         return;
 
     setProcessedAmount(Bytes, 0);
-    setTotalAmount(Files, 1);
-    Q_EMIT description(this, i18n("Receiving file over KDE Connect"),
-                        { i18nc("File transfer origin", "From"), m_from },
-                        { i18nc("File transfer destination", "To"), m_destination.toLocalFile() });
 
     QNetworkRequest req(m_destination);
     if (m_size >= 0) {
@@ -126,7 +119,6 @@ void FileTransferJob::transferFinished()
     //TODO: MD5-check the file
     if (m_size == m_written) {
         qCDebug(KDECONNECT_CORE) << "Finished transfer" << m_destination;
-        setProcessedAmount(Files, 1);
         emitResult();
     } else {
         qCDebug(KDECONNECT_CORE) << "Received incomplete file ("<< m_written << "/" << m_size << "bytes ), deleting";

core/filetransferjob.h

@@ -32,6 +32,7 @@
 
 #include "kdeconnectcore_export.h"
 
+class NetworkPacket;
 /**
  * @short It will stream a device into a url destination
  *
@@ -49,10 +50,11 @@ public:
      * @p size specifies the expected size of the stream we're reading.
      * @p destination specifies where these contents should be stored
      */
-    FileTransferJob(const QSharedPointer<QIODevice>& origin, qint64 size, const QUrl& destination);
+    FileTransferJob(const NetworkPacket* np, const QUrl& destination);
     void start() override;
     QUrl destination() const { return m_destination; }
     void setOriginName(const QString& from) { m_from = from; }
+    const NetworkPacket* networkPacket() { return m_np;}
 
 private Q_SLOTS:
     void doStart();
@@ -74,6 +76,7 @@ private:
     qulonglong m_speedBytes;
     qint64 m_written;
     qint64 m_size;
+    const NetworkPacket* m_np;
 };
 
 #endif

core/networkpacket.cpp

@@ -176,6 +176,6 @@ bool NetworkPacket::unserialize(const QByteArray& a, NetworkPacket* np)
 
 FileTransferJob* NetworkPacket::createPayloadTransferJob(const QUrl& destination) const
 {
-    return new FileTransferJob(payload(), payloadSize(), destination);
+    return new FileTransferJob(this, destination);
 }
 

plugins/share/shareplugin.cpp

@@ -26,7 +26,6 @@
 #include <QDir>
 #include <QDesktopServices>
 #include <QDBusConnection>
-#include <QDebug>
 #include <QTemporaryFile>
 #include <QDateTime>
 
@@ -44,6 +43,7 @@ Q_LOGGING_CATEGORY(KDECONNECT_PLUGIN_SHARE, "kdeconnect.plugin.share")
 
 SharePlugin::SharePlugin(QObject* parent, const QVariantList& args)
     : KdeConnectPlugin(parent, args)
+    , m_compositeJob()
 {
 }
 
@@ -121,11 +121,20 @@ bool SharePlugin::receivePacket(const NetworkPacket& np)
 
         if (np.hasPayload()) {
             qint64 dateModified = np.get<qint64>(QStringLiteral("lastModified"), QDateTime::currentMSecsSinceEpoch());
+
+            if (!m_compositeJob) {
+                m_compositeJob = new CompositeFileTransferJob(device()->id());
+                KIO::getJobTracker()->registerJob(m_compositeJob);
+            }
+
             FileTransferJob* job = np.createPayloadTransferJob(destination);
             job->setOriginName(device()->name() + ": " + filename);
             connect(job, &KJob::result, this, [this, dateModified] (KJob* job) -> void { finished(job, dateModified); });
-            KIO::getJobTracker()->registerJob(job);
+            m_compositeJob->addSubjob(job);
-            job->start();
+
+            if (!m_compositeJob->isRunning()) {
+                m_compositeJob->start();
+            }
         } else {
             QFile file(destination.toLocalFile());
             file.open(QIODevice::WriteOnly);

plugins/share/shareplugin.h

@@ -21,9 +21,12 @@
 #ifndef SHAREPLUGIN_H
 #define SHAREPLUGIN_H
 
+#include <QPointer>
+
 #include <KIO/Job>
 
 #include <core/kdeconnectplugin.h>
+#include <core/compositefiletransferjob.h>
 
 #define PACKET_TYPE_SHARE_REQUEST QStringLiteral("kdeconnect.share.request")
 #define PACKET_TYPE_SHARE_REQUEST_UPDATE QStringLiteral("kdeconnect.share.request.update")
@@ -60,5 +63,7 @@ private:
     QUrl destinationDir() const;
     QUrl getFileDestination(const QString filename) const;
     void setDateModified(const QUrl& destination, const qint64 timestamp);
+
+    QPointer<CompositeFileTransferJob> m_compositeJob;
 };
 #endif