http.cpp

@@ -1,7 +1,6 @@
 #include "http.h"
 #include "curl/curl.h"
 #include "curl/easy.h"
-#include "curl/multi.h"
 #include <spdlog/spdlog.h>
 #include <spdlog/sinks/stdout_color_sinks.h>
 
@@ -46,31 +45,13 @@ HttpClient::~HttpClient() {
   });
 }
 
-void HttpClient::cancel_request(request_id id) {
+bool HttpClient::send_request(std::string method, std::string url, HttpOptions opts, ResponseCallback cb) {
-  CURL *requestHandle = reinterpret_cast<CURL*>(id);
-  auto request = m_requests.find(requestHandle);
-  if (request == m_requests.end()) {
-    m_logger->warn("cancel_request: not found");
-    return;
-  }
-  curl_multi_remove_handle(m_curlMulti, requestHandle);
-  if (request->second.socketData->pollHandle) {
-    m_logger->debug("closing poll handle");
-    uv_poll_stop(request->second.socketData->pollHandle);
-    uv_close((uv_handle_t*)request->second.socketData->pollHandle, [](uv_handle_t *h){
-      delete h;
-    });
-  }
-  m_requests.erase(request);
-}
-
-request_id HttpClient::send_request(std::string method, std::string url, HttpOptions opts, ResponseCallback cb) {
   m_logger->debug("send request {} {}", method, url);
   CURL *requestHandle = curl_easy_init();
   std::pair<decltype(m_requests)::iterator, bool> insertResult = m_requests.emplace(requestHandle, this);
   if (!insertResult.second) {
     curl_easy_cleanup(requestHandle);
-    return nullptr;
+    return false;
   }
   auto requestData = insertResult.first;
   requestData->second.callback = cb;
@@ -102,8 +83,7 @@ request_id HttpClient::send_request(std::string method, std::string url, HttpOpt
   if (opts.body) {
     curl_easy_setopt(requestHandle, CURLOPT_POSTFIELDS, opts.body->c_str());
   }
-  curl_multi_add_handle(m_curlMulti, requestHandle);
+  return CURLM_OK == curl_multi_add_handle(m_curlMulti, requestHandle);
-  return reinterpret_cast<void*>(requestHandle);
 }
 
 int HttpClient::curl_socket_cb(CURL *curl, curl_socket_t curlSocket, int action, HttpClient *self, void *socketPtr) {

http.h

@@ -9,8 +9,6 @@
 #include <vector>
 
 namespace http {
-  typedef void *request_id;
-  
   struct HttpResponse {
     int status;
     std::string body;
@@ -26,8 +24,7 @@ namespace http {
     HttpClient(HttpClient&&) = delete;
     HttpClient(HttpClient&) = delete;
     ~HttpClient();
-    request_id send_request(std::string method, std::string url, HttpOptions opts, ResponseCallback cb);
+    bool send_request(std::string method, std::string url, HttpOptions opts, ResponseCallback cb);
-    void cancel_request(request_id id);
   private:
     void check_curl_messages();
     static int curl_socket_cb(CURL *curl, curl_socket_t curlSocket, int action, HttpClient *self, void *socketPtr);

manager.cpp

@@ -17,8 +17,7 @@ const unsigned long REPOST_INTERVAL = 2000;
 const unsigned long VK_CHECK_INTERVAL = 600000;
 
 RepostManager::RepostManager(uv_loop_t *eventLoop, tg::AuthCodeProvider tgCodeProvider, tg::PasswordProvider tgPasswordProvider, state::AppState *appState, config::AppConfig *config)
- : m_vk(eventLoop), m_tg(eventLoop, config->tgApiId, config->tgApiHash, config->tgPhoneNumber),
+ : m_vk(eventLoop), m_tg(eventLoop, config->tgApiId, config->tgApiHash, config->tgPhoneNumber) {
-   m_fetcher(this) {
   m_appState = appState;
   m_appConfig = config;
   m_tg.authCodeProvider = tgCodeProvider;
@@ -41,7 +40,6 @@ RepostManager::~RepostManager() {
   }
   if (m_checkTimer) {
     uv_timer_stop(m_checkTimer);
-    m_checkTimerStarted = false;
     uv_close((uv_handle_t*)m_checkTimer, [](uv_handle_t *h){ delete h; });
   }
 }
@@ -60,12 +58,6 @@ void RepostManager::load_more_telegram_chats() {
 
 void RepostManager::start() {
   m_nRequiredChats = 100500;//m_appConfig->tgSources.size() + 1;
-  for (auto &appState : m_appState->vkRepostState) {
-    appState.lastLoadedPostDate = appState.lastForwardedPostDate;
-  }
-  for (auto &appState : m_appState->tgRepostState) {
-    appState.lastLoadedPostDate = appState.lastForwardedPostDate;
-  }
   m_tg.add_update_handler([this](void*, td_api::Object &obj){
     if (obj.get_id() == td_api::updateAuthorizationState::ID) {
       auto &authState = (td_api::updateAuthorizationState&)obj;
@@ -75,6 +67,7 @@ void RepostManager::start() {
       }
     } else if (obj.get_id() == td_api::updateNewChat::ID) {
       auto &update = (td_api::updateNewChat&)obj;
+      spdlog::debug("chat {}", update.chat_->id_);
       if (update.chat_->id_ == m_appConfig->tgDestinationId) {
         ++m_nLoadedRequiredChats;
         spdlog::info("destination chat {} loaded (loaded {}/{} chats)", m_appConfig->tgDestinationId, m_nLoadedRequiredChats, m_nRequiredChats);
@@ -102,39 +95,24 @@ void RepostManager::start() {
   m_tg.start();
 }
 
-void NewPostFetcher::fetch(bool fetchVk, bool fetchTg, decltype(onDone) onDone, decltype(onError) onError) {
+NewPostFetcher::NewPostFetcher(RepostManager *m, bool fetchVk, bool fetchTg) : mgr(m) {
-  if (working) return;
-  working = true;
   if (fetchVk) {
-    for (int i = 0; i < mgr->m_appConfig->vkSources.size(); ++i) {
+    for (int i = 0; i < m->m_appConfig->vkSources.size(); ++i) {
       fetcher_state &&state {};
       state.sourceIndex = i;
       vkState.emplace_back(state);
     }
   }
   if (fetchTg) {
-    for (int i = 0; i < mgr->m_appConfig->tgSources.size(); ++i) {
+    for (int i = 0; i < m->m_appConfig->tgSources.size(); ++i) {
       fetcher_state &&state {};
       state.sourceIndex = i;
       tgState.emplace_back(state);
     }
   }
-  this->onDone = onDone;
-  this->onError = onError;
-  continue_fetch();
 }
 
-void NewPostFetcher::reset_state() {
+void NewPostFetcher::fetch() {
-  if (!working) return;
-  for (int i = 0; i < vkState.size(); ++i) {
-    vkState[i] = fetcher_state();
-  }
-  for (int i = 0; i < tgState.size(); ++i) {
-    tgState[i] = fetcher_state();
-  }
-}
-
-void NewPostFetcher::continue_fetch() {
   bool vkReady = true;
   spdlog::info("fetch called");
   for (int i = 0; i < vkState.size(); ++i) {
@@ -169,7 +147,7 @@ void NewPostFetcher::continue_fetch() {
           for (auto i = aposts.begin(), e = aposts.end(); i != e; ++i) {
             state.posts.emplace_back(std::move(*i));
           }
-          continue_fetch();
+          fetch();
         });
       }
     }
@@ -189,10 +167,8 @@ void NewPostFetcher::continue_fetch() {
           for (auto &p : posts) {
             spdlog::debug("[tg:{}] got post dated {}", i, p->date_);
           }
-          if (posts.size() > 0) {
+          state.offset = posts[posts.size() - 1]->id_;
-            state.offset = posts[posts.size() - 1]->id_;
+          spdlog::info("[tg:{}] setting from to id {}", i, posts[posts.size() - 1]->id_);
-            spdlog::info("[tg:{}] setting from to id {}", i, posts[posts.size() - 1]->id_);
-          }
           check_tg_posts(i, std::move(posts));
         });
         state.count = state.count * 3 / 2;
@@ -210,7 +186,7 @@ void NewPostFetcher::continue_fetch() {
           for (auto i = aposts.begin(), e = aposts.end(); i != e; ++i) {
             state.posts.emplace_back(std::move(*i));
           }
-          continue_fetch();
+          fetch();
         });
       }
     }
@@ -227,7 +203,7 @@ void NewPostFetcher::continue_fetch() {
     int total =
       std::accumulate(vkState.begin(), vkState.end(), 0, addPostCount)
        + std::accumulate(tgState.begin(), tgState.end(), 0, addPostCount);
-    merged.reserve(total + mgr->m_unprocessedTgPosts.size());
+    merged.reserve(total);
 
     std::vector<int> indexes;
     indexes.reserve(nLists);
@@ -265,29 +241,6 @@ void NewPostFetcher::continue_fetch() {
       --indexes[minPostListIdx];
       ++k;
     }
-
-    while (mgr->m_unprocessedTgPosts.size() > 0) {
-      AbstractPost tgPost = mgr->m_unprocessedTgPosts.front();
-      bool duplicate = false, inserted = false;
-      for (int i = 0; i < merged.size(); ++i) {
-        if (merged[i].date == tgPost.date) {
-          spdlog::debug("not inserting duplicate unproc tg post");
-          duplicate = true;
-          break;
-        } else if (merged[i].date > tgPost.date) {
-          spdlog::debug("inserting unproc tg post at pos {}", i);
-          merged.insert(merged.begin() + i, std::move(tgPost));
-          inserted = true;
-          break;
-        }
-      }
-      if (!duplicate && !inserted) {
-        spdlog::debug("appending unproc tg post");
-        merged.push_back(std::move(tgPost));
-      }
-      mgr->m_unprocessedTgPosts.pop();
-    }
-    working = false;
     onDone(std::move(merged));
   }
 }
@@ -299,26 +252,23 @@ void NewPostFetcher::check_vk_posts(int index, std::vector<vk::Post> posts) {
   long oldLastPostDate = appState.lastLoadedPostDate;
   if (posts.size() > 0) {
     spdlog::info("[vk:{}] last post date is now {}", index, posts[0].date);
-    std::vector<AbstractPost> aposts = mgr->to_abstract_posts(posts, state.sourceIndex);
+  }
-    spdlog::info("[vk:{}] looking for date {}, have {} - {}", index, oldLastPostDate, aposts[0].date, aposts[aposts.size() - 1].date);
+  std::vector<AbstractPost> aposts = mgr->to_abstract_posts(posts, state.sourceIndex);
-    if (mgr->drop_posts_older_than(aposts, oldLastPostDate)) {
+  spdlog::info("[vk:{}] looking for date {}, have {} - {}", index, oldLastPostDate, aposts[0].date, aposts[aposts.size() - 1].date);
-      spdlog::info("[vk:{}] found last remembered post", index);
+  if (mgr->drop_posts_older_than(aposts, oldLastPostDate)) {
-      state.ready = true;
+    spdlog::info("[vk:{}] found last remembered post", index);
-    }
-    state.posts.reserve(state.posts.size() + aposts.size());
-    for (auto i = aposts.begin(), e = aposts.end(); i != e; ++i) {
-      state.posts.emplace_back(std::move(*i));
-    }
-    state.needRequest = true;
-    if (state.ready && !state.posts.empty()) {
-      spdlog::debug("[vk:{}] last loaded post date is now {}", index, state.posts[0].date);
-      appState.lastLoadedPostDate = state.posts[0].date;
-    }
-  } else {
     state.ready = true;
   }
-
+  state.posts.reserve(state.posts.size() + aposts.size());
-  continue_fetch();
+  for (auto i = aposts.begin(), e = aposts.end(); i != e; ++i) {
+    state.posts.emplace_back(std::move(*i));
+  }
+  state.needRequest = true;
+  if (state.ready && !state.posts.empty()) {
+    spdlog::debug("[vk:{}] last loaded post date is now {}", index, state.posts[0].date);
+    appState.lastLoadedPostDate = state.posts[0].date;
+  }
+  fetch();
 }
 
 void NewPostFetcher::check_tg_posts(int index, std::vector<td::tl::unique_ptr<td_api::message>> posts) {
@@ -328,50 +278,52 @@ void NewPostFetcher::check_tg_posts(int index, std::vector<td::tl::unique_ptr<td
   long oldLastPostDate = appState.lastLoadedPostDate;
   if (posts.size() > 0) {
     spdlog::info("[tg:{}] last post date is now {}", index, posts[0]->date_);
-    std::vector<AbstractPost> aposts = mgr->to_abstract_posts(posts, state.sourceIndex);
+  }
-    if (mgr->drop_posts_older_than(aposts, oldLastPostDate)) {
+  std::vector<AbstractPost> aposts = mgr->to_abstract_posts(posts, state.sourceIndex);
-      spdlog::info("[tg:{}] found last remembered post", index);
+  if (mgr->drop_posts_older_than(aposts, oldLastPostDate)) {
-      state.ready = true;
+    spdlog::info("[tg:{}] found last remembered post", index);
-    }
-    state.posts.reserve(state.posts.size() + aposts.size());
-    for (auto i = aposts.begin(), e = aposts.end(); i != e; ++i) {
-      state.posts.emplace_back(std::move(*i));
-    }
-    state.needRequest = true;
-    if (state.ready && !state.posts.empty()) {
-      spdlog::debug("[tg:{}] last loaded post date is now {}", index, state.posts[0].date);
-      appState.lastLoadedPostDate = state.posts[0].id;
-    }
-  } else {
     state.ready = true;
   }
-
+  state.posts.reserve(state.posts.size() + aposts.size());
-  continue_fetch();
+  for (auto i = aposts.begin(), e = aposts.end(); i != e; ++i) {
+    state.posts.emplace_back(std::move(*i));
+  }
+  state.needRequest = true;
+  if (state.ready && !state.posts.empty()) {
+    spdlog::debug("[tg:{}] last loaded post date is now {}", index, state.posts[0].date);
+    appState.lastLoadedPostDate = state.posts[0].id;
+  }
+  fetch();
 }
 
 void RepostManager::on_clients_ready() {
+  for (auto &appState : m_appState->vkRepostState) {
+    appState.lastLoadedPostDate = appState.lastForwardedPostDate;
+  }
+  for (auto &appState : m_appState->tgRepostState) {
+    appState.lastLoadedPostDate = appState.lastForwardedPostDate;
+  }
 
-  spdlog::info("checking all sources");
+  NewPostFetcher *f = new NewPostFetcher(this, true, true);
-  m_fetcher.fetch(
+
-      true, true,
+  f->onDone = [this, f](auto posts){
-      [this](auto posts){
+    delete f;
-        on_new_posts(posts);
+    on_new_posts(posts);
-      },
+  };
-      [](){
+  
-        // TODO error handling
+  f->onError = [f](){
-        spdlog::error("first post check failed");
+    delete f;
-      });
+  };
+
+  f->fetch();
 }
 
 void RepostManager::on_new_posts(std::vector<AbstractPost> posts) {
   spdlog::info("collected {} new posts", posts.size());
   enqueue_for_repost(posts);
 
-  if (!m_checkTimerStarted) {
+  spdlog::info("scheduling next check");
-    spdlog::info("scheduling next check");
+  uv_timer_start(m_checkTimer, &RepostManager::check_timer_callback, VK_CHECK_INTERVAL, 0);
-    uv_timer_start(m_checkTimer, &RepostManager::check_timer_callback, VK_CHECK_INTERVAL, 0);
-    m_checkTimerStarted = true;
-  }
 }
 
 void RepostManager::collect_all_vk_posts(const std::variant<long, std::string> wall, std::function<void(std::vector<vk::Post>)> callback) {
@@ -469,19 +421,6 @@ bool RepostManager::drop_posts_older_than(std::vector<AbstractPost> &posts, long
   }
 }
 
-std::optional<AbstractPost> RepostManager::to_abstract_post(const vk::Post &post, int sourceIndex) {
-  return { AbstractPost(posts::SRC_VK, sourceIndex, post.id, post.date, post.text) };
-}
-
-std::optional<AbstractPost> RepostManager::to_abstract_post(const td_api::message &post, int sourceIndex) {
-  if (post.content_->get_id() == td_api::messageText::ID) {
-    auto &content = (td_api::messageText&) *post.content_;
-    return { AbstractPost(posts::SRC_TELEGRAM, sourceIndex, post.id_, post.date_, content.text_->text_) };
-  } else {
-    return {};
-  }
-}
-
 std::vector<AbstractPost> RepostManager::to_abstract_posts(std::vector<vk::Post> &posts, int sourceIndex) {
   std::vector<AbstractPost> result;
   result.reserve(posts.size());
@@ -535,23 +474,21 @@ void RepostManager::check_timer_callback(uv_timer_t *h) {
   self->recheck_vk_posts({});
 }
 
-bool RepostManager::recheck_vk_posts(std::function<void()> onDone) {
+void RepostManager::recheck_vk_posts(std::function<void()> onDone) {
-  if (m_fetcher.working) {
-    spdlog::error("can't recheck VK posts: another check is in progress");
-    return false;
-  }
   spdlog::info("checking VK posts");
-  auto onFetchDone = [this, onDone](std::vector<AbstractPost> &&posts){
+  NewPostFetcher *f = new NewPostFetcher(this, true, false);
+  f->onDone = [this, f, onDone](std::vector<AbstractPost> &&posts){
     spdlog::info("checked VK posts");
     this->on_new_posts(posts);
     if (onDone)
       onDone();
+    delete f;
   };
-  auto onFetchError = [](){
+  f->onError = [f](){
+    delete f;
     spdlog::error("failed to check VK posts");
   };
-  m_fetcher.fetch(true, false, onFetchDone, onFetchError);
+  f->fetch();
-  return true;
 }
 
 void RepostManager::repost(AbstractPost &post) {
@@ -593,22 +530,11 @@ void RepostManager::on_tg_message(td_api::updateNewMessage &update) {
   if (sourceIndex == m_appConfig->tgSources.size()) {
     return;
   }
-  std::optional<AbstractPost> post = to_abstract_post(*update.message_, sourceIndex);
-  if (!post) {
-    spdlog::debug("tg message is not a post");
-    return;
-  }
-
-  spdlog::debug("adding tg post to the unprocessed tg post queue");
-  m_unprocessedTgPosts.push(*post);
   
-  if (!m_fetcher.working) {
+  uv_timer_stop(m_checkTimer);
-    if (m_checkTimerStarted) {
+  std::vector<td::tl::unique_ptr<td_api::message>> v;
-      uv_timer_stop(m_checkTimer);
+  v.push_back(std::move(update.message_));
-      m_checkTimerStarted = false;
+  recheck_vk_posts([this, post = to_abstract_posts(v, sourceIndex)](){
-    }
+    on_new_posts(post);
-    std::vector<AbstractPost> posts = { *post };
+  });
-    spdlog::debug("rechecking vk posts before processing the new tg post");
-    recheck_vk_posts({});
-  }
 }

manager.h

@@ -3,7 +3,6 @@
 #include "config.h"
 #include "posts.h"
 #include "state.h"
-#include "td/tl/TlObject.h"
 #include "tg.h"
 #include "vk.h"
 #include <functional>
@@ -26,17 +25,13 @@ namespace manager {
       std::vector<AbstractPost> posts;
     };
 
-    bool working = false;
     RepostManager *mgr;
     std::vector<fetcher_state> vkState, tgState;
     std::function<void(std::vector<AbstractPost>&&)> onDone;
     std::function<void()> onError;
 
-    inline NewPostFetcher(RepostManager *m) : mgr(m) {};
+    NewPostFetcher(RepostManager *m, bool fetchVk, bool fetchTg);
-    void fetch(bool fetchVk, bool fetchTg, decltype(onDone) onDone, decltype(onError) onError);
+    void fetch();
-  private:
-    void reset_state();
-    void continue_fetch();
     void check_vk_posts(int index, std::vector<vk::Post> posts);
     void check_tg_posts(int index, std::vector<td::tl::unique_ptr<td_api::message>> posts);
   };
@@ -53,7 +48,7 @@ namespace manager {
     void load_more_telegram_chats();
     void on_new_posts(std::vector<AbstractPost> posts);
     void on_tg_message(td_api::updateNewMessage &update);
-    bool recheck_vk_posts(std::function<void()> onDone);
+    void recheck_vk_posts(std::function<void()> onDone);
 
     void collect_all_vk_posts(const std::variant<long, std::string> wall, std::function<void(std::vector<vk::Post>)> callback);
     void collect_all_tg_posts(long channel, std::function<void(std::vector<td::tl::unique_ptr<td_api::message>>)> callback);
@@ -67,8 +62,6 @@ namespace manager {
 
     bool drop_posts_older_than(std::vector<AbstractPost> &posts, long lastPostId);
     
-    std::optional<AbstractPost> to_abstract_post(const vk::Post &post, int sourceIndex);
-    std::optional<AbstractPost> to_abstract_post(const td_api::message &post, int sourceIndex);
     std::vector<AbstractPost> to_abstract_posts(std::vector<vk::Post> &posts, int sourceIndex);
     std::vector<AbstractPost> to_abstract_posts(std::vector<td::tl::unique_ptr<td_api::message>> &posts, int sourceIndex);
 
@@ -81,11 +74,8 @@ namespace manager {
     config::AppConfig *m_appConfig;
     vk::VKClient m_vk;
     tg::TelegramClient m_tg;
-    NewPostFetcher m_fetcher;
     std::queue<AbstractPost> m_repostQueue;
-    std::queue<AbstractPost> m_unprocessedTgPosts;
     uv_timer_t *m_repostTimer = nullptr;
-    bool m_checkTimerStarted = false;
     uv_timer_t *m_checkTimer = nullptr;
     int m_nRequiredChats;
     int m_nLoadedRequiredChats = 0;