observable.hpp

// Copyright Take Vos 2021.
// 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)
 
#pragma once
 
#include "unfair_mutex.hpp"
#include "concepts.hpp"
#include "notifier.hpp"
#include <memory>
#include <vector>
#include <atomic>
#include <concepts>
#include <type_traits>
#include <mutex>
 
namespace hi::inline v1 {
template<typename T>
class observable;
 
namespace detail {
 
template<typename T>
struct observable_impl {
    using value_type = T;
    using owner_type = observable<value_type>;
 
    struct proxy_type {
        constexpr proxy_type(proxy_type const&) noexcept = delete;
        constexpr proxy_type& operator=(proxy_type const&) noexcept = delete;
 
        constexpr proxy_type(proxy_type&& other) noexcept :
            actual(std::exchange(other.actual, nullptr)), old_value(std::exchange(other.old_value, {}))
        {
        }
 
        constexpr proxy_type& operator=(proxy_type&& other) noexcept
        {
            if (actual) {
#if HI_BUILD_TYPE == HI_BT_DEBUG
                // This proxy will not be used anymore, so notifier_owners may cause new proxies to be opened.
                actual->rw_count = false;
#endif
                if (old_value != actual->value) {
                    actual->notifier_owners();
                }
            }
 
            actual = std::exchange(other.actual, nullptr);
            old_value = std::exchange(other.old_value, {});
        }
 
        constexpr proxy_type() noexcept : actual(nullptr), old_value() {}
 
        constexpr proxy_type(observable_impl *actual) noexcept : actual(actual), old_value(actual->value)
        {
#if HI_BUILD_TYPE == HI_BT_DEBUG
            // Cannot open a read-write proxy when something already has a read proxy open.
            hi_assert(actual->ro_count == 0);
            // There may only be one read-write proxy.
            hi_assert(not std::exchange(actual->rw_count, true));
#endif
        }
 
        constexpr ~proxy_type()
        {
            if (actual) {
#if HI_BUILD_TYPE == HI_BT_DEBUG
                // This proxy will not be used anymore, so notifier_owners may cause new proxies to be opened.
                actual->rw_count = false;
#endif
                if (old_value != actual->value) {
                    actual->notify_owners();
                }
            }
        }
 
        constexpr operator value_type &() const noexcept
        {
            hi_axiom(actual);
            return actual->value;
        }
 
        constexpr value_type *operator->() const noexcept
        {
            hi_axiom(actual);
            return std::addressof(actual->value);
        }
 
        constexpr value_type& operator*() const noexcept
        {
            hi_axiom(actual);
            return actual->value;
        }
 
        constexpr value_type *operator&() const noexcept
        {
            hi_axiom(actual);
            return std::addressof(actual->value);
        }
 
        observable_impl *actual;
        value_type old_value;
    };
 
    struct const_proxy_type {
        constexpr const_proxy_type(const_proxy_type const& other) noexcept : actual(other.actual)
        {
#if HI_BUILD_TYPE == HI_BT_DEBUG
            if (actual) {
                hi_assert(actual->rw_count == false);
                hi_assert(actual->ro_count != 0);
                ++actual->ro_count;
            }
#endif
        }
 
        constexpr const_proxy_type(const_proxy_type&& other) noexcept : actual(std::exchange(other.actual, nullptr))
        {
#if HI_BUILD_TYPE == HI_BT_DEBUG
            if (actual) {
                hi_assert(actual->rw_count == false);
                hi_assert(actual->ro_count != 0);
            }
#endif
        }
 
        constexpr const_proxy_type& operator=(const_proxy_type const& other) noexcept
        {
#if HI_BUILD_TYPE == HI_BT_DEBUG
            if (actual) {
                hi_assert(actual->rw_count == false);
                hi_assert(actual->ro_count != 0);
                --actual->ro_count;
            }
#endif
            actual = other.actual;
#if HI_BUILD_TYPE == HI_BT_DEBUG
            if (actual) {
                hi_assert(actual->rw_count == false);
                hi_assert(actual->ro_count != 0);
                ++actual->ro_count;
            }
#endif
        }
 
        constexpr const_proxy_type& operator=(const_proxy_type&& other) noexcept
        {
#if HI_BUILD_TYPE == HI_BT_DEBUG
            if (actual) {
                hi_assert(actual->rw_count == false);
                hi_assert(actual->ro_count != 0);
                --actual->ro_count;
            }
#endif
            actual = std::exchange(other.actual, nullptr);
        }
 
        constexpr const_proxy_type(proxy_type&& other) noexcept : actual(std::exchange(other.actual, nullptr))
        {
#if HI_BUILD_TYPE == HI_BT_DEBUG
            if (actual) {
                hi_assert(actual->rw_count == true);
                hi_assert(actual->ro_count == 0);
                actual->rw_count = false;
                ++actual->ro_count;
            }
#endif
        }
 
