tdesktop/Telegram/SourceFiles/mtproto/mtproto_concurrent_sender.cpp

225 lines
5.6 KiB
C++
Raw Normal View History

2018-06-02 14:29:21 +00:00
/*
This file is part of Telegram Desktop,
the official desktop application for the Telegram messaging service.
For license and copyright information please follow this link:
https://github.com/telegramdesktop/tdesktop/blob/master/LEGAL
*/
2019-12-02 13:10:19 +00:00
#include "mtproto/mtproto_concurrent_sender.h"
2018-06-02 14:29:21 +00:00
#include "mtproto/mtp_instance.h"
2019-11-18 09:28:14 +00:00
#include "mtproto/mtproto_rpc_sender.h"
2019-11-13 08:31:12 +00:00
#include "mtproto/facade.h"
2018-06-02 14:29:21 +00:00
namespace MTP {
class ConcurrentSender::RPCDoneHandler : public RPCAbstractDoneHandler {
public:
RPCDoneHandler(
not_null<ConcurrentSender*> sender,
Fn<void(FnMut<void()>)> runner);
2018-06-02 14:29:21 +00:00
2019-07-18 14:06:38 +00:00
bool operator()(
2018-06-02 14:29:21 +00:00
mtpRequestId requestId,
const mtpPrime *from,
const mtpPrime *end) override;
private:
base::weak_ptr<ConcurrentSender> _weak;
Fn<void(FnMut<void()>)> _runner;
2018-06-02 14:29:21 +00:00
};
class ConcurrentSender::RPCFailHandler : public RPCAbstractFailHandler {
public:
RPCFailHandler(
not_null<ConcurrentSender*> sender,
Fn<void(FnMut<void()>)> runner,
2018-06-02 14:29:21 +00:00
FailSkipPolicy skipPolicy);
bool operator()(
mtpRequestId requestId,
const RPCError &error) override;
private:
base::weak_ptr<ConcurrentSender> _weak;
Fn<void(FnMut<void()>)> _runner;
2018-06-02 14:29:21 +00:00
FailSkipPolicy _skipPolicy = FailSkipPolicy::Simple;
};
ConcurrentSender::RPCDoneHandler::RPCDoneHandler(
not_null<ConcurrentSender*> sender,
Fn<void(FnMut<void()>)> runner)
2018-06-02 14:29:21 +00:00
: _weak(sender)
, _runner(std::move(runner)) {
2018-06-02 14:29:21 +00:00
}
2019-07-18 14:06:38 +00:00
bool ConcurrentSender::RPCDoneHandler::operator()(
2018-06-02 14:29:21 +00:00
mtpRequestId requestId,
const mtpPrime *from,
const mtpPrime *end) {
auto response = gsl::make_span(
from,
end - from);
_runner([=, weak = _weak, moved = bytes::make_vector(response)]() mutable {
2018-06-02 14:29:21 +00:00
if (const auto strong = weak.get()) {
strong->senderRequestDone(requestId, std::move(moved));
}
});
2019-07-18 14:06:38 +00:00
return true;
2018-06-02 14:29:21 +00:00
}
ConcurrentSender::RPCFailHandler::RPCFailHandler(
not_null<ConcurrentSender*> sender,
Fn<void(FnMut<void()>)> runner,
2018-06-02 14:29:21 +00:00
FailSkipPolicy skipPolicy)
: _weak(sender)
, _runner(std::move(runner))
2018-06-02 14:29:21 +00:00
, _skipPolicy(skipPolicy) {
}
bool ConcurrentSender::RPCFailHandler::operator()(
mtpRequestId requestId,
const RPCError &error) {
if (_skipPolicy == FailSkipPolicy::Simple) {
if (MTP::isDefaultHandledError(error)) {
return false;
}
} else if (_skipPolicy == FailSkipPolicy::HandleFlood) {
if (MTP::isDefaultHandledError(error) && !MTP::isFloodError(error)) {
return false;
}
}
_runner([=, weak = _weak, error = error]() mutable {
2018-06-02 14:29:21 +00:00
if (const auto strong = weak.get()) {
strong->senderRequestFail(requestId, std::move(error));
}
});
return true;
}
template <typename Method>
auto ConcurrentSender::with_instance(Method &&method)
-> std::enable_if_t<is_callable_v<Method, not_null<Instance*>>> {
crl::on_main([method = std::forward<Method>(method)]() mutable {
if (const auto instance = MainInstance()) {
std::move(method)(instance);
}
});
}
ConcurrentSender::RequestBuilder::RequestBuilder(
not_null<ConcurrentSender*> sender,
details::SerializedRequest &&serialized) noexcept
2018-06-02 14:29:21 +00:00
: _sender(sender)
, _serialized(std::move(serialized)) {
}
void ConcurrentSender::RequestBuilder::setToDC(ShiftedDcId dcId) noexcept {
_dcId = dcId;
}
void ConcurrentSender::RequestBuilder::setCanWait(crl::time ms) noexcept {
2018-06-02 14:29:21 +00:00
_canWait = ms;
}
void ConcurrentSender::RequestBuilder::setFailSkipPolicy(
FailSkipPolicy policy) noexcept {
_failSkipPolicy = policy;
}
void ConcurrentSender::RequestBuilder::setAfter(
mtpRequestId requestId) noexcept {
_afterRequestId = requestId;
}
mtpRequestId ConcurrentSender::RequestBuilder::send() {
2019-12-02 13:10:19 +00:00
const auto requestId = details::GetNextRequestId();
2018-06-02 14:29:21 +00:00
const auto dcId = _dcId;
const auto msCanWait = _canWait;
const auto afterRequestId = _afterRequestId;
_sender->senderRequestRegister(requestId, std::move(_handlers));
_sender->with_instance([
=,
request = std::move(_serialized),
done = std::make_shared<RPCDoneHandler>(_sender, _sender->_runner),
2018-06-02 14:29:21 +00:00
fail = std::make_shared<RPCFailHandler>(
_sender,
_sender->_runner,
2018-06-02 14:29:21 +00:00
_failSkipPolicy)
](not_null<Instance*> instance) mutable {
instance->sendSerialized(
requestId,
std::move(request),
RPCResponseHandler(std::move(done), std::move(fail)),
dcId,
msCanWait,
afterRequestId);
});
return requestId;
}
ConcurrentSender::ConcurrentSender(Fn<void(FnMut<void()>)> runner)
: _runner(runner) {
2018-06-02 14:29:21 +00:00
}
ConcurrentSender::~ConcurrentSender() {
senderRequestCancelAll();
}
void ConcurrentSender::senderRequestRegister(
mtpRequestId requestId,
Handlers &&handlers) {
_requests.emplace(requestId, std::move(handlers));
}
void ConcurrentSender::senderRequestDone(
mtpRequestId requestId,
bytes::const_span result) {
if (auto handlers = _requests.take(requestId)) {
2019-07-18 14:06:38 +00:00
if (!handlers->done(requestId, result)) {
2019-01-13 13:28:05 +00:00
handlers->fail(
requestId,
RPCError::Local(
"RESPONSE_PARSE_FAILED",
2019-07-18 14:06:38 +00:00
"ConcurrentSender::senderRequestDone"));
}
2018-06-02 14:29:21 +00:00
}
}
void ConcurrentSender::senderRequestFail(
mtpRequestId requestId,
RPCError &&error) {
if (auto handlers = _requests.take(requestId)) {
2019-01-13 13:28:05 +00:00
handlers->fail(requestId, std::move(error));
2018-06-02 14:29:21 +00:00
}
}
void ConcurrentSender::senderRequestCancel(mtpRequestId requestId) {
2018-06-21 00:54:59 +00:00
senderRequestDetach(requestId);
2018-06-02 14:29:21 +00:00
with_instance([=](not_null<Instance*> instance) {
instance->cancel(requestId);
});
}
void ConcurrentSender::senderRequestCancelAll() {
auto list = std::vector<mtpRequestId>(_requests.size());
for (const auto &pair : base::take(_requests)) {
list.push_back(pair.first);
2018-06-02 14:29:21 +00:00
}
with_instance([list = std::move(list)](not_null<Instance*> instance) {
for (const auto requestId : list) {
instance->cancel(requestId);
}
});
}
2018-06-21 00:54:59 +00:00
void ConcurrentSender::senderRequestDetach(mtpRequestId requestId) {
_requests.erase(requestId);
}
2018-06-02 14:29:21 +00:00
} // namespace MTP