Utils: Introduce FileStreamer

The class is responsible for asynchronous read / write
of file contents. The file may be local or remote.
It's also able to do an asynchronous copy of files
between different devices.

Change-Id: I65e4325b6b7f98bfc17286c9a72b0018db472a16
Reviewed-by: Qt CI Bot <qt_ci_bot@qt-project.org>
Reviewed-by: hjk <hjk@qt.io>

7 files changed, 926 insertions, 6 deletions

diff --git a/src/libs/utils/CMakeLists.txt b/src/libs/utils/CMakeLists.txt
index d6b6efb4f2..e8adad413e 100644
--- a/src/libs/utils/CMakeLists.txt
+++ b/src/libs/utils/CMakeLists.txt
@@ -55,6 +55,7 @@ add_qtc_library(Utils
     filepath.cpp filepath.h
     filepathinfo.h
     filesearch.cpp filesearch.h
+    filestreamer.cpp filestreamer.h
     filesystemmodel.cpp filesystemmodel.h
     filesystemwatcher.cpp filesystemwatcher.h
     fileutils.cpp fileutils.h
diff --git a/src/libs/utils/asynctask.h b/src/libs/utils/asynctask.h
index 84cf1ee838..5beaf400ea 100644
--- a/src/libs/utils/asynctask.h
+++ b/src/libs/utils/asynctask.h
@@ -24,13 +24,18 @@ class QTCREATOR_UTILS_EXPORT AsyncTaskBase : public QObject
 signals:
     void started();
     void done();
+    void resultReadyAt(int index);
 };
 
 template <typename ResultType>
 class AsyncTask : public AsyncTaskBase
 {
 public:
-    AsyncTask() { connect(&m_watcher, &QFutureWatcherBase::finished, this, &AsyncTaskBase::done); }
+    AsyncTask() {
+        connect(&m_watcher, &QFutureWatcherBase::finished, this, &AsyncTaskBase::done);
+        connect(&m_watcher, &QFutureWatcherBase::resultReadyAt,
+                this, &AsyncTaskBase::resultReadyAt);
+    }
     ~AsyncTask()
     {
         if (isDone())
@@ -72,6 +77,7 @@ public:
 
     QFuture<ResultType> future() const { return m_watcher.future(); }
     ResultType result() const { return m_watcher.result(); }
+    ResultType resultAt(int index) const { return m_watcher.resultAt(index); }
     QList<ResultType> results() const { return future().results(); }
     bool isResultAvailable() const { return future().resultCount(); }
 
diff --git a/src/libs/utils/filestreamer.cpp b/src/libs/utils/filestreamer.cpp
new file mode 100644
index 0000000000..222488a510
--- /dev/null
+++ b/src/libs/utils/filestreamer.cpp
@@ -0,0 +1,489 @@
+// Copyright (C) 2023 The Qt Company Ltd.
+// SPDX-License-Identifier: LicenseRef-Qt-Commercial OR GPL-3.0-only WITH Qt-GPL-exception-1.0
+
+#include "filestreamer.h"
+
+#include "asynctask.h"
+#include "qtcprocess.h"
+
+#include <QFile>
+#include <QMutex>
+#include <QMutexLocker>
+#include <QWaitCondition>
+
+namespace Utils {
+
+using namespace Tasking;
+
+// TODO: Adjust according to time spent on single buffer read so that it's not more than ~50 ms
+// in case of local read / write. Should it be adjusted dynamically / automatically?
+static const qint64 s_bufferSize = 0x1 << 20; // 1048576
+
+class FileStreamBase : public QObject
+{
+    Q_OBJECT
+
+public:
+    void setFilePath(const FilePath &filePath) { m_filePath = filePath; }
+    void start() {
+        QTC_ASSERT(!m_taskTree, return);
+
+        const TaskItem task = m_filePath.needsDevice() ? remoteTask() : localTask();
+        m_taskTree.reset(new TaskTree({task}));
+        const auto finalize = [this](bool success) {
+            m_taskTree.release()->deleteLater();
+            emit done(success);
+        };
+        connect(m_taskTree.get(), &TaskTree::done, this, [=] { finalize(true); });
+        connect(m_taskTree.get(), &TaskTree::errorOccurred, this, [=] { finalize(false); });
+        m_taskTree->start();
+    }
+
+signals:
+    void done(bool success);
+
+protected:
+    FilePath m_filePath;
+    std::unique_ptr<TaskTree> m_taskTree;
+
+private:
+    virtual TaskItem remoteTask() = 0;
+    virtual TaskItem localTask() = 0;
+};
+
+static void localRead(QPromise<QByteArray> &promise, const FilePath &filePath)
+{
+    if (promise.isCanceled())
+        return;
+
+    QFile file(filePath.path());
+    if (!file.exists()) {
+        promise.future().cancel();
+        return;
+    }
+
+    if (!file.open(QFile::ReadOnly)) {
+        promise.future().cancel();
+        return;
+    }
+
+    while (int chunkSize = qMin(s_bufferSize, file.bytesAvailable())) {
+        if (promise.isCanceled())
+            return;
+        promise.addResult(file.read(chunkSize));
+    }
+}
+
+class FileStreamReader : public FileStreamBase
+{
+    Q_OBJECT
+
+signals:
+    void readyRead(const QByteArray &newData);
+
+private:
+    TaskItem remoteTask() final {
+        const auto setup = [this](QtcProcess &process) {
+            const QStringList args = {"if=" + m_filePath.path()};
+            const FilePath dd = m_filePath.withNewPath("dd");
+            process.setCommand({dd, args, OsType::OsTypeLinux});
+            QtcProcess *processPtr = &process;
+            connect(processPtr, &QtcProcess::readyReadStandardOutput, this, [this, processPtr] {
+                emit readyRead(processPtr->readAllRawStandardOutput());
+            });
+        };
+        return Process(setup);
+    }
+    TaskItem localTask() final {
+        const auto setup = [this](AsyncTask<QByteArray> &async) {
+            async.setConcurrentCallData(localRead, m_filePath);
+            AsyncTask<QByteArray> *asyncPtr = &async;
+            connect(asyncPtr, &AsyncTaskBase::resultReadyAt, this, [=](int index) {
+                emit readyRead(asyncPtr->resultAt(index));
+            });
+        };
+        return Async<QByteArray>(setup);
+    }
+};
+
+class WriteBuffer : public QObject
+{
+    Q_OBJECT
+
+public:
+    WriteBuffer(bool isConcurrent, QObject *parent)
+        : QObject(parent)
+        , m_isConcurrent(isConcurrent) {}
+    struct Data {
+        QByteArray m_writeData;
+        bool m_closeWriteChannel = false;
+        bool m_canceled = false;
+        bool hasNewData() const { return m_closeWriteChannel || !m_writeData.isEmpty(); }
+    };
+
+    void write(const QByteArray &newData) {
+        if (m_isConcurrent) {
+            QMutexLocker locker(&m_mutex);
+            QTC_ASSERT(!m_data.m_closeWriteChannel, return);
+            QTC_ASSERT(!m_data.m_canceled, return);
+            m_data.m_writeData += newData;
+            m_waitCondition.wakeOne();
+            return;
+        }
+        emit writeRequested(newData);
+    }
+    void closeWriteChannel() {
+        if (m_isConcurrent) {
+            QMutexLocker locker(&m_mutex);
+            QTC_ASSERT(!m_data.m_canceled, return);
+            m_data.m_closeWriteChannel = true;
+            m_waitCondition.wakeOne();
+            return;
+        }
+        emit closeWriteChannelRequested();
+    }
+    void cancel() {
+        if (m_isConcurrent) {
+            QMutexLocker locker(&m_mutex);
+            m_data.m_canceled = true;
+            m_waitCondition.wakeOne();
+            return;
+        }
+        emit closeWriteChannelRequested();
+    }
+    Data waitForData() {
+        QTC_ASSERT(m_isConcurrent, return {});
+        QMutexLocker locker(&m_mutex);
+        if (!m_data.hasNewData())
+            m_waitCondition.wait(&m_mutex);
+        return std::exchange(m_data, {});
+    }
+
+signals:
+    void writeRequested(const QByteArray &newData);
+    void closeWriteChannelRequested();
+
+private:
+    QMutex m_mutex;
+    QWaitCondition m_waitCondition;
+    Data m_data;
+    bool m_isConcurrent = false; // Depends on whether FileStreamWriter::m_writeData is empty or not
+};
+
+static void localWrite(QPromise<void> &promise, const FilePath &filePath,
+                       const QByteArray &initialData, WriteBuffer *buffer)
+{
+    if (promise.isCanceled())
+        return;
+
+    QFile file(filePath.path());
+
+    if (!file.open(QFile::WriteOnly | QFile::Truncate)) {
+        promise.future().cancel();
+        return;
+    }
+
+    if (!initialData.isEmpty()) {
+        const qint64 res = file.write(initialData);
+        if (res != initialData.size())
+            promise.future().cancel();
+        return;
+    }
+
+    while (true) {
+        if (promise.isCanceled()) {
+            promise.future().cancel();
+            return;
+        }
+        const WriteBuffer::Data data = buffer->waitForData();
+        if (data.m_canceled || promise.isCanceled()) {
+            promise.future().cancel();
+            return;
+        }
+        if (!data.m_writeData.isEmpty()) {
+            // TODO: Write in chunks of s_bufferSize and check for promise.isCanceled()
+            const qint64 res = file.write(data.m_writeData);
+            if (res != data.m_writeData.size()) {
+                promise.future().cancel();
+                return;
+            }
+        }
+        if (data.m_closeWriteChannel)
+            return;
+    }
+}
+
+class FileStreamWriter : public FileStreamBase
+{
+    Q_OBJECT
+
+public:
+    ~FileStreamWriter() { // TODO: should d'tor remove unfinished file write leftovers?
+        if (m_writeBuffer && isBuffered())
+            m_writeBuffer->cancel();
+    }
+
+    void setWriteData(const QByteArray &writeData) {
+        QTC_ASSERT(!m_taskTree, return);
+        m_writeData = writeData;
+    }
+    void write(const QByteArray &newData) {
+        QTC_ASSERT(m_taskTree, return);
+        QTC_ASSERT(m_writeData.isEmpty(), return);
+        QTC_ASSERT(m_writeBuffer, return);
+        m_writeBuffer->write(newData);
+    }
+    void closeWriteChannel() {
+        QTC_ASSERT(m_taskTree, return);
+        QTC_ASSERT(m_writeData.isEmpty(), return);
+        QTC_ASSERT(m_writeBuffer, return);
+        m_writeBuffer->closeWriteChannel();
+    }
+
+signals:
+    void started();
+
+private:
+    TaskItem remoteTask() final {
+        const auto setup = [this](QtcProcess &process) {
+            m_writeBuffer = new WriteBuffer(false, &process);
+            connect(m_writeBuffer, &WriteBuffer::writeRequested, &process, &QtcProcess::writeRaw);
+            connect(m_writeBuffer, &WriteBuffer::closeWriteChannelRequested,
+                    &process, &QtcProcess::closeWriteChannel);
+            const QStringList args = {"of=" + m_filePath.path()};
+            const FilePath dd = m_filePath.withNewPath("dd");
+            process.setCommand({dd, args, OsType::OsTypeLinux});
+            if (isBuffered())
+                process.setProcessMode(ProcessMode::Writer);
+            else
+                process.setWriteData(m_writeData);
+            connect(&process, &QtcProcess::started, this, [this] { emit started(); });
+        };
+        const auto finalize = [this](const QtcProcess &) {
+            delete m_writeBuffer;
+            m_writeBuffer = nullptr;
+        };
+        return Process(setup, finalize, finalize);
+    }
+    TaskItem localTask() final {
+        const auto setup = [this](AsyncTask<void> &async) {
+            m_writeBuffer = new WriteBuffer(isBuffered(), &async);
+            async.setConcurrentCallData(localWrite, m_filePath, m_writeData, m_writeBuffer);
+            emit started();
+        };
+        const auto finalize = [this](const AsyncTask<void> &) {
+            delete m_writeBuffer;
+            m_writeBuffer = nullptr;
+        };
+        return Async<void>(setup, finalize, finalize);
+    }
+
+    bool isBuffered() const { return m_writeData.isEmpty(); }
+    QByteArray m_writeData;
+    WriteBuffer *m_writeBuffer = nullptr;
+};
+
+class FileStreamReaderAdapter : public Utils::Tasking::TaskAdapter<FileStreamReader>
+{
+public:
+    FileStreamReaderAdapter() { connect(task(), &FileStreamBase::done, this, &TaskInterface::done); }
+    void start() override { task()->start(); }
+};
+
+class FileStreamWriterAdapter : public Utils::Tasking::TaskAdapter<FileStreamWriter>
+{
+public:
+    FileStreamWriterAdapter() { connect(task(), &FileStreamBase::done, this, &TaskInterface::done); }
+    void start() override { task()->start(); }
+};
+
+} // namespace Utils
+
+QTC_DECLARE_CUSTOM_TASK(Reader, Utils::FileStreamReaderAdapter);
+QTC_DECLARE_CUSTOM_TASK(Writer, Utils::FileStreamWriterAdapter);
+
+namespace Utils {
+
+static Group interDeviceTransfer(const FilePath &source, const FilePath &destination)
+{
+    struct TransferStorage { QPointer<FileStreamWriter> writer; };
+    Condition condition;
+    TreeStorage<TransferStorage> storage;
+
+    const auto setupReader = [=](FileStreamReader &reader) {
+        reader.setFilePath(source);
+        QTC_CHECK(storage->writer != nullptr);
+        QObject::connect(&reader, &FileStreamReader::readyRead,
+                         storage->writer, &FileStreamWriter::write);
+    };
+    const auto finalizeReader = [=](const FileStreamReader &) {
+        QTC_CHECK(storage->writer != nullptr);
+        storage->writer->closeWriteChannel();
+    };
+    const auto setupWriter = [=](FileStreamWriter &writer) {
+        writer.setFilePath(destination);
+        ConditionActivator *activator = condition.activator();
+        QObject::connect(&writer, &FileStreamWriter::started,
+                         &writer, [activator] { activator->activate(); });
+        QTC_CHECK(storage->writer == nullptr);
+        storage->writer = &writer;
+    };
+
+    const Group root {
+        parallel,
+        Storage(storage),
+        Writer(setupWriter),
+        Group {
+            WaitFor(condition),
+            Reader(setupReader, finalizeReader, finalizeReader)
+        }
+    };
+
+    return root;
+}
+
+static void transfer(QPromise<void> &promise, const FilePath &source, const FilePath &destination)
+{
+    if (promise.isCanceled())
+        return;
+
+    std::unique_ptr<TaskTree> taskTree(new TaskTree(interDeviceTransfer(source, destination)));
+
+    QEventLoop eventLoop;
+    bool finalized = false;
+    const auto finalize = [loop = &eventLoop, &taskTree, &finalized](int exitCode) {
+        if (finalized) // finalize only once
+            return;
+        finalized = true;
+        // Give the tree a chance to delete later all tasks that have finished and caused
+        // emission of tree's done or errorOccurred signal.
+        // TODO: maybe these signals should be sent queued already?
+        QMetaObject::invokeMethod(loop, [loop, &taskTree, exitCode] {
+            taskTree.reset();
+            loop->exit(exitCode);
+        }, Qt::QueuedConnection);
+    };
+    QTimer timer;
+    timer.setInterval(50);
+    QObject::connect(&timer, &QTimer::timeout, [&promise, finalize] {
+        if (promise.isCanceled())
+            finalize(2);
+    });
+    QObject::connect(taskTree.get(), &TaskTree::done, &eventLoop, [=] { finalize(0); });
+    QObject::connect(taskTree.get(), &TaskTree::errorOccurred, &eventLoop, [=] { finalize(1); });
+    taskTree->start();
+    timer.start();
+    if (eventLoop.exec())
+        promise.future().cancel();
+}
+
+class FileStreamerPrivate : public QObject
+{
+public:
+    StreamMode m_streamerMode = StreamMode::Transfer;
+    FilePath m_source;
+    FilePath m_destination;
+    QByteArray m_readBuffer;
+    QByteArray m_writeBuffer;
+    StreamResult m_streamResult = StreamResult::FinishedWithError;
+    std::unique_ptr<TaskTree> m_taskTree;
+
+    TaskItem task() {
+        if (m_streamerMode == StreamMode::Reader)
+            return readerTask();
+        if (m_streamerMode == StreamMode::Writer)
+            return writerTask();
+        return transferTask();
+    }
+
+private:
+    TaskItem readerTask() {
+        const auto setup = [this](FileStreamReader &reader) {
+            m_readBuffer.clear();
+            reader.setFilePath(m_source);
+            connect(&reader, &FileStreamReader::readyRead, this, [this](const QByteArray &data) {
+                m_readBuffer += data;
+            });
+        };
+        return Reader(setup);
+    }
+    TaskItem writerTask() {
+        const auto setup = [this](FileStreamWriter &writer) {
+            writer.setFilePath(m_destination);
+            writer.setWriteData(m_writeBuffer);
+        };
+        return Writer(setup);
+    }
+    TaskItem transferTask() {
+        const auto setup = [this](AsyncTask<void> &async) {
+            async.setConcurrentCallData(transfer, m_source, m_destination);
+        };
+        return Async<void>(setup);
+    }
+};
+
+FileStreamer::FileStreamer(QObject *parent)
+    : QObject(parent)
+    , d(new FileStreamerPrivate)
+{
+}
+
+FileStreamer::~FileStreamer()
+{
+    delete d;
+}
+
+void FileStreamer::setSource(const FilePath &source)
+{
+    d->m_source = source;
+}
+
+void FileStreamer::setDestination(const FilePath &destination)
+{
+    d->m_destination = destination;
+}
+
+void FileStreamer::setStreamMode(StreamMode mode)
+{
+    d->m_streamerMode = mode;
+}
+
+QByteArray FileStreamer::readData() const
+{
+    return d->m_readBuffer;
+}
+
+void FileStreamer::setWriteData(const QByteArray &writeData)
+{
+    d->m_writeBuffer = writeData;
+}
+
+StreamResult FileStreamer::result() const
+{
+    return d->m_streamResult;
+}
+
+void FileStreamer::start()
+{
+    // TODO: Preliminary check if local source exists?
+    QTC_ASSERT(!d->m_taskTree, return);
+    d->m_taskTree.reset(new TaskTree({d->task()}));
+    const auto finalize = [this](bool success) {
+        d->m_streamResult = success ? StreamResult::FinishedWithSuccess
+                                    : StreamResult::FinishedWithError;
+        d->m_taskTree.release()->deleteLater();
+        emit done();
+    };
+    connect(d->m_taskTree.get(), &TaskTree::done, this, [=] { finalize(true); });
+    connect(d->m_taskTree.get(), &TaskTree::errorOccurred, this, [=] { finalize(false); });
+    d->m_taskTree->start();
+}
+
+void FileStreamer::stop()
+{
+    d->m_taskTree.reset();
+}
+
+} // namespace Utils
+
+#include "filestreamer.moc"
diff --git a/src/libs/utils/filestreamer.h b/src/libs/utils/filestreamer.h
new file mode 100644
index 0000000000..b572e910a2
--- /dev/null
+++ b/src/libs/utils/filestreamer.h
@@ -0,0 +1,62 @@
+// Copyright (C) 2023 The Qt Company Ltd.
+// SPDX-License-Identifier: LicenseRef-Qt-Commercial OR GPL-3.0-only WITH Qt-GPL-exception-1.0
+
+#pragma once
+
+#include "utils_global.h"
+
+#include "filepath.h"
+#include "tasktree.h"
+
+#include <QObject>
+
+QT_BEGIN_NAMESPACE
+class QByteArray;
+QT_END_NAMESPACE
+
+namespace Utils {
+
+enum class StreamMode { Reader, Writer, Transfer };
+
+enum class StreamResult { FinishedWithSuccess, FinishedWithError };
+
+class QTCREATOR_UTILS_EXPORT FileStreamer final : public QObject
+{
+    Q_OBJECT
+
+public:
+    FileStreamer(QObject *parent = nullptr);
+    ~FileStreamer();
+
+    void setSource(const FilePath &source);
+    void setDestination(const FilePath &destination);
+    void setStreamMode(StreamMode mode); // Transfer by default
+
+    // Only for Reader mode
+    QByteArray readData() const;
+    // Only for Writer mode
+    void setWriteData(const QByteArray &writeData);
+
+    StreamResult result() const;
+
+    void start();
+    void stop();
+
+signals:
+    void done();
+
+private:
+    class FileStreamerPrivate *d = nullptr;
+};
+
+class FileStreamerAdapter : public Utils::Tasking::TaskAdapter<FileStreamer>
+{
+public:
+    FileStreamerAdapter() { connect(task(), &FileStreamer::done, this,
+                [this] { emit done(task()->result() == StreamResult::FinishedWithSuccess); }); }
+    void start() override { task()->start(); }
+};
+
+} // namespace Utils
+
+QTC_DECLARE_CUSTOM_TASK(Streamer, Utils::FileStreamerAdapter);
diff --git a/src/libs/utils/utils.qbs b/src/libs/utils/utils.qbs
index cf952ae62d..8c0312ed03 100644
--- a/src/libs/utils/utils.qbs
+++ b/src/libs/utils/utils.qbs
@@ -127,6 +127,8 @@ Project {
             "filepath.h",
             "filesearch.cpp",
             "filesearch.h",
+            "filestreamer.cpp",
+            "filestreamer.h",
             "filesystemmodel.cpp",
             "filesystemmodel.h",
             "filesystemwatcher.cpp",
diff --git a/src/plugins/remotelinux/filesystemaccess_test.cpp b/src/plugins/remotelinux/filesystemaccess_test.cpp
index fcc8477974..fa3f654f38 100644
--- a/src/plugins/remotelinux/filesystemaccess_test.cpp
+++ b/src/plugins/remotelinux/filesystemaccess_test.cpp
@@ -9,6 +9,7 @@
 #include <projectexplorer/devicesupport/filetransfer.h>
 #include <projectexplorer/devicesupport/sshparameters.h>
 #include <utils/filepath.h>
+#include <utils/filestreamer.h>
 #include <utils/processinterface.h>
 
 #include <QDebug>
@@ -86,6 +87,44 @@ void FileSystemAccessTest::initTestCase()
     QVERIFY(!filePath.exists());
     QVERIFY(filePath.createDir());
     QVERIFY(filePath.exists());
+
+    const QString streamerDir("streamerDir");
+    const QString sourceDir("source");
+    const QString destDir("dest");
+    const QString localDir("local");
+    const QString remoteDir("remote");
+    const FilePath localRoot;
+    const FilePath remoteRoot = m_device->rootPath();
+    const FilePath localTempDir = *localRoot.tmpDir();
+    const FilePath remoteTempDir = *remoteRoot.tmpDir();
+    m_localStreamerDir = localTempDir / streamerDir;
+    m_remoteStreamerDir = remoteTempDir / streamerDir;
+    m_localSourceDir = m_localStreamerDir / sourceDir;
+    m_remoteSourceDir = m_remoteStreamerDir / sourceDir;
+    m_localDestDir = m_localStreamerDir / destDir;
+    m_remoteDestDir = m_remoteStreamerDir / destDir;
+    m_localLocalDestDir = m_localDestDir / localDir;
+    m_localRemoteDestDir = m_localDestDir / remoteDir;
+    m_remoteLocalDestDir = m_remoteDestDir / localDir;
+    m_remoteRemoteDestDir = m_remoteDestDir / remoteDir;
+
+    QVERIFY(m_localSourceDir.createDir());
+    QVERIFY(m_remoteSourceDir.createDir());
+    QVERIFY(m_localDestDir.createDir());
+    QVERIFY(m_remoteDestDir.createDir());
+    QVERIFY(m_localLocalDestDir.createDir());
+    QVERIFY(m_localRemoteDestDir.createDir());
+    QVERIFY(m_remoteLocalDestDir.createDir());
+    QVERIFY(m_remoteRemoteDestDir.createDir());
+
+    QVERIFY(m_localSourceDir.exists());
+    QVERIFY(m_remoteSourceDir.exists());
+    QVERIFY(m_localDestDir.exists());
+    QVERIFY(m_remoteDestDir.exists());
+    QVERIFY(m_localLocalDestDir.exists());
+    QVERIFY(m_localRemoteDestDir.exists());
+    QVERIFY(m_remoteLocalDestDir.exists());
+    QVERIFY(m_remoteRemoteDestDir.exists());
 }
 
 void FileSystemAccessTest::cleanupTestCase()
@@ -94,6 +133,12 @@ void FileSystemAccessTest::cleanupTestCase()
         return;
     QVERIFY(baseFilePath().exists());
     QVERIFY(baseFilePath().removeRecursively());
+
+    QVERIFY(m_localStreamerDir.removeRecursively());
+    QVERIFY(m_remoteStreamerDir.removeRecursively());
+
+    QVERIFY(!m_localStreamerDir.exists());
+    QVERIFY(!m_remoteStreamerDir.exists());
 }
 
 void FileSystemAccessTest::testCreateRemoteFile_data()
@@ -102,13 +147,13 @@ void FileSystemAccessTest::testCreateRemoteFile_data()
 
     QTest::newRow("Spaces") << QByteArray("Line with spaces");
     QTest::newRow("Newlines") << QByteArray("Some \n\n newlines \n");
-    QTest::newRow("Carriage return") << QByteArray("Line with carriage \r return");
+    QTest::newRow("CarriageReturn") << QByteArray("Line with carriage \r return");
     QTest::newRow("Tab") << QByteArray("Line with \t tab");
     QTest::newRow("Apostrophe") << QByteArray("Line with apostrophe's character");
-    QTest::newRow("Quotation marks") << QByteArray("Line with \"quotation marks\"");
-    QTest::newRow("Backslash 1") << QByteArray("Line with \\ backslash");
-    QTest::newRow("Backslash 2") << QByteArray("Line with \\\" backslash");
-    QTest::newRow("Command output") << QByteArray("The date is: $(date +%D)");
+    QTest::newRow("QuotationMarks") << QByteArray("Line with \"quotation marks\"");
+    QTest::newRow("Backslash1") << QByteArray("Line with \\ backslash");
+    QTest::newRow("Backslash2") << QByteArray("Line with \\\" backslash");
+    QTest::newRow("CommandOutput") << QByteArray("The date is: $(date +%D)");
 
     const int charSize = sizeof(char) * 0x100;
     QByteArray charString(charSize, Qt::Uninitialized);
@@ -201,6 +246,8 @@ void FileSystemAccessTest::testFileTransfer_data()
     QTest::addColumn<FileTransferMethod>("fileTransferMethod");
 
     QTest::addRow("Sftp") << FileTransferMethod::Sftp;
+    // TODO: By default rsync doesn't support creating target directories,
+    // needs to be done manually - see RsyncDeployService.
 //    QTest::addRow("Rsync") << FileTransferMethod::Rsync;
 }
 