        constexpr const_proxy_type& operator=(proxy_type&& other) noexcept
        {
#if HI_BUILD_TYPE == HI_BT_DEBUG
            if (actual) {
                hi_assert(actual->rw_count == false);
                hi_assert(actual->ro_count != 0);
                --actual->ro_count;
            }
#endif
            actual = std::exchange(other.actual, nullptr);
#if HI_BUILD_TYPE == HI_BT_DEBUG
            if (actual) {
                hi_assert(actual->rw_count == true);
                hi_assert(actual->ro_count == 0);
                actual->rw_count = false;
                ++actual->ro_count;
            }
#endif
        }
 
        constexpr const_proxy_type() noexcept : actual(nullptr) {}
 
        constexpr const_proxy_type(observable_impl *actual) noexcept : actual(actual)
        {
#if HI_BUILD_TYPE == HI_BT_DEBUG
            if (actual) {
                // Cannot open a read-only proxy with a read-write proxy.
                hi_assert(actual->rw_count == false);
                ++actual->ro_count;
            }
#endif
        }
 
        constexpr ~const_proxy_type()
        {
#if HI_BUILD_TYPE == HI_BT_DEBUG
            if (actual) {
                hi_assert(actual->rw_count == false);
                hi_assert(actual->ro_count != 0);
                --actual->ro_count;
            }
#endif
        }
 
        constexpr operator value_type const &() const noexcept
        {
            hi_axiom(actual);
            return actual->value;
        }
 
        constexpr value_type const *operator->() const noexcept
        {
            hi_axiom(actual);
            return std::addressof(actual->value);
        }
 
        constexpr value_type const& operator*() const noexcept
        {
            hi_axiom(actual);
            return actual->value;
        }
 
        constexpr value_type const *operator&() const noexcept
        {
            hi_axiom(actual);
            return std::addressof(actual->value);
        }
 
        observable_impl *actual;
    };
 
    value_type value;
    std::vector<owner_type *> owners;
#if HI_BUILD_TYPE == HI_BT_DEBUG
    size_t ro_count = 0;
    bool rw_count = false;
#endif
 
    ~observable_impl()
    {
        hi_axiom(owners.empty());
    }
 
    observable_impl(observable_impl const&) = delete;
    observable_impl(observable_impl&&) = delete;
    observable_impl& operator=(observable_impl const&) = delete;
    observable_impl& operator=(observable_impl&&) = delete;
 
    observable_impl() noexcept : value() {}
    observable_impl(std::convertible_to<value_type> auto&& value) noexcept : value(hi_forward(value)) {}
 
    proxy_type proxy() noexcept
    {
        return this;
    }
 
    const_proxy_type const_proxy() noexcept
    {
        return this;
    }
 
    void notify_owners() const noexcept
    {
        for (hilet& owner : owners) {
            owner->_notifier(value);
        }
    }
 
    void add_owner(owner_type& owner) noexcept
    {
        hi_axiom(std::find(owners.cbegin(), owners.cend(), &owner) == owners.cend());
 
        owners.push_back(&owner);
    }
 
    void remove_owner(owner_type& owner) noexcept
    {
        hilet nr_erased = std::erase(owners, &owner);
        hi_axiom(nr_erased == 1);
    }
 
    void reseat_owners(std::shared_ptr<observable_impl> const& new_impl) noexcept
    {
        hi_axiom(not owners.empty());
 
        auto keep_this_alive = owners.front()->_pimpl;
        hilet values_are_same = keep_this_alive->value == new_impl->value;
 
        for (auto owner : owners) {
            owner->_pimpl = new_impl;
            new_impl->owners.push_back(owner);
            if (not values_are_same) {
                owner->_notifier(value);
            }
        }
        owners.clear();
    }
};
 
} // namespace detail
 
template<typename T>
class observable {
public:
    using value_type = T;
    using impl_type = detail::observable_impl<value_type>;
    using proxy_type = impl_type::proxy_type;
    using const_proxy_type = impl_type::const_proxy_type;
    using reference = value_type&;
    using const_reference = value_type const&;
    using notifier_type = notifier<void(value_type)>;
    using token_type = notifier_type::token_type;
    using awaiter_type = notifier_type::awaiter_type;
 
    ~observable()
    {
        _pimpl->remove_owner(*this);
    }
 
    observable() noexcept : _pimpl(std::make_shared<impl_type>())
    {
        _pimpl->add_owner(*this);
    }
 
    observable(observable const& other) noexcept : _pimpl(other._pimpl)
    {
        _pimpl->add_owner(*this);
    }
 
