notifier.hpp

// Copyright Take Vos 2022.
// 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 "../utility/utility.hpp"
#include "../concurrency/concurrency.hpp"
#include "../macros.hpp"
#include <vector>
#include <tuple>
#include <functional>
#include <coroutine>
#include <mutex>
 
hi_export_module(hikogui.dispatch : notifier);
 
hi_export namespace hi::inline v1 {
 
template<typename T = void()>
class notifier;
 
// The partial template specialization allows the use of a `std::function`-like template
// argument, that looks like a function prototype.
template<typename R, typename... Args>
class notifier<R(Args...)> {
public:
    static_assert(std::is_same_v<R, void>, "Result of a notifier must be void.");
 
    using result_type = R;
    using callback_proto = R(Args...);
 
    using callback_type = callback<R(Args...)>;
    using weak_callback_type = weak_callback<R(Args...)>;
 
    class awaiter_type {
    public:
        constexpr awaiter_type() noexcept = default;
        constexpr awaiter_type(awaiter_type&&) noexcept = default;
        constexpr awaiter_type& operator=(awaiter_type const&) noexcept = delete;
 
        constexpr awaiter_type(awaiter_type const& other) noexcept : _notifier(other._notifier) {}
 
        constexpr awaiter_type& operator=(awaiter_type&&other) noexcept
        {
            _cbt = nullptr;
            _args = {};
            _notifier = other._notifier;
        }
 
        constexpr awaiter_type(notifier& notifier) noexcept : _notifier(&notifier) {}
 
        [[nodiscard]] constexpr bool await_ready() const noexcept
        {
            return false;
        }
 
        void await_suspend(std::coroutine_handle<> handle) noexcept
        {
            hi_assert_not_null(_notifier);
 
            // We can use the this pointer in the callback, as `await_suspend()` is called by
            // the co-routine on the same object as `await_resume()`.
            _cbt = _notifier->subscribe(
                [this, handle](Args... args) {
                    // Copy the arguments received from the notifier into the awaitable object
                    // So that it can be read using `await_resume()`.
                    _args = {std::forward<Args>(args)...};
 
                    // Resume the co-routine.
                    handle.resume();
                },
                callback_flags::main | callback_flags::once);
        }
 
        constexpr void await_resume() const noexcept
            requires(sizeof...(Args) == 0)
        {
        }
 
        constexpr auto await_resume() const noexcept
            requires(sizeof...(Args) == 1)
        {
            return std::get<0>(_args);
        }
 
        constexpr auto await_resume() const noexcept
            requires(sizeof...(Args) > 1)
        {
            return _args;
        }
 
    private:
        notifier *_notifier = nullptr;
        callback<R(Args...)> _cbt;
        std::tuple<Args...> _args;
    };
 
    constexpr notifier() noexcept = default;
    constexpr notifier(notifier&&) noexcept = delete;
    constexpr notifier(notifier const&) noexcept = delete;
    constexpr notifier& operator=(notifier&&) noexcept = delete;
    constexpr notifier& operator=(notifier const&) noexcept = delete;
 
    awaiter_type operator co_await() const noexcept
    {
        return awaiter_type{const_cast<notifier&>(*this)};
    }
 
    template<forward_of<callback_proto> Func>
    [[nodiscard]] callback_type subscribe(Func&& func, callback_flags flags = callback_flags::synchronous) noexcept
    {
        auto callback = callback_type{std::forward<Func>(func)};
        auto const lock = std::scoped_lock(_mutex);
        _callbacks.emplace_back(callback, flags);
        return callback;
    }
 
    template<forward_of<void()> F>
    void loop_local_post_function(F&&) const noexcept;
    template<forward_of<void()> F>
    void loop_main_post_function(F&&) const noexcept;
    template<forward_of<void()> F>
    void loop_timer_post_function(F&&) const noexcept;
 
    void operator()(Args... args) const noexcept
    {
        auto const lock = std::scoped_lock(_mutex);
 
        for (auto& [callback, flags] : _callbacks) {
            if (is_synchronous(flags)) {
                if (auto cb = callback.lock()) {
                    cb(std::forward<Args>(args)...);
                }
 
            } else if (is_local(flags)) {
                loop_local_post_function([=] {
                    // The callback object here is captured by-copy, so that
                    // the loop can check if it was expired.
                    if (auto cb = callback.lock()) {
                        // The captured arguments are now plain copies so we do
                        // not forward them in the call.
                        cb(args...);
                    }
                });
 
            } else if (is_main(flags)) {
                loop_main_post_function([=] {
                    // The callback object here is captured by-copy, so that
                    // the loop can check if it was expired.
                    if (auto cb = callback.lock()) {
                        // The captured arguments are now plain copies so we do
                        // not forward them in the call.
                        cb(args...);
                    }
                });
 
            } else if (is_timer(flags)) {
                loop_timer_post_function([=] {
                    // The callback object here is captured by-copy, so that
                    // the loop can check if it was expired.
                    if (auto cb = callback.lock()) {
                        // The captured arguments are now plain copies so we do
                        // not forward them in the call.
                        cb(args...);
                    }
                });
 
            } else {
                hi_no_default();
            }
 
            // If the callback should only be triggered once, like inside an awaitable.
            // Then reset the weak_ptr in _callbacks so that it will be cleaned up.
            // In the lambda above the weak_ptr is copied first so that it callback will get executed
            // as long as the shared_ptr's use count does not go to zero.
            if (is_once(flags)) {
                callback.reset();
            }
        }
        clean_up();
    }
 
private:
    mutable unfair_mutex _mutex;
 
    mutable std::vector<std::pair<weak_callback_type, callback_flags>> _callbacks;
 
    void clean_up() const noexcept
    {
        hi_axiom(_mutex.is_locked());
 
        // Cleanup all callbacks that have expired, or when they may only be triggered once.
        std::erase_if(_callbacks, [](auto const& item) {
            return item.first.expired();
        });
    }
 
#ifndef NDEBUG
    mutable bool _notifying = false;
#endif
};
 
} // namespace hi::inline v1