feat: do not load entire data to memory in libarchive operations

14 files changed, 479 insertions, 187 deletions

diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index db8ee2db..b4d6f68d 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -69,9 +69,9 @@ find_package(OpenSSL REQUIRED)
 # Introduce Qt
 if (QT5_ENV_SUPPORT)
     # Support Qt version: 6.x, 5.12.x and 5.15.x
-    find_package(Qt6 6 COMPONENTS Core Test Widgets PrintSupport Network Core5Compat)
+    find_package(Qt6 6 COMPONENTS Core Test Widgets PrintSupport Network Core5Compat Concurrent)
     if(NOT Qt6_DIR)
-        find_package(Qt5 5 COMPONENTS Core Test Widgets PrintSupport Network REQUIRED)
+        find_package(Qt5 5 COMPONENTS Core Test Widgets PrintSupport Network Concurrent REQUIRED)
         message(STATUS "Use Qt5 for application building ${Qt5_DIR}")
     else()
         message(STATUS "Use Qt6 for application building ${Qt6_DIR}")
diff --git a/src/core/CMakeLists.txt b/src/core/CMakeLists.txt
index ee95b061..c4eb8049 100644
--- a/src/core/CMakeLists.txt
+++ b/src/core/CMakeLists.txt
@@ -101,9 +101,9 @@ target_link_libraries(gpgfrontend_core
         
 # link Qt core
 if(Qt6_DIR)
-    target_link_libraries(gpgfrontend_core PUBLIC Qt6::Core)
+    target_link_libraries(gpgfrontend_core PUBLIC Qt6::Core Qt6::Concurrent)
 else()
-    target_link_libraries(gpgfrontend_core PUBLIC Qt5::Core)
+    target_link_libraries(gpgfrontend_core PUBLIC Qt5::Core Qt5::Concurrent)
 endif()
 
 # set up pch
diff --git a/src/core/function/ArchiveFileOperator.cpp b/src/core/function/ArchiveFileOperator.cpp
index c8cd99f4..b252f3a0 100644
--- a/src/core/function/ArchiveFileOperator.cpp
+++ b/src/core/function/ArchiveFileOperator.cpp
@@ -30,6 +30,7 @@
 
 #include <archive.h>
 #include <archive_entry.h>
+#include <sys/fcntl.h>
 
 #include <fstream>
 
@@ -37,14 +38,6 @@
 
 namespace GpgFrontend {
 
-struct ArchiveStruct {
-  struct archive *archive;
-  struct archive_entry *entry;
-  int fd;
-  bool is_open;
-  std::string name;
-};
-
 auto CopyData(struct archive *ar, struct archive *aw) -> int {
   int r;
   const void *buff;
@@ -68,126 +61,185 @@ auto CopyData(struct archive *ar, struct archive *aw) -> int {
   }
 }
 
-void ArchiveFileOperator::NewArchive2Fd(
-    const std::filesystem::path &target_directory, int fd,
-    const OperationCallback &cb) {
+struct ArchiveReadClientData {
+  GFDataExchanger *ex;
+  std::array<std::byte, 1024> buf;
+  const std::byte *p_buf = buf.data();
+};
+
+auto ArchiveReadCallback(struct archive *, void *client_data,
+                         const void **buffer) -> ssize_t {
+  auto *rdata = static_cast<ArchiveReadClientData *>(client_data);
+  *buffer = reinterpret_cast<const void *>(rdata->p_buf);
+  return rdata->ex->Read(rdata->buf.data(), rdata->buf.size());
+}
+
+auto ArchiveWriteCallback(struct archive *, void *client_data,
+                          const void *buffer, size_t length) -> ssize_t {
+  auto *ex = static_cast<GFDataExchanger *>(client_data);
+  return ex->Write(static_cast<const std::byte *>(buffer), length);
+}
+
+auto ArchiveCloseWriteCallback(struct archive *, void *client_data) -> int {
+  auto *ex = static_cast<GFDataExchanger *>(client_data);
+  ex->CloseWrite();
+  return 0;
+}
+
+void ArchiveFileOperator::NewArchive2DataExchanger(
+    const std::filesystem::path &target_directory,
+    std::shared_ptr<GFDataExchanger> exchanger, const OperationCallback &cb) {
   RunIOOperaAsync(
       [=](const DataObjectPtr &data_object) -> GFError {
-        struct archive *archive;
-        struct archive_entry *entry;
-        std::array<char, 8192> buff{};
+        std::array<char, 1024> buff{};
+        auto ret = 0;
+        const auto base_path = target_directory.parent_path();
 
-        archive = archive_write_new();
+        auto *archive = archive_write_new();
         archive_write_add_filter_none(archive);
         archive_write_set_format_pax_restricted(archive);
-        archive_write_open_fd(archive, fd);
 
-        for (const auto &file :
-             std::filesystem::recursive_directory_iterator(target_directory)) {
-          entry = archive_entry_new();
+        archive_write_open(archive, exchanger.get(), nullptr,
+                           ArchiveWriteCallback, ArchiveCloseWriteCallback);
 
-          auto file_path = file.path().string();
-          archive_entry_set_pathname(entry, file_path.c_str());
-          archive_entry_set_filetype(entry, AE_IFREG);
-          archive_entry_set_perm(entry, 0644);
+        auto *disk = archive_read_disk_new();
+        archive_read_disk_set_standard_lookup(disk);
+        auto r = archive_read_disk_open(disk, target_directory.c_str());
 
-          std::ifstream target_file(file_path, std::ifstream::binary);
-          if (!target_file) {
-            SPDLOG_ERROR("cannot open file: {}, abort...", file_path);
-            archive_entry_free(entry);
-            continue;
+        if (r != ARCHIVE_OK) {
+          SPDLOG_ERROR("archive_read_disk_open() failed: {}, abort...",
+                       archive_error_string(disk));
+          archive_read_free(disk);
+          archive_write_free(archive);
+          return -1;
+        }
+
+        for (;;) {
+          auto *entry = archive_entry_new();
+          r = archive_read_next_header2(disk, entry);
+          if (r == ARCHIVE_EOF) break;
+          if (r != ARCHIVE_OK) {
+            SPDLOG_ERROR(
+                "archive_read_next_header2() failed, ret: {}, explain: {}", r,
+                archive_error_string(disk));
+            ret = -1;
+            break;
           }
 
-          target_file.seekg(0, std::ios::end);
-          auto file_size = target_file.tellg();
-          target_file.seekg(0, std::ios::beg);
-          archive_entry_set_size(entry, file_size);
+          archive_read_disk_descend(disk);
 
-          archive_write_header(archive, entry);
+          // turn absolute path to relative path
+          archive_entry_set_pathname(
+              entry, std::filesystem::relative(
+                         std::filesystem::path(archive_entry_pathname(entry)),
+                         base_path)
+                         .c_str());
 
-          while (!target_file.eof()) {
-            target_file.read(buff.data(), buff.size());
-            std::streamsize const bytes_read = target_file.gcount();
-            archive_write_data(archive, buff.data(), bytes_read);
+          r = archive_write_header(archive, entry);
+          if (r < ARCHIVE_OK) {
+            SPDLOG_ERROR("archive_write_header() failed, ret: {}, explain: {} ",
+                         r, archive_error_string(archive));
+            continue;
           }
 
+          if (r == ARCHIVE_FATAL) {
+            SPDLOG_ERROR(
+                "archive_write_header() failed, ret: {}, explain: {}, "
+                "abort ...",
+                r, archive_error_string(archive));
+            ret = -1;
+            break;
+          }
+
+          if (r > ARCHIVE_FAILED) {
+            auto fd = open(archive_entry_sourcepath(entry), O_RDONLY);
+            auto len = read(fd, buff.data(), buff.size());
+            while (len > 0) {
+              archive_write_data(archive, buff.data(), len);
+              len = read(fd, buff.data(), buff.size());
+            }
+            close(fd);
+          }
           archive_entry_free(entry);
         }
 
-        archive_write_close(archive);
+        archive_read_free(disk);
         archive_write_free(archive);
-        close(fd);
-        return 0;
+        return ret;
       },
       cb, "archive_write_new");
 }
 
-void ArchiveFileOperator::ExtractArchiveFromFd(
-    int fd, const std::filesystem::path &target_path,
-    const OperationCallback &cb) {
-  SPDLOG_DEBUG("extract archive from fd start, cuurent thread: {}",
-               QThread::currentThread()->currentThreadId());
+void ArchiveFileOperator::ExtractArchiveFromDataExchanger(
+    std::shared_ptr<GFDataExchanger> ex,
+    const std::filesystem::path &target_path, const OperationCallback &cb) {
   RunIOOperaAsync(
       [=](const DataObjectPtr &data_object) -> GFError {
-        SPDLOG_DEBUG("extract archive from fd processing, cuurent thread: {}",
-                     QThread::currentThread()->currentThreadId());
-        struct archive *archive;
-        struct archive *ext;
-        struct archive_entry *entry;
-
-        archive = archive_read_new();
-        ext = archive_write_disk_new();
-        archive_write_disk_set_options(ext, 0);
-
-#ifndef NO_BZIP2_EXTRACT
-        archive_read_support_filter_bzip2(archive);
-#endif
-#ifndef NO_GZIP_EXTRACT
-        archive_read_support_filter_gzip(archive);
-#endif
-#ifndef NO_COMPRESS_EXTRACT
-        archive_read_support_filter_compress(archive);
-#endif
-#ifndef NO_TAR_EXTRACT
-        archive_read_support_format_tar(archive);
-#endif
-#ifndef NO_CPIO_EXTRACT
-        archive_read_support_format_cpio(archive);
-#endif
-#ifndef NO_LOOKUP
-        archive_write_disk_set_standard_lookup(ext);
-#endif
-
-        archive_read_open_fd(archive, fd, 8192);
-        SPDLOG_ERROR("archive_read_open_fd() failed: {}",
-                     archive_error_string(archive));
-
-        while (archive_read_next_header(archive, &entry) == ARCHIVE_OK) {
-          SPDLOG_DEBUG("add file: {}, size: {}, bytes: {}, file type: {}",
-                       archive_entry_pathname_utf8(entry),
-                       archive_entry_size(entry),
-                       archive_entry_filetype(entry));
-
-          auto file_path =
-              std::filesystem::path(archive_entry_pathname_utf8(entry));
-          auto target_file_path = target_path / file_path;
-          archive_entry_set_pathname(entry, file_path.c_str());
-
-          auto r = archive_write_header(ext, entry);
+        auto *archive = archive_read_new();
+        auto *ext = archive_write_disk_new();
+
+        auto r = archive_read_support_filter_all(archive);
+        if (r != ARCHIVE_OK) {
+          SPDLOG_ERROR("archive_read_support_filter_all(), ret: {}, reason: {}",
+                       r, archive_error_string(archive));
+          return r;
+        }
+
+        r = archive_read_support_format_all(archive);
+        if (r != ARCHIVE_OK) {
+          SPDLOG_ERROR("archive_read_support_format_all(), ret: {}, reason: {}",
+                       r, archive_error_string(archive));
+          return r;
+        }
+
+        auto rdata = ArchiveReadClientData{};
+        rdata.ex = ex.get();
+
+        r = archive_read_open(archive, &rdata, nullptr, ArchiveReadCallback,
+                              nullptr);
+        if (r != ARCHIVE_OK) {
+          SPDLOG_ERROR("archive_read_open(), ret: {}, reason: {}", r,
+                       archive_error_string(archive));
+          return r;
+        }
+
+        r = archive_write_disk_set_options(ext, 0);
+        if (r != ARCHIVE_OK) {
+          SPDLOG_ERROR("archive_write_disk_set_options(), ret: {}, reason: {}",
+                       r, archive_error_string(archive));
+          return r;
+        }
+
+        for (;;) {
+          struct archive_entry *entry;
+          r = archive_read_next_header(archive, &entry);
+          if (r == ARCHIVE_EOF) break;
           if (r != ARCHIVE_OK) {
-            SPDLOG_ERROR("archive_write_header() failed: {}",
-                         archive_error_string(ext));
+            SPDLOG_ERROR("archive_read_next_header(), ret: {}, reason: {}", r,
+                         archive_error_string(archive));
+            break;
           }
 
-          r = CopyData(archive, ext);
+          r = archive_write_header(ext, entry);
           if (r != ARCHIVE_OK) {
-            SPDLOG_ERROR("copy_data() failed: {}", archive_error_string(ext));
+            SPDLOG_ERROR("archive_write_header(), ret: {}, reason: {}", r,
+                         archive_error_string(archive));
+          } else {
+            r = CopyData(archive, ext);
           }
         }
 
-        archive_read_free(archive);
-        archive_write_free(ext);
-        close(fd);
+        r = archive_read_free(archive);
+        if (r != ARCHIVE_OK) {
+          SPDLOG_ERROR("archive_read_free(), ret: {}, reason: {}", r,
+                       archive_error_string(archive));
+        }
+        r = archive_write_free(ext);
+        if (r != ARCHIVE_OK) {
+          SPDLOG_ERROR("archive_read_free(), ret: {}, reason: {}", r,
+                       archive_error_string(archive));
+        }
+
         return 0;
       },
       cb, "archive_read_new");
diff --git a/src/core/function/ArchiveFileOperator.h b/src/core/function/ArchiveFileOperator.h
index 86a3070d..ce3022bd 100644
--- a/src/core/function/ArchiveFileOperator.h
+++ b/src/core/function/ArchiveFileOperator.h
@@ -29,6 +29,7 @@
 #pragma once
 
 #include "core/GpgFrontendCore.h"
+#include "core/model/GFDataExchanger.h"
 #include "core/typedef/CoreTypedef.h"
 #include "core/utils/IOUtils.h"
 
@@ -51,8 +52,9 @@ class GPGFRONTEND_CORE_EXPORT ArchiveFileOperator {
    * @param compress
    * @param files
    */
-  static void NewArchive2Fd(const std::filesystem::path &target_directory,
-                            int fd, const OperationCallback &cb);
+  static void NewArchive2DataExchanger(
+      const std::filesystem::path &target_directory,
+      std::shared_ptr<GFDataExchanger>, const OperationCallback &cb);
 
   /**
    * @brief
@@ -60,8 +62,8 @@ class GPGFRONTEND_CORE_EXPORT ArchiveFileOperator {
    * @param archive_path
    * @param base_path
    */
-  static void ExtractArchiveFromFd(int fd,
-                                   const std::filesystem::path &target_path,
-                                   const OperationCallback &cb);
+  static void ExtractArchiveFromDataExchanger(
+      std::shared_ptr<GFDataExchanger> fd,
+      const std::filesystem::path &target_path, const OperationCallback &cb);
 };
 }  // namespace GpgFrontend
diff --git a/src/core/function/CacheManager.cpp b/src/core/function/CacheManager.cpp
index 87fe5e7c..80d5aeb4 100644
--- a/src/core/function/CacheManager.cpp
+++ b/src/core/function/CacheManager.cpp
@@ -226,7 +226,7 @@ class CacheManager::Impl : public SingletonFunctionObject<CacheManager::Impl> {
   void flush_cache_storage() {
     for (const auto& cache : cache_storage_.mirror()) {
       auto key = get_data_object_key(cache.first);
-      SPDLOG_DEBUG("save cache into filesystem, key {}, value size: {}", key,
+      SPDLOG_TRACE("save cache into filesystem, key {}, value size: {}", key,
                    cache.second.size());
       GpgFrontend::DataObjectOperator::GetInstance().SaveDataObj(key,
                                                                  cache.second);
diff --git a/src/core/function/DataObjectOperator.cpp b/src/core/function/DataObjectOperator.cpp
index a55fe79a..01625736 100644
--- a/src/core/function/DataObjectOperator.cpp
+++ b/src/core/function/DataObjectOperator.cpp
@@ -95,8 +95,7 @@ auto DataObjectOperator::SaveDataObj(const std::string& _key,
                             QAESEncryption::Padding::ISO);
   auto encoded =
       encryption.encode(QByteArray::fromStdString(to_string(value)), hash_key_);
-
-  SPDLOG_DEBUG("saving data object {} to {} , size: {} bytes", hash_obj_key,
+  SPDLOG_TRACE("saving data object {} to {} , size: {} bytes", hash_obj_key,
                obj_path.u8string(), encoded.size());
 
   WriteFileStd(obj_path.u8string(), encoded.toStdString());
diff --git a/src/core/function/gpg/GpgFileOpera.cpp b/src/core/function/gpg/GpgFileOpera.cpp
index 929c02ea..a8e33725 100644
--- a/src/core/function/gpg/GpgFileOpera.cpp
+++ b/src/core/function/gpg/GpgFileOpera.cpp
@@ -27,8 +27,6 @@
  */
 #include "GpgFileOpera.h"
 
-#include <unistd.h>
-
 #include "core/function/ArchiveFileOperator.h"
 #include "core/function/gpg/GpgBasicOperator.h"
 #include "core/model/GFBuffer.h"
@@ -77,37 +75,35 @@ void GpgFileOpera::EncryptDirectory(std::vector<GpgKey> keys,
                                     bool ascii,
                                     const std::filesystem::path& out_path,
                                     const GpgOperationCallback& cb) {
+  std::shared_ptr<GFDataExchanger> ex = std::make_shared<GFDataExchanger>(8192);
+
   RunGpgOperaAsync(
       [=](const DataObjectPtr& data_object) -> GpgError {
-        std::array<int, 2> pipe_fds;
-        if (pipe(pipe_fds.data()) != 0) {
-          SPDLOG_ERROR(
-              "cannot create pipe for directory archive and encryt process");
-          return GPG_ERR_EPIPE;
-        }
-
-        ArchiveFileOperator::NewArchive2Fd(
-            in_path, pipe_fds[1], [](GFError err, const DataObjectPtr&) {
-              SPDLOG_DEBUG("new archive 2 fd operation, err: {}", err);
-            });
-
         std::vector<gpgme_key_t> recipients(keys.begin(), keys.end());
 
         // Last entry data_in array has to be nullptr
         recipients.emplace_back(nullptr);
 
-        GpgData data_in(pipe_fds[0]);
+        GpgData data_in(ex);
         GpgData data_out(out_path, false);
 
+        SPDLOG_DEBUG("encrypt directory start");
+
         auto* ctx = ascii ? ctx_.DefaultContext() : ctx_.BinaryContext();
         auto err = CheckGpgError(gpgme_op_encrypt(ctx, recipients.data(),
                                                   GPGME_ENCRYPT_ALWAYS_TRUST,
                                                   data_in, data_out));
         data_object->Swap({GpgEncryptResult(gpgme_op_encrypt_result(ctx))});
 
+        SPDLOG_DEBUG("encrypt directory finished, err: {}", err);
         return err;
       },
       cb, "gpgme_op_encrypt", "2.1.0");
+
+  ArchiveFileOperator::NewArchive2DataExchanger(
+      in_path, ex, [=](GFError err, const DataObjectPtr&) {
+        SPDLOG_DEBUG("new archive 2 fd operation, err: {}", err);
+      });
 }
 
 void GpgFileOpera::DecryptFile(const std::filesystem::path& in_path,
@@ -131,31 +127,20 @@ void GpgFileOpera::DecryptFile(const std::filesystem::path& in_path,
 void GpgFileOpera::DecryptArchive(const std::filesystem::path& in_path,
                                   const std::filesystem::path& out_path,
                                   const GpgOperationCallback& cb) {
-  SPDLOG_DEBUG("decrypt archive start, cuurent thread: {}",
-               QThread::currentThread()->currentThreadId());
-  RunGpgOperaAsync(
-      [=](const DataObjectPtr& data_object) -> GpgError {
-        std::array<int, 2> pipe_fds;
-        if (pipe(pipe_fds.data()) != 0) {
-          SPDLOG_ERROR(
-              "cannot create pipe for directory archive and encryt process");
-          return GPG_ERR_EPIPE;
-        }
+  std::shared_ptr<GFDataExchanger> ex = std::make_shared<GFDataExchanger>(8192);
 
-        SPDLOG_DEBUG("decrypt archive processing, cuurent thread: {}",
-                     QThread::currentThread()->currentThreadId());
-        ArchiveFileOperator::ExtractArchiveFromFd(
-            pipe_fds[0], out_path, [](GFError err, const DataObjectPtr&) {
-              SPDLOG_DEBUG("extract archive from fd operation, err: {}", err);
-            });
+  ArchiveFileOperator::ExtractArchiveFromDataExchanger(
+      ex, out_path, [](GFError err, const DataObjectPtr&) {
+        SPDLOG_DEBUG("extract archive from fd operation, err: {}", err);
+      });
 
+  RunGpgOperaAsync(
+      [=](const DataObjectPtr& data_object) -> GpgError {
         GpgData data_in(in_path, true);
-        GpgData data_out(pipe_fds[1]);
+        GpgData data_out(ex);
 
-        SPDLOG_DEBUG("start to decrypt archive: {}", in_path.string());
         auto err = CheckGpgError(
             gpgme_op_decrypt(ctx_.DefaultContext(), data_in, data_out));
-        SPDLOG_DEBUG("decryption of archive done: {}", in_path.string());
 
         data_object->Swap(
             {GpgDecryptResult(gpgme_op_decrypt_result(ctx_.DefaultContext()))});
diff --git a/src/core/model/GFDataExchanger.cpp b/src/core/model/GFDataExchanger.cpp
new file mode 100644
index 00000000..621a18f1
--- /dev/null
+++ b/src/core/model/GFDataExchanger.cpp
@@ -0,0 +1,88 @@
+/**
+ * Copyright (C) 2021 Saturneric <[email protected]>
+ *
+ * This file is part of GpgFrontend.
+ *
+ * GpgFrontend 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 3 of the License, or
+ * (at your option) any later version.
+ *
+ * GpgFrontend 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 GpgFrontend. If not, see <https://www.gnu.org/licenses/>.
+ *
+ * The initial version of the source code is inherited from
+ * the gpg4usb project, which is under GPL-3.0-or-later.
+ *
+ * All the source code of GpgFrontend was modified and released by
+ * Saturneric <[email protected]> starting on May 12, 2021.
+ *
+ * SPDX-License-Identifier: GPL-3.0-or-later
+ *
+ */
+
+#include "GFDataExchanger.h"
+
+namespace GpgFrontend {
+
+auto GFDataExchanger::Write(const std::byte* buffer, size_t size) -> ssize_t {
+  std::unique_lock<std::mutex> lock(mutex_);
+  if (close_) return -1;
+
+  if (size == 0) return 0;
+  ssize_t write_bytes = 0;
+
+  try {
+    for (size_t i = 0; i < size; i++) {
+      if (queue_.size() == queue_max_size_) not_empty_.notify_all();
+      not_full_.wait(lock,
+                     [=] { return queue_.size() < queue_max_size_ || close_; });
+      if (close_) return -1;
+
+      queue_.push(buffer[i]);
+      write_bytes++;
+    }
+  } catch (...) {
+    return write_bytes;
+  }
+
+  if (!queue_.empty()) not_empty_.notify_all();
+  return write_bytes;
+}
+
+auto GFDataExchanger::Read(std::byte* buffer, size_t size) -> ssize_t {
+  std::unique_lock<std::mutex> lock(mutex_);
+
+  if (size <= 0 || (close_ && queue_.empty())) return 0;
+  ssize_t read_bytes = 0;
+
+  for (size_t i = 0; i < size; ++i) {
+    if (queue_.empty()) not_full_.notify_all();
+    not_empty_.wait(lock, [=] { return !queue_.empty() || close_; });
+
+    if (close_ && queue_.empty()) return 0;
+    buffer[i] = queue_.front();
+    queue_.pop();
+    read_bytes++;
+  }
+
+  if (queue_.size() < queue_max_size_) not_full_.notify_all();
+  return read_bytes;
+}
+
+void GFDataExchanger::CloseWrite() {
+  std::unique_lock<std::mutex> const lock(mutex_);
+
+  close_ = true;
+
+  not_full_.notify_one();
+}
+
+GFDataExchanger::GFDataExchanger(ssize_t size) : queue_max_size_(size) {}
+
+}  // namespace GpgFrontend
\ No newline at end of file
diff --git a/src/core/model/GFDataExchanger.h b/src/core/model/GFDataExchanger.h
new file mode 100644
index 00000000..70ed5263
--- /dev/null
+++ b/src/core/model/GFDataExchanger.h
@@ -0,0 +1,53 @@
+/**
+ * Copyright (C) 2021 Saturneric <[email protected]>
+ *
+ * This file is part of GpgFrontend.
+ *
+ * GpgFrontend 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 3 of the License, or
+ * (at your option) any later version.
+ *
+ * GpgFrontend 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 GpgFrontend. If not, see <https://www.gnu.org/licenses/>.
+ *
+ * The initial version of the source code is inherited from
+ * the gpg4usb project, which is under GPL-3.0-or-later.
+ *
+ * All the source code of GpgFrontend was modified and released by
+ * Saturneric <[email protected]> starting on May 12, 2021.
+ *
+ * SPDX-License-Identifier: GPL-3.0-or-later
+ *
+ */
+
+#pragma once
+
+#include <queue>
+
+namespace GpgFrontend {
+
+class GFDataExchanger {
+ public:
+  explicit GFDataExchanger(ssize_t size);
+
+  auto Write(const std::byte* buffer, size_t size) -> ssize_t;
+
+  auto Read(std::byte* buffer, size_t size) -> ssize_t;
+
+  void CloseWrite();
+
+ private:
+  std::condition_variable not_full_, not_empty_;
+  std::queue<std::byte> queue_;
+  std::mutex mutex_;
+  const ssize_t queue_max_size_;
+  bool close_ = false;
+};
+
+}  // namespace GpgFrontend
\ No newline at end of file
diff --git a/src/core/model/GpgData.cpp b/src/core/model/GpgData.cpp
index 6a6105eb..15b01595 100644
--- a/src/core/model/GpgData.cpp
+++ b/src/core/model/GpgData.cpp
@@ -30,12 +30,30 @@
 
 #include <unistd.h>
 
+#include <utility>
+
+#include "core/model/GFDataExchanger.h"
 #include "core/typedef/GpgTypedef.h"
 
 namespace GpgFrontend {
 
 constexpr size_t kBufferSize = 32 * 1024;
 
+auto GFReadExCb(void* handle, void* buffer, size_t size) -> ssize_t {
+  auto* ex = static_cast<GFDataExchanger*>(handle);
+  return ex->Read(static_cast<std::byte*>(buffer), size);
+}
+
+auto GFWriteExCb(void* handle, const void* buffer, size_t size) -> ssize_t {
+  auto* ex = static_cast<GFDataExchanger*>(handle);
+  return ex->Write(static_cast<const std::byte*>(buffer), size);
+}
+
+void GFReleaseExCb(void* handle) {
+  auto* ex = static_cast<GFDataExchanger*>(handle);
+  ex->CloseWrite();
+}
+
 GpgData::GpgData() {
   gpgme_data_t data;
 
@@ -65,10 +83,10 @@ GpgData::GpgData(const void* buffer, size_t size, bool copy) {
   data_ref_ = std::unique_ptr<struct gpgme_data, DataRefDeleter>(data);
 }
 
-GpgData::GpgData(int fd) : fd_(fd) {
+GpgData::GpgData(int fd) : fd_(fd), data_cbs_() {
   gpgme_data_t data;
 
-  auto err = gpgme_data_new_from_fd(&data, fd);
+  auto err = gpgme_data_new_from_fd(&data, fd_);
   assert(gpgme_err_code(err) == GPG_ERR_NO_ERROR);
 
   data_ref_ = std::unique_ptr<struct gpgme_data, DataRefDeleter>(data);
@@ -84,6 +102,21 @@ GpgData::GpgData(const std::filesystem::path& path, bool read) {
   data_ref_ = std::unique_ptr<struct gpgme_data, DataRefDeleter>(data);
 }
 
+GpgData::GpgData(std::shared_ptr<GFDataExchanger> ex)
+    : data_cbs_(), data_ex_(std::move(ex)) {
+  gpgme_data_t data;
+
+  data_cbs_.read = GFReadExCb;
+  data_cbs_.write = GFWriteExCb;
+  data_cbs_.seek = nullptr;
+  data_cbs_.release = GFReleaseExCb;
+
+  auto err = gpgme_data_new_from_cbs(&data, &data_cbs_, data_ex_.get());
+  assert(gpgme_err_code(err) == GPG_ERR_NO_ERROR);
+
+  data_ref_ = std::unique_ptr<struct gpgme_data, DataRefDeleter>(data);
+}
+
 GpgData::~GpgData() {
   if (fp_ != nullptr) {
     fclose(fp_);
diff --git a/src/core/model/GpgData.h b/src/core/model/GpgData.h
index 4abde270..9a55a18c 100644
--- a/src/core/model/GpgData.h
+++ b/src/core/model/GpgData.h
@@ -33,6 +33,9 @@
 #include "core/typedef/CoreTypedef.h"
 
 namespace GpgFrontend {
+
+class GFDataExchanger;
+
 /**
  * @brief
  *
@@ -64,6 +67,13 @@ class GPGFRONTEND_CORE_EXPORT GpgData {
   /**
    * @brief Construct a new Gpg Data object
    *
+   * @param fd
+   */
+  explicit GpgData(std::shared_ptr<GFDataExchanger>);
+
+  /**
+   * @brief Construct a new Gpg Data object
+   *
    * @param path
    */
   explicit GpgData(const std::filesystem::path& path, bool read);
@@ -117,6 +127,9 @@ class GPGFRONTEND_CORE_EXPORT GpgData {
   std::unique_ptr<struct gpgme_data, DataRefDeleter> data_ref_ = nullptr;  ///<
   FILE* fp_ = nullptr;
   int fd_ = -1;
+
+  struct gpgme_data_cbs data_cbs_;
+  std::shared_ptr<GFDataExchanger> data_ex_;
 };
 
 }  // namespace GpgFrontend
diff --git a/src/core/thread/TaskRunner.cpp b/src/core/thread/TaskRunner.cpp
index 6d5edd93..72266d20 100644
--- a/src/core/thread/TaskRunner.cpp
+++ b/src/core/thread/TaskRunner.cpp
@@ -28,10 +28,11 @@
 
 #include "core/thread/TaskRunner.h"
 
-#include <qobject.h>
-#include <qobjectdefs.h>
-#include <qthread.h>
-#include <qthreadpool.h>
+#include <QtConcurrent>
+#include <boost/uuid/uuid.hpp>
+#include <boost/uuid/uuid_generators.hpp>
+#include <boost/uuid/uuid_io.hpp>
+#include <utility>
 
 #include "core/thread/Task.h"
 
@@ -55,6 +56,61 @@ class TaskRunner::Impl : public QThread {
     task->SafelyRun();
   }
 
+  static void PostTask(const Task::TaskRunnable& runner,
+                       const Task::TaskCallback& cb, DataObjectPtr p_obj) {
+    auto* callback_thread = QThread::currentThread();
+    auto data_object = std::move(p_obj);
+    const auto task_uuid = generate_uuid();
+
+    QtConcurrent::run(runner, data_object).then([=](int rtn) {
+      if (!cb) {
+        SPDLOG_TRACE("task {} doesn't have a callback function", task_uuid);
+        return;
+      }
+
+      if (callback_thread == QThread::currentThread()) {
+        SPDLOG_TRACE("for task {}, the callback thread is the same thread: {}",
+                     task_uuid, static_cast<void*>(callback_thread));
+
+        cb(rtn, data_object);
+
+        // raise signal, announcing this task comes to an end
+        SPDLOG_TRACE(
+            "for task {}, its life comes to an end in the same thread "
+            "after its callback executed.",
+            task_uuid);
+      } else {
+        SPDLOG_TRACE("for task {}, callback thread is a different thread: {}",
+                     task_uuid, static_cast<void*>(callback_thread));
+        if (!QMetaObject::invokeMethod(callback_thread, [=]() {
+              SPDLOG_TRACE("calling callback of task {}", task_uuid);
+              try {
+                cb(rtn, data_object);
+              } catch (...) {
+                SPDLOG_ERROR(
+                    "unknown exception was caught when execute "
+                    "callback of task {}",
+                    task_uuid);
+              }
+              // raise signal, announcing this task comes to an end
+              SPDLOG_TRACE(
+                  "for task {}, its life comes to an end whether its "
+                  "callback function fails or not.",
+                  task_uuid);
+            })) {
+          SPDLOG_ERROR(
+              "task {} had failed to invoke the callback function to "
+              "target thread",
+              task_uuid);
+          SPDLOG_TRACE(
+              "for task {}, its life must come to an end now, although it "
+              "has something not done yet.",
+              task_uuid);
+        }
+      }
+    });
+  }
+
   void PostConcurrentTask(Task* task) {
     if (task == nullptr) {
       SPDLOG_ERROR("task posted is null");
@@ -80,22 +136,29 @@ class TaskRunner::Impl : public QThread {
     if (task == nullptr) return;
     // TODO
   }
+
+ private:
+  static auto generate_uuid() -> std::string {
+    return boost::uuids::to_string(boost::uuids::random_generator()());
+  }
 };
 
-GpgFrontend::Thread::TaskRunner::TaskRunner()
-    : p_(SecureCreateUniqueObject<Impl>()) {}
+TaskRunner::TaskRunner() : p_(SecureCreateUniqueObject<Impl>()) {}
 
-GpgFrontend::Thread::TaskRunner::~TaskRunner() {
+TaskRunner::~TaskRunner() {
   if (p_->isRunning()) {
     Stop();
   }
 }
 
-void GpgFrontend::Thread::TaskRunner::PostTask(Task* task) {
-  p_->PostTask(task);
+void TaskRunner::PostTask(Task* task) { p_->PostTask(task); }
+
+void TaskRunner::PostTask(const Task::TaskRunnable& runner,
+                          const Task::TaskCallback& cb, DataObjectPtr p_obj) {
+  p_->PostTask(runner, cb, p_obj);
 }
 
-void GpgFrontend::Thread::TaskRunner::PostConcurrentTask(Task* task) {
+void TaskRunner::PostConcurrentTask(Task* task) {
   p_->PostConcurrentTask(task);
 }
 
diff --git a/src/core/thread/TaskRunner.h b/src/core/thread/TaskRunner.h
index f70c8211..9b06057b 100644
--- a/src/core/thread/TaskRunner.h
+++ b/src/core/thread/TaskRunner.h
@@ -30,11 +30,10 @@
 
 #include "core/GpgFrontendCore.h"
 #include "core/function/SecureMemoryAllocator.h"
+#include "core/thread/Task.h"
 
 namespace GpgFrontend::Thread {
 
-class Task;
-
 class GPGFRONTEND_CORE_EXPORT TaskRunner : public QObject {
   Q_OBJECT
  public:
@@ -89,6 +88,15 @@ class GPGFRONTEND_CORE_EXPORT TaskRunner : public QObject {
   /**
    * @brief
    *
+   * @param runner
+   * @param cb
+   */
+  void PostTask(const Task::TaskRunnable& runner, const Task::TaskCallback& cb,
+                DataObjectPtr p_obj);
+
+  /**
+   * @brief
+   *
    * @param task
    */
   void PostConcurrentTask(Task* task);
diff --git a/src/core/utils/AsyncUtils.cpp b/src/core/utils/AsyncUtils.cpp
index 4de7fa1e..d414ac99 100644
--- a/src/core/utils/AsyncUtils.cpp
+++ b/src/core/utils/AsyncUtils.cpp
@@ -49,43 +49,39 @@ void RunGpgOperaAsync(GpgOperaRunnable runnable, GpgOperationCallback callback,
     return;
   }
 
-  auto* task = new Thread::Task(
-      [=](const DataObjectPtr& data_object) -> int {
-        auto custom_data_object = TransferParams();
-        GpgError err = runnable(custom_data_object);
-
-        data_object->Swap({err, custom_data_object});
-        return 0;
-      },
-      operation, TransferParams(),
-      [=](int, const DataObjectPtr& data_object) {
-        callback(ExtractParams<GpgError>(data_object, 0),
-                 ExtractParams<DataObjectPtr>(data_object, 1));
-      });
-
   Thread::TaskRunnerGetter::GetInstance()
       .GetTaskRunner(Thread::TaskRunnerGetter::kTaskRunnerType_GPG)
-      ->PostTask(task);
+      ->PostTask(
+          [=](const DataObjectPtr& data_object) -> int {
+            auto custom_data_object = TransferParams();
+            GpgError err = runnable(custom_data_object);
+
+            data_object->Swap({err, custom_data_object});
+            return 0;
+          },
+          [=](int, const DataObjectPtr& data_object) {
+            callback(ExtractParams<GpgError>(data_object, 0),
+                     ExtractParams<DataObjectPtr>(data_object, 1));
+          },
+          TransferParams());
 }
 
 void RunIOOperaAsync(OperaRunnable runnable, OperationCallback callback,
                      const std::string& operation) {
-  auto* task = new Thread::Task(
-      [=](const DataObjectPtr& data_object) -> int {
-        auto custom_data_object = TransferParams();
-        GpgError err = runnable(custom_data_object);
-
-        data_object->Swap({err, custom_data_object});
-        return 0;
-      },
-      operation, TransferParams(),
-      [=](int, const DataObjectPtr& data_object) {
-        callback(ExtractParams<GFError>(data_object, 0),
-                 ExtractParams<DataObjectPtr>(data_object, 1));
-      });
-
   Thread::TaskRunnerGetter::GetInstance()
       .GetTaskRunner(Thread::TaskRunnerGetter::kTaskRunnerType_IO)
-      ->PostTask(task);
+      ->PostTask(
+          [=](const DataObjectPtr& data_object) -> int {
+            auto custom_data_object = TransferParams();
+            GpgError err = runnable(custom_data_object);
+
+            data_object->Swap({err, custom_data_object});
+            return 0;
+          },
+          [=](int, const DataObjectPtr& data_object) {
+            callback(ExtractParams<GFError>(data_object, 0),
+                     ExtractParams<DataObjectPtr>(data_object, 1));
+          },
+          TransferParams());
 }
 }  // namespace GpgFrontend
\ No newline at end of file