    observable& operator=(observable const& other) noexcept
    {
        if (this == &other or _pimpl == other._pimpl) {
            return *this;
        }
 
        hi_axiom(_pimpl);
        _pimpl->reseat_owners(other._pimpl);
        return *this;
    }
 
    token_type subscribe(callback_flags flags, std::invocable<value_type> auto&& callback) noexcept
    {
        return _notifier.subscribe(flags, hi_forward(callback));
    }
 
    token_type subscribe(std::invocable<value_type> auto&& callback) noexcept
    {
        return _notifier.subscribe(callback_flags::synchronous, hi_forward(callback));
    }
 
    awaiter_type operator co_await() const noexcept
    {
        return _notifier.operator co_await();
    }
 
    const_proxy_type const_proxy() const noexcept
    {
        hi_axiom(_pimpl);
        return _pimpl->const_proxy();
    }
 
    proxy_type proxy() const noexcept
    {
        hi_axiom(_pimpl);
        return _pimpl->const_proxy();
    }
 
    proxy_type proxy() noexcept
    {
        hi_axiom(_pimpl);
        return _pimpl->proxy();
    }
 
    const_reference operator*() const noexcept
    {
        return *const_proxy();
    }
 
    const_proxy_type operator->() const noexcept
    {
        return const_proxy();
    }
 
    observable(std::convertible_to<value_type> auto&& value) noexcept : _pimpl(std::make_shared<impl_type>(hi_forward(value)))
    {
        _pimpl->add_owner(*this);
    }
 
    observable& operator=(std::convertible_to<value_type> auto&& value) noexcept
    {
        hi_axiom(_pimpl);
        *proxy() = hi_forward(value);
        return *this;
    }
 
    auto operator++() noexcept requires(requires(value_type a) { {++a}; })
    {
        return ++*proxy();
    }
 
    auto operator--() noexcept requires(requires(value_type a) { {--a}; })
    {
        return --*proxy();
    }
 
    auto operator++(int) noexcept requires(requires(value_type a) { {a++}; })
    {
        return (*proxy())++;
    }
 
    auto operator--(int) noexcept requires(requires(value_type a) { {a--}; })
    {
        return (*proxy())--;
    }
 
    auto operator+=(auto&& rhs) noexcept requires(requires(value_type a, decltype(rhs) b) { {a += hi_forward(b)}; })
    {
        return *proxy() += hi_forward(rhs);
    }
 
    auto operator-=(auto&& rhs) noexcept requires(requires(value_type a, decltype(rhs) b) { {a -= hi_forward(b)}; })
    {
        return *proxy() -= hi_forward(rhs);
    }
 
    auto operator*=(auto&& rhs) noexcept requires(requires(value_type a, decltype(rhs) b) { {a *= hi_forward(b)}; })
    {
        return *proxy() *= hi_forward(rhs);
    }
 
    auto operator/=(auto&& rhs) noexcept requires(requires(value_type a, decltype(rhs) b) { {a /= hi_forward(b)}; })
    {
        return *proxy() /= hi_forward(rhs);
    }
 
    auto operator%=(auto&& rhs) noexcept requires(requires(value_type a, decltype(rhs) b) { {a %= hi_forward(b)}; })
    {
        return *proxy() %= hi_forward(rhs);
    }
 
    auto operator&=(auto&& rhs) noexcept requires(requires(value_type a, decltype(rhs) b) { {a &= hi_forward(b)}; })
    {
        return *proxy() &= hi_forward(rhs);
    }
 
    auto operator|=(auto&& rhs) noexcept requires(requires(value_type a, decltype(rhs) b) { {a |= hi_forward(b)}; })
    {
        return *proxy() |= hi_forward(rhs);
    }
 
    auto operator^=(auto&& rhs) noexcept requires(requires(value_type a, decltype(rhs) b) { {a ^= hi_forward(b)}; })
    {
        return *proxy() ^= hi_forward(rhs);
    }
 
    auto operator<<=(auto&& rhs) noexcept requires(requires(value_type a, decltype(rhs) b) { {a <<= hi_forward(b)}; })
    {
        return *proxy() <<= hi_forward(rhs);
    }
 
    auto operator>>=(auto&& rhs) noexcept requires(requires(value_type a, decltype(rhs) b) { {a >>= hi_forward(b)}; })
    {
        return *proxy() >>= hi_forward(rhs);
    }
 
private : std::shared_ptr<impl_type> _pimpl;
    notifier_type _notifier;
    friend impl_type;
};
 
template<typename T>
struct observable_argument {
    using type = T;
};
 
template<typename T>
struct observable_argument<observable<T>> {
    using type = typename observable<T>::value_type;
};
 
template<typename T>
using observable_argument_t = typename observable_argument<T>::type;
 
} // namespace hi::inline v1