timeout.hpp

//                  ReactivePlusPlus library
//
//          Copyright Aleksey Loginov 2022 - present.
// Distributed under the Boost Software License, Version 1.0.
//    (See accompanying file LICENSE_1_0.txt or copy at
//          https://www.boost.org/LICENSE_1_0.txt)
//
// Project home: https://github.com/victimsnino/ReactivePlusPlus
//
 
#pragma once
 
#include <rpp/operators/lift.hpp>          // required due to operator uses lift
#include <rpp/operators/details/early_unsubscribe.hpp>
#include <rpp/operators/details/serialized_subscriber.hpp>
#include <rpp/operators/details/subscriber_with_state.hpp>
#include <rpp/operators/fwd/timeout.hpp>
#include <rpp/subscribers/constraints.hpp>
#include <rpp/utils/exceptions.hpp>
#include <rpp/sources/error.hpp>
 
#include <rpp/utils/spinlock.hpp>
 
#include <atomic>
 
IMPLEMENTATION_FILE(timeout_tag);
 
namespace rpp::details
{
template<constraint::observable FallbackObs>
struct timeout_state : early_unsubscribe_state
{
    timeout_state(const FallbackObs& fallback_obs, const composite_subscription& subscription_of_subscriber)
        : early_unsubscribe_state(subscription_of_subscriber)
        , fallback_obs{fallback_obs} {}
 
    FallbackObs                         fallback_obs;
    std::atomic<schedulers::time_point> last_emission_time{};
 
    static constexpr schedulers::time_point s_timeout_reached = schedulers::time_point::min();
};
 
template<constraint::observable FallbackObs, typename Worker>
struct timeout_on_next
{
    template<typename Value>
    void operator()(Value&& v, const auto& subscriber, const std::shared_ptr<timeout_state<FallbackObs>>& state) const
    {
        if (state->last_emission_time.exchange(Worker::now(), std::memory_order_acq_rel) != timeout_state<FallbackObs>::s_timeout_reached)
            subscriber.on_next(std::forward<Value>(v));
    }
};
 
using timeout_on_error     = early_unsubscribe_on_error;
using timeout_on_completed = early_unsubscribe_on_completed;
 
template<constraint::observable FallbackObs>
struct timeout_state_with_serialized_spinlock : timeout_state<FallbackObs>
{
    using timeout_state<FallbackObs>::timeout_state;
 
    // spinlock because most part of time there is only one thread would be active
    utils::spinlock spinlock{};
};
 
template<constraint::decayed_type Type, constraint::observable_of_type<Type> FallbackObs, schedulers::constraint::scheduler TScheduler>
struct timeout_impl
{
    schedulers::duration period;
    FallbackObs          fallback_obs;
    TScheduler           scheduler;
 
    template<constraint::subscriber_of_type<Type> TSub>
    auto operator()(TSub&& in_subscriber) const
    {
        auto state = std::make_shared<timeout_state_with_serialized_spinlock<FallbackObs>>(fallback_obs, in_subscriber.get_subscription());
        // change subscriber to serialized to avoid manual using of mutex
        auto subscriber = make_serialized_subscriber(std::forward<TSub>(in_subscriber),
                                                     std::shared_ptr<utils::spinlock>{state, &state->spinlock});
 
        const auto worker = scheduler.create_worker(state->children_subscriptions);
        state->last_emission_time.store(worker.now(), std::memory_order_relaxed);
 
        const auto last_emission_time = state->last_emission_time.load(std::memory_order_relaxed);
        worker.schedule(last_emission_time + period,
                        [period = period, prev_emission_time = last_emission_time, subscriber, state]() mutable -> schedulers::optional_duration
                        {
                            while (true)
                            {
                                // last emission time still same value -> timeout reached, else -> prev_emission_time
                                // would be update to actual emission time
                                if (state->last_emission_time.compare_exchange_strong(prev_emission_time,
                                                                                      timeout_state<FallbackObs>::s_timeout_reached,
                                                                                      std::memory_order_acq_rel))
                                    return time_is_out(state, subscriber);
 
                                // if we still need to wait a bit more -> let's wait
                                if (const auto diff_to_schedule = (prev_emission_time + period) - decltype(worker)::now();
                                    diff_to_schedule > rpp::schedulers::duration{0})
                                    return diff_to_schedule;
 
                                // okay, we here because:
                                // 1) last_emission_time was not equal to prev_emission_time
                                // 2) last_emission_time + period before now -> we are still in timeout state
                                // 3) prev_emission_time updated to last_emission_time
                                // So we can return to begin
                            }
                        });
 
        return create_subscriber_with_state<Type>(state->children_subscriptions,
                                                  timeout_on_next<FallbackObs, decltype(worker)>{},
                                                  timeout_on_error{},
                                                  timeout_on_completed{},
                                                  std::move(subscriber),
                                                  std::move(state));
    }
 
private:
    static schedulers::optional_duration time_is_out(const auto& state, const auto& subscriber)
    {
        state->children_subscriptions.unsubscribe();
        state->fallback_obs.subscribe(subscriber);
        return std::nullopt;
    }
};
} // namespace rpp::details