@@ -282,5 +329,303 @@ void FileSystemAccessTest::testFileTransfer()
     QVERIFY2(remoteDir.removeRecursively(&errorString), qPrintable(errorString));
 }
 
+void FileSystemAccessTest::testFileStreamer_data()
+{
+    QTest::addColumn<QString>("fileName");
+    QTest::addColumn<QByteArray>("data");
+
+    const QByteArray spaces("Line with spaces");
+    const QByteArray newlines("Some \n\n newlines \n");
+    const QByteArray carriageReturn("Line with carriage \r return");
+    const QByteArray tab("Line with \t tab");
+    const QByteArray apostrophe("Line with apostrophe's character");
+    const QByteArray quotationMarks("Line with \"quotation marks\"");
+    const QByteArray backslash1("Line with \\ backslash");
+    const QByteArray backslash2("Line with \\\" backslash");
+    const QByteArray commandOutput("The date is: $(date +%D)");
+
+    const int charSize = sizeof(char) * 0x100;
+    QByteArray charString(charSize, Qt::Uninitialized);
+    char *data = charString.data();
+    for (int c = 0; c < charSize; ++c)
+        data[c] = c;
+
+    const int bigSize = 1024 * 1024; // = 256 * 1024 * 1024 = 268.435.456 bytes
+    QByteArray bigString;
+    for (int i = 0; i < bigSize; ++i)
+        bigString += charString;
+
+    QTest::newRow("Spaces") << QString("spaces") << spaces;
+    QTest::newRow("Newlines") << QString("newlines") << newlines;
+    QTest::newRow("CarriageReturn") << QString("carriageReturn") << carriageReturn;
+    QTest::newRow("Tab") << QString("tab") << tab;
+    QTest::newRow("Apostrophe") << QString("apostrophe") << apostrophe;
+    QTest::newRow("QuotationMarks") << QString("quotationMarks") << quotationMarks;
+    QTest::newRow("Backslash1") << QString("backslash1") << backslash1;
+    QTest::newRow("Backslash2") << QString("backslash2") << backslash2;
+    QTest::newRow("CommandOutput") << QString("commandOutput") << commandOutput;
+    QTest::newRow("AllCharacters") << QString("charString") << charString;
+    QTest::newRow("BigString") << QString("bigString") << bigString;
+}
+
+void FileSystemAccessTest::testFileStreamer()
+{
+    QElapsedTimer timer;
+    timer.start();
+
+    QFETCH(QString, fileName);
+    QFETCH(QByteArray, data);
+
+    const FilePath localSourcePath = m_localSourceDir / fileName;
+    const FilePath remoteSourcePath = m_remoteSourceDir / fileName;
+    const FilePath localLocalDestPath = m_localDestDir / "local" / fileName;
+    const FilePath localRemoteDestPath = m_localDestDir / "remote" / fileName;
+    const FilePath remoteLocalDestPath = m_remoteDestDir / "local" / fileName;
+    const FilePath remoteRemoteDestPath = m_remoteDestDir / "remote" / fileName;
+
+    localSourcePath.removeFile();
+    remoteSourcePath.removeFile();
+    localLocalDestPath.removeFile();
+    localRemoteDestPath.removeFile();
+    remoteLocalDestPath.removeFile();
+    remoteRemoteDestPath.removeFile();
+
+    QVERIFY(!localSourcePath.exists());
+    QVERIFY(!remoteSourcePath.exists());
+    QVERIFY(!localLocalDestPath.exists());
+    QVERIFY(!localRemoteDestPath.exists());
+    QVERIFY(!remoteLocalDestPath.exists());
+    QVERIFY(!remoteRemoteDestPath.exists());
+
+    std::optional<QByteArray> localData;
+    std::optional<QByteArray> remoteData;
+    std::optional<QByteArray> localLocalData;
+    std::optional<QByteArray> localRemoteData;
+    std::optional<QByteArray> remoteLocalData;
+    std::optional<QByteArray> remoteRemoteData;
+
+    using namespace Tasking;
+
+    const auto localWriter = [&] {
+        const auto setup = [&](FileStreamer &streamer) {
+            streamer.setStreamMode(StreamMode::Writer);
+            streamer.setDestination(localSourcePath);
+            streamer.setWriteData(data);
+        };
+        return Streamer(setup);
+    };
+    const auto remoteWriter = [&] {
+        const auto setup = [&](FileStreamer &streamer) {
+            streamer.setStreamMode(StreamMode::Writer);
+            streamer.setDestination(remoteSourcePath);
+            streamer.setWriteData(data);
+        };
+        return Streamer(setup);
+    };
+    const auto localReader = [&] {
+        const auto setup = [&](FileStreamer &streamer) {
+            streamer.setStreamMode(StreamMode::Reader);
+            streamer.setSource(localSourcePath);
+        };
+        const auto onDone = [&](const FileStreamer &streamer) {
+            localData = streamer.readData();
+        };
+        return Streamer(setup, onDone);
+    };
+    const auto remoteReader = [&] {
+        const auto setup = [&](FileStreamer &streamer) {
+            streamer.setStreamMode(StreamMode::Reader);
+            streamer.setSource(remoteSourcePath);
+        };
+        const auto onDone = [&](const FileStreamer &streamer) {
+            remoteData = streamer.readData();
+        };
+        return Streamer(setup, onDone);
+    };
+    const auto transfer = [](const FilePath &source, const FilePath &dest,
+                             std::optional<QByteArray> *result) {
+        const auto setupTransfer = [=](FileStreamer &streamer) {
+            streamer.setSource(source);
+            streamer.setDestination(dest);
+        };
+        const auto setupReader = [=](FileStreamer &streamer) {
+            streamer.setStreamMode(StreamMode::Reader);
+            streamer.setSource(dest);
+        };
+        const auto onReaderDone = [result](const FileStreamer &streamer) {
+            *result = streamer.readData();
+        };
+        const Group root {
+            Streamer(setupTransfer),
+            Streamer(setupReader, onReaderDone)
+        };
+        return root;
+    };
+
+    // In total: 5 local reads, 3 local writes, 5 remote reads, 3 remote writes
+    const Group root {
+        Group {
+            parallel,
+            localWriter(),
+            remoteWriter()
+        },
+        Group {
+            parallel,
+            localReader(),
+            remoteReader()
+        },
+        Group {
+            parallel,
+            transfer(localSourcePath, localLocalDestPath, &localLocalData),
+            transfer(remoteSourcePath, localRemoteDestPath, &localRemoteData),
+            transfer(localSourcePath, remoteLocalDestPath, &remoteLocalData),
+            transfer(remoteSourcePath, remoteRemoteDestPath, &remoteRemoteData),
+        }
+    };
+
+    QEventLoop eventLoop;
+    TaskTree taskTree(root);
+    int doneCount = 0;
+    int errorCount = 0;
+    connect(&taskTree, &TaskTree::done, this, [&doneCount, &eventLoop] {
+        ++doneCount;
+        eventLoop.quit();
+    });
+    connect(&taskTree, &TaskTree::errorOccurred, this, [&errorCount, &eventLoop] {
+        ++errorCount;
+        eventLoop.quit();
+    });
+    taskTree.start();
+    QVERIFY(taskTree.isRunning());
+
+    QTimer timeoutTimer;
+    bool timedOut = false;
+    connect(&timeoutTimer, &QTimer::timeout, &eventLoop, [&eventLoop, &timedOut] {
+        timedOut = true;
+        eventLoop.quit();
+    });
+    timeoutTimer.setInterval(10000);
+    timeoutTimer.setSingleShot(true);
+    timeoutTimer.start();
+    eventLoop.exec();
+    QCOMPARE(timedOut, false);
+    QCOMPARE(taskTree.isRunning(), false);
+    QCOMPARE(doneCount, 1);
+    QCOMPARE(errorCount, 0);
+
+    QVERIFY(localData);
+    QCOMPARE(*localData, data);
+    QVERIFY(remoteData);
+    QCOMPARE(*remoteData, data);
+
+    QVERIFY(localLocalData);
+    QCOMPARE(*localLocalData, data);
+    QVERIFY(localRemoteData);
+    QCOMPARE(*localRemoteData, data);
+    QVERIFY(remoteLocalData);
+    QCOMPARE(*remoteLocalData, data);
+    QVERIFY(remoteRemoteData);
+    QCOMPARE(*remoteRemoteData, data);
+
+    qDebug() << "Elapsed time:" << timer.elapsed() << "ms.";
+}
+
+void FileSystemAccessTest::testBlockingTransfer_data()
+{
+    testFileStreamer_data();
+}
+
+void FileSystemAccessTest::testBlockingTransfer()
+{
+    QElapsedTimer timer;
+    timer.start();
+
+    QFETCH(QString, fileName);
+    QFETCH(QByteArray, data);
+
+    const FilePath localSourcePath = m_localSourceDir / fileName;
+    const FilePath remoteSourcePath = m_remoteSourceDir / fileName;
+    const FilePath localLocalDestPath = m_localDestDir / "local" / fileName;
+    const FilePath localRemoteDestPath = m_localDestDir / "remote" / fileName;
+    const FilePath remoteLocalDestPath = m_remoteDestDir / "local" / fileName;
+    const FilePath remoteRemoteDestPath = m_remoteDestDir / "remote" / fileName;
+
+    localSourcePath.removeFile();
+    remoteSourcePath.removeFile();
+    localLocalDestPath.removeFile();
+    localRemoteDestPath.removeFile();
+    remoteLocalDestPath.removeFile();
+    remoteRemoteDestPath.removeFile();
+
+    QVERIFY(!localSourcePath.exists());
+    QVERIFY(!remoteSourcePath.exists());
+    QVERIFY(!localLocalDestPath.exists());
+    QVERIFY(!localRemoteDestPath.exists());
+    QVERIFY(!remoteLocalDestPath.exists());
+    QVERIFY(!remoteRemoteDestPath.exists());
+
+    bool writerHit = false;
+    const auto writeChecker = [&](const expected_str<qint64> &res) {
+        writerHit = true;
+        QVERIFY(res);
+    };
+
+    bool readerHit = false;
+    const auto readChecker = [&](const QByteArray &expected) {
+        return [&](const expected_str<QByteArray> &result) {
+            readerHit = true;
+            QCOMPARE(result, expected);
+        };
+    };
+
+    bool dummyHit = false;
+    const auto dummy = [&](const expected_str<void> &res) {
+        dummyHit = true;
+        QVERIFY(res);
+    };
+
+    localSourcePath.asyncWriteFileContents(writeChecker, data);
+    localSourcePath.asyncFileContents(readChecker(data));
+    QVERIFY(writerHit);
+    QVERIFY(readerHit);
+
+    writerHit = false;
+    readerHit = false;
+    remoteSourcePath.asyncWriteFileContents(writeChecker, data);
+    remoteSourcePath.asyncFileContents(readChecker(data));
+    QVERIFY(writerHit);
+    QVERIFY(readerHit);
+
+    dummyHit = false;
+    readerHit = false;
+    localSourcePath.asyncCopyFile(dummy, localLocalDestPath);
+    localLocalDestPath.asyncFileContents(readChecker(data));
+    QVERIFY(dummyHit);
+    QVERIFY(readerHit);
+
+    dummyHit = false;
+    readerHit = false;
+    remoteSourcePath.asyncCopyFile(dummy, localRemoteDestPath);
+    localRemoteDestPath.asyncFileContents(readChecker(data));
+    QVERIFY(dummyHit);
+    QVERIFY(readerHit);
+
+    dummyHit = false;
+    readerHit = false;
+    localSourcePath.asyncCopyFile(dummy, remoteLocalDestPath);
+    remoteLocalDestPath.asyncFileContents(readChecker(data));
+    QVERIFY(dummyHit);
+    QVERIFY(readerHit);
+
+    dummyHit = false;
+    readerHit = false;
+    remoteSourcePath.asyncCopyFile(dummy, remoteRemoteDestPath);
+    remoteRemoteDestPath.asyncFileContents(readChecker(data));
+    QVERIFY(dummyHit);
+    QVERIFY(readerHit);
+
+    qDebug() << "Elapsed time:" << timer.elapsed() << "ms.";
+}
+
 } // Internal
 } // RemoteLinux
