notifier.hpp

// Copyright Take Vos 2020.
// 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 "required.hpp"
#include "unfair_recursive_mutex.hpp"
#include "coroutine.hpp"
#include <mutex>
#include <vector>
#include <tuple>
#include <functional>
 
namespace tt {
 
template<typename T>
class notifier {
};
 
template<typename Result, typename... Args>
class notifier<Result(Args...)> {
public:
    static_assert(std::is_same_v<Result,void>, "Result of a notifier must be void.");
 
    using result_type = Result;
    using callback_type = std::function<Result(Args const &...)>;
    using callback_ptr_type = std::shared_ptr<callback_type>;
 
    callback_ptr_type subscribe(callback_ptr_type const &callback_ptr) noexcept
    {
        auto lock = std::scoped_lock(_mutex);
 
        ttlet i = std::find_if(_callbacks.cbegin(), _callbacks.cend(), [&callback_ptr](ttlet &item) {
            return item.lock() == callback_ptr;
        });
 
        if (i == _callbacks.cend()) {
            _callbacks.emplace_back(callback_ptr);
        }
        return callback_ptr;
    }
 
    template<typename Callback> requires (std::is_invocable_v<Callback>)
    [[nodiscard]] callback_ptr_type subscribe(Callback &&callback) noexcept
    {
        auto callback_ptr = std::make_shared<callback_type>(std::forward<decltype(callback)>(callback));
 
        auto lock = std::scoped_lock(_mutex);
        _callbacks.emplace_back(callback_ptr);
        return callback_ptr;
    }
 
    void unsubscribe(callback_ptr_type const &callback_ptr) noexcept
    {
        auto lock = std::scoped_lock(_mutex);
 
        ttlet new_end = std::remove_if(_callbacks.begin(), _callbacks.end(), [&callback_ptr](ttlet &item) {
            return item.expired() || item.lock() == callback_ptr;
        });
        _callbacks.erase(new_end, _callbacks.cend());
    }
 
    std::vector<std::weak_ptr<callback_type>> callbacks() const noexcept
    {
        auto lock = std::scoped_lock(_mutex);
 
        // Clean up all the callbacks that expired.
        std::erase_if(_callbacks, [](auto &x){ return x.expired(); });
 
        return _callbacks;
    }
 
    void operator()(Args const &...args) const noexcept requires(std::is_same_v<result_type, void>)
    {
        auto callbacks_ = callbacks();
        for (auto &callback : callbacks_) {
            if (auto callback_ = callback.lock()) {
                (*callback_)(args...);
            }
        };
    }
 
    generator<result_type> operator()(Args const &...args) const noexcept requires(not std::is_same_v<result_type,void>)
    {
        auto callbacks_ = callbacks();
        for (auto &callback : callbacks_) {
            if (auto callback_ = callback.lock()) {
                co_yield (*callback_)(args...);
            }
        };
    }
 
private:
    mutable unfair_recursive_mutex _mutex;
    mutable std::vector<std::weak_ptr<callback_type>> _callbacks;
};
 
} // namespace tt