DownloadManager: draft

CMakeLists.txt

@@ -313,6 +313,7 @@ set(TDLIB_SOURCE
   td/telegram/DialogSource.cpp
   td/telegram/Document.cpp
   td/telegram/DocumentsManager.cpp
+  td/telegram/DownloadManager.cpp
   td/telegram/DraftMessage.cpp
   td/telegram/FileReferenceManager.cpp
   td/telegram/files/FileBitmask.cpp

td/telegram/DownloadManager.cpp

@@ -0,0 +1,198 @@
+//
+// Copyright Aliaksei Levin (levlam@telegram.org), Arseny Smirnov (arseny30@gmail.com) 2014-2022
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+#include "DownloadManager.h"
+#include "td/utils/algorithm.h"
+#include "td/utils/FlatHashMap.h"
+namespace td {
+class DownloadManagerImpl final : public DownloadManager {
+ public:
+  void set_callback(unique_ptr<Callback> callback) final {
+    callback_ = std::move(callback);
+    loop();
+  }
+
+  Status toggle_is_paused(FileId file_id, bool is_paused) final {
+    if (!callback_) {
+      return Status::OK();
+    }
+    auto it = active_files_.find(file_id);
+    if (it == active_files_.end()) {
+      return Status::Error(400, "Can't find file");
+    }
+
+    auto &file_info = it->second;
+    if (is_paused != file_info.is_paused) {
+      file_info.is_paused = is_paused;
+      if (is_paused) {
+        callback_->pause_file(file_info.internal_file_id);
+      } else {
+        callback_->start_file(file_info.internal_file_id, file_info.priority);
+      }
+    }
+
+    // TODO: update db
+
+    return Status::OK();
+  }
+
+  void toggle_all_is_paused(bool is_paused) final {
+    if (!callback_) {
+      return;
+    }
+    for (auto &it : active_files_) {
+      toggle_is_paused(it.key(), is_paused);
+    }
+
+    // TODO: update db
+  }
+
+  void remove_file(FileId file_id, FileSourceId file_source_id, bool delete_from_cache) final {
+    if (!callback_) {
+      return;
+    }
+    auto it = active_files_.find(file_id);
+    if (it != active_files_.end() && (!file_source_id.is_valid() || file_source_id == it->second.file_source_id)) {
+      auto &file_info = it->second;
+      if (!file_info.is_paused) {
+        callback_->pause_file(file_info.internal_file_id);
+      }
+      if (delete_from_cache) {
+        callback_->delete_file(file_info.internal_file_id);
+      }
+      by_internal_file_id_.erase(file_info.internal_file_id);
+
+      active_files_.erase(it);
+    }
+    // TODO: remove from db
+  }
+
+  void remove_all_files(bool only_active, bool only_completed, bool delete_from_cache) final {
+    if (!callback_) {
+      return;
+    }
+    if (!only_completed) {
+      td::table_remove_if(active_files_, [&](auto &it) {
+        FileInfo &file_info = it.second;
+        if (!file_info.is_paused) {
+          callback_->pause_file(file_info.internal_file_id);
+        }
+        if (delete_from_cache) {
+          callback_->delete_file(file_info.internal_file_id);
+        }
+        return true;
+      });
+    }
+
+    // TODO: remove from db. should respect only_active
+    // TODO: if delete_from_cache, should iterate all files in db
+  }
+
+  void add_file(FileId file_id, FileSourceId file_source_id, std::string search_by, int8 priority) final {
+    if (!callback_) {
+      return;
+    }
+    FileInfo file_info;
+    file_info.internal_file_id = callback_->dup_file_id(file_id);
+    file_info.file_source_id = file_source_id;
+    file_info.is_paused = false;
+    file_info.priority = priority;
+    by_internal_file_id_[file_info.internal_file_id] = file_id;
+
+    if (active_files_.count(file_id) == 0) {
+      counters_.total_count++;
+      callback_->update_counters(counters_);
+    }
+    active_files_[file_id] = file_info;
+    callback_->start_file(file_info.internal_file_id, file_info.priority);
+
+    // TODO: add file to db
+  }
+
+  FoundFileDownloads search(std::string query, bool only_active, bool only_completed, std::string offset,
+                            int32 limit) final {
+    if (!callback_) {
+      return {};
+    }
+    // TODO: query to database
+    return FoundFileDownloads();
+  }
+
+  void update_file_download_state(FileId internal_file_id, int64 download_size, int64 size, bool is_paused) final {
+    if (!callback_) {
+      return;
+    }
+
+    auto by_internal_file_id_it = by_internal_file_id_.find(internal_file_id);
+    if (by_internal_file_id_it == by_internal_file_id_.end()) {
+      return;
+    }
+    auto it = active_files_.find(by_internal_file_id_it->second);
+    CHECK(it != active_files_.end());
+    auto &file_info = it->second;
+    counters_.downloaded_size -= file_info.downloaded_size;
+    counters_.total_count -= file_info.size;
+    file_info.size = size;
+    file_info.downloaded_size = download_size;
+    counters_.downloaded_size += file_info.downloaded_size;
+    counters_.total_count += file_info.size;
+    file_info.is_paused = is_paused;
+
+    if (download_size == size) {
+      active_files_.erase(it);
+      by_internal_file_id_.erase(by_internal_file_id_it);
+
+      if (active_files_.empty()) {
+        counters_ = {};
+      }
+    }
+    callback_->update_counters(counters_);
+  }
+
+  void update_file_deleted(FileId internal_file_id) final {
+    if (!callback_) {
+      return;
+    }
+
+    auto it = by_internal_file_id_.find(internal_file_id);
+    if (it == by_internal_file_id_.end()) {
+      return;
+    }
+    remove_file(it->second, {}, false);
+  }
+
+ private:
+  unique_ptr<Callback> callback_;
+  struct FileInfo {
+    int8 priority;
+    bool is_paused{};
+    FileId internal_file_id{};
+    FileSourceId file_source_id{};
+
+    int64 size{};
+    int64 downloaded_size{};
+  };
+  FlatHashMap<FileId, FileInfo, FileIdHash> active_files_;
+  FlatHashMap<FileId, FileId, FileIdHash> by_internal_file_id_;
+
+  Counters counters_;
+
+  void loop() override {
+    if (!callback_) {
+      return;
+    }
+    // TODO: ???
+    // TODO: load active files from db
+  }
+  void tear_down() override {
+    callback_.reset();
+  }
+};
+
+unique_ptr<DownloadManager> DownloadManager::create() {
+  return make_unique<DownloadManagerImpl>();
+}
+}  // namespace td

td/telegram/DownloadManager.h

@@ -0,0 +1,75 @@
+//
+// Copyright Aliaksei Levin (levlam@telegram.org), Arseny Smirnov (arseny30@gmail.com) 2014-2022
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+#pragma once
+#include "td/actor/actor.h"
+#include "td/telegram/files/FileId.h"
+#include "td/telegram/files/FileSourceId.h"
+#include "td/utils/common.h"
+
+namespace td {
+class DownloadManager : public td::Actor {
+ public:
+  // creates, but do not stats the actor
+  static td::unique_ptr<DownloadManager> create();
+
+  struct Counters {
+    int64 total_size{};
+    int32 total_count{};
+    int64 downloaded_size{};
+  };
+
+  struct FileDownload {
+    FileId file_id;
+    FileSourceId file_source_id;
+    bool is_paused{};
+    int32 add_date{};
+    int32 complete_date{};
+  };
+
+  struct FoundFileDownloads {
+    int32 total_count{};
+    std::vector<FileDownload> file_downloads;
+    std::string offset;
+  };
+
+  // Trying to make DownloadManager testable, so all interactions with G() will be hidden is this probably monstrous interface
+  class Callback {
+   public:
+    virtual ~Callback() = default;
+    virtual void update_counters(Counters counters) = 0;
+    virtual void start_file(FileId file_id, int8 priority) = 0;
+    virtual void pause_file(FileId file_id) = 0;
+    virtual void delete_file(FileId file_id) = 0;
+    virtual FileId dup_file_id(FileId file_id) = 0;
+
+    virtual std::string get_unique_file_id(FileId file_id) = 0;
+    virtual std::string get_file_source_serialized(FileSourceId file_source_id) = 0;
+  };
+
+  //
+  // public interface for user
+  //
+
+  // sets callback to handle all updates
+  virtual void set_callback(unique_ptr<Callback> callback) = 0;
+
+  virtual Status toggle_is_paused(FileId, bool is_paused) = 0;
+  virtual void toggle_all_is_paused(bool is_paused) = 0;
+  virtual void remove_file(FileId file_id, FileSourceId file_source_id, bool delete_from_cache) = 0;
+  virtual void remove_all_files(bool only_active, bool only_completed, bool delete_from_cache) = 0;
+  // Files are always added in is_paused = false state
+  virtual void add_file(FileId file_id, FileSourceId file_source_id, std::string search_by, int8 priority) = 0;
+  virtual FoundFileDownloads search(std::string query, bool only_active, bool only_completed, std::string offset,
+                                    int32 limit) = 0;
+
+  //
+  // private interface to handle all kinds of updates
+  //
+  virtual void update_file_download_state(FileId file_id, int64 download_size, int64 size, bool is_paused) = 0;
+  virtual void update_file_deleted(FileId file_id) = 0;
+};
+};  // namespace td

td/telegram/Td.cpp

@@ -35,6 +35,7 @@
 #include "td/telegram/DialogParticipant.h"
 #include "td/telegram/DialogSource.h"
 #include "td/telegram/DocumentsManager.h"
+#include "td/telegram/DownloadManager.h"
 #include "td/telegram/FileReferenceManager.h"
 #include "td/telegram/files/FileGcParameters.h"
 #include "td/telegram/files/FileId.h"
@@ -3312,6 +3313,8 @@ void Td::dec_actor_refcnt() {
       LOG(DEBUG) << "CountryInfoManager was cleared" << timer;
       documents_manager_.reset();
       LOG(DEBUG) << "DocumentsManager was cleared" << timer;
+      download_manager_.reset();
+      LOG(DEBUG) << "DownloadManager was cleared" << timer;
       file_manager_.reset();
       LOG(DEBUG) << "FileManager was cleared" << timer;
       file_reference_manager_.reset();
@@ -3498,6 +3501,8 @@ void Td::clear() {
   LOG(DEBUG) << "ContactsManager actor was cleared" << timer;
   country_info_manager_actor_.reset();
   LOG(DEBUG) << "CountryInfoManager actor was cleared" << timer;
+  download_manger_actor_.reset();
+  LOG(DEBUG) << "DownloadManager actor was cleared" << timer;
   file_manager_actor_.reset();
   LOG(DEBUG) << "FileManager actor was cleared" << timer;
   file_reference_manager_actor_.reset();
@@ -3965,6 +3970,45 @@ void Td::init_managers() {
   G()->set_contacts_manager(contacts_manager_actor_.get());
   country_info_manager_ = make_unique<CountryInfoManager>(this, create_reference());
   country_info_manager_actor_ = register_actor("CountryInfoManager", country_info_manager_.get());
+  download_manager_ = DownloadManager::create();
+  download_manger_actor_ = register_actor("DownloadManager", download_manager_.get());
+  // TODO: move this callback somewhere else
+  class DownloadManagerCallback final : public DownloadManager::Callback {
+   public:
+    DownloadManagerCallback(ActorShared<> parent) : parent_(std::move(parent)) {
+      // TODO
+    }
+    void update_counters(DownloadManager::Counters counters) final {
+      // TODO
+    }
+    void start_file(FileId file_id, int8 priority) final {
+      // TODO
+    }
+    void pause_file(FileId file_id) final {
+      // TODO
+    }
+    void delete_file(FileId file_id) final {
+      // TODO
+    }
+    FileId dup_file_id(FileId file_id) final {
+      // TODO
+      return FileId();
+    }
+    string get_unique_file_id(FileId file_id) final {
+      // TODO
+      return std::string();
+    }
+    string get_file_source_serialized(FileSourceId file_source_id) final {
+      // TODO
+      return std::string();
+    }
+
+   private:
+    ActorShared<> parent_;
+  };
+  send_closure_later(download_manger_actor_, &DownloadManager::set_callback,
+                     make_unique<DownloadManagerCallback>(create_reference()));
+
   game_manager_ = make_unique<GameManager>(this, create_reference());
   game_manager_actor_ = register_actor("GameManager", game_manager_.get());
   G()->set_game_manager(game_manager_actor_.get());

td/telegram/Td.h

@@ -48,6 +48,7 @@ class ContactsManager;
 class CountryInfoManager;
 class DeviceTokenManager;
 class DocumentsManager;
+class DownloadManager;
 class FileManager;
 class FileReferenceManager;
 class GameManager;
@@ -145,6 +146,8 @@ class Td final : public Actor {
   ActorOwn<ContactsManager> contacts_manager_actor_;
   unique_ptr<CountryInfoManager> country_info_manager_;
   ActorOwn<CountryInfoManager> country_info_manager_actor_;
+  unique_ptr<DownloadManager> download_manager_;
+  ActorOwn<DownloadManager> download_manger_actor_;
   unique_ptr<FileManager> file_manager_;
   ActorOwn<FileManager> file_manager_actor_;
   unique_ptr<FileReferenceManager> file_reference_manager_;

tdutils/td/utils/FlatHashMapLinear.h

@@ -110,6 +110,7 @@ template <class KeyT>
 struct SetNode {
   using public_key_type = KeyT;
   using public_type = KeyT;
+  using second_type = KeyT;  // TODO: remove second_type?
 
   KeyT first{};