subscribable_futures.hpp

Go to the documentation of this file.

#pragma once
 
/// @file userver/drivers/subscribable_futures.hpp
/// @brief @copybrief drivers::SubscribableFutureWrapper
 
#include <boost/intrusive_ptr.hpp>
#include <boost/smart_ptr/intrusive_ref_counter.hpp>
 
#include <userver/engine/deadline.hpp>
#include <userver/engine/exception.hpp>
#include <userver/engine/future_status.hpp>
#include <userver/engine/single_consumer_event.hpp>
#include <userver/engine/task/cancel.hpp>
#include <userver/utils/make_intrusive_ptr.hpp>
 
USERVER_NAMESPACE_BEGIN
 
namespace drivers {
 
namespace impl {
 
struct EventHolder final : boost::intrusive_ref_counter<EventHolder> {
    engine::SingleConsumerEvent event{engine::SingleConsumerEvent::NoAutoReset{}};
};
 
}  // namespace impl
 
/// @ingroup userver_concurrency
///
/// @brief An adaptor for working with certain external futures.
///
/// The `SubscribableFuture` must have a `Subscribe(Callback)` method that calls
/// the callback when the future is ready. If the future is already ready, then
/// `Subscribe` must call the callback immediately. If the promise is dropped
/// and will never be fulfilled properly, then the `SubscribableFuture` should
/// call the callback anyway.
///
/// @see drivers::WaitForSubscribableFuture
/// @see drivers::TryWaitForSubscribableFuture
template <typename SubscribableFuture>
class SubscribableFutureWrapper final {
public:
    explicit SubscribableFutureWrapper(SubscribableFuture&& future)
        : original_future_(static_cast<SubscribableFuture&&>(future)),
          event_holder_(utils::make_intrusive_ptr<impl::EventHolder>()) {
        original_future_.Subscribe([event_holder = event_holder_](auto&) { event_holder->event.Send(); });
    }
 
    /// @returns the original future
    SubscribableFuture& GetFuture() { return original_future_; }
 
    /// @brief Wait for the future. The result can be retrieved from the original
    /// future using GetFuture once ready.
    /// @throws engine::WaitInterruptedException on task cancellation
    void Wait() {
        if (TryWaitUntil(engine::Deadline{}) != engine::FutureStatus::kReady) {
            throw engine::WaitInterruptedException(engine::current_task::CancellationReason());
        }
    }
 
    /// @brief Wait for the future. The result can be retrieved from the original
    /// future using GetFuture once ready.
    /// @returns an error code if deadline is exceeded or task is cancelled
    [[nodiscard]] engine::FutureStatus TryWaitUntil(engine::Deadline deadline) {
        if (event_holder_->event.WaitForEventUntil(deadline)) {
            return engine::FutureStatus::kReady;
        }
        return engine::current_task::ShouldCancel() ? engine::FutureStatus::kCancelled : engine::FutureStatus::kTimeout;
    }
 
private:
    SubscribableFuture original_future_;
    boost::intrusive_ptr<impl::EventHolder> event_holder_;
};
 
/// @ingroup userver_concurrency
///
/// @brief Waits on the given future as described
/// on drivers::SubscribableFutureWrapper.
///
/// The result can be retrieved from the original future once ready.
///
/// @throws engine::WaitInterruptedException on task cancellation
template <typename SubscribableFuture>
void WaitForSubscribableFuture(SubscribableFuture&& future) {
    SubscribableFutureWrapper<SubscribableFuture&>{future}.Wait();
}
 
/// @ingroup userver_concurrency
/// @overload
/// @returns an error code if deadline is exceeded or task is cancelled
///
/// @warning Repeatedly waiting again after `deadline` expiration leads to a
/// memory leak, use drivers::SubscribableFutureWrapper instead.
template <typename SubscribableFuture>
[[nodiscard]] engine::FutureStatus
TryWaitForSubscribableFuture(SubscribableFuture&& future, engine::Deadline deadline) {
    return SubscribableFutureWrapper<SubscribableFuture&>{future}.TryWaitUntil(deadline);
}
 
}  // namespace drivers
 
USERVER_NAMESPACE_END