diff --git a/src/plugins/remotelinux/filesystemaccess_test.h b/src/plugins/remotelinux/filesystemaccess_test.h
index 9684cbc926..f7101af390 100644
--- a/src/plugins/remotelinux/filesystemaccess_test.h
+++ b/src/plugins/remotelinux/filesystemaccess_test.h
@@ -4,6 +4,7 @@
 #pragma once
 
 #include <projectexplorer/devicesupport/idevicefactory.h>
+#include <utils/filepath.h>
 
 namespace RemoteLinux {
 namespace Internal {
@@ -28,6 +29,10 @@ private slots:
     void testFileActions();
     void testFileTransfer_data();
     void testFileTransfer();
+    void testFileStreamer_data();
+    void testFileStreamer();
+    void testBlockingTransfer_data();
+    void testBlockingTransfer();
 
     void cleanupTestCase();
 
@@ -35,6 +40,16 @@ private:
     TestLinuxDeviceFactory m_testLinuxDeviceFactory;
     bool m_skippedAtWhole = false;
     ProjectExplorer::IDeviceConstPtr m_device;
+    Utils::FilePath m_localStreamerDir;
+    Utils::FilePath m_remoteStreamerDir;
+    Utils::FilePath m_localSourceDir;
+    Utils::FilePath m_remoteSourceDir;
+    Utils::FilePath m_localDestDir;
+    Utils::FilePath m_remoteDestDir;
+    Utils::FilePath m_localLocalDestDir;
+    Utils::FilePath m_localRemoteDestDir;
+    Utils::FilePath m_remoteLocalDestDir;
+    Utils::FilePath m_remoteRemoteDestDir;
 };
 
 } // Internal