This repository has been archived on 2020-05-25. You can view files and clone it, but cannot push or open issues or pull requests.
tdlib-fork/td/telegram/QueryCombiner.cpp

100 lines
2.7 KiB
C++
Raw Normal View History

//
// Copyright Aliaksei Levin (levlam@telegram.org), Arseny Smirnov (arseny30@gmail.com) 2014-2019
//
// 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 "td/telegram/QueryCombiner.h"
#include "td/utils/Time.h"
namespace td {
void QueryCombiner::add_query(int64 query_id, Promise<Promise<Unit>> &&send_query, Promise<Unit> &&promise) {
LOG(INFO) << "Add query " << query_id;
auto &query = queries_[query_id];
if (promise) {
query.promises.push_back(std::move(promise));
} else if (min_delay_ > 0 && !query.is_sent) {
// if there is no promise, than no one waits for response
// we can delay query to not exceed any flood limit
if (query.send_query) {
// the query is already delayed
return;
}
query.send_query = std::move(send_query);
delayed_queries_.push(query_id);
loop();
return;
}
if (query.is_sent) {
// just wait for the result
return;
}
if (!query.send_query) {
query.send_query = std::move(send_query);
}
do_send_query(query_id, query);
}
void QueryCombiner::do_send_query(int64 query_id, QueryInfo &query) {
LOG(INFO) << "Send query " << query_id;
CHECK(query.send_query);
query.is_sent = true;
auto send_query = std::move(query.send_query);
CHECK(!query.send_query);
next_query_time_ = Time::now() + min_delay_;
query_count_++;
send_query.set_value(PromiseCreator::lambda([actor_id = actor_id(this), query_id](Result<Unit> &&result) {
send_closure(actor_id, &QueryCombiner::on_get_query_result, query_id, std::move(result));
}));
}
void QueryCombiner::on_get_query_result(int64 query_id, Result<Unit> &&result) {
LOG(INFO) << "Get result of query " << query_id << (result.is_error() ? " error" : " success");
query_count_--;
auto it = queries_.find(query_id);
CHECK(it != queries_.end());
CHECK(it->second.is_sent);
auto promises = std::move(it->second.promises);
queries_.erase(it);
for (auto &promise : promises) {
if (result.is_ok()) {
promise.set_value(Unit());
} else {
promise.set_error(result.error().clone());
}
}
loop();
}
void QueryCombiner::loop() {
auto now = Time::now();
if (now < next_query_time_) {
set_timeout_in(next_query_time_ - now + 0.001);
return;
}
if (query_count_ != 0) {
return;
}
while (!delayed_queries_.empty()) {
auto query_id = delayed_queries_.front();
delayed_queries_.pop();
auto it = queries_.find(query_id);
if (it == queries_.end()) {
continue;
}
auto &query = it->second;
if (query.is_sent) {
continue;
}
do_send_query(query_id, query);
break;
}
}
} // namespace td