Skip to content
This repository was archived by the owner on Jul 4, 2025. It is now read-only.
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 20 additions & 18 deletions engine/services/download_service.cc
Original file line number Diff line number Diff line change
Expand Up @@ -222,7 +222,6 @@ cpp::result<std::string, std::string> DownloadService::StopTask(
CTL_INF("Stopping task: " << task_id);
auto cancelled = task_queue_.cancelTask(task_id);
if (cancelled) {
EmitTaskStopped(task_id);
return task_id;
}
CTL_INF("Not found in pending task, try to find task " + task_id +
Expand All @@ -236,7 +235,6 @@ cpp::result<std::string, std::string> DownloadService::StopTask(
std::lock_guard<std::mutex> lock(stop_mutex_);
tasks_to_stop_.insert(task_id);
}
EmitTaskStopped(task_id);
return task_id;
}

Expand Down Expand Up @@ -356,7 +354,15 @@ void DownloadService::ProcessTask(DownloadTask& task, int worker_id) {
fclose(file);
}

if (!result.has_error()) {
if (result.has_error()) {
if (result.error().type == DownloadEventType::DownloadStopped) {
RemoveTaskFromStopList(task.id);
EmitTaskStopped(task.id);
} else {
EmitTaskError(task.id);
}
} else {
// success
// if the download has error, we are not run the callback
ExecuteCallback(task);
EmitTaskCompleted(task.id);
Expand All @@ -369,7 +375,7 @@ void DownloadService::ProcessTask(DownloadTask& task, int worker_id) {
worker_data->downloading_data_map.clear();
}

cpp::result<void, std::string> DownloadService::ProcessMultiDownload(
cpp::result<void, ProcessDownloadFailed> DownloadService::ProcessMultiDownload(
DownloadTask& task, CURLM* multi_handle,
const std::vector<std::pair<CURL*, FILE*>>& handles) {
auto still_running = 0;
Expand All @@ -379,25 +385,21 @@ cpp::result<void, std::string> DownloadService::ProcessMultiDownload(

auto result = ProcessCompletedTransfers(multi_handle);
if (result.has_error()) {
EmitTaskError(task.id);
{
std::lock_guard<std::mutex> lock(event_emit_map_mutex);
event_emit_map_.erase(task.id);
}
return cpp::fail(result.error());
return cpp::fail(ProcessDownloadFailed{
.message = result.error(),
.task_id = task.id,
.type = DownloadEventType::DownloadError,
});
}

if (IsTaskTerminated(task.id) || stop_flag_) {
CTL_INF("IsTaskTerminated " + std::to_string(IsTaskTerminated(task.id)));
CTL_INF("stop_flag_ " + std::to_string(stop_flag_));
{
std::lock_guard<std::mutex> lock(event_emit_map_mutex);
event_emit_map_.erase(task.id);
}
CTL_INF("Emit task stopped: " << task.id);
EmitTaskStopped(task.id);
RemoveTaskFromStopList(task.id);
return cpp::fail("Task " + task.id + " cancelled");
return cpp::fail(ProcessDownloadFailed{
.message = result.error(),
.task_id = task.id,
.type = DownloadEventType::DownloadStopped,
});
}
} while (still_running);
return {};
Expand Down
8 changes: 7 additions & 1 deletion engine/services/download_service.h
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,12 @@
#include "common/event.h"
#include "utils/result.hpp"

struct ProcessDownloadFailed {
std::string message;
std::string task_id;
cortex::event::DownloadEventType type;
};

class DownloadService {
private:
static constexpr int MAX_CONCURRENT_TASKS = 4;
Expand Down Expand Up @@ -48,7 +54,7 @@ class DownloadService {

void ProcessTask(DownloadTask& task, int worker_id);

cpp::result<void, std::string> ProcessMultiDownload(
cpp::result<void, ProcessDownloadFailed> ProcessMultiDownload(
DownloadTask& task, CURLM* multi_handle,
const std::vector<std::pair<CURL*, FILE*>>& handles);

Expand Down
Loading