// // Copyright Aliaksei Levin (levlam@telegram.org), Arseny Smirnov (arseny30@gmail.com) 2014-2024 // // 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/utils/Closure.h" #include "td/utils/common.h" #include "td/utils/invoke.h" #include "td/utils/Promise.h" #include "td/utils/ScopeGuard.h" #include "td/utils/Status.h" #include #include namespace td { namespace detail { class EventPromise final : public PromiseInterface { public: void set_value(Unit &&) final { ok_.try_emit(); fail_.clear(); } void set_error(Status &&) final { do_set_error(); } EventPromise(const EventPromise &) = delete; EventPromise &operator=(const EventPromise &) = delete; EventPromise(EventPromise &&) = delete; EventPromise &operator=(EventPromise &&) = delete; ~EventPromise() final { do_set_error(); } EventPromise() = default; explicit EventPromise(EventFull ok) : ok_(std::move(ok)), use_ok_as_fail_(true) { } EventPromise(EventFull ok, EventFull fail) : ok_(std::move(ok)), fail_(std::move(fail)), use_ok_as_fail_(false) { } private: EventFull ok_; EventFull fail_; bool use_ok_as_fail_ = false; void do_set_error() { if (use_ok_as_fail_) { ok_.try_emit(); } else { ok_.clear(); fail_.try_emit(); } } }; class SendClosure { public: template void operator()(ArgsT &&...args) const { send_closure(std::forward(args)...); } }; } // namespace detail inline Promise create_event_promise(EventFull &&ok) { return Promise(td::make_unique(std::move(ok))); } inline Promise create_event_promise(EventFull ok, EventFull fail) { return Promise(td::make_unique(std::move(ok), std::move(fail))); } template class FutureActor; template class PromiseActor; template class ActorTraits> { public: static constexpr bool need_context = false; static constexpr bool need_start_up = false; }; template class PromiseActor final : public PromiseInterface { friend class FutureActor; enum State { Waiting, Hangup }; public: PromiseActor() = default; PromiseActor(const PromiseActor &) = delete; PromiseActor &operator=(const PromiseActor &) = delete; PromiseActor(PromiseActor &&) = default; PromiseActor &operator=(PromiseActor &&) = default; ~PromiseActor() final { close(); } void set_value(T &&value) final; void set_error(Status &&error) final; void close() { future_id_.reset(); } // NB: if true is returned no further events will be sent bool is_hangup() { if (state_ == State::Hangup) { return true; } if (!future_id_.is_alive()) { state_ = State::Hangup; future_id_.release(); event_.clear(); return true; } return false; } template friend void init_promise_future(PromiseActor *promise, FutureActor *future); bool empty_promise() { return future_id_.empty(); } bool empty() { return future_id_.empty(); } private: ActorOwn> future_id_; EventFull event_; State state_ = State::Hangup; void init() { state_ = State::Waiting; event_.clear(); } }; template class FutureActor final : public Actor { friend class PromiseActor; public: enum State { Waiting, Ready }; static constexpr int HANGUP_ERROR_CODE = 426487; FutureActor() = default; FutureActor(const FutureActor &) = delete; FutureActor &operator=(const FutureActor &) = delete; FutureActor(FutureActor &&) = default; FutureActor &operator=(FutureActor &&) = default; ~FutureActor() final = default; bool is_ok() const { return is_ready() && result_.is_ok(); } bool is_error() const { CHECK(is_ready()); return is_ready() && result_.is_error(); } T move_as_ok() { return move_as_result().move_as_ok(); } Status move_as_error() TD_WARN_UNUSED_RESULT { return move_as_result().move_as_error(); } Result move_as_result() TD_WARN_UNUSED_RESULT { CHECK(is_ready()); SCOPE_EXIT { do_stop(); }; return std::move(result_); } bool is_ready() const { return !empty() && state_ == State::Ready; } void close() { event_.clear(); result_.clear(); do_stop(); } void set_event(EventFull &&event) { CHECK(!empty()); event_ = std::move(event); if (state_ != State::Waiting) { event_.try_emit_later(); } } State get_state() const { return state_; } template friend void init_promise_future(PromiseActor *promise, FutureActor *future); private: EventFull event_; Result result_ = Status::Error(500, "Empty FutureActor"); State state_ = State::Waiting; void set_value(T &&value) { set_result(std::move(value)); } void set_error(Status &&error) { set_result(std::move(error)); } void set_result(Result &&result) { CHECK(state_ == State::Waiting); result_ = std::move(result); state_ = State::Ready; event_.try_emit_later(); } void hangup() final { set_error(Status::Error()); } void start_up() final { // empty } void init() { CHECK(empty()); state_ = State::Waiting; event_.clear(); } }; template void PromiseActor::set_value(T &&value) { if (state_ == State::Waiting && !future_id_.empty()) { send_closure(std::move(future_id_), &FutureActor::set_value, std::move(value)); } } template void PromiseActor::set_error(Status &&error) { if (state_ == State::Waiting && !future_id_.empty()) { send_closure(std::move(future_id_), &FutureActor::set_error, std::move(error)); } } template void init_promise_future(PromiseActor *promise, FutureActor *future) { promise->init(); future->init(); promise->future_id_ = register_actor("FutureActor", future); CHECK(future->get_info() != nullptr); } template class PromiseFuture { public: PromiseFuture() { init_promise_future(&promise_, &future_); } PromiseActor &promise() { return promise_; } FutureActor &future() { return future_; } PromiseActor &&move_promise() { return std::move(promise_); } FutureActor &&move_future() { return std::move(future_); } private: PromiseActor promise_; FutureActor future_; }; template FutureActor send_promise_immediately(ActorId actor_id, ResultT (ActorBT::*func)(PromiseActor &&, DestArgsT...), ArgsT &&...args) { PromiseFuture pf; Scheduler::instance()->send_closure_immediately( std::move(actor_id), create_immediate_closure(func, pf.move_promise(), std::forward(args)...)); return pf.move_future(); } template FutureActor send_promise_later(ActorId actor_id, ResultT (ActorBT::*func)(PromiseActor &&, DestArgsT...), ArgsT &&...args) { PromiseFuture pf; Scheduler::instance()->send_closure_later( std::move(actor_id), create_immediate_closure(func, pf.move_promise(), std::forward(args)...)); return pf.move_future(); } template auto promise_send_closure(ArgsT &&...args) { return [t = std::make_tuple(std::forward(args)...)](auto &&res) mutable { call_tuple(detail::SendClosure(), std::tuple_cat(std::move(t), std::make_tuple(std::forward(res)))); }; } template Promise create_promise_from_promise_actor(PromiseActor &&from) { return Promise(td::make_unique>(std::move(from))); } } // namespace td