BlockingCache.hpp

#pragma once
 
#include "util/Assert.hpp"
#include "util/Mutex.hpp"
#include "util/Spawn.hpp"
 
#include <boost/asio/error.hpp>
#include <boost/asio/spawn.hpp>
#include <boost/asio/steady_timer.hpp>
#include <boost/signals2/connection.hpp>
#include <boost/signals2/signal.hpp>
#include <boost/signals2/variadic_signal.hpp>
 
#include <atomic>
#include <concepts>
#include <expected>
#include <functional>
#include <optional>
#include <shared_mutex>
#include <utility>
 
namespace util {

template <typename ValueType, typename ErrorType>
    requires(not std::same_as<ValueType, ErrorType>)
class BlockingCache {
public:
    enum class State { NoValue, Updating, HasValue };
 
private:
    std::atomic<State> state_{State::NoValue};
    util::Mutex<std::optional<ValueType>, std::shared_mutex> value_;
    boost::signals2::signal<void(std::expected<ValueType, ErrorType>)> updateFinished_;
 
public:
    BlockingCache() = default;

    explicit BlockingCache(ValueType initialValue)
        : state_{State::HasValue}, value_(std::move(initialValue))
    {
    }
 
    BlockingCache(BlockingCache&&) = delete;
    BlockingCache(BlockingCache const&) = delete;
    BlockingCache&
    operator=(BlockingCache&&) = delete;
    BlockingCache&
    operator=(BlockingCache const&) = delete;

    using Updater = std::function<std::expected<ValueType, ErrorType>(boost::asio::yield_context)>;

    using Verifier = std::function<bool(ValueType const&)>;

    [[nodiscard]] std::expected<ValueType, ErrorType>
    asyncGet(boost::asio::yield_context yield, Updater updater, Verifier verifier)
    {
        switch (state_) {
            case State::Updating: {
                return wait(yield, std::move(updater), std::move(verifier));
            }
            case State::HasValue: {
                auto const value = value_.template lock<std::shared_lock>();
                ASSERT(value->has_value(), "Value should be presented when the cache is full");
                return value->value();
            }
            case State::NoValue: {
                return update(yield, std::move(updater), std::move(verifier));
            }
        };
        std::unreachable();
    }

    [[nodiscard]] std::expected<ValueType, ErrorType>
    update(boost::asio::yield_context yield, Updater updater, Verifier verifier)
    {
        if (state_ == State::Updating) {
            return asyncGet(yield, std::move(updater), std::move(verifier));
        }
        state_ = State::Updating;
 
        auto const result = updater(yield);
        auto const shouldBeCached = result.has_value() and verifier(result.value());
 
        if (shouldBeCached) {
            value_.lock().get() = result.value();
            state_ = State::HasValue;
        } else {
            state_ = State::NoValue;
            value_.lock().get() = std::nullopt;
        }
 
        updateFinished_(result);
        return result;
    }

    void
    invalidate()
    {
        if (state_ == State::HasValue) {
            state_ = State::NoValue;
            value_.lock().get() = std::nullopt;
        }
    }

    [[nodiscard]] State
    state() const
    {
        return state_;
    }
 
private:
    std::expected<ValueType, ErrorType>
    wait(boost::asio::yield_context yield, Updater updater, Verifier verifier)
    {
        struct SharedContext {
            SharedContext(boost::asio::yield_context y)
                : timer(y.get_executor(), boost::asio::steady_timer::duration::max())
            {
            }
 
            boost::asio::steady_timer timer;
            std::optional<std::expected<ValueType, ErrorType>> result;
        };
 
        auto sharedContext = std::make_shared<SharedContext>(yield);
        boost::system::error_code errorCode;
 
        boost::signals2::scoped_connection const slot =
            updateFinished_.connect([yield,
                                     sharedContext](std::expected<ValueType, ErrorType> value) {
                util::spawn(
                    yield,
                    [sharedContext = std::move(sharedContext), value = std::move(value)](auto&&) {
                        sharedContext->result = std::move(value);
                        sharedContext->timer.cancel();
                    }
                );
            });
 
        if (state_ == State::Updating) {
            sharedContext->timer.async_wait(yield[errorCode]);
            ASSERT(sharedContext->result.has_value(), "There should be some value after waiting");
            return std::move(sharedContext->result).value();
        }
        return asyncGet(yield, std::move(updater), std::move(verifier));
    }
};
 
}  // namespace util