Merge pull request #1542 from janhq/j/update-download-percentage-event

fix: incorrect downloadedBytes callback

Author: namchuai

engine/services/download_service.cc

@@ -228,7 +228,7 @@ cpp::result<bool, std::string> DownloadService::Download(
 
 curl_off_t DownloadService::GetLocalFileSize(
     const std::filesystem::path& path) const {
-  FILE* file = fopen(path.string().c_str(), "r");
+  auto file = fopen(path.string().c_str(), "r");
   if (!file) {
     return -1;
   }
@@ -237,7 +237,7 @@ curl_off_t DownloadService::GetLocalFileSize(
     return -1;
   }
 
-  curl_off_t file_size = ftell64(file);
+  auto file_size = ftell64(file);
   fclose(file);
   return file_size;
 }
@@ -264,11 +264,7 @@ void DownloadService::ProcessTask(DownloadTask& task) {
   CTL_INF("Processing task: " + task.id);
   std::vector<std::pair<CURL*, FILE*>> task_handles;
 
-  downloading_data_ = std::make_shared<DownloadingData>(DownloadingData{
-      .item_id = "",
-      .download_task = &task,
-      .event_queue = event_queue_.get(),
-  });
+  active_task_ = std::make_shared<DownloadTask>(task);
 
   for (auto& item : task.items) {
     CURL* handle = curl_easy_init();
@@ -284,7 +280,13 @@ void DownloadService::ProcessTask(DownloadTask& task) {
       CTL_ERR("Failed to open output file " + item.localPath.string());
       return;
     }
-    downloading_data_->item_id = item.id;
+
+    auto dl_data_ptr = std::make_shared<DownloadingData>(DownloadingData{
+        .item_id = item.id,
+        .download_service = this,
+    });
+    downloading_data_map_.insert(std::make_pair(item.id, dl_data_ptr));
+
     if (auto headers = CreateHeaders(item.downloadUrl); headers) {
       curl_easy_setopt(handle, CURLOPT_HTTPHEADER, headers);
     }
@@ -294,7 +296,7 @@ void DownloadService::ProcessTask(DownloadTask& task) {
     curl_easy_setopt(handle, CURLOPT_FOLLOWLOCATION, 1L);
     curl_easy_setopt(handle, CURLOPT_NOPROGRESS, 0L);
     curl_easy_setopt(handle, CURLOPT_XFERINFOFUNCTION, ProgressCallback);
-    curl_easy_setopt(handle, CURLOPT_XFERINFODATA, downloading_data_.get());
+    curl_easy_setopt(handle, CURLOPT_XFERINFODATA, dl_data_ptr.get());
 
     curl_multi_add_handle(multi_handle_, handle);
     task_handles.push_back(std::make_pair(handle, file));
@@ -329,6 +331,8 @@ void DownloadService::ProcessTask(DownloadTask& task) {
       fclose(pair.second);
     }
 
+    active_task_.reset();
+    downloading_data_map_.clear();
     return;
   }
 
@@ -338,7 +342,8 @@ void DownloadService::ProcessTask(DownloadTask& task) {
     curl_easy_cleanup(pair.first);
     fclose(pair.second);
   }
-  downloading_data_.reset();
+  downloading_data_map_.clear();
+  active_task_.reset();
 
   RemoveTaskFromStopList(task.id);
 

engine/services/download_service.h

@@ -80,8 +80,7 @@ class DownloadService {
  private:
   struct DownloadingData {
     std::string item_id;
-    DownloadTask* download_task;
-    EventQueue* event_queue;
+    DownloadService* download_service;
   };
 
   cpp::result<void, std::string> VerifyDownloadTask(
@@ -113,7 +112,9 @@ class DownloadService {
       callbacks_;
   std::mutex callbacks_mutex_;
 
-  std::shared_ptr<DownloadingData> downloading_data_;
+  std::shared_ptr<DownloadTask> active_task_;
+  std::unordered_map<std::string, std::shared_ptr<DownloadingData>>
+      downloading_data_map_;
 
   void WorkerThread();
 
@@ -131,13 +132,23 @@ class DownloadService {
 
   static int ProgressCallback(void* ptr, curl_off_t dltotal, curl_off_t dlnow,
                               curl_off_t ultotal, curl_off_t ulnow) {
-    auto* downloading_data = static_cast<DownloadingData*>(ptr);
-    auto& event_queue = *downloading_data->event_queue;
-    auto& download_task = *downloading_data->download_task;
+    auto downloading_data = static_cast<DownloadingData*>(ptr);
+    if (downloading_data == nullptr) {
+      return 0;
+    }
+    const auto dl_item_id = downloading_data->item_id;
+    if (dltotal <= 0) {
+      return 0;
+    }
+
+    auto dl_srv = downloading_data->download_service;
+    auto active_task = dl_srv->active_task_;
+    if (active_task == nullptr) {
+      return 0;
+    }
 
-    // update the download task with corresponding download item
-    for (auto& item : download_task.items) {
-      if (item.id == downloading_data->item_id) {
+    for (auto& item : active_task->items) {
+      if (item.id == dl_item_id) {
         item.downloadedBytes = dlnow;
         item.bytes = dltotal;
         break;
@@ -154,10 +165,10 @@ class DownloadService {
 
     // throttle event by 1 sec
     if (time_since_last_event >= 1000) {
-      event_queue.enqueue(
+      dl_srv->event_queue_->enqueue(
           EventType::DownloadEvent,
           DownloadEvent{.type_ = DownloadEventType::DownloadUpdated,
-                        .download_task_ = download_task});
+                        .download_task_ = *active_task});
 
       // Update the last event time
       last_event_time = current